-
Notifications
You must be signed in to change notification settings - Fork 5
Expand file tree
/
Copy pathmake_radio.py
More file actions
169 lines (148 loc) · 6.05 KB
/
Copy pathmake_radio.py
File metadata and controls
169 lines (148 loc) · 6.05 KB
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
162
163
164
165
166
167
168
169
# This work is loosley based on work by Barry Byford,
# https://github.com/ukBaz the document I read
# https://ukbaz.github.io/howto/ubit_radio.html had a Creative Commons
# Attribution-ShareAlike 4.0 International License.
# (https://creativecommons.org/licenses/by-sa/4.0/)
#
# This piece of work has completed the psuedo-python and Barry Byford has
# given me his permission to license this code seperately in communication
# https://github.com/ukBaz/ukBaz.github.io/issues/29
#
# Author - Phil Hall - https://github.com/rhubarbdog
# - Barry Byford - https://github.com/ukBaz
# License - MIT March 2019 - see file LICENSE in this repository
# First Published - 2019-03-22
#
# Modifications
# Phil Hall
# 2018-12-20
# 1) Added broadcast rate and channel and others to radio.config
# 2) added endian to .to_bytes() method
# 3) changed some variable names
# 4) Added receive_packet method
# 5) Added support for packet type 4 float
# 6) Limit message length to 19
# 7) Limit send_value: name length to 8
# 8) Added support for packet type 5 value with float
# 9) Added on() and off() methods for completeness
# 10) Strip 0 bytes as per issue 1
import radio
import microbit
import ustruct
class MakeRadio:
def __init__(self, group, power = 6, queue = 3):
radio.config(group = group, data_rate = radio.RATE_1MBIT,\
channel = 7, power = power, queue = queue)
radio.on()
self.dal_header = b'\x01' + group.to_bytes(1, 'little') + b'\x01'
def on(self):
radio.on()
def off(self):
radio.off()
def send_number(self, number):
time_stamp = microbit.running_time().to_bytes(4, 'little')
serial_num = int(0).to_bytes(4, 'little')
if number <= 2147483647 and number >= -2147483648 and\
type(number) is int:
number_bytes = number.to_bytes(4, 'little')
packet_type = int(0).to_bytes(1, 'little')
else:
number_bytes = ustruct.pack('<d',number)
packet_type = int(4).to_bytes(1, 'little')
raw_bytes = (self.dal_header +
packet_type +
time_stamp +
serial_num +
number_bytes)
radio.send_bytes(raw_bytes)
def send_value(self, name, value):
if len(name) > 8:
name = name[:8]
time_stamp = microbit.running_time().to_bytes(4, 'little')
serial_num = int(0).to_bytes(4, 'little')
if value <= 2147483647 and value >= -2147483648 and\
type(value) is int:
number = int(value).to_bytes(4, 'little')
packet_type = int(1).to_bytes(1, 'little')
else:
number = ustruct.pack('<d', value)
packet_type = int(5).to_bytes(1, 'little')
name_bytes = bytes(str(name), 'utf8')
name_length = len(name_bytes).to_bytes(1, 'little')
raw_bytes = (self.dal_header +
packet_type +
time_stamp +
serial_num +
number +
name_length +
name_bytes)
radio.send_bytes(raw_bytes)
def send_string(self, message):
if len(message) > 19:
message = message[:19]
packet_type = int(2).to_bytes(1, 'little')
time_stamp = microbit.running_time().to_bytes(4, 'little')
serial_num = int(0).to_bytes(4, 'little')
message_bytes = bytes(str(message), 'utf8')
message_length = len(message_bytes).to_bytes(1, 'little')
raw_bytes = (self.dal_header +
packet_type +
time_stamp +
serial_num +
message_length +
message_bytes)
radio.send_bytes(raw_bytes)
def receive_packet(self):
data = radio.receive_bytes()
return self._parse_packet(data)
def _parse_packet(self, data):
if data is None:
return None
if data[:3] != self.dal_header:
pass
#raise Exception('Bad header on this channel. ' + str(data[:3]))
packet_type = int.from_bytes(data[3:4], 'little')
if packet_type == 0: # number
number = ustruct.unpack('<i', data[-4:])[0]
return number
elif packet_type == 1: # value
value = ustruct.unpack('<i', data[12:16])[0]
name = str(data[17:], 'utf8')
return (name.rstrip('\x00'), value)
elif packet_type == 2: # string
message = str(data[13:], 'utf8')
return message.rstrip('\x00')
elif packet_type == 4: # floating point number
float_ = ustruct.unpack('<d',data[-8:])[0]
return float_
elif packet_type == 5: # value with float
float_ = ustruct.unpack('<d',data[12:20])[0]
l = len(data)
if l > 29:
l = 29
name = str(data[21:l], 'ascii')
return (name.rstrip('\x00'), float_)
return None # raise Exception('Unknown packet type')
def _sniff(self):
data = radio.receive_bytes()
if data is None:
return None
packet_type = int.from_bytes(data[3:4], 'little')
packet_type = ustruct.unpack('<B', data[3:4])[0]
print("packet id\t:",packet_type)
time_stamp = int.from_bytes(data[4:8], 'little')
print('time stamp\t:',time_stamp)
serial = int.from_bytes(data[8:12], 'little')
print('serial \t:', serial)
if packet_type == 0:
number = ustruct.unpack('<i',data[-4:])
print("number \t:",number)
if packet_type == 1:
number = ustruct.unpack('<i',data[12:16])
print("number \t:",number)
if packet_type == 4:
print('decimal \t', ustruct.unpack('<d',data[-8:]))
if packet_type == 5:
print('decimal \t', ustruct.unpack('<d',data[12:20]))
print("received\t:", self._parse_packet(data))
return data