141 lines
3.6 KiB
Python
141 lines
3.6 KiB
Python
import json
|
|
from urllib.parse import quote
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Collection, Any
|
|
|
|
|
|
type Numeric = int | float
|
|
|
|
|
|
@dataclass
|
|
class SensorData:
|
|
mac: str
|
|
value: Numeric
|
|
# actually `time` is either decimal int,
|
|
# hex int or <date>T<time>+/-<timezone>
|
|
time: int | None = None
|
|
name: str | None = None
|
|
|
|
|
|
@dataclass
|
|
class GeoPosition:
|
|
lat: Numeric
|
|
lon: Numeric
|
|
alt: Numeric
|
|
|
|
|
|
@dataclass
|
|
class Device:
|
|
mac: str
|
|
name: str | None = None
|
|
owner: str | None = None
|
|
geopos: GeoPosition | None = None
|
|
sensors: Collection[SensorData] = ()
|
|
|
|
def nm_req_data(self) -> str:
|
|
'''Generate a NarodMon format (#mac#value) request body'''
|
|
|
|
res = '#' + self.mac
|
|
if self.name is not None:
|
|
res += '#' + self.name
|
|
res += '\n'
|
|
|
|
if self.owner is not None:
|
|
res += '#OWNER#' + self.owner + '\n'
|
|
|
|
for s in self.sensors:
|
|
res += f'#{s.mac}#{s.value}'
|
|
if s.time is not None:
|
|
res += f'#{s.time}'
|
|
if s.name is not None:
|
|
res += '#' + s.name
|
|
res += '\n'
|
|
|
|
if self.geopos is not None:
|
|
res += (
|
|
f'#LAT#{self.geopos.lat}\n'
|
|
f'#LON#{self.geopos.lon}\n'
|
|
f'#ALT#{self.geopos.alt}\n'
|
|
)
|
|
|
|
res += '##'
|
|
|
|
return res
|
|
|
|
def http_req_data(self) -> str:
|
|
'''Generate an x-www-form-urlencoded format request body'''
|
|
|
|
res = 'ID=' + self.mac
|
|
# TODO: i didn't get how to pass device name in an HTTP request
|
|
# because narodmon docs are written incredibly well /sarcasm
|
|
|
|
for s in self.sensors:
|
|
res += f'&{s.mac}={s.value}'
|
|
# if s.time is not None:
|
|
# res += f'&time={s.time}'
|
|
# if s.name is not None:
|
|
# res += f'&name={quote(s.name)}'
|
|
|
|
if self.sensors:
|
|
s = self.sensors[0]
|
|
if s.time is not None:
|
|
res += f'&time={s.time}'
|
|
|
|
if self.name is not None:
|
|
res += '&name=' + quote(self.name)
|
|
if self.owner is not None:
|
|
res += '&owner=' + quote(self.owner)
|
|
|
|
if self.geopos is not None:
|
|
res += (
|
|
f'&lat={self.geopos.lat}'
|
|
f'&lon={self.geopos.lon}'
|
|
f'&alt={self.geopos.alt}'
|
|
)
|
|
|
|
return res
|
|
|
|
def json_req_data(self) -> str:
|
|
'''Generate a JSON format request body'''
|
|
|
|
obj = {
|
|
'devices': [
|
|
self._json_device_obj(),
|
|
]
|
|
}
|
|
|
|
return json.dumps(obj)
|
|
|
|
def _json_device_obj(self) -> dict[str, Any]:
|
|
'''Create a device JSON object for json_req_data().
|
|
Moved to a separate function to allow performing tests
|
|
with multiple devices in a request at once.'''
|
|
|
|
dev = {
|
|
'mac': self.mac,
|
|
'sensors': [],
|
|
}
|
|
|
|
if self.name is not None:
|
|
dev['name'] = self.name
|
|
if self.owner is not None:
|
|
dev['owner'] = self.owner
|
|
if self.geopos is not None:
|
|
dev['lat'] = self.geopos.lat
|
|
dev['lon'] = self.geopos.lon
|
|
dev['alt'] = self.geopos.alt
|
|
|
|
sl: list[dict[str, Any]] = dev['sensors']
|
|
for s in self.sensors:
|
|
sobj = {
|
|
'id': s.mac,
|
|
'value': s.value,
|
|
}
|
|
if s.time is not None:
|
|
sobj['time'] = s.time
|
|
if s.name is not None:
|
|
sobj['name'] = s.name
|
|
sl.append(sobj)
|
|
|
|
return dev
|