Verified Commit da641385 authored by Clayton Craft's avatar Clayton Craft
Browse files

Revert "stm_agps: add support for slower serial devices by using trio-serial"

This reverts commit a3aa4cf8.

trio-serial doesn't work with kernel gnss 'devices', e.g. /dev/gnss0
parent 8d84dbcb
......@@ -7,23 +7,19 @@ import os
import pynmea2
import trio
from datetime import datetime
from trio_serial import SerialStream
from .logger import LoggedException
class STM_AGPS:
def __init__(self, serial_port, baud=9600):
def __init__(self, serial_port):
self.__log = logging.getLogger(__name__)
if not os.path.exists(serial_port):
raise LoggedException("Serial port does not exist: "
self._ser_port = serial_port
self._location = b""
# reminder: bytearrays are mutable
self._buf = bytearray()
self._baud = baud
async def __aenter__(self):
......@@ -37,39 +33,24 @@ class STM_AGPS:
async def open(self):
self._ser = SerialStream(self._ser_port, baudrate=self._baud)
await self._ser.aopen()
self._ser = await trio.open_file(self._ser_port,
"w+b", buffering=0)
except Exception as e:
raise LoggedException(e)
async def readline(self):
# based on this implementation of readline:
idx = self._buf.find(b'\n')
if idx >= 0:
line = self._buf[:idx+1]
self._buf = bytearray(self._buf[idx+1:])
return bytes(line)
while True:
data = await self._ser.receive_some(10)
idx = data.find(b'\n')
if idx >= 0:
line = self._buf + data[:idx+1]
self._buf = bytearray(data[idx+1:])
return bytes(line)
return await self._ser.readline()
async def _serial_write_cmd(self, cmd, expect=None):
# number of times to poll serial output for ACK after sending command
polling_loops = 50"cmd: {cmd}")
await self._ser.send_all(str(cmd).encode("ascii"))
await self._ser.send_all(b'\r\n')
await self._ser.write(str(cmd).encode("ascii"))
await self._ser.write(b'\r\n')
if expect:
for i in range(polling_loops):
line = await self.readline()
line = await self._ser.readline()
line = line[:-1]"read: {line}")
if expect.encode("ascii") in line:
......@@ -81,7 +62,7 @@ class STM_AGPS:
# wait for cmd completion
for i in range(polling_loops):
line = await self.readline()
line = await self._ser.readline()
line = line[:-1]"read: {line}")
if str(cmd).encode("ascii") in line:
......@@ -101,7 +82,7 @@ class STM_AGPS:
if line.startswith(ack.encode()):
await f.write(line + b'\n')
line = await self.readline()
line = await self._ser.readline()
line = line[:-1]
async def _load_from_file(self, ack, file):
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment