jono/mqttdevicemanager
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

working parsed output as properties

Browse Source
This commit is contained in:
Jono Targett 2026-03-15 00:16:34 +10:30
parent cc2e5cf613
commit 07123f2725
1 changed files with 12 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
assethandler.py
View File

@ -5,6 +5,7 @@ import uuid
import aioserial import aioserial
import aiomqtt import aiomqtt
import pyubx2 import pyubx2
from datetime import datetime, timedelta
SERIAL_PORT = "/tmp/ttyV0" SERIAL_PORT = "/tmp/ttyV0"
BAUD = 115200 BAUD = 115200
@ -12,11 +13,7 @@ BAUD = 115200
SERIAL_PUBLISH_TOPIC = "serial/data" SERIAL_PUBLISH_TOPIC = "serial/data"
SERIAL_COMMAND_TOPIC = "serial/command" SERIAL_COMMAND_TOPIC = "serial/command"
async def parse_serial(ser): async def parse_serial(ser):
"""
Async generator: read raw bytes from serial and yield parsed UBX/NMEA messages.
"""
buffer = bytearray() buffer = bytearray()
# pyubx2 requires a file-like object, so we'll use a memoryview wrapper # pyubx2 requires a file-like object, so we'll use a memoryview wrapper
@ -32,39 +29,34 @@ async def parse_serial(ser):
ubr = pyubx2.UBXReader(StreamWrapper(), parsing=True) ubr = pyubx2.UBXReader(StreamWrapper(), parsing=True)
while True: while True:
# read available data from serial (small chunks)
chunk = await ser.read_async(200) chunk = await ser.read_async(200)
if chunk: if chunk:
buffer.extend(chunk) buffer.extend(chunk)
# attempt to parse as many messages as possible
try: try:
while True: while True:
raw, parsed = ubr.read() raw, parsed = ubr.read()
if raw is None: if raw is None:
break break
yield raw, parsed yield raw, parsed
except (BlockingIOError, Exception): except (pyubx2.UBXStreamError, BlockingIOError, Exception):
# incomplete message; wait for more bytes
pass pass
except pyubx2.UBXStreamError:
pass # silently ignore incomplete packets
else: else:
await asyncio.sleep(0) # yield control await asyncio.sleep(0) # yield control
async def serial_reader(ser, mqtt_client): async def serial_reader(ser, mqtt_client):
"""
Coroutine wrapper around async generator parse_serial.
Feeds parsed UBX/NMEA messages to MQTT.
"""
async for raw, parsed in parse_serial(ser): async for raw, parsed in parse_serial(ser):
# For example, publish raw bytes to MQTT if isinstance(parsed, pyubx2.UBXMessage):
print(parsed) for name, value in vars(parsed).items():
await mqtt_client.publish("serial/parsed", str(parsed)) if name.startswith('_'):
await mqtt_client.publish("serial/raw", raw) continue
# Optional: inspect parsed fields await mqtt_client.publish(f"ubx/{parsed.identity}/{name}", value, retain=True)
# print(parsed) else:
print(parsed.identity, parsed)
await mqtt_client.publish(f"ubx/raw", raw)
await mqtt_client.publish(f"ubx/parsed", str(parsed))
async def mqtt_command_writer(ser, client): async def mqtt_command_writer(ser, client):
async for message in client.messages: async for message in client.messages: