motian_mavbridge/tests/test_aircraft.py

from types import SimpleNamespace
import unittest

from motian_bridge import Config
from motian_bridge.aircraft import ParamEntry, ParamTable
from motian_bridge.dialects.v20 import VKFly
from motian_bridge.telemetry import Telemetry


def cfg(**kwargs):
    fields = dict(system_id="UAV000001", uav_type=7, serial_port="loop://",
                  serial_baudrate=57600, mqtt_host="localhost")
    fields.update(kwargs)
    return Config(**fields)


def sender(system=1, component=1):
    return VKFly.MAVLink(None, srcSystem=system, srcComponent=component)


def heartbeat(mav, armed=False, mode=4):
    return mav.heartbeat_encode(
        VKFly.MAV_TYPE_QUADROTOR, VKFly.MAV_AUTOPILOT_GENERIC,
        VKFly.MAV_MODE_FLAG_SAFETY_ARMED if armed else 0,
        mode, VKFly.MAV_STATE_ACTIVE).pack(mav)


def param_value_frame(mav, param_id, value, index=0, count=1,
                      type_=VKFly.MAV_PARAM_TYPE_REAL32):
    return mav.param_value_encode(param_id, value, type_, count, index).pack(mav)


class AircraftTests(unittest.TestCase):
    def test_aircraft_created_from_heartbeat(self):
        telemetry = Telemetry(cfg())
        self.assertIsNone(telemetry.aircraft)
        self.assertIsNone(telemetry.source)
        telemetry.feed(heartbeat(sender(4, 1)), now=0)
        aircraft = telemetry.aircraft
        self.assertIsNotNone(aircraft)
        self.assertEqual(aircraft.source, (4, 1))
        self.assertEqual(telemetry.source, (4, 1))
        self.assertEqual(aircraft.uav_type, VKFly.MAV_TYPE_QUADROTOR)

    def test_param_table_accumulates_and_completes(self):
        telemetry = Telemetry(cfg())
        mav = sender()
        telemetry.feed(heartbeat(mav), now=0)
        table = telemetry.aircraft.params
        self.assertFalse(table.complete())
        telemetry.feed(param_value_frame(mav, b"FS_CONF_A", 4660, index=0, count=2), now=0)
        self.assertFalse(table.complete())
        telemetry.feed(param_value_frame(mav, b"ALT_LIM_UP1", 120, index=1, count=2), now=0)
        self.assertTrue(table.complete())
        self.assertEqual(table.get("FS_CONF_A"), 4660)
        self.assertEqual(table.get("ALT_LIM_UP1"), 120)
        self.assertEqual(telemetry.param_value("FS_CONF_A"), 4660)
        self.assertIsNone(telemetry.param_value("MISSING"))
        self.assertIn("FS_CONF_A", table)

    def test_duplicate_index_and_unknown_index_do_not_complete(self):
        table = ParamTable()
        known = SimpleNamespace(param_id=b"P", param_value=1.0, param_type=9,
                                param_count=2, param_index=0)
        table.update(known)
        table.update(known)
        self.assertEqual(len(table.received), 1)
        self.assertEqual(len(table), 1)
        unknown = SimpleNamespace(param_id=b"Q", param_value=2.0, param_type=9,
                                  param_count=2, param_index=0xFFFF)
        table.update(unknown)
        self.assertEqual(table.received, {0})
        self.assertFalse(table.complete())

    def test_param_table_setitem_accepts_values_and_entries(self):
        table = ParamTable()
        table["X"] = 5.0
        table["Y"] = ParamEntry(3.0, VKFly.MAV_PARAM_TYPE_UINT8, 1)
        self.assertEqual(table.get("X"), 5.0)
        self.assertEqual(table.get("Y"), 3.0)
        self.assertEqual(table.entries["Y"].param_type, VKFly.MAV_PARAM_TYPE_UINT8)


if __name__ == "__main__":
    unittest.main()