-
Notifications
You must be signed in to change notification settings - Fork 14
/
radioreceiver.py
214 lines (189 loc) · 6.94 KB
/
radioreceiver.py
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
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
import datetime
import glob
import json
import os
import subprocess
import sys
import serial
from django.core.serializers.json import DjangoJSONEncoder
from rest_framework import status
from rest_framework.response import Response
from rest_framework.views import APIView
from mercury.models import AGEvent
from mercury.serializers import AGMeasurementSerializer
def serial_ports():
""" Lists serial port names
:raises EnvironmentError:
On unsupported or unknown platforms
:returns:
A list of the serial ports available on the system
"""
if sys.platform.startswith("win"):
ports = ["COM%s" % (i + 1) for i in range(256)]
elif sys.platform.startswith("darwin"):
ports = glob.glob("/dev/tty.*")
elif sys.platform.startswith("linux") or sys.platform.startswith("cygwin"):
# this excludes your current terminal "/dev/tty"
ports = glob.glob("/dev/tty[A-Za-z]*")
else:
raise EnvironmentError("Unsupported platform")
result = []
for port in ports:
try:
s = serial.Serial(port)
s.close()
result.append(port)
except (OSError, serial.SerialException):
pass
if "TRAVIS" in os.environ:
result = ["dev/tty.USB"]
return result
def call_script(uuid, port, fake):
"""
Run a shell script to receive radio sensor data from the vehicle
This script will call local server to store all data received
Command Example:
python3 ./scripts/radioport.py --uuid d81cac8d-26e1-4983-a942-1922e54a943d
--port /dev/tty.SOC --fake
--data '{"sensor_id": 1, "values": {"power": "2", "speed": 1},
"date": "2020-02-02T20:21:22"}'
"""
command = "python3 ./scripts/radioport.py --uuid {} --port {} ".format(
str(uuid), str(port)
)
if fake:
command = command + "--fake --data "
data = {
"sensor_id": 1,
"values": {"power": "2", "speed": 1},
"date": datetime.datetime(2020, 2, 2, 20, 21, 22),
}
data = json.dumps(data, cls=DjangoJSONEncoder)
command = command + "'" + data + "'"
print(command)
subprocess.call(command, shell=True)
def check_port(ser):
"""
Package ser.is_open for test purpose
TODO: Add other checks on Serial port
"""
return ser.is_open
class RadioReceiverView(APIView):
"""
This is a Django REST API supporting user to send GET request fetching the RADIO
module configuration settings from backend and to send POST request sending
JSON-formatted data.
"""
def get(self, request, event_uuid=None):
"""
The get request sent from web to determine the parameters of the serial port
Url Sample:
https://localhost:8000/radioreceiver/d81cac8d-26e1-4983-a942-1922e54a943d?
eventid=1&enable=1&baudrate=8000&bytesize=8&parity=N&stopbits=1&timeout=None&fake=1
uuid: event_uuid
enable: must define, set the port on if 1, off if 0
baudrate: Optional, default 9600
bytesize: Optional, default 8 bits
parity: Optional, default no parity
stop bits: Optional, default one stop bit
timeout: Optional, default 1 second
fake: Optional, send fake data for test only
"""
# First check event_uuid exists
try:
event = AGEvent.objects.get(event_uuid=event_uuid)
except AGEvent.DoesNotExist:
event = False
if event is False:
return Response("Wrong uuid in url", status=status.HTTP_400_BAD_REQUEST)
# Check Serial port parameters
params = request.query_params
enable = params.get("enable")
if enable is None:
return Response(
"Missing enable value in url", status=status.HTTP_400_BAD_REQUEST
)
enable = int(enable)
ser = serial.Serial()
valid_ports = serial_ports()
if len(valid_ports) > 0:
ser.port = valid_ports[0]
else:
return Response("No valid ports on the backend", status=status.HTTP_200_OK)
res = {"enable": enable}
if params.get("baudrate"):
ser.baudrate = params.get("baudrate")
if params.get("bytesize"):
bytesize = int(params.get("bytesize"))
ser.bytesize = bytesize
if params.get("parity"):
ser.parity = params.get("parity")
if params.get("stopbits"):
ser.stopbits = int(params.get("stopbits"))
if params.get("timeout"):
timeout = int(params.get("timeout"))
ser.timeout = timeout
fake = params.get("fake")
if fake:
call_script(event_uuid, ser.port, fake)
elif enable:
try:
ser.open()
if check_port(ser):
# Call Script
call_script(event_uuid, ser.port, fake)
except serial.serialutil.SerialException:
pass
else:
if check_port(ser):
ser.close()
# Response data
res["baudrate"] = ser.baudrate
res["bytesize"] = ser.bytesize
res["parity"] = ser.parity
res["stopbits"] = ser.stopbits
res["timeout"] = ser.timeout
return Response(json.dumps(res), status=status.HTTP_200_OK)
def post(self, request, event_uuid=None):
"""
The post receives sensor data through internet
Url example:
http://localhost:8000/radioreceiver/d81cac8d-26e1-4983-a942-1922e54a943d
Post Json Data Example
{
"sensor_id": 1,
"values": {
"power" : "1",
"speed" : "2",
}
"date" : 2020-03-11T20:20+01:00
}
"""
# First check event_uuid exists
try:
event = AGEvent.objects.get(event_uuid=event_uuid)
except AGEvent.DoesNotExist:
event = False
if event is False:
return Response("Wrong uuid in url", status=status.HTTP_400_BAD_REQUEST)
json_data = request.data
if isinstance(json_data, str):
json_data = json.loads(json_data)
res = {"measurement_event": event_uuid}
dic = {
"measurement_timestamp": "date",
"measurement_sensor": "sensor_id",
"measurement_value": "values",
}
for d in dic:
if json_data.get(dic[d]) is None:
return Response(
"Missing required params " + dic[d],
status=status.HTTP_400_BAD_REQUEST,
)
res[d] = json_data[dic[d]]
serializer = AGMeasurementSerializer(data=res)
if serializer.is_valid():
serializer.save()
return Response("Saved Successfully", status=status.HTTP_200_OK)
return Response("The model fails to save", status=status.HTTP_400_BAD_REQUEST)