diff --git a/CHANGELOG.md b/CHANGELOG.md index 52d318f..68358f7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,10 @@ +## DroneCOT 2.2.3 + +- SerialWorker: handle MAVLink **ADSB_VEHICLE** messages (in addition to + OPEN_DRONE_ID_MESSAGE_PACK), so Remote ID receivers that emit ADS-B-style + frames — e.g. the **BlueMark DroneScout Bridge (DS101)** in ADS-B output mode + — are decoded to CoT. Adds a test for the ADSB_VEHICLE -> RID conversion. + ## DroneCOT 2.2.2 - Fix UTC timestamp handling on Python 3.9 and 3.10. diff --git a/src/dronecot/VERSION b/src/dronecot/VERSION index b1b25a5..5859406 100644 --- a/src/dronecot/VERSION +++ b/src/dronecot/VERSION @@ -1 +1 @@ -2.2.2 +2.2.3 diff --git a/src/dronecot/classes.py b/src/dronecot/classes.py index 18d7bc8..33dd6fb 100644 --- a/src/dronecot/classes.py +++ b/src/dronecot/classes.py @@ -87,6 +87,35 @@ def _mavlink_pack_to_parse_payload_schema(self, messages, pack_size) -> dict: parsed = dronecot.odid.message_pack_to_dict(messages, pack_size) return dronecot.rid_normalize.odid_parsed_to_rid_dict(parsed) + @staticmethod + def _adsb_vehicle_to_rid_dict(msg) -> dict: + """Convert a MAVLink ADSB_VEHICLE to a RIDWorker dict. + + The DroneScout Bridge (and similar Remote ID receivers) can emit detected + Remote ID as MAVLink ADSB_VEHICLE instead of OPEN_DRONE_ID_MESSAGE_PACK. + Units: lat/lon 1e-7 deg, altitude mm, hor/ver velocity cm/s, heading cdeg. + Returns {} when there is no position. + """ + lat = getattr(msg, "lat", 0) / 1e7 + lon = getattr(msg, "lon", 0) / 1e7 + if not lat and not lon: + return {} + callsign = getattr(msg, "callsign", "") or "" + if isinstance(callsign, bytes): + callsign = callsign.decode("ascii", "ignore") + callsign = callsign.replace("\x00", "").strip() + icao = int(getattr(msg, "ICAO_address", 0) or 0) + return { + "BasicID": callsign or f"ICAO-{icao:06X}", + "Latitude": lat, + "Longitude": lon, + "AltitudeGeo": getattr(msg, "altitude", 0) / 1000.0, # mm -> m + "SpeedHorizontal": getattr(msg, "hor_velocity", 0) / 100.0, # cm/s -> m/s + "SpeedVertical": getattr(msg, "ver_velocity", 0) / 100.0, + "Direction": getattr(msg, "heading", 0) / 100.0, # cdeg -> deg + "data": {"type": "MAVLink ADSB_VEHICLE"}, + } + async def run(self, _=-1) -> None: """Read MAVLink messages from serial and enqueue decoded ODID payloads.""" self._logger.info("Running SerialWorker") @@ -154,6 +183,10 @@ async def run(self, _=-1) -> None: msg.messages, msg.msg_pack_size ) await self.put_queue(parsed_payload) + elif msg_type == "ADSB_VEHICLE": + rid = self._adsb_vehicle_to_rid_dict(msg) + if rid: + await self.put_queue(rid) elif msg_type == "HEARTBEAT": self._logger.debug( "MAVLink heartbeat received at %s", diff --git a/tests/test_adsb_vehicle.py b/tests/test_adsb_vehicle.py new file mode 100644 index 0000000..3763b70 --- /dev/null +++ b/tests/test_adsb_vehicle.py @@ -0,0 +1,43 @@ +#!/usr/bin/env python3 +# Copyright Sensors & Signals LLC https://www.snstac.com/ +# SPDX-License-Identifier: Apache-2.0 +"""Tests for SerialWorker MAVLink ADSB_VEHICLE -> RID dict (DroneScout Bridge).""" + +import types +import unittest + +from dronecot.classes import SerialWorker + + +def _adsb(**kw): + return types.SimpleNamespace(**kw) + + +class AdsbVehicleTestCase(unittest.TestCase): + def test_full_conversion(self): + rid = SerialWorker._adsb_vehicle_to_rid_dict( + _adsb( + lat=407128000, lon=-740060000, altitude=120500, + hor_velocity=825, ver_velocity=-50, heading=27000, + callsign="DRONE1\x00\x00", ICAO_address=0x123456, + ) + ) + self.assertAlmostEqual(rid["Latitude"], 40.7128, places=6) + self.assertAlmostEqual(rid["Longitude"], -74.0060, places=6) + self.assertEqual(rid["AltitudeGeo"], 120.5) # mm -> m + self.assertEqual(rid["SpeedHorizontal"], 8.25) # cm/s -> m/s + self.assertEqual(rid["Direction"], 270.0) # cdeg -> deg + self.assertEqual(rid["BasicID"], "DRONE1") + + def test_no_callsign_falls_back_to_icao(self): + rid = SerialWorker._adsb_vehicle_to_rid_dict( + _adsb(lat=407128000, lon=-740060000, callsign="", ICAO_address=0x123456) + ) + self.assertEqual(rid["BasicID"], "ICAO-123456") + + def test_no_position_dropped(self): + self.assertEqual(SerialWorker._adsb_vehicle_to_rid_dict(_adsb(lat=0, lon=0)), {}) + + +if __name__ == "__main__": + unittest.main()