BDS08 and BDS09 encoders

pyModeS/encoder/__init__.py

@@ -0,0 +1,68 @@
+from .bds.bds08 import me08
+from .bds.bds09 import me09
+from pyModeS import common
+
+
+def encode_adsb(**kwargs):
+    """Encode ADS-B message.
+
+    Args:
+        icao (string): Transponder ICAO address (6 hexdigits)
+        capability (int): Transponder capability, between 0 and 7
+        typecode (int): Typecode, less than 32
+
+        callsign (string): Callsign (6 hexdigits)
+        category (int): Aircraft category, between 0 and 7, Default to 0.
+
+        speed (int): Speed in knots.
+        angle (float): Track angle or heading angle in degrees.
+        vertical_rate (int): vertical rate in feet/minute
+        intent_change (int): Intent change flag, 0 or 1. Default to 0.
+        ifr_capability (int): IFR capability flag, 0 or 1. Default to 1.
+        navigation_quality (int): NUC (ver 0) or NACv (ver 1, 2), between 0 and 7.
+            Default to 0.
+        supersonic (bool): Is this a supersonic flight? Default to False.
+        speed_type (str): Speed type: GS, IAS, or TAS. Default to GS.
+        vertical_rate_source (str): GNSS or BARO. Default to BARO.
+        gnss_baro_alt_diff (int): Different between GNSS and barometric altitude in feet.
+            Negative value indicates GNSS altitude below barometric altitude. Default to 0
+
+    Returns:
+        string: 28 hexdigits raw message
+
+    """
+    tc = kwargs.get("typecode")
+
+    if 1 <= tc <= 4:
+        me = me08(**kwargs)
+    elif tc == 19:
+        me = me09(**kwargs)
+
+    msg = _constuct(**dict(kwargs, me=me))
+    return msg
+
+
+def _constuct(**kwargs):
+    icao = kwargs.get("icao")
+    me = kwargs.get("me")
+    capability = kwargs.get("capability", 6)
+
+    if icao is None or len(icao) != 6:
+        raise Exception("Transponder address must be 6 hexadecimal characters.")
+
+    if me is None or len(me) != 14:
+        raise Exception("Message be 14 hexadecimal characters.")
+
+    if capability > 6:
+        raise Exception("Transponder capability must be smaller than 7.")
+
+    header_bin = "10001" + "{0:03b}".format(capability)
+    header_hex = "{0:02X}".format(int(header_bin, 2))
+
+    msg = header_hex + icao + me + "000000"
+
+    pi = common.crc(msg, encode=True)
+    pi_hex = "{0:06X}".format(pi)
+
+    msg = msg[:-6] + pi_hex
+    return msg

pyModeS/encoder/bds/bds05.py

@@ -0,0 +1,5 @@
+# ------------------------------------------
+#   BDS 0,5
+#   ADS-B TC=9-18
+#   Airborn position
+# ------------------------------------------

pyModeS/encoder/bds/bds06.py

@@ -0,0 +1,5 @@
+# ------------------------------------------
+#   BDS 0,6
+#   ADS-B TC=5-8
+#   Surface position
+# ------------------------------------------

pyModeS/encoder/bds/bds08.py

@@ -0,0 +1,40 @@
+# ------------------------------------------
+#   BDS 0,8
+#   ADS-B TC=1-4
+#   Aircraft identitification and category
+# ------------------------------------------
+
+from pyModeS import common
+
+charmap = "#ABCDEFGHIJKLMNOPQRSTUVWXYZ##### ###############0123456789######"
+
+
+def me08(callsign, **kwargs):
+    cs = callsign
+    tc = kwargs.get("typecode")
+    cat = kwargs.get("category", 0)
+
+    if len(cs) > 8:
+        raise Exception("callsign must contain less than 9 characters")
+
+    if tc > 4:
+        raise Exception("typecode must be less 5")
+
+    if cat > 7:
+        raise Exception("category must be less 8")
+
+    if not cs.isalnum():
+        raise Exception("callsign must only contain alphanumeric characters")
+
+    cs = "{:<8}".format(cs.upper())
+
+    idx = [charmap.index(c) for c in cs]
+    me_bin = (
+        "{0:05b}".format(tc)
+        + "{0:03b}".format(cat)
+        + "".join("{0:06b}".format(i) for i in idx)
+    )
+
+    me_hex = "{0:04X}".format(int(me_bin, 2))
+
+    return me_hex

pyModeS/encoder/bds/bds09.py

@@ -0,0 +1,119 @@
+# ------------------------------------------
+#   BDS 0,9
+#   ADS-B TC=19
+#   Aircraft Airborn velocity
+# ------------------------------------------
+
+import numpy as np
+
+
+def me09(speed, angle, vertical_rate, **kwargs):
+    spd = speed
+    agl = angle
+    vr = vertical_rate
+
+    tc = kwargs.get("typecode")
+    intent = kwargs.get("intent_change", 0)
+    ifr = kwargs.get("ifr_capability", 1)
+    navq = kwargs.get("navigation_quality", 0)
+    supersonic = kwargs.get("supersonic", False)
+    spd_type = kwargs.get("speed_type", "gs").lower()
+    vr_source = kwargs.get("vertical_rate_source", "baro").lower()
+    alt_diff = kwargs.get("gnss_baro_alt_diff", 0)
+
+    if tc != 19:
+        raise Exception("Typecode must be 19.")
+
+    if intent not in (0, 1):
+        raise Exception("Intent change flag must be 0 or 1.")
+
+    if ifr not in (0, 1):
+        raise Exception("IFR capability flag must be 0 or 1.")
+
+    if type(supersonic) != bool:
+        raise Exception("Subsonic flag must be True or False.")
+
+    if navq > 7:
+        raise Exception("Navigation quality indicator must be smaller than 8.")
+
+    if spd_type not in ["gs", "tas"]:
+        raise Exception("Speed type must be 'gs', 'ias', or 'tas'.")
+
+    if vr_source not in ["baro", "gnss"]:
+        raise Exception("Vertical rate source must be 'baro' or 'gnss'.")
+
+    me_bin = ""
+
+    # typecode
+    me_bin += "{0:05b}".format(tc)
+
+    # sub-type
+    if supersonic:
+        if spd_type == "gs":
+            me_bin += "010"
+        else:
+            me_bin += "100"
+    else:
+        if spd_type == "gs":
+            me_bin += "001"
+        else:
+            me_bin += "011"
+
+    # intent, ifr, navigation quality
+    me_bin += str(intent) + str(ifr) + "{0:03b}".format(navq)
+
+    # speed and angle part
+    if spd_type == "gs":
+        vx = spd * np.sin(np.radians(agl))
+        vy = spd * np.cos(np.radians(agl))
+
+        if supersonic:
+            vx /= 4
+            vy /= 4
+
+        vx = int(round(vx))
+        vy = int(round(vy))
+
+        sew = "0" if vx >= 0 else "1"
+        sns = "0" if vy >= 0 else "1"
+        vew = "{0:010b}".format(min(abs(vx), 1023) + 1)
+        vns = "{0:010b}".format(min(abs(vy), 1023) + 1)
+
+        me_bin += sew + vew + sns + vns
+
+    elif spd_type == "ias" or spd_type == "tas":
+        hdg = int(round(agl * 1024 / 360))
+        hdg = min(hdg, 1023)
+
+        air_type = "1" if spd_type == "tas" else "0"
+
+        if supersonic:
+            spd /= 4
+
+        spd = min(int(round(spd)), 1023)
+
+        me_bin += "1" + "{0:010b}".format(hdg) + air_type + "{0:010b}".format(spd)
+
+    # vertical rate source
+    me_bin += "1" if vr_source == "baro" else "0"
+
+    # vertical rate
+    me_bin += "0" if vr > 0 else "1"
+    vr = int(round((abs(vr) / 64 + 1)))
+    vr = min(vr, 511)
+    me_bin += "{0:09b}".format(vr)
+
+    # reserved
+    me_bin += "00"
+
+    # altitude difference
+    me_bin += "1" if alt_diff < 0 else "0"
+    alt_diff = int(round(abs(alt_diff) / 25 + 1))
+    alt_diff = min(alt_diff, 127)
+    me_bin += "{0:07b}".format(alt_diff)
+    print(me_bin)
+
+    # convert to hexdigits
+    me_hex = "{0:04X}".format(int(me_bin, 2))
+
+    return me_hex

tests/test_encoder.py

@@ -0,0 +1,23 @@
+from pyModeS import encoder
+
+
+def test_identification():
+    msg = encoder.encode_adsb(
+        icao="406B90", typecode=4, capability=5, category=0, callsign="EZY85MH"
+    )
+    assert msg == "8D406B902015A678D4D220AA4BDA"
+
+
+def test_speed():
+    msg = encoder.encode_adsb(
+        icao="485020",
+        typecode=19,
+        capability=5,
+        speed_type="gs",
+        speed=159,
+        angle=182.88,
+        vertical_rate=-832,
+        vertical_rate_source="gnss",
+        gnss_baro_alt_diff=550,
+    )
+    assert msg == "8D485020994409940838175B284F"