AX25 encoding fixes.

Github user dscp46 discovered a potential data corruption in AX25
frame encoding and the control byte.

In Python3, you can not to some bit manipulations on the str data type
as that has to contain valid Unicode characters.  One of the bytes types
must be used instead.

d-rats_in_mobaxterm_install.sh:
  Reports that newerverions of mobaxterm do not have busybox
  pre-installed.

d-rats/agw.py:
d-rats/comm.py:
d-rats/wlk.py:
  Change to use bytes type for properly encoding binary.

Others:
  Spelling and pylint fixes found.

Author: wb8tyw

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()