import serial import time import threading from config import cfg SEND_COM_INTERVAL = 0.02 #s READ_COM_RETRY_COUNT = 3 #retry count of read data from com interface, after send measure command READ_COM_RETRY_INTERVAL = 0.02 #s DEVICE_PROTECT_INTERVAL = 0.2 #s, used to protect connecting device, like motors RELAY_SET_RETRY_COUNT = 3 #if in setting, wait times RELAY_SET_RETRY_INTERVAL = 0.02 #s, if in setting, wait time interval RELAY_THREAD_INTERVAL = 1.0 #s, relay thread working(check newstatus) interval class Command(): def __init__(self): self.QUERY = bytes.fromhex("FF") self.LOW = [ bytes.fromhex("A0 01 00 A1"), bytes.fromhex("A0 02 00 A2"), bytes.fromhex("A0 03 00 A3"), bytes.fromhex("A0 04 00 A4"), bytes.fromhex("A0 05 00 A5"), bytes.fromhex("A0 06 00 A6"), bytes.fromhex("A0 07 00 A7"), bytes.fromhex("A0 08 00 A8") ] self.HIGH = [ bytes.fromhex("A0 01 01 A2"), bytes.fromhex("A0 02 01 A3"), bytes.fromhex("A0 03 01 A4"), bytes.fromhex("A0 04 01 A5"), bytes.fromhex("A0 05 01 A6"), bytes.fromhex("A0 06 01 A7"), bytes.fromhex("A0 07 01 A8"), bytes.fromhex("A0 08 01 A9") ] class _relaycontroller(): def __init__(self): self.ttyname = "/dev/%s" % cfg.get('relay','device') # device can be find in folder '/dev/', like 'ttyUSB0' self.bps = cfg.getint('relay','bps') # bits per second self.timeout = cfg.getint('relay','timeout') # connect timeout self.CMD = Command() self.engine = None self.IsSetting = False self.PreSetTime = time.time() def __del__(self): if self.IsOpen(): self.LowAll() self.Close() def __sendonestatus(self, onestatus): try: self.engine.write(onestatus) time.sleep(SEND_COM_INTERVAL) return True except Exception as e: print("Zhi Device Error: ", e) return False def __getstatus(self): try: self.engine.read_all() # clean buffer before write command and read data self.engine.write(self.CMD.QUERY) for i in range(0, READ_COM_RETRY_COUNT): time.sleep(READ_COM_RETRY_INTERVAL) data = self.engine.read(8) if data : return (True, data) except Execption as e: print("Zhi Device Error: ", e) return (False, None) def IsOpen(self): if self.engine is None: return False return self.engine.is_open def Open(self): if self.IsOpen(): return True try: self.engine = serial.Serial(self.ttyname, self.bps, timeout=self.timeout) if self.engine.is_open : return True self.engine.open() if self.engine.is_open : return True else: return False except Exception as e: print("Zhi Device Error: ", e) return False def Close(self): if not self.IsOpen(): return True self.engine.close() if self.engine.is_open : return False else: return True def GetStatus(self): if not self.IsOpen(): return (False, None) (hr, statusbytes) = self.__getstatus() if hr: status = 0 for i in range(0, 8): if statusbytes[i] > 0 : status += 2**i return (True, status) else: return (False, None) def SetStatus(self, newstatus): if not self.IsOpen(): return False if self.IsSetting: return False # set issetting tag self.IsSetting = True (hr, status) = self.GetStatus() if not hr: self.IsSetting = False return False elif newstatus == status: self.IsSetting = False return True # protect connecting device, like motors，each set should have time interval curinterval = time.time() - self.PreSetTime if curinterval < DEVICE_PROTECT_INTERVAL : time.sleep(DEVICE_PROTECT_INTERVAL - curinterval) # set relay self.PreSetTime = time.time() div=status newdiv=newstatus for i in range(0, 8): div, rem = divmod(div, 2) newdiv, newrem = divmod(newdiv, 2) if rem != newrem: cmd = self.CMD.HIGH[i] if newrem == 1 else self.CMD.LOW[i] if not self.__sendonestatus(cmd): self.IsSetting = False return False self.IsSetting = False return True Controller = _relaycontroller() _relaythreadsinglelock = False class _relaythread(threading.Thread): def __init__(self): threading.Thread.__init__(self) self.Status = 0 self.Endtime = 0 # the timestemp when finish this Status( back to Stop Status). If 0, keep this status forever. self.running = False def StopThread(self): self.running = False while _relaythreadsinglelock: time.sleep(1) def SetNewStatus(self, newstatus, newduration): if not Controller.Open() or not self.running : return False if newstatus == self.Status : # if new status equal current status, just update Endtime self._updateEndtime(newduration) return True if Controller.IsSetting : if not self._waitPreSetFinish(): return False hr = Controller.SetStatus(newstatus) # else set new status to relay and update Endtime if not hr: return False self.Status = newstatus self._updateEndtime(newduration) return True def _waitPreSetFinish(self): for i in range(0, RELAY_SET_RETRY_COUNT): time.sleep(RELAY_SET_RETRY_INTERVAL) if not Controller.IsSetting: return True return False def _updateEndtime(self, duration): if duration <= 0.0 : self.Endtime = 0.0 else: curtime = time.time() self.Endtime = curtime + duration def run(self): if not Controller.Open(): print('Zhi Device Error: Can not open device ttyUSB0.') return global _relaythreadsinglelock if(_relaythreadsinglelock): return _relaythreadsinglelock = True self.running = True while(self.running): if(self.Endtime == 0): # if 0, continue forever, just sleep and loop time.sleep(RELAY_THREAD_INTERVAL) else: curtime = time.time() duration = self.Endtime - curtime if duration > RELAY_THREAD_INTERVAL : #if duration is more than RELAY_THREAD_INTERVAL, just sleep this interval and loop time.sleep(RELAY_THREAD_INTERVAL) elif duration > 0: #if duration is less than RELAY_THREAD_INTERVAL, just sleep this duration and loop time.sleep(duration) else: #if duration is less then 0, set Stop Status and loop self.SetNewStatus(0, 0.0) Controller.SetStatus(0) self.running = False _relaythreadsinglelock = False RThread = _relaythread()