GH1: seatalk datagram refactoring

seatalk/__init__.py

@@ -0,0 +1,43 @@
+from seatalk.seatalk import SeatalkDevice
+
+from seatalk.datagrams.s00_depth import Depth
+from seatalk.datagrams.s01_equipment_id import EquipmentID1
+from seatalk.datagrams.s10_apparent_wind_speed_and_angle import ApparentWindAngle
+from seatalk.datagrams.s11_apparent_wind_speed import ApparentWindSpeed
+from seatalk.datagrams.s20_speed1 import Speed1
+from seatalk.datagrams.s21_trip_mileage import TripMileage
+from seatalk.datagrams.s22_total_mileage import TotalMileage
+from seatalk.datagrams.s23_water_temperature import WaterTemperature1
+from seatalk.datagrams.s24_display_units_mileage_speed import DisplayUnitsMileageSpeed
+from seatalk.datagrams.s25_TotalTripLog import TotalTripLog
+from seatalk.datagrams.s26_speed2 import Speed2
+from seatalk.datagrams.s27_water_temperature2 import WaterTemperature2
+from seatalk.datagrams.s30_set_lamp_intensity1 import SetLampIntensity1
+from seatalk.datagrams.s36_cancel_mob import CancelMOB
+from seatalk.datagrams.s38_code_lock_data import CodeLockData
+from seatalk.datagrams.s50_latitude_position import LatitudePosition
+from seatalk.datagrams.s51_longitude_position import LongitudePosition
+from seatalk.datagrams.s52_speed_over_ground import SpeedOverGround
+from seatalk.datagrams.s53_course_over_ground import CourseOverGround
+from seatalk.datagrams.s54_gmt_time import GMT_Time
+from seatalk.datagrams.s55_key_stroke1 import KeyStroke1
+from seatalk.datagrams.s56_date import Date
+from seatalk.datagrams.s57_sat_info import SatInfo
+from seatalk.datagrams.s58_position import Position
+from seatalk.datagrams.s59_countdown_timer import CountdownTimer
+from seatalk.datagrams.s60_e80_initialization import E80Initialization
+from seatalk.datagrams.s65_select_fathom import SelectFathom
+from seatalk.datagrams.s66_wind_alarm import WindAlarm
+from seatalk.datagrams.s68_alarm_acknowledgement import AlarmAcknowledgement
+from seatalk.datagrams.s6c_equipment_id import EquipmentID2
+from seatalk.datagrams.s6e_man_over_board import ManOverBoard
+from seatalk.datagrams.s80_set_lamp_intensity2 import SetLampIntensity2
+from seatalk.datagrams.s81_course_computer_setup import CourseComputerSetup
+from seatalk.datagrams.s82_target_waypoint_name import TargetWaypointName
+from seatalk.datagrams.s86_key_stroke2 import KeyStroke2
+from seatalk.datagrams.s87_set_response_level import SetResponseLevel
+from seatalk.datagrams.s90_device_identification1 import DeviceIdentification1
+from seatalk.datagrams.s91_set_rudder_gain import SetRudderGain
+from seatalk.datagrams.s93_enter_ap_setup import EnterAPSetup
+from seatalk.datagrams.s99_compass_variation import CompassVariation
+from seatalk.datagrams.sa4_device_identification2 import DeviceIdentification2

seatalk/datagrams/s00_depth.py

@@ -0,0 +1,49 @@
+from common.helper import UnitConverter
+from nmea import nmea_datagram
+from seatalk.datagrams.seatalk_datagram import SeatalkDatagram
+
+
+class Depth(SeatalkDatagram, nmea_datagram.DepthBelowKeel):   # NMEA: dbt
+    """
+    00  02  YZ  XX XX  Depth below transducer: XXXX/10 feet
+                   Flags in Y: Y&8 = 8: Anchor Alarm is active
+                               Y&4 = 4: Metric display units or
+                                          Fathom display units if followed by command 65
+                               Y&2 = 2: Used, unknown meaning
+                   Flags in Z: Z&4 = 4: Transducer defective
+                               Z&2 = 2: Deep Alarm is active
+                               Z&1 = 1: Shallow Depth Alarm is active
+                   Corresponding NMEA sentences: DPT, DBT
+    """
+    seatalk_id = 0x00
+    data_length = 2
+
+    def __init__(self, anchor_alarm_active=None, metric_display_units=None, transducer_defective=None, depth_alarm_active=None, shallow_alarm_active=None, *args, **kwargs):
+        SeatalkDatagram.__init__(self)
+        nmea_datagram.DepthBelowKeel.__init__(self, *args, **kwargs)
+        self.anchor_alarm_active = anchor_alarm_active
+        self.metric_display_units = metric_display_units
+        self.transducer_defective = transducer_defective
+        self.depth_alarm_active = depth_alarm_active
+        self.shallow_alarm_active = shallow_alarm_active
+
+    def process_datagram(self, first_half_byte, data):
+        self.anchor_alarm_active =  (data[0] & 0x80) != 0
+        self.metric_display_units = (data[0] & 0x40) != 0
+        self.transducer_defective = (data[0] & 0x04) != 0
+        self.depth_alarm_active =   (data[0] & 0x02) != 0
+        self.shallow_alarm_active = (data[0] & 0x01) != 0
+
+        feet = self.get_value(data[1:]) / 10.0
+        self.depth_m = UnitConverter.feet_to_meter(feet)
+
+    def get_seatalk_datagram(self):
+        feet_value = UnitConverter.meter_to_feet(self.depth_m) * 10
+        flags = 0
+        flags |= 0x80 if self.anchor_alarm_active  else 0x00
+        flags |= 0x40 if self.metric_display_units else 0x00
+        flags |= 0x04 if self.transducer_defective else 0x00
+        flags |= 0x02 if self.depth_alarm_active   else 0x00
+        flags |= 0x01 if self.shallow_alarm_active else 0x00
+
+        return bytearray([self.seatalk_id, self.data_length, flags]) + self.set_value(feet_value)

seatalk/datagrams/s01_equipment_id.py

@@ -0,0 +1,39 @@
+import enum
+
+from common.helper import TwoWayDict
+from seatalk.datagrams.seatalk_datagram import _TwoWayDictDatagram
+
+
+class EquipmentID1(_TwoWayDictDatagram):
+    """
+    01  05  XX XX XX XX XX XX  Equipment ID, sent at power on, reported examples:
+    01  05  00 00 00 60 01 00  Course Computer 400G
+    01  05  04 BA 20 28 01 00  ST60 Tridata
+    01  05  70 99 10 28 01 00  ST60 Log
+    01  05  F3 18 00 26 0F 06  ST80 Masterview
+    01  05  FA 03 00 30 07 03  ST80 Maxi Display
+    01  05  FF FF FF D0 00 00  Smart Controller Remote Control Handset
+    """
+    seatalk_id = 0x01
+    data_length = 5
+
+    class Equipments(enum.IntEnum):
+        Course_Computer_400G = enum.auto()
+        ST60_Tridata = enum.auto()
+        ST60_Tridata_Plus = enum.auto()
+        ST60_Log = enum.auto()
+        ST80_Masterview = enum.auto()
+        ST80_Maxi_Display = enum.auto()
+        Smart_Controller_Remote_Control_Handset = enum.auto()
+
+    def __init__(self, set_key: Equipments = None):
+        equipment_map = TwoWayDict({
+            bytes([0x00, 0x00, 0x00, 0x60, 0x01, 0x00]): self.Equipments.Course_Computer_400G,
+            bytes([0x04, 0xBA, 0x20, 0x28, 0x01, 0x00]): self.Equipments.ST60_Tridata,
+            bytes([0x87, 0x72, 0x25, 0x28, 0x01, 0x00]): self.Equipments.ST60_Tridata_Plus,
+            bytes([0x70, 0x99, 0x10, 0x28, 0x01, 0x00]): self.Equipments.ST60_Log,
+            bytes([0xF3, 0x18, 0x00, 0x26, 0x0F, 0x06]): self.Equipments.ST80_Masterview,
+            bytes([0xFA, 0x03, 0x00, 0x30, 0x07, 0x03]): self.Equipments.ST80_Maxi_Display,
+            bytes([0xFF, 0xFF, 0xFF, 0xD0, 0x00, 0x00]): self.Equipments.Smart_Controller_Remote_Control_Handset,
+        })
+        _TwoWayDictDatagram.__init__(self, map=equipment_map, set_key=set_key)

seatalk/datagrams/s10_apparent_wind_speed_and_angle.py

@@ -0,0 +1,21 @@
+from seatalk.datagrams.seatalk_datagram import SeatalkDatagram
+
+
+class ApparentWindAngle(SeatalkDatagram):  # TODO nmea mwv with ApparentWindSpeed
+    """
+    10  01  XX  YY  Apparent Wind Angle: XXYY/2 degrees right of bow
+                Used for autopilots Vane Mode (WindTrim)
+                Corresponding NMEA sentence: MWV
+    """
+    seatalk_id = 0x10
+    data_length = 1
+
+    def __init__(self, angle_degree=None):
+        SeatalkDatagram.__init__(self)
+        self.angle_degree = angle_degree
+
+    def process_datagram(self, first_half_byte, data):
+        self.angle_degree = self.get_value(data) / 2  # TODO maybe some validation for <0° or >360° ?
+
+    def get_seatalk_datagram(self):
+        return bytearray([self.seatalk_id, self.data_length]) + self.set_value(int(self.angle_degree * 2))

seatalk/datagrams/s11_apparent_wind_speed.py

@@ -0,0 +1,34 @@
+from common.helper import UnitConverter, byte_to_str
+from seatalk.datagrams.seatalk_datagram import SeatalkDatagram
+from seatalk.seatalk_exceptions import DataValidationException
+
+
+class ApparentWindSpeed(SeatalkDatagram):  # TODO nmea mwv with ApparentWindAngle
+    """
+    11  01  XX  0Y  Apparent Wind Speed: (XX & 0x7F) + Y/10 Knots
+                Units flag: XX&0x80=0    => Display value in Knots
+                            XX&0x80=0x80 => Display value in Meter/Second
+                Corresponding NMEA sentence: MWV
+    """
+    seatalk_id = 0x11
+    data_length = 1
+
+    def __init__(self, speed_knots=None):
+        SeatalkDatagram.__init__(self)
+        self.speed_knots = speed_knots
+
+    def process_datagram(self, first_half_byte, data):
+        if data[1] & 0xF0:  # 0Y <- the 0 is important
+            raise DataValidationException(f"{type(self).__name__}: Byte 1 is bigger than 0x0F {byte_to_str(data[1])}")
+
+        speed = (data[0] & 0x7F) + data[1] / 10
+
+        if data[0] & 0x80:  # Meter/Second
+            self.speed_knots = UnitConverter.meter_to_nm(speed * 60 * 60)
+        else:  # Knots
+            self.speed_knots = speed
+
+    def get_seatalk_datagram(self):
+        x_byte = int(self.speed_knots)
+        y_byte = int((round(self.speed_knots, 1) - x_byte) * 10)
+        return bytearray([self.seatalk_id, self.data_length, x_byte, y_byte])

seatalk/datagrams/s20_speed1.py

@@ -0,0 +1,21 @@
+from nmea import nmea_datagram
+from seatalk.datagrams.seatalk_datagram import SeatalkDatagram
+
+
+class Speed1(SeatalkDatagram, nmea_datagram.SpeedThroughWater):  # NMEA: vhw
+    """
+    20  01  XX  XX  Speed through water: XXXX/10 Knots
+                     Corresponding NMEA sentence: VHW
+    """
+    seatalk_id = 0x20
+    data_length = 1
+
+    def __init__(self, *args, **kwargs):
+        SeatalkDatagram.__init__(self)
+        nmea_datagram.SpeedThroughWater.__init__(self, *args, **kwargs)
+
+    def process_datagram(self, first_half_byte, data):
+        self.speed_knots = self.get_value(data) / 10
+
+    def get_seatalk_datagram(self):
+        return bytearray([self.seatalk_id, self.data_length]) + self.set_value(self.speed_knots * 10)

seatalk/datagrams/s21_trip_mileage.py

@@ -0,0 +1,21 @@
+from seatalk.datagrams.seatalk_datagram import SeatalkDatagram
+
+
+class TripMileage(SeatalkDatagram):
+    """
+    21  02  XX  XX  0X  Trip Mileage: XXXXX/100 nautical miles
+    """
+    seatalk_id = 0x21
+    data_length = 2
+
+    def __init__(self, mileage_miles=None):
+        SeatalkDatagram.__init__(self)
+        self.mileage_miles = mileage_miles
+
+    def process_datagram(self, first_half_byte, data):
+        value = (data[2] & 0x0F) << 16 | data[1] << 8 | data[0]
+        self.mileage_miles = value / 100
+
+    def get_seatalk_datagram(self):
+        data = int(self.mileage_miles * 100).to_bytes(3, "little")
+        return bytearray([self.seatalk_id, self.data_length]) + data

seatalk/datagrams/s22_total_mileage.py

@@ -0,0 +1,19 @@
+from seatalk.datagrams.seatalk_datagram import SeatalkDatagram
+
+
+class TotalMileage(SeatalkDatagram):
+    """
+    22  02  XX  XX  00  Total Mileage: XXXX/10 nautical miles
+    """
+    seatalk_id = 0x22
+    data_length = 2
+
+    def __init__(self, mileage_miles=None):
+        SeatalkDatagram.__init__(self)
+        self.mileage_miles = mileage_miles
+
+    def process_datagram(self, first_half_byte, data):
+        self.mileage_miles = self.get_value(data[:2]) / 10
+
+    def get_seatalk_datagram(self):
+        return bytearray([self.seatalk_id, self.data_length]) + self.set_value(self.mileage_miles * 10) + bytearray([0x00])

seatalk/datagrams/s23_water_temperature.py

@@ -0,0 +1,27 @@
+from common.helper import UnitConverter
+from nmea import nmea_datagram
+from seatalk.datagrams.seatalk_datagram import SeatalkDatagram
+
+
+class WaterTemperature1(SeatalkDatagram, nmea_datagram.WaterTemperature):  # NMEA: mtw
+    """
+    23  Z1  XX  YY  Water temperature (ST50): XX deg Celsius, YY deg Fahrenheit
+                 Flag Z&4: Sensor defective or not connected (Z=4)
+                 Corresponding NMEA sentence: MTW
+    """
+    seatalk_id = 0x23
+    data_length = 1
+
+    def __init__(self, sensor_defective=None, *args, **kwargs):
+        SeatalkDatagram.__init__(self)
+        nmea_datagram.WaterTemperature.__init__(self, *args, **kwargs)
+        self.sensor_defective = sensor_defective
+
+    def process_datagram(self, first_half_byte, data):
+        self.sensor_defective = first_half_byte & 4 == 4
+        self.temperature_c = data[0]
+
+    def get_seatalk_datagram(self):
+        fahrenheit = UnitConverter.celsius_to_fahrenheit(self.temperature_c)
+        first_half_byte = (self.sensor_defective << 6) | self.data_length
+        return bytearray([self.seatalk_id, first_half_byte, int(self.temperature_c), int(fahrenheit)])

seatalk/datagrams/s24_display_units_mileage_speed.py

@@ -0,0 +1,26 @@
+import enum
+
+from common.helper import TwoWayDict
+from seatalk.datagrams.seatalk_datagram import _TwoWayDictDatagram
+
+
+class DisplayUnitsMileageSpeed(_TwoWayDictDatagram):
+    """
+    24  02  00  00  XX  Display units for Mileage & Speed
+                    XX: 00=nm/knots, 06=sm/mph, 86=km/kmh
+    """
+    seatalk_id = 0x24
+    data_length = 2
+
+    class Unit(enum.IntEnum):
+        Knots = enum.auto()
+        Mph = enum.auto()
+        Kph = enum.auto()
+
+    def __init__(self, unit: Unit = None):
+        unit_map = TwoWayDict({
+            bytes([0x00, 0x00, 0x00]): self.Unit.Knots,
+            bytes([0x00, 0x00, 0x06]): self.Unit.Mph,
+            bytes([0x00, 0x00, 0x86]): self.Unit.Kph,
+        })
+        _TwoWayDictDatagram.__init__(self, map=unit_map, set_key=unit)

seatalk/datagrams/s25_TotalTripLog.py

@@ -0,0 +1,51 @@
+from seatalk.datagrams.seatalk_datagram import SeatalkDatagram
+
+
+class TotalTripLog(SeatalkDatagram):
+    """
+    25  Z4  XX  YY  UU  VV AW  Total & Trip Log
+                     total= (XX+YY*256+Z* 4096)/ 10 [max=104857.5] nautical miles
+                     trip = (UU+VV*256+W*65536)/100 [max=10485.75] nautical miles
+
+
+    https://github.com/mariokonrad/marnav/blob/master/src/marnav/seatalk/message_25.cpp
+
+    total= (XX+YY*256+Z*65536)/ 10 [max=104857.5] nautical miles
+    (the factor for Z in the description from Thomas Knauf is wrong)
+
+    (Shifting and other logical operations are faster than division and additions. Maybe some compilers would see that, but this looks way more straight forward and prettier ;-) )
+    """
+    seatalk_id = 0x25
+    data_length = 4
+
+    def __init__(self, total_miles=None, trip_miles=None):
+        SeatalkDatagram.__init__(self)
+        self.total_miles = total_miles
+        self.trip_miles = trip_miles
+
+    def process_datagram(self, first_half_byte, data):
+        # * 256   <=> <<  8
+        # * 4096  <=> << 12
+        # * 65536 <=> << 16
+        total_nibble = first_half_byte
+        trip_nibble = data[4] & 0x0F  # What is the "A" for?
+
+        #                       Z                   YY             XX
+        self.total_miles = (total_nibble << 16 | data[1] << 8 | data[0]) / 10
+
+        #                       W               VV              UU
+        self.trip_miles = (trip_nibble << 16 | data[3] << 8 | data[2]) / 100
+
+    def get_seatalk_datagram(self):
+        raw_total = int(self.total_miles * 10)
+        z = raw_total >> 16
+        xx = raw_total & 0x0000FF
+        yy = (raw_total >> 8) & 0x0000FF
+
+        raw_trip = int(self.trip_miles * 100)
+        aw = raw_trip >> 16
+        uu = raw_trip & 0x0000FF
+        vv = (raw_trip >> 8) & 0x0000FF
+
+        first_byte = (z << 4) | self.data_length
+        return bytearray([self.seatalk_id, first_byte, xx, yy, uu, vv, aw])

seatalk/datagrams/s26_speed2.py

@@ -0,0 +1,27 @@
+from nmea import nmea_datagram
+from seatalk.datagrams.seatalk_datagram import SeatalkDatagram
+
+
+class Speed2(SeatalkDatagram, nmea_datagram.SpeedThroughWater):  # NMEA: vhw
+    """
+    26  04  XX  XX  YY  YY DE  Speed through water:
+                     XXXX/100 Knots, sensor 1, current speed, valid if D&4=4
+                     YYYY/100 Knots, average speed (trip/time) if D&8=0
+                              or data from sensor 2 if D&8=8
+                     E&1=1: Average speed calculation stopped
+                     E&2=2: Display value in MPH
+                     Corresponding NMEA sentence: VHW
+    """
+    seatalk_id = 0x26
+    data_length = 4
+
+    def __init__(self, *args, **kwargs):
+        SeatalkDatagram.__init__(self)
+        nmea_datagram.SpeedThroughWater.__init__(self, *args, **kwargs)
+
+    def process_datagram(self, first_half_byte, data):
+        # TODO Y and E flag
+        self.speed_knots = self.get_value(data) / 100.0
+
+    def get_seatalk_datagram(self):
+        return bytearray([self.seatalk_id, self.data_length]) + self.set_value(self.speed_knots * 100) + bytearray([0x00, 0x00, 0x00])