#buildingFrame = 0; #if it is a start of new frame
class Const:
- HEADER = 0x02
- SNS_BITS = 5
- LGT_SNS = 0x00
- ACC_SNS = 0x01
- MAG_SNS = 0x02
- TCH_SNS = 0x03
- ADC_SNS_1 = 0x04
-
- INT = 0
- LONG = 1
- FLOAT = 2
- DOUBLE = 3
- TAIL = 0x0A
-
- START = 0
- STOP = 1
- SET_INT = 2
- buildingFrame = 0
-
- snsCode = {'light':LGT_SNS,'touch':TCH_SNS,'acc':ACC_SNS,'mag':MAG_SNS,'a2d':ADC_SNS_1}
- cmdCode = {'start':START, 'stop':STOP, 'set':SET_INT}
- sizeMap = {INT:2, FLOAT:4, LONG:4, DOUBLE:8}
- typeMap = {0:INT, 1:LONG, FLOAT:2, DOUBLE:3}
+ HEADER = 0x02
+ SNS_BITS = 5
+ LGT_SNS = 0x00
+ ACC_SNS = 0x01
+ MAG_SNS = 0x02
+ TCH_SNS = 0x03
+ ADC_SNS_1 = 0x04
+
+ INT = 0
+ LONG = 1
+ FLOAT = 2
+ DOUBLE = 3
+ TAIL = 0x0A
+
+ START = 0
+ STOP = 1
+ SET_INT = 2
+ buildingFrame = 0
+
+ snsCode = {'light':LGT_SNS,'touch':TCH_SNS,'acc':ACC_SNS,'mag':MAG_SNS,'a2d':ADC_SNS_1}
+ cmdCode = {'start':START, 'stop':STOP, 'set':SET_INT}
+ sizeMap = {INT:2, FLOAT:4, LONG:4, DOUBLE:8}
+ typeMap = {0:INT, 1:LONG, FLOAT:2, DOUBLE:3}
class Static:
- count = 0
+ count = 0
class State: #information stored
class Device:
# Constructors
def __init__(self, config):
- print("IN")
+ print("IN")
self.config = config
self.serial = None
# Methods
def connect(self):
- print("C")
- buildingFrame = 0
+ print("C")
+ buildingFrame = 0
try:
self.inbuf = []
self.serial = Serial(self.config.device, \
return str(ex)
def disconnect(self):
- print("DC")
+ print("DC")
if self.serial and self.serial.isOpen():
self.serial.close()
return False
return True
- def control(self):#################
- print("CT")
- frame = [None]*7
+ def control(self):
+ print("CT")
+ frame = [None]*7
for key in list(self.config.enable.keys()):
state = self.config.enable[key]
rate = self.config.rate[key]
sns = Const.snsCode[key]
frame[0] = chr(Const.HEADER)
frame[1] = chr(cmd<<(Const.SNS_BITS)|(0x1F&sns))
- frame[2:6] = pack('f',float(rate))
+ frame[2:6] = pack('f',float(rate))
frame[6] = Const.TAIL
self.serial.write(frame)
print('[SEND1] ',frame)
items = []
count = 0
limit = 1000
- if not self.running():
- return items;
- if self.serial.readable():
- buildingFrame = 0
- while (self.serial.inWaiting() or count<(self.frame_len(self.inbuf)-1)): ######
- char = self.serial.read()
- count +=1
- if char == chr(Const.TAIL) and buildingFrame and (len(self.inbuf))== self.frame_len(self.inbuf)-1:
- self.inbuf.append(char)
- #print("[TAIL]")
- line = "".join(self.inbuf)
- print(self.inbuf)
- time.sleep(0.001)
- #self.printHex(line)
- item = self._parse_ascii(line) # analyze the received data
- items.append(item)
- buildingFrame = 0 # finished building one frame
- #print ("BF set to 0")
- self.inbuf = []
- count = 0
- elif char == chr(Const.HEADER) and buildingFrame==0:
- self.inbuf = []
- buildingFrame = 1
- #print ("BF set to 1")
- self.inbuf.append(char)
- #print("[HEADER]")
- #count +=1
- elif buildingFrame:
- self.inbuf.append(char)
- #print("[DT]")
- #count +=1
- else:
- #print("[ERROR] Byte Going Nowhere")
- count = 0
- buildingFrame = 0 # reset the construction
- #print ("BF set to 0!")
- self.inbuf = []
- if count > limit:
- count = 0
- print("[ERROR] Exceeded Read Limit")
- break
- return items
-
-
- # auxilary function
- def frame_len(self, frame):
- if len(frame) < 3:
- return -1
- dataType_snsType = ord(frame[1])
- dataNum = ord(frame[2])
- dataType = (0xE0&dataType_snsType)>>Const.SNS_BITS
- snsType = 0x1F&dataType_snsType
- #print(dataType_snsType)
- #print(dataType)
- #print(snsType)
- #print(dataNum)
- dataSize = Const.sizeMap[Const.typeMap[dataType]]
- return (dataSize*dataNum+4)
-
-
- def printHex(self, frame):
- #print("PH")
- frameLen = self.frame_len(frame)
- i = 0
- while i<frameLen:
- print(hex(ord(frame[i])))
- i+=1
+ if not self.running():
+ return items;
+ if self.serial.readable():
+ buildingFrame = 0
+ while (self.serial.inWaiting() or count<(self.frame_len(self.inbuf)-1)): ######
+ char = self.serial.read()
+ count +=1
+ if char == chr(Const.TAIL) and buildingFrame and (len(self.inbuf))== self.frame_len(self.inbuf)-1:
+ self.inbuf.append(char)
+ #print("[TAIL]")
+ line = "".join(self.inbuf)
+ print(self.inbuf)
+ time.sleep(0.001)
+ #self.printHex(line)
+ item = self._parse_ascii(line) # analyze the received data
+ items.append(item)
+ buildingFrame = 0 # finished building one frame
+ #print ("BF set to 0")
+ self.inbuf = []
+ count = 0
+ elif char == chr(Const.HEADER) and buildingFrame==0:
+ self.inbuf = []
+ buildingFrame = 1
+ #print ("BF set to 1")
+ self.inbuf.append(char)
+ #print("[HEADER]")
+ #count +=1
+ elif buildingFrame:
+ self.inbuf.append(char)
+ #print("[DT]")
+ #count +=1
+ else:
+ #print("[ERROR] Byte Going Nowhere")
+ count = 0
+ buildingFrame = 0 # reset the construction
+ #print ("BF set to 0!")
+ self.inbuf = []
+ if count > limit:
+ count = 0
+ print("[ERROR] Exceeded Read Limit")
+ break
+ return items
+
+
+ # auxilary function
+ def frame_len(self, frame):
+ if len(frame) < 3:
+ return -1
+ dataType_snsType = ord(frame[1])
+ dataNum = ord(frame[2])
+ dataType = (0xE0&dataType_snsType)>>Const.SNS_BITS
+ snsType = 0x1F&dataType_snsType
+ #print(dataType_snsType)
+ #print(dataType)
+ #print(snsType)
+ #print(dataNum)
+ dataSize = Const.sizeMap[Const.typeMap[dataType]]
+ return (dataSize*dataNum+4)
+
+
+ def printHex(self, frame):
+ #print("PH")
+ frameLen = self.frame_len(frame)
+ i = 0
+ while i<frameLen:
+ print(hex(ord(frame[i])))
+ i+=1
# Private methods
def _write_ascii(self, line): # to be changed
- #print("WA")
+ #print("WA")
if self.serial:
print('write: [' + line + ']')
self.serial.write(line + '\n')
time.sleep(0.1)
def _parse_ascii(self, line):##############
- #print("PA")
- dataType_snsType = ord(line[1]);
- dataNum = ord(line[2]);
- dataType = (0xE0&dataType_snsType)>>Const.SNS_BITS
- snsType = 0x1F&dataType_snsType
- state = State()
- if snsType == Const.ACC_SNS:
- line = line[3:15]
- state.acc = unpack('3f',line)
- elif snsType == Const.MAG_SNS:
- line = line[3:9]
- state.mag = unpack('3h',line)
- elif snsType == Const.LGT_SNS:
- state.light[0] = ord(line[3])
- elif snsType == Const.TCH_SNS:
- line = line[3:11]
- state.touch = sunpack('2f',line)
- elif snsType == Const.ADC_SNS_1:
- line = line[3:15]
- state.a2d = unpack('6h', line)
- else:
- print('[ERROR] Nothing Happened!')
- return state
+ #print("PA")
+ dataType_snsType = ord(line[1]);
+ dataNum = ord(line[2]);
+ dataType = (0xE0&dataType_snsType)>>Const.SNS_BITS
+ snsType = 0x1F&dataType_snsType
+ state = State()
+ if snsType == Const.ACC_SNS:
+ line = line[3:15]
+ state.acc = unpack('3f',line)
+ elif snsType == Const.MAG_SNS:
+ line = line[3:9]
+ state.mag = unpack('3h',line)
+ elif snsType == Const.LGT_SNS:
+ state.light[0] = ord(line[3])
+ elif snsType == Const.TCH_SNS:
+ line = line[3:11]
+ state.touch = sunpack('2f',line)
+ elif snsType == Const.ADC_SNS_1:
+ line = line[3:15]
+ state.a2d = unpack('6h', line)
+ else:
+ print('[ERROR] Nothing Happened!')
+ return state
projects / ~andy / csm213a-hw / blobdiff