FilesExpand file tree

 master

/

code.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

130 lines (94 loc) · 2.55 KB

 master

/

code.py

Copy path

Top

File metadata and controls

• Code
• Blame

130 lines (94 loc) · 2.55 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

import board

from busio import UART

import sys

Packet_lengths_spaceorb = {

'D': 13,

'K': 6,

'.': 4,

'C': 4,

'F': 4,

'M': 5,

'N': 4,

'P': 6,

'\v': 1,

'\n': 1,

'\r': 1,

'@': 62,

'E': 5,

'Z': 14,

'd': 26,

'k': 5

}

class Observable:

def __init__(self):

self.observers = []

def attach(self, observer):

self.observers.append(observer)

def detach(self, observer):

self.observers.remove(observer)

def emit(self, msg):

for observer in self.observers:

observer.receive(msg)

class Observer:

def __init__(self):

pass

def receive(self, msg):

pass

class PipelineStage(Observable, Observer):

def __init__(self):

super().__init__()

class UARTSource(Observable):

def __init__(self):

super().__init__()

self.uart = UART(board.TX, board.RX, baudrate=9600)

def tick(self):

self.emit(self.uart.read(self.uart.in_waiting))

class Packetizer(PipelineStage):

def __init__(self, packet_lengths):

super().__init__()

self.packet_lengths = packet_lengths

self.buf = bytearray(max(self.packet_lengths.values()))

self.in_packet = False

self.cursor = 0

def current_packet_length(self):

if self.in_packet:

return self.packet_lengths.get(chr(self.buf[0]), 0)

else:

return 0

def add_byte(self, b):

if self.in_packet:

self.buf[self.cursor] = b

self.cursor = self.cursor + 1

else:

if chr(b) in self.packet_lengths:

self.buf[0] = b

self.cursor = 1

self.in_packet = True

if self.in_packet and self.cursor >= self.current_packet_length():

self.emit(self.buf[0:self.cursor])

self.in_packet = False

self.cursor = 0

def receive(self, bs):

for b in bs:

self.add_byte(b)

class StdOutObserver(Observer):

def __init__(self, label):

super().__init__()

self.label = label

def receive(self, msg):

print(self.label, ": ", msg)

def connect_pipeline(stages):

pairs = [(stages[i], stages[i+1]) for i in range(len(stages)-1)]

for pair in pairs:

pair[0].attach(pair[1])

return stages

def main():

pipeline = connect_pipeline([

UARTSource(),

Packetizer(Packet_lengths_spaceorb),

StdOutObserver("Packet")

])

while True:

pipeline[0].tick()

if __name__ == "__main__":

main()