AX25 encoding fixes.

changes/298.bugfix

@@ -0,0 +1,2 @@
+Need to use AX.25 Unnumbered Information frames.
+Fix possible Filename corruption for AX.25 frames.

changes/304.bugfix

@@ -0,0 +1 @@
+Make sure busybox is installed for MobaXterm installs

changes/305.bugfix

@@ -0,0 +1 @@
+Possible data loss if special binary sequence sent over D-Star radio links

d-rats_in_mobaxterm_install.sh

@@ -16,7 +16,7 @@ export DEBIAN_FRONTEND
 $apt_get -y upgrade
 $apt_get -y install \
   aspell aspell-de aspell-en aspell-es aspell-it \
-  ffmpeg \
+  busybox ffmpeg \
   gedit gettext gettext-devel git gcc-core \
   libgtk3_0 libjpeg-devel libportaudio-devel \
   python39 python39-devel python39-gi python39-lxml \

d-rats_repeater.py

@@ -42,10 +42,10 @@
     import gettext
     _ = gettext.gettext
 
-import gi
+import gi  # type: ignore
 gi.require_version("Gtk", "3.0")
-from gi.repository import Gtk
-from gi.repository import GObject
+from gi.repository import Gtk  # type: ignore
+from gi.repository import GObject  # type: ignore
 
 # Make sure no one tries to run this with privileges.
 try:

d_rats/agw.py

@@ -637,30 +637,50 @@ def ssid(call):
     try:
         sid = int(sid)
     except ValueError:
-        raise InvalidCallsignError("Invalid SSID `%s'" % sid)
+        raise InvalidCallsignError("Invalid SSID `%s'" % sid) from None
 
     if sid < 0 or sid > 7:
         raise InvalidCallsignError("Invalid SSID `%i'" % sid)
 
     return callsign, sid
 
 
-def encode_ssid(sid, last=False):
+def encode_ssid(sid, last=False) -> bytes:
     '''
     Encode SSID for transmission.
 
     :param sid: Station ID number
     :type sid: int
-    :param last: last flag, unused
+    :param last: last flag, indicates last call in a list of calls.
     :type last: bool
     :returns: Encoded Station ID character
-    :rtype: str
+    :rtype: bytes
     '''
     if last:
         mask = 0x61
     else:
         mask = 0x60
-    return chr((sid << 1) | mask)
+    result = (sid << 1) | mask
+    return result.to_bytes(1,'little')
+
+
+def encode_call(callsign: str, last=False) -> bytes:
+    '''
+    Encode callsign to AX25 call sign field.
+
+    :param callsign: Call sign with SSID
+    :type callsign: str
+    :param last: last flag, indicates last call in a list of calls.
+    :type last: bool
+    :returns: Encoded call sign
+    :rtype: bytes
+    '''
+    call, sid = ssid(callsign)
+    # Encode the call by grabbing each character and shifting
+    # left one bit
+    ax25_call = b"".join([chr(ord(x) << 1) for x in call])
+    ax25_call += encode_ssid(sid=sid, last=last)
+    return ax25_call
 
 
 def transmit_data(conn, dcall, spath, data):
@@ -677,26 +697,18 @@ def transmit_data(conn, dcall, spath, data):
     :type data: bytes
     '''
     global_logger.info("transmit_data:")
-    call, sid = ssid(dcall)
 
-    # Encode the call by grabbing each character and shifting
-    # left one bit
-    dst_str = "".join([chr(ord(x) << 1) for x in call])
-    dst_str += encode_ssid(sid)
-    dst = dst_str.encode('utf-8', 'replace')
+    dst = encode_call(callsign=dcall)
 
-    src_str = ""
     for scall in spath:
-        call, sid = ssid(scall)
-        src_str += "".join([chr(ord(x) << 1) for x in call])
-        src_str += encode_ssid(sid, spath[-1] == scall)
-        src = src_str.encode('utf-8', 'replace')
+        last = spath[-1] == scall
+        src = encode_call(callsign=scall, last=last)
 
     data_frame = struct.pack("!B7s%isBB" % len(src),
                              0x00,    # Space for flag (?)
                              dst,     # Dest Call
                              src,     # Source Path
-                             0x3E,    # Info
+                             0x03,    # Unnumbered Info
                              0xF0)    # PID: No layer 3
     data_frame += data
 
@@ -747,6 +759,7 @@ def test_server(host="127.0.0.1", port=8000):
     '''
     # Quick and dirty simulator for agwpe unit tests.
     global_logger.info("test_server: starting %s:%i", host, port)
+    # pylint: disable=import-outside-toplevel
     from time import sleep
 
     agw = AGWConnection(host, port, timeout=10, server=True)
@@ -775,6 +788,7 @@ def main():
                         datefmt="%m/%d/%Y %H:%M:%S",
                         level=logging.INFO)
 
+    # pylint: disable=import-outside-toplevel
     from time import sleep
     server = threading.Thread(target=test_server)
     server.start()

d_rats/aprs_dprs.py

@@ -35,7 +35,7 @@ class AprsDprsCodes:
     APRS_ALTERNATE_SYMBOL_TABLE = '\\'
     APRS_ADVISORY_CODE = '\\<'
     APRS_CAR_CODE = '/>'
-    APRS_DIGI_CODD = '/#'
+    APRS_DIGI_CODE = '/#'
     APRS_DOT_CODE = '//'
     APRS_FALLBACK_CODE = '//'
     APRS_INFO_KIOSK_CODE = '\\?'
@@ -85,7 +85,7 @@ def _init_dprs_to_aprs(cls):
         and a list of regular APRS symbols
 
         `DPRS information <https://www.aprs-is.net/DPRS.aspx>`_
-        `DPRS Caldulator <http://www.aprs-is.net/DPRSCalc.aspx>_
+        `DPRS calculator <http://www.aprs-is.net/DPRSCalc.aspx>_
         '''
 
         if cls._dprs_to_aprs:

d_rats/comm.py

@@ -13,7 +13,7 @@
 # Needed for python2+python3 support
 import sys
 
-import serial
+import serial  # type: ignore
 
 from . import utils
 from . import agw
@@ -394,15 +394,15 @@ def is_xon(self):
             self.dsr_seen = True
         time.sleep(0.01)
         self.state = True  # Pending re-write
-        # Code below is not optimmal, for some reason it almost works
+        # Code below is not optimal, for some reason it almost works
         #
         # Using the serial driver xon/xoff protocol handling instead of
         # this has been found not work in some cases.
         #
         # Bug #1 No code is present to detect if an XOFF or XON appears in
         # the middle of a data stream.
         # Bug #2 The XOFF/XON is not guaranteed to show up at the
-        # beginnng of a read packet.
+        # beginning of a read packet.
         # Apparently the serial device is expected to send at least XON/XOFF to
         # back to D-rats for every packet sent to it.
         if self.in_waiting == 0:
@@ -947,19 +947,13 @@ def write(self, buf):
         :type buf: bytes
         '''
         spath = [self.__call,] + self.__path.split(",")
-        src = ""
         for scall in spath:
-            call, sid = agw.ssid(scall)
-            src += "".join([chr(ord(x) << 1) for x in call])
-            src += agw.encode_ssid(sid, spath[-1] == scall)
+            last = spath[-1] == scall
+            send_src = agw.encode_call(callsign=scall, last=last)
 
-        call, sid = agw.ssid("DRATS")
-        dst = "".join([chr(ord(x) << 1) for x in call])
-        dst += agw.encode_ssid(sid)
-        send_dst = dst.encode('utf-8', 'replace')
-        send_src = src.encode('utf-8', 'replace')
+        send_dst = agw.encode_call(callsign="DRATS")
 
-        hdr = struct.pack("!7s%isBB" % len(src),
+        hdr = struct.pack("!7s%isBB" % len(send_src),
                           send_dst,     # Dest call
                           send_src,     # Source path
                           0x03,         # Control

d_rats/gps.py

@@ -30,7 +30,8 @@
 import socket
 
 from math import pi, cos, acos, sin, atan2
-import serial
+import gettext
+import serial  # type: ignore
 
 from .dplatform import Platform
 
@@ -39,11 +40,7 @@
 from .aprs_icons import APRSicons
 from .dratsexception import DPRSException
 
-# This makes pylance happy with out overriding settings
-# from the invoker of the class
-if not '_' in locals():
-    import gettext
-    _ = gettext.gettext
+_ = gettext.gettext
 
 TEST = "$GPGGA,180718.02,4531.3740,N,12255.4599,W,1,07,1.4," \
        "50.6,M,-21.4,M,,*63 KE7JSS  ,440.350+ PL127.3"
@@ -213,7 +210,7 @@ def calc(buf):
         for _char in buf:
             char = ord(_char)
             for _indx in range(0, 8):
-                xorflag = (((icomcrc ^ char) & 0x01) == 0x01)
+                xorflag = ((icomcrc ^ char) & 0x01) == 0x01
                 icomcrc = (icomcrc >> 1) & 0x7fff
                 if xorflag:
                     icomcrc ^= 0x8408
@@ -707,14 +704,14 @@ def to_aprs(self, dest="APRATS",
         '''
         To APRS.
 
-        :param dest: Destination, default "APRSATS"
+        :param dest: Destination, default "APRATS"
         :param dest: str
         :param code: APRS code, default APRS_CAR_CODE
         :type code: str
         :returns: a GPS-A (APRS-compliant) string
         :rtype: str
         '''
-        symtab = code[0]
+        symbol_table = code[0]
         symbol = code[1]
         stamp = time.strftime("%H%M%S", time.gmtime())
 
@@ -726,25 +723,25 @@ def to_aprs(self, dest="APRATS",
         sta_str = "%s>%s,DSTAR*:/%sh" % (sta, dest, stamp)
 
         if self.latitude > 0:
-            northsouth = "N"
+            north_south = "N"
             lat_m = 1
         else:
-            northsouth = "S"
+            north_south = "S"
             lat_m = -1
 
         if self.longitude > 0:
-            eastwest = "E"
+            east_west = "E"
             lon_m = 1
         else:
-            eastwest = "W"
+            east_west = "W"
             lon_m = -1
 
         sta_str += "%07.2f%s%s%08.2f%s%s" % \
                    (deg2nmea(self.latitude * lat_m),
-                    northsouth,
-                    symtab,
+                    north_south,
+                    symbol_table,
                     deg2nmea(self.longitude * lon_m),
-                    eastwest,
+                    east_west,
                     symbol)
         if self.speed and self.direction:
             sta_str += "%03.0f/%03.0f" % \
@@ -891,7 +888,7 @@ def __init__(self, sentence, station=None):
         else:
             self.logger.info("Unsupported GPS sentence type: %s", sentence)
 
-    def _test_checksum(self, string, csum):
+    def _test_checksum(self, string, checksum):
         try:
             idx = string.index("*")
         except ValueError:
@@ -901,14 +898,14 @@ def _test_checksum(self, string, csum):
 
         segment = string[1:idx]
 
-        csum = csum.upper()
-        calc_csum = nmea_checksum(segment).upper()
+        checksum = checksum.upper()
+        calc_checksum = nmea_checksum(segment).upper()
 
-        if csum != calc_csum:
+        if checksum != calc_checksum:
             self.logger.info("_test_checksum: Failed checksum: %s != %s",
-                             csum, calc_csum)
+                             checksum, calc_checksum)
 
-        return csum == calc_csum
+        return checksum == calc_checksum
 
     def _parse_gpgga(self, string):
         elements = string.split(",", 14)
@@ -934,7 +931,7 @@ def _parse_gpgga(self, string):
         if not match:
             raise GpsGppgaChecksumError("No checksum (%s)" % elements[14])
 
-        csum = match.group(2)
+        checksum = match.group(2)
         if "," in match.group(3):
             sta, com = match.group(3).split(",", 1)
             if not sta.strip().startswith("$"):
@@ -945,7 +942,7 @@ def _parse_gpgga(self, string):
         if len(self.comment) >= 7 and "*" in self.comment[-3:-1]:
             self._parse_dprs_comment()
 
-        self.valid = self._test_checksum(string, csum)
+        self.valid = self._test_checksum(string, checksum)
 
     def _parse_gprmc(self, string):
         if "\r\n" in string:
@@ -984,7 +981,7 @@ def _parse_gprmc(self, string):
             self.logger.info("_parse_gprmc: Invalid end: %s", elements[end])
             return
 
-        csum = match.group(1)
+        checksum = match.group(1)
         if "," in station:
             sta, com = station.split(",", 1)
             self.station = utils.filter_to_ascii(sta.strip())
@@ -1000,7 +997,7 @@ def _parse_gprmc(self, string):
                              elements[2])
         else:
             self.logger.info("_parse_gprmc: GPRMC is valid")
-            self.valid = self._test_checksum(string, csum)
+            self.valid = self._test_checksum(string, checksum)
 
     def _from_nmea_gpgga(self, string):
         string = string.replace('\r', ' ')
@@ -1378,7 +1375,7 @@ def connect(self):
         :rtype: bool
         '''
         try:
-            _, host, port = self.port.split(":", 3)
+            _proto, host, port = self.port.split(":", 3)
             port = int(port)
         except ValueError as err:
             self.logger.info("connect: Unable to parse %s (%s)",

d_rats/wl2k.py

@@ -11,11 +11,11 @@
 import time
 import re
 import random
-import gi
+import gi  # type: ignore
 
 gi.require_version("Gtk", "3.0")
-from gi.repository import GLib
-from gi.repository import GObject
+from gi.repository import GLib  # type: ignore
+from gi.repository import GObject  # type: ignore
 
 from d_rats import version
 from d_rats import formgui
@@ -393,8 +393,9 @@ def send_to_socket(self, sock):
         data = self.__lzh_content
 
         # filename \0 length(0) \0
-        header_str = self.__name + "\x00" + chr(len(data) & 0xFF) + "\x00"
-        header = header_str.encode('utf-8', 'replace')
+        data_len = chr(len(data) & 0xFF)
+        header = self.__name.encode('utf-8', 'replace') + b'\x00' + \
+            data_len.to_bytes(1,'little') + b'\x00'
         sock.send(struct.pack("<BB", FBB_BLOCK_HDR, len(header)) + header)
 
         checksum = 0
@@ -680,7 +681,7 @@ def send_messages(self, messages):
             cs_octet += ord("\r")
             self._send(proposal)
 
-        cs_octet = ((~cs_octet & 0xFF) + 1)
+        cs_octet = (~cs_octet & 0xFF) + 1
         data_str = "F> %02X" % cs_octet
         self._send(data_str.encode('utf-8', 'replace'))
         resp = self._recv()