FilesExpand file tree

 master

/

SerialComms.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

161 lines (150 loc) · 5.36 KB

 master

/

SerialComms.py

Copy path

Top

File metadata and controls

• Code
• Blame

161 lines (150 loc) · 5.36 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

91

92

93

94

95

96

97

98

99

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

##

## Message Format

## Byte 0: Start Character - PROTOCOL_STX = 0xFE

## Byte 1: Message Length = Message Type + Size of Message

## Byte 2: Message Type

## Byte 3-n: Message Data.

##

import serial

import struct

import time

PROTOCOL_STX = 0xFE

WAITING_FOR_STX = 0

WAITING_FOR_LENGTH = 1

WAITING_FOR_MSGTYPE = 2

WAITING_FOR_MESSAGE = 3

class SerialComms:

def __init__(self,verbose=0):

"""

Typical Usage:

msg = SerialComms()

msg = SerialComms(1) <- Shows Error/Status Messages

"""

self._state = WAITING_FOR_STX

self.msg_length = 0

self.msg_type = 0

self.message = []

self._msg_count = 0

self.verbose = verbose

def _debug(self,message):

"""

If the class is initialised with the verbose flag to

1, then this class with print status messages.

"""

if self.verbose:

print message

def process(self,c):

"""

Populates a message from a datastream.

@Reads: A single character from a datastream

@Returns: True, if a complete message has been recieved.

False, if a complete message hasn't been recieved yet.

Completed Message contained within self.message

"""

if self._state == WAITING_FOR_STX:

if ord(c) == PROTOCOL_STX:

self._debug("Start Recieved")

self._state=WAITING_FOR_LENGTH

self._msg_count = 0

self.message = []

elif self._state == WAITING_FOR_LENGTH:

self._debug("Reading Length")

self.msg_length = ord(c)

self._debug("Length: "+str(self.msg_length))

self._state = WAITING_FOR_MSGTYPE

elif self._state == WAITING_FOR_MSGTYPE:

self._debug("Reading Msg Type")

self.msg_type = ord(c)

self._state = WAITING_FOR_MESSAGE

elif self._state == WAITING_FOR_MESSAGE:

#if self.msg_length - 1 == 0:

# return True

#if not (self._msg_count > self.msg_length - 1):

# self.message.append(ord(c))

# self._msg_count += 1

# if self._msg_count == self.msg_length-1:

# self._debug("Message Complete")

# self._state = WAITING_FOR_STX

# return True

if (self._msg_count < self.msg_length):

self.message.append(ord(c))

self._msg_count += 1

return False

if (self._msg_count == self.msg_length):

self._debug("Message Complete")

self._state = WAITING_FOR_STX

return True

else:

self._state=WAITING_FOR_STX

return False

def hexdump(src, length=16):

result = []

digits = 4 if isinstance(src, unicode) else 2

for i in xrange(0, len(src), length):

s = src[i:i+length]

hexa = b' '.join(["%0*X" % (digits, ord(x)) for x in s])

text = b''.join([x if 0x20 <= ord(x) < 0x7F else b'.' for x in s])

result.append( b"%04X %-*s %s" % (i, length*(digits + 1), hexa, text) )

return b'\n'.join(result)

if __name__ == "__main__":

recieved = ""

port = serial.Serial("/dev/ttyUSB0",9600,timeout=1)

time.sleep(3)

bus = SerialComms()

quit = False

while (not quit):

command = raw_input(">").lower()

if command == "on":

port.write(struct.pack('BBBBB', 0xFE,2,1,2,1))

elif command == "off":

port.write(struct.pack('BBBBB', 0xFE,2,1,2,0))

elif command == "ping":

port.write(struct.pack('BBBB', 0xFE,1,255,1))

elif command == "quit":

quit = True

while (port.inWaiting()>0):

c = port.read(1)

bus.process(c)

# #print hexdump(c)

if ord(c) == 0xFE:

print hexdump(recieved)

recieved = ""

recieved += c

else:

recieved += c

#port.write(struct.pack('BBBBB', 0xFE,2,1,0,1))

#time.sleep(1)

#port.write(struct.pack('BBBBB', 0xFE,2,1,0,0))

#time.sleep(1)

#port.write(struct.pack('BBBBB', 0xFE,2,1,0,1))

#time.sleep(1)

#port.write(struct.pack('BBBBB', 0xFE,2,1,0,0))

#port.write(struct.pack('BBBB', 0xFE,1,1,9))

#port.write(struct.pack('BBBBB',0xFE,2,2,3,1))

#port.write(struct.pack('BBBBB',0xFE,2,2,2,0))

#port.write(struct.pack('BBBBB',0xFE,2,2,5,0))

#port.write(struct.pack('BBB', 0xFE,0,3))

#port.write(struct.pack('BBBB', 0xFE,1,1,8))

#port.write(struct.pack('BBBB', 0xFE,1,1,3))

#time.sleep(0.5)

#while(port.inWaiting()>0):

#print hexdump(recieved)

#print hexdump(port.read())

#if (bus.process(port.read(1))):

#if bus.msg_type == 1:

#print "IR Send Request Recieved",

#print bus.message

#if bus.msg_type == 2:

#print "Set SSPC Request Recieved",

### print bus.message

#if bus.message[0] < 4 and bus.message[1] < 2:

#print "Setting SSPC %i to" % (bus.message[0]),

#if bus.message[1] == 1:

#print "On"

#else:

#print "Off"

#else:

#print "Invalid SSPC Set Request"

#if bus.msg_type == 3:

#print "SSPC Status Request Recieved"

port.close()