add a LineBasedModule for the common task of parsing line-by-line input

csdr/module/__init__.py

@@ -110,6 +110,42 @@ class PickleModule(ThreadModule):
         pass
 
 
+class LineBasedModule(ThreadModule, metaclass=ABCMeta):
+    def __init__(self):
+        self.retained = bytes()
+        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:
+                self.retained += data
+                lines = self.retained.split(b"\n")
+
+                # keep the last line
+                # this should either be empty if the last char was \n
+                # or an incomplete line if the read returned early
+                self.retained = lines[-1]
+
+                # log all completed lines
+                for line in lines[0:-1]:
+                    parsed = self.process(line)
+                    if parsed is not None:
+                        self.writer.write(pickle.dumps(parsed))
+
+    @abstractmethod
+    def process(self, line: bytes) -> any:
+        pass
+
+
 class PopenModule(AutoStartModule, metaclass=ABCMeta):
     def __init__(self):
         self.process = None

csdr/module/msk144.py

@@ -1,6 +1,6 @@
 from pycsdr.types import Format
 from pycsdr.modules import ExecModule
-from csdr.module import ThreadModule
+from csdr.module import LineBasedModule
 from owrx.wsjt import WsjtParser, Msk144Profile
 import pickle
 
@@ -17,40 +17,17 @@ class Msk144Module(ExecModule):
         )
 
 
-class ParserAdapter(ThreadModule):
+class ParserAdapter(LineBasedModule):
     def __init__(self):
-        self.retained = bytes()
         self.parser = WsjtParser()
         self.dialFrequency = 0
+        self.profile = Msk144Profile()
         super().__init__()
 
-    def run(self):
+    def process(self, line: bytes):
-        profile = Msk144Profile()
+        # actual messages from msk144decoder should start with "*** "
-
+        if line[0:4] == b"*** ":
-        while self.doRun:
+            return self.parser.parse(self.profile, self.dialFrequency, line[4:])
-            data = self.reader.read()
-            if data is None:
-                self.doRun = False
-            else:
-                self.retained += data
-                lines = self.retained.split(b"\n")
-
-                # keep the last line
-                # this should either be empty if the last char was \n
-                # or an incomplete line if the read returned early
-                self.retained = lines[-1]
-
-                # parse all completed lines
-                for line in lines[0:-1]:
-                    # actual messages from msk144decoder should start with "*** "
-                    if line[0:4] == b"*** ":
-                        self.writer.write(pickle.dumps(self.parser.parse(profile, self.dialFrequency, line[4:])))
-
-    def getInputFormat(self) -> Format:
-        return Format.CHAR
-
-    def getOutputFormat(self) -> Format:
-        return Format.CHAR
 
     def setDialFrequency(self, frequency: int) -> None:
         self.dialFrequency = frequency

owrx/adsb/dump1090.py

@@ -1,9 +1,8 @@
 from pycsdr.modules import ExecModule, Writer, TcpSource
 from pycsdr.types import Format
-from csdr.module import LogWriter, ThreadModule
+from csdr.module import LogWriter, LineBasedModule
 from owrx.socket import getAvailablePort
 import time
-import pickle
 
 import logging
 
@@ -51,38 +50,8 @@ class Dump1090Module(ExecModule):
             self.tcpSource.setWriter(writer)
 
 
-class RawDeframer(ThreadModule):
+class RawDeframer(LineBasedModule):
-    def __init__(self):
+    def process(self, line: bytes):
-        self.retained = bytes()
-        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:
-                self.retained += data
-                lines = self.retained.split(b"\n")
-
-                # keep the last line
-                # this should either be empty if the last char was \n
-                # or an incomplete line if the read returned early
-                self.retained = lines[-1]
-
-                # log all completed lines
-                for line in lines[0:-1]:
-                    parsed = self.parse(line)
-                    if parsed is not None:
-                        self.writer.write(pickle.dumps(parsed))
-
-    def parse(self, line):
         if line.startswith(b'*') and line.endswith(b';') and len(line) in [16, 30]:
             return bytes.fromhex(line[1:-1].decode())
         else: