openwebrx-clone/owrx/aprs/kiss.py

56 lines
1.4 KiB
Python
Raw Normal View History

2021-09-06 13:05:33 +00:00
from pycsdr.modules import Reader
from pycsdr.types import Format
2021-09-06 18:00:14 +00:00
from csdr.module import ThreadModule
2021-09-06 13:05:33 +00:00
import pickle
2021-09-02 09:00:57 +00:00
import logging
logger = logging.getLogger(__name__)
FEND = 0xC0
FESC = 0xDB
TFEND = 0xDC
TFESC = 0xDD
2021-09-06 18:00:14 +00:00
class KissDeframer(ThreadModule):
2021-09-02 09:00:57 +00:00
def __init__(self):
self.escaped = False
self.buf = bytearray()
2021-09-06 13:05:33 +00:00
super().__init__()
def getInputFormat(self) -> Format:
return Format.CHAR
def getOutputFormat(self) -> Format:
return Format.CHAR
def run(self):
while self.doRun:
data = self.reader.read()
if data is None:
self.doRun = False
else:
for frame in self.parse(data):
self.writer.write(pickle.dumps(frame))
2021-09-02 09:00:57 +00:00
def parse(self, input):
for b in input:
if b == FESC:
self.escaped = True
elif self.escaped:
if b == TFEND:
self.buf.append(FEND)
elif b == TFESC:
self.buf.append(FESC)
else:
logger.warning("invalid escape char: %s", str(input[0]))
self.escaped = False
2021-09-06 13:05:33 +00:00
elif b == FEND:
2021-09-02 09:00:57 +00:00
# data frames start with 0x00
if len(self.buf) > 1 and self.buf[0] == 0x00:
2021-09-06 13:05:33 +00:00
yield self.buf[1:]
2021-09-02 09:00:57 +00:00
self.buf = bytearray()
else:
self.buf.append(b)