add reading KISS files

SatsDecoder/ui.py

@@ -586,6 +586,7 @@ def named_conn_btn(self, _=None, **kw):
             utils.ConnMode.TCP_CLI: ('Connect', 'Disconnect'),
             utils.ConnMode.TCP_SRV: ('Start', 'Stop'),
             utils.ConnMode.HEX: ('Run', 'Stop'),
+            utils.ConnMode.KISS_FILES: ('Open', 'Stop'),
         }
         self.con_btn.config(text=d[m]['d' in kw])
 
@@ -618,6 +619,8 @@ def con(self):
         m = utils.ConnMode(self.conn_mode.current())
         if m == utils.ConnMode.HEX:
             self._hex_values()
+        elif m == utils.ConnMode.KISS_FILES:
+            self._kiss_files()
         else:
             self.stop() if self.sk else self._start()
 
@@ -690,6 +693,11 @@ def _finish(ok=0):
 
         ask_hex.update()
 
+    def _kiss_files(self):
+        for fn in filedialog.askopenfilenames(filetypes=[('KISS', ['*.kss']), ('All files', '*.*')]):
+            for t, data in utils.kiss_read(pathlib.Path(fn)):
+                self.feed(data, t)
+
     def _start(self):
         curr_mode = utils.ConnMode(self.conn_mode.current())
         self.is_server = curr_mode == utils.ConnMode.TCP_SRV
@@ -811,7 +819,7 @@ def _receive(self, conn):
 
         return self.feed(frame)
 
-    def feed(self, frame):
+    def feed(self, frame, t=None):
         try:
             data = frame[self.frame_off:]
             for i in self.decoder.recognize(data):
@@ -820,7 +828,7 @@ def feed(self, frame):
                 if '\0' in name:
                     # TODO: log it, '`null` in name not allowed'
                     continue
-                date = 0
+                date = t
 
                 if ty == 'img':
                     ir_ret, img = packet
@@ -830,7 +838,7 @@ def feed(self, frame):
 
                 elif ty == 'tlm':
                     packet, tlm = packet
-                    date = getattr(tlm, 'Time', dt.datetime.utcnow())
+                    date = getattr(tlm, 'Time', t or dt.datetime.utcnow())
                     name = ('%s_%s_%s_%s.txt' % (name, self.proto, tlm._name, date)).replace(
                         ' ', '_').replace(':', '-')
                     fp = pathlib.Path(self.out_dir_v.get()) / name

SatsDecoder/utils.py

@@ -21,13 +21,15 @@ class ConnMode(enum.IntEnum):
     TCP_CLI = 1
     TCP_SRV = 2
     HEX = 3
+    KISS_FILES = 4
 
 
 con_mode_names = {
     ConnMode.AGWPE_CLI: 'AGWPE Client',
     ConnMode.TCP_CLI: 'TCP Client',
     ConnMode.TCP_SRV: 'TCP Server',
     ConnMode.HEX: 'HEX values',
+    ConnMode.KISS_FILES: 'KISS files',
 }
 
 
@@ -394,6 +396,45 @@ def gps_to_utc(week, sec):
     return x + dt.timedelta(seconds=sec)
 
 
+# KISS basic info: https://www.ax25.net/kiss.aspx
+# The basic implementation is taken from the kiss module from gr-satellites
+KISS_FEND = b'\xc0'
+KISS_FESC = b'\xdb'
+KISS_TFEND = b'\xdc'
+KISS_TFESC = b'\xdd'
+KISS_CMD_DATA = 0
+KISS_CMD_TS = 9
+
+
+def kiss_unescape(frame: bytes):
+    frame = frame.replace(KISS_FESC + KISS_TFEND, KISS_FEND)
+    frame = frame.replace(KISS_FESC + KISS_TFESC, KISS_FESC)
+    return frame
+
+
+def kiss_read(fp):
+    epoch = dt.datetime(1970, 1, 1)
+    with fp.open('rb') as kf:
+        frames = kf.read().split(KISS_FEND)
+        if frames[0]:
+            raise ValueError('no frame start: %s' % frames[0])
+
+        t = None
+        for fr in frames[1:]:
+            if not fr:
+                continue
+            if fr[0] == KISS_CMD_TS:
+                # timestamp
+                ts, = struct.unpack('>Q', kiss_unescape(fr[1:]))
+                t = epoch + dt.timedelta(seconds=ts / 1000)
+            elif fr[0] == KISS_CMD_DATA:
+                # data frame
+                yield t, kiss_unescape(fr[1:])
+            else:
+                # TODO: unknown, what to do?
+                pass
+
+
 seqs_map = {
     '\x72\x73\x32\x30\x73': 'aHR0cHM6Ly91cGxvYWQud2lraW1lZGlhLm9yZy93aWtpcGVkaWEvY29tbW9u'
                             'cy90aHVtYi80LzRhL0N1YmVTYXRfR2Vvc2Nhbi1FZGVsdmVpc19lbWJsZW0u'