uscg.py

#!/usr/bin/env python
"""Connect to a socket and forward what is received to another port.

Filter to a list of AIS receivers/basestations.


TODO: For speed, provide functions that only parse the timestamp, station, etc.
"""

import datetime
import doctest
import re
import sys
import time
import unittest

from BitVector import BitVector

from . import binary, nmea, sqlhelp

######################################################################
# NEW Regular Expression Parsing Style

# FIX: make the field names be name1_name2 rather than camel case

# USCG has some receivers that emit corrupted fields, so loosen from this
#  | (,s(?P<s_rssi>\d*))
#  | (,d(?P<signal_strength>[-0-9]*))
#  | (,t(?P<t_recver_hhmmss>(?P<t_hour>\d\d)(?P<t_min>\d\d)(?P<t_sec>\d\d.\d*)))
#  | (,T(?P<time_of_arrival>[0-9.]*))
#  | (,x(?P<x_station_counter>[0-9]*))
#  | (,(?P<station>(?P<station_type>[rbB])[a-zA-Z0-9]*))

# AIVDO might not have a channel associated.
uscg_ais_nmea_regex_str = r"""[!$](?P<talker>AI)(?P<stringType>VD[MO])
,(?P<total>\d?)
,(?P<senNum>\d?)
,(?P<seqId>[0-9]?)
,(?P<chan>[AB]?)
,(?P<body>[;:=@a-zA-Z0-9<>\?\'\`]*)
,(?P<fillBits>\d)\*(?P<checksum>[0-9A-F][0-9A-F])
(
  (,S(?P<slot>\d*))
  | (,s(?P<s_rssi>\d*))
  | (,d(?P<signal_strength>[-0-9]*))
  | (,t(?P<t_recver_hhmmss>(?P<t_hour>\d\d)(?P<t_min>\d\d)(?P<t_sec>\d\d.\d*)))
  | (,T(?P<time_of_arrival>[^,]*))
  | (,x(?P<x_station_counter>[0-9]*))
  | (,(?P<station>(?P<station_type>[rbB])[a-zA-Z0-9_-]*))
)*
,(?P<timeStamp>\d+([.]\d+)?)?
"""
"""
Regular expression for parsing a USCG.

* s - receive signal strength indicator (RSSI)
* d - dBm
* S - slot number the message was received in
* x - index counter incremented for each sentence sent by remote site
* T - time of arrival with the minute.  Related to slot number
* r - receive station  r[^Bb] is a receiver, rR receiver, rB basestation, b is basestation
* t - HHMMSS.SS at the receiver (presume this is UTC from the local GPS)

@note: probably not complete
@todo: What is the x field?
@todo: make all fields explicit so errors do not parse
@bug: does not match this>???? !AIVDM,1,1,,B,85MwqdAKf=Wsd5sKUfl@u>DMk70JwpQ2hjnTHlbfcWj<2n<jRtHd,0*7E,x151038,r003669947,1222129826
@bug: is S missing a comma?
@bug: make this conform to the python coding style guides... time_stamp
"""

uscg_ais_nmea_regex = re.compile(uscg_ais_nmea_regex_str, re.VERBOSE)
"""Use this regex to parse USCG NMEA strings fields after the checksum."""


def write_uscg_nmea_fields(nmea_str, out=sys.stdout, indent="\t"):
    """Write out the fields of a USCG nmea string.

    Args:
        nmea_str: USCG style nmea string
        out: stream object to write to
        indent: how to indent each field
    """
    match_obj = uscg_ais_nmea_regex.search(nmea_str)
    if not match_obj:
        return
    out.write(indent + "         prefix = " + str(match_obj.group("talker")) + "\n")
    out.write(indent + "     stringType = " + str(match_obj.group("stringType")) + "\n")
    out.write(indent + "          total = " + str(match_obj.group("total")) + "\n")
    out.write(indent + "         senNum = " + str(match_obj.group("senNum")) + "\n")
    out.write(indent + "          seqId = " + str(match_obj.group("seqId")) + "\n")
    out.write(indent + "           chan = " + str(match_obj.group("chan")) + "\n")
    out.write(indent + "           body = " + str(match_obj.group("body")) + "\n")
    out.write(indent + "       fillBits = " + str(match_obj.group("fillBits")) + "\n")
    out.write(indent + "       checksum = " + str(match_obj.group("checksum")) + "\n")
    out.write(indent + "           slot = " + str(match_obj.group("slot")) + "\n")
    out.write(indent + "              s = " + str(match_obj.group("s_rssi")) + "\n")
    out.write(
        indent + "signal_strength = " + str(match_obj.group("signal_strength")) + "\n"
    )
    out.write(
        indent + "time_of_arrival = " + str(match_obj.group("time_of_arrival")) + "\n"
    )
    out.write(
        indent + "              x = " + str(match_obj.group("x_station_counter")) + "\n"
    )
    out.write(indent + "        station = " + str(match_obj.group("station")) + "\n")
    out.write(
        indent + "   station_type = " + str(match_obj.group("station_type")) + "\n"
    )
    out.write(indent + "      timeStamp = " + str(match_obj.group("timeStamp")) + "\n")


######################################################################
# OLD Style


def get_station(nmeaStr):
    """Return the station without doing anything else.  Try to be fast"""
    fields = nmeaStr.split(",")
    station = None
    for i in range(len(fields) - 1, 5, -1):
        if len(fields[i]) == 0:
            continue  # maybe it should throw a parse exception instead?
        if fields[i][0] in ("b", "r"):
            station = fields[i]
            continue
    return station


def get_contents(nmeaStr):
    """Return the AIS msg string.  AIS goo"""
    return nmeaStr.split(",")[5]


class UscgNmea:
    __slots__ = (
        "aisChannel",
        "cg_sec",
        "checksumStr",
        "contents",
        "fillbits",
        "msgTypeChar",
        "nmeaType",
        "rssi",
        "sentenceNum",
        "sequentialMsgId",
        "signalStrength",
        "slotNumber",
        "sqlTimestampStr",
        "station",
        "stationTypeCode",
        "timeOfArrival",
        "timestamp",
        "totalSentences",
        "x",
    )

    @classmethod
    def from_nmea(cls, nmea_str: str) -> UscgNmea:
        """Factory constructor to instantiate UscgNmea from a sentence string."""
        return cls(nmea_str=nmea_str)

    def __init__(
        self,
        nmea_str: str | None = None,
        *,
        nmeaStr: str | None = None,
    ):
        """Initialize UscgNmea from sentence string.

        Args:
            nmea_str: Standard NMEA sentence string (positional or keyword).
            nmeaStr: Legacy keyword alias for nmea_str.
        """
        nmea_str = nmea_str or nmeaStr
        if nmea_str is not None:
            fields = nmea_str.split(",")
            self.cg_sec = float(fields[-1])
            self.timestamp = datetime.datetime.utcfromtimestamp(self.cg_sec)
            self.sqlTimestampStr = sqlhelp.sec2timestamp(self.cg_sec)
            # See 80_330e_PAS
            self.nmeaType = fields[0][1:]
            self.totalSentences = int(fields[1])
            self.sentenceNum = int(fields[2])
            tmp = fields[3]
            if len(tmp) > 0:
                self.sequentialMsgId = int(tmp)
            else:
                self.sequentialMsgId = None
            # FIX: make an int if the above is set
            self.aisChannel = fields[4]  # 'A' or 'B'
            self.contents = fields[5]
            self.fillbits = int(fields[6].split("*")[0])
            self.checksumStr = fields[6].split("*")[
                1
            ]  # FIX: this is a hex string.  Convert?

            if self.sentenceNum == 1:
                self.msgTypeChar = fields[5][0]
            else:
                self.msgTypeChar = None

            self.station = None
            for i in range(len(fields) - 1, 5, -1):
                if len(fields[i]) == 0:
                    continue  # maybe it should throw a parse exception instead?
                f = fields[i]
                c = f[0]  # first character determines what the field is
                if c in ("b", "r", "B", "R"):
                    self.station = (
                        f  # FIX: think we want to keep the code in the first char
                    )
                    self.stationTypeCode = self.station[0]
                    continue
                    # break # Found it so ditch the for loop
                if c == "s":
                    self.rssi = int(f[1:])
                    continue
                if c == "d":
                    self.signalStrength = int(f[1:])
                    continue
                if c == "T":
                    try:
                        self.timeOfArrival = float(f[1:])
                    except ValueError:
                        # print 'warning: bogus time of arrival: %s' % (f[1:],)
                        pass
                    continue
                if c == "S":
                    self.slotNumber = int(f[1:])
                    continue
                if c == "x":
                    # I don't know what x is
                    self.x = int(f[1:])
                    continue

    def getBitVector(self):
        """
        Returns:
            bits for the payload (even if this is a multipart)
            BitVector
        """
        return binary.ais6tobitvec(self.contents)

    def __eq__(self, other):
        # Try to be smart for speed
        if getattr(self, "cg_sec", None) != getattr(other, "cg_sec", None):
            return False
        if getattr(self, "contents", None) != getattr(other, "contents", None):
            return False
        return str(self) == str(other)

    def __ne__(self, other):
        return not self.__eq__(other)

    def __str__(self):
        return self.buildNmea()

    def buildNmea(self):
        """Use the values in this message to reconstruct a single line nmea string"""

        parts = ["!" + self.nmeaType, str(self.totalSentences), str(self.sentenceNum)]
        if getattr(self, "sequentialMsgId", None) is None:
            parts.append("")
        else:
            parts.append(str(self.sequentialMsgId))
        parts.append(self.aisChannel)
        parts.append(self.contents)
        parts.append(str(self.fillbits) + "*" + self.checksumStr)

        if getattr(self, "rssi", None) is not None:
            parts.append("s" + str(self.rssi))
        if getattr(self, "signalStrength", None) is not None:
            parts.append("d" + str(self.signalStrength))
        if getattr(self, "timeOfArrival", None) is not None:
            parts.append("T" + str(self.timeOfArrival))
        if getattr(self, "slotNumber", None) is not None:
            parts.append("S" + str(self.slotNumber))
        if getattr(self, "x", None) is not None:
            parts.append("x" + str(self.x))

        if getattr(self, "station", None):
            parts.append(self.station)
        parts.append(str(getattr(self, "cg_sec", "")))  # Always last
        return ",".join(parts)


#    def getDriver(self):
#        """
#        Return the python module that handles this message type
#        """
# FIX: where did I do this nicely?


class TestUscgNmea(unittest.TestCase):
    def testUscgNmea(self):
        un = UscgNmea(
            "!AIVDM,1,1,,B,15Cjtd0Oj;Jp7ilG7=UkKBoB0<06,0*63,s1234,d-119,T12.34567123,r003669958,S4321,1085889680"
        )

        assert un.nmeaType == "AIVDM"
        assert un.totalSentences == 1
        assert un.sentenceNum == 1
        assert un.sequentialMsgId is None
        assert un.aisChannel == "B"
        assert un.fillbits == 0
        assert un.checksumStr == "63"

        assert un.rssi == 1234
        assert un.signalStrength == -119
        assert un.timeOfArrival == 12.34567123
        assert un.slotNumber == 4321
        assert un.station == "r003669958"
        assert un.stationTypeCode == "r"
        assert un.cg_sec == float(1085889680)
        print(un.timestamp)
        print(un.sqlTimestampStr)  # Hmmm... they look the same

    def testEquality(self):
        m1 = UscgNmea(
            "!AIVDM,1,1,,B,15Cjtd0Oj;Jp7ilG7=UkKBoB0<06,0*63,s1234,d-119,T12.34567123,r003669958,S4321,1085889680"
        )
        m1same = UscgNmea(
            "!AIVDM,1,1,,B,15Cjtd0Oj;Jp7ilG7=UkKBoB0<06,0*63,s1234,d-119,T12.34567123,r003669958,S4321,1085889680"
        )
        # A whole bunch of mangled fields
        m2 = UscgNmea(
            "!AIVDM,2,1,,B,15Cjtd0Oj;Jp7ilG7=UkKBoB0<06,0*63,s1234,d-119,T12.34567123,r003669958,S4321,1085889680"
        )
        m3 = UscgNmea(
            "!AIVDM,1,2,,B,15Cjtd0Oj;Jp7ilG7=UkKBoB0<06,0*63,s1234,d-119,T12.34567123,r003669958,S4321,1085889680"
        )
        m4 = UscgNmea(
            "!AIVDM,1,1,7,B,15Cjtd0Oj;Jp7ilG7=UkKBoB0<06,0*63,s1234,d-119,T12.34567123,r003669958,S4321,1085889680"
        )
        m5 = UscgNmea(
            "!AIVDM,1,1,,A,15Cjtd0Oj;Jp7ilG7=UkKBoB0<06,0*63,s1234,d-119,T12.34567123,r003669958,S4321,1085889680"
        )
        m6 = UscgNmea(
            "!AIVDM,1,1,,B,25Cjtd0Oj;Jp7ilG7=UkKBoB0<06,0*63,s1234,d-119,T12.34567123,r003669958,S4321,1085889680"
        )
        m7 = UscgNmea(
            "!AIVDM,1,1,,B,15Cjtd0Oj;Jp7ilG7=UkKBoB0<06,1*63,s1234,d-119,T12.34567123,r003669958,S4321,1085889680"
        )
        m8 = UscgNmea(
            "!AIVDM,1,1,,B,15Cjtd0Oj;Jp7ilG7=UkKBoB0<06,0*64,s1234,d-119,T12.34567123,r003669958,S4321,1085889680"
        )
        # m9 = UscgNmea('!AIVDM,1,1,,B,15Cjtd0Oj;Jp7ilG7=UkKBoB0<06,0*63,s123,d-119,T12.34567123,r003669958,S4321,1085889680')
        # m10 = UscgNmea('!AIVDM,1,1,,B,15Cjtd0Oj;Jp7ilG7=UkKBoB0<06,0*63,s1234,d-120,T12.34567123,r003669958,S4321,1085889680')
        # m11 = UscgNmea('!AIVDM,1,1,,B,15Cjtd0Oj;Jp7ilG7=UkKBoB0<06,0*63,s1234,d-119,T11.34567123,r003669958,S4321,1085889680')
        m12 = UscgNmea(
            "!AIVDM,1,1,,B,15Cjtd0Oj;Jp7ilG7=UkKBoB0<06,0*63,s1234,d-119,T12.34567123,r003669959,S4321,1085889680"
        )
        # m13 = UscgNmea('!AIVDM,1,1,,B,15Cjtd0Oj;Jp7ilG7=UkKBoB0<06,0*63,s1234,d-119,T12.34567123,r003669958,S432,1085889680')
        m14 = UscgNmea(
            "!AIVDM,1,1,,B,15Cjtd0Oj;Jp7ilG7=UkKBoB0<06,0*63,s1234,d-119,T12.34567123,r003669958,S4321,1085889681"
        )
        assert m1 == m1
        assert m1 == m1same
        assert m1 != m2
        assert m1 != m3
        assert m1 != m4
        assert m1 != m5
        assert m1 != m6
        assert m1 != m7
        assert m1 != m8
        # self.failUnless(m1!=m9)
        # self.failUnless(m1!=m10)
        # self.failUnless(m1!=m11)
        assert m1 != m12
        # self.failUnless(m1!=m13)
        assert m1 != m14


def create_nmea(
    bits,
    nmeaType="!AIVDM",  # Could also use $.
    totalSentences=None,
    sentenceNum=None,
    sequentialMsgId=None,
    aisChannel="A",
    station="runknown",
    cg_sec=None,
):
    """Build a NMEA string for an AIS binary message payload.

    e.g. !AIVDM,1,1,,B,13UIAT001mmL=vhP1Sa:?8>l06A<,0*37,s24467,rNDBC46001,1202235568

    >>> bv=BitVector.from_bitstring('0010000001010000100110000110000011001100010110'
                               '1110111111001100101011000101101101011110011111'
                               '0010001100110011000001110100011001000000000000'
                               '000000000000111001111000000000')
    >>> create_nmea(bv,cg_sec=1202235568)
    'AIVDM,1,1,,A,852HH<iKgk:iKG_j<k1lI0000qp0,0*13,runknown,1202235568'

    @type bits: BitVector
    Args:
        cg_sec: seconds since the epoch in UTC or if None, the current time will be used
        totalSentences: defaults to 1.  Eventually will calculated for multi-line
    @type totalSentences: positive int
        sentences: defaults to 1.  Eventually will calculated for multi-line
    @type sentenceNum: positive int
        sequentialMsgId: the number 1..9 for this group of messages.  Defaults to blank
    @type sequentialMsgId: positive int

    @todo: handle the rest of the uscg fields
    @todo: handle multi-line messages
    """
    bitLen = len(bits)
    # If larger, this needs to be a multi-line set of nmea msgs.
    assert bitLen <= 168

    if totalSentences is not None:
        # FIX: what is the right max number for a 5 slot message?
        assert totalSentences < 5
    else:
        # FIX: for multi-line, calculate this
        totalSentences = 1
    if sentenceNum is not None:
        # FIX: should not be done here for multi-line
        assert sentenceNum < totalSentences
    else:
        sentenceNum = 1

    if sequentialMsgId is not None:
        assert sequentialMsgId < 10
        sequentialMsgId = str(sequentialMsgId)
    else:
        sequentialMsgId = ""

    pad = 6 - (bitLen % 6)
    if pad == 6:
        pad = 0
    if pad:
        # Pad out to multiple of 6
        bits = bits + BitVector(size=(6 - (bitLen % 6)))
    payload = binary.bitvectoais6(bits)[0]

    fields = [
        nmeaType,
    ]
    fields.append(str(totalSentences))
    fields.append(str(sentenceNum))
    fields.append(sequentialMsgId)
    fields.append(aisChannel)
    fields.append(payload)
    fields.append(str(pad))
    firstStr = ",".join(fields)
    checksum = nmea.checksumStr(firstStr)
    fields = [
        firstStr + "*" + checksum,
    ]
    fields.append(station)
    if cg_sec is None:
        cg_sec = time.time()
    fields.append(str(cg_sec))

    return ",".join(fields)


def test():
    print("doctests ...")
    numfail, _ = doctest.testmod()
    if not numfail:
        print("ok")
    else:
        print("FAILED")


if __name__ == "__main__":
    test()