Compass: Add --compass=on option to enable compass

Author: amotl

CHANGES.rst

@@ -9,6 +9,7 @@ in progress
 - SignalK telemetry: Use Radian (rad) unit for angle values
 - SignalK telemetry: Use Kelvin unit for temperature
 - SignalK telemetry: Use ratio (percentage) for battery level
+- Compass: Add ``--compass=on`` option to enable compass
 
 
 2022-08-03 0.5.1

README.rst

@@ -113,6 +113,9 @@ default, the Bluetooth adapter ``hci0`` will be used.
     # Get device reading.
     calypso-anemometer read
 
+    # Get device reading, with compass (roll, pitch, heading).
+    calypso-anemometer read --compass=on
+
     # Get device readings, continuously at 4 Hz (default).
     calypso-anemometer read --subscribe
 

calypso_anemometer/cli.py

@@ -8,7 +8,7 @@
 
 from calypso_anemometer.core import CalypsoDeviceApi
 from calypso_anemometer.engine import handler_factory, run_engine
-from calypso_anemometer.model import CalypsoDeviceDataRate, CalypsoDeviceMode, Settings
+from calypso_anemometer.model import CalypsoDeviceCompassStatus, CalypsoDeviceDataRate, CalypsoDeviceMode, Settings
 from calypso_anemometer.util import EnumChoice, make_sync, setup_logging
 
 logger = logging.getLogger(__name__)
@@ -66,6 +66,12 @@ def cli(ctx, quiet: t.Optional[bool], verbose: t.Optional[bool], debug: t.Option
     required=False,
     help="Set device data rate to one of HZ_1, HZ_4, or HZ_8.",
 )
+compass_option = click.option(
+    "--compass",
+    type=EnumChoice(CalypsoDeviceCompassStatus, case_sensitive=False),
+    required=False,
+    help="Set device compass status to one of OFF, ON.",
+)
 subscribe_option = click.option("--subscribe", is_flag=True, required=False, help="Continuously receive readings")
 target_option = click.option("--target", type=str, required=False, help="Submit telemetry data to target")
 
@@ -128,6 +134,7 @@ async def explore(
     help="Set device mode to one of SLEEP, LOW_POWER, or NORMAL.",
 )
 @rate_option
+@compass_option
 @click.pass_context
 @make_sync
 async def set_option(
@@ -138,6 +145,7 @@ async def set_option(
     ble_connect_timeout: t.Optional[float] = None,
     mode: t.Optional[CalypsoDeviceMode] = None,
     rate: t.Optional[CalypsoDeviceDataRate] = None,
+    compass: t.Optional[CalypsoDeviceCompassStatus] = None,
 ):
     async def handler(calypso: CalypsoDeviceApi):
         if mode is not None:
@@ -146,6 +154,9 @@ async def handler(calypso: CalypsoDeviceApi):
         if rate is not None:
             logger.info(f"Setting device data rate to {rate}")
             await calypso.set_datarate(rate)
+        if compass is not None:
+            logger.info(f"Setting device compass status to {compass}")
+            await calypso.set_compass(compass)
         await calypso.about()
 
     settings = Settings(
@@ -165,6 +176,7 @@ async def handler(calypso: CalypsoDeviceApi):
 @subscribe_option
 @target_option
 @rate_option
+@compass_option
 @click.pass_context
 @make_sync
 async def read(
@@ -176,6 +188,7 @@ async def read(
     subscribe: t.Optional[bool] = False,
     target: t.Optional[str] = None,
     rate: t.Optional[CalypsoDeviceDataRate] = None,
+    compass: t.Optional[CalypsoDeviceCompassStatus] = None,
 ):
     quiet = ctx.parent.params.get("quiet")
     settings = Settings(
@@ -184,26 +197,28 @@ async def read(
         ble_discovery_timeout=ble_discovery_timeout,
         ble_connect_timeout=ble_connect_timeout,
     )
-    handler = await handler_factory(subscribe=subscribe, target=target, rate=rate, quiet=quiet)
+    handler = await handler_factory(subscribe=subscribe, target=target, rate=rate, compass=compass, quiet=quiet)
     await run_engine(workhorse=CalypsoDeviceApi, settings=settings, handler=handler)
 
 
 @click.command()
 @subscribe_option
 @target_option
 @rate_option
+@compass_option
 @click.pass_context
 @make_sync
 async def fake(
     ctx,
     subscribe: bool = False,
     target: t.Optional[str] = None,
     rate: t.Optional[CalypsoDeviceDataRate] = None,
+    compass: t.Optional[CalypsoDeviceCompassStatus] = None,
 ):
     from calypso_anemometer.fake import CalypsoDeviceApiFake
 
     quiet = ctx.parent.params.get("quiet")
-    handler = await handler_factory(subscribe=subscribe, target=target, rate=rate, quiet=quiet)
+    handler = await handler_factory(subscribe=subscribe, target=target, rate=rate, compass=compass, quiet=quiet)
     await run_engine(workhorse=CalypsoDeviceApiFake, handler=handler)
 
 

calypso_anemometer/core.py

@@ -27,6 +27,7 @@
 )
 from calypso_anemometer.model import (
     BleCharSpec,
+    CalypsoDeviceCompassStatus,
     CalypsoDeviceDataRate,
     CalypsoDeviceInfo,
     CalypsoDeviceInfoCharacteristic,
@@ -212,6 +213,14 @@ async def set_datarate(self, rate: CalypsoDeviceDataRate):
             CalypsoDeviceStatusCharacteristic.rate.value.uuid, data=bytes([rate.value]), response=True
         )
 
+    async def set_compass(self, compass: CalypsoDeviceCompassStatus):
+        if compass is None:
+            return
+        logger.info(f"Setting compass status to {compass}")
+        await self.client.write_gatt_char(
+            CalypsoDeviceStatusCharacteristic.compass.value.uuid, data=bytes([compass.value]), response=True
+        )
+
     async def get_reading(self):
         logger.info("Requesting reading")
         data: bytearray = await self.client.read_gatt_char(CalypsoDeviceReadingCharacteristic.data.value.uuid)

calypso_anemometer/engine.py

@@ -7,7 +7,7 @@
 
 from calypso_anemometer.core import CalypsoDeviceApi
 from calypso_anemometer.exception import CalypsoError
-from calypso_anemometer.model import CalypsoDeviceDataRate, CalypsoReading, Settings
+from calypso_anemometer.model import CalypsoDeviceCompassStatus, CalypsoDeviceDataRate, CalypsoReading, Settings
 from calypso_anemometer.telemetry.adapter import TelemetryAdapter
 from calypso_anemometer.util import wait_forever
 
@@ -32,6 +32,7 @@ async def handler_factory(
     subscribe: bool = False,
     target: t.Optional[str] = None,
     rate: t.Optional[CalypsoDeviceDataRate] = None,
+    compass: t.Optional[CalypsoDeviceCompassStatus] = None,
     quiet: bool = False,
 ) -> t.Callable:
     """
@@ -40,6 +41,7 @@ async def handler_factory(
     :param subscribe: Whether to run in one-shot or continuous mode.
     :param target: Where to submit telemetry data to, and how.
     :param rate: At which rate to sample the readings.
+    :param compass: If the compass should be enabled or not.
     :param quiet: Do not print to stdout or stderr.
 
     :return: An asynchronous handler function accepting a reference to a workhorse instance.
@@ -66,6 +68,9 @@ def process_reading(reading: CalypsoReading):
 
     # Main handler, which receives readings.
     async def handler(calypso: CalypsoDeviceApi):
+        # Optionally enable compass.
+        await calypso.set_compass(compass)
+
         # One-shot reading.
         if not subscribe:
             reading = await calypso.get_reading()

calypso_anemometer/fake.py

@@ -8,7 +8,7 @@
 
 import aiorate
 
-from calypso_anemometer.model import CalypsoDeviceDataRate, CalypsoReading, Settings
+from calypso_anemometer.model import CalypsoDeviceCompassStatus, CalypsoDeviceDataRate, CalypsoReading, Settings
 
 logger = logging.getLogger(__name__)
 
@@ -43,6 +43,7 @@ def __init__(self, settings: Optional[Settings] = None, ble_address: Optional[st
         self.settings = settings
         self.ble_address = settings.ble_address
         self.datarate: CalypsoDeviceDataRate = CalypsoDeviceDataRate.HZ_4
+        self.compass: CalypsoDeviceCompassStatus = CalypsoDeviceCompassStatus.OFF
         self.reading: Optional[CalypsoReading] = None
 
     async def __aenter__(self):
@@ -67,6 +68,9 @@ async def disconnect(self):
     async def set_datarate(self, rate: CalypsoDeviceDataRate):
         self.datarate = rate
 
+    async def set_compass(self, compass: CalypsoDeviceCompassStatus):
+        self.compass = compass
+
     async def get_reading(self):
         logger.info("Producing reading")
         return await self.produce_fake_reading()

testing/test_cli.py

@@ -147,13 +147,13 @@ def test_cli_read_stdout_success(caplog):
 @mock.patch("calypso_anemometer.core.BleakClient.write_gatt_char", AsyncMock(return_value=None))
 @mock.patch("calypso_anemometer.core.BleakClient.start_notify", AsyncMock(return_value=None))
 @mock.patch("calypso_anemometer.engine.wait_forever", AsyncMock(return_value=None))
-def test_cli_subscribe_stdout_success(caplog):
+def test_cli_subscribe_rate_compass_stdout_success(caplog):
     """
     Test successful `calypso-anemometer read --subscribe`
     """
 
     runner = CliRunner()
-    result = runner.invoke(cli, ["read", "--subscribe", "--rate=HZ_8"], catch_exceptions=False)
+    result = runner.invoke(cli, ["read", "--subscribe", "--rate=HZ_8", "--compass=on"], catch_exceptions=False)
     assert result.exit_code == 0
 
     # TODO: Currently no reading is emitted and processed, because `start_notify` is mocked
@@ -164,6 +164,7 @@ def test_cli_subscribe_stdout_success(caplog):
     assert "Found device at address: bar: foo" in caplog.messages
     assert "Connecting to device at 'bar' with adapter 'hci0'" in caplog.messages
     assert "Setting data rate to 8" in caplog.messages
+    assert "Setting compass status to 1" in caplog.messages
     assert "Subscribing to readings" in caplog.messages
     assert "Disconnecting" in caplog.messages
 

testing/test_core_write.py

@@ -13,7 +13,7 @@
 
 import calypso_anemometer
 from calypso_anemometer.core import CalypsoDeviceApi
-from calypso_anemometer.model import CalypsoDeviceDataRate
+from calypso_anemometer.model import CalypsoDeviceCompassStatus, CalypsoDeviceDataRate
 
 
 @pytest.mark.asyncio
@@ -30,3 +30,35 @@ async def test_set_datarate_success(mocker: MockerFixture, caplog):
     assert "Connecting to device at 'bar' with adapter 'hci0'" in caplog.messages
     assert "Setting data rate to 8" in caplog.messages
     assert "Disconnecting" in caplog.messages
+
+
+@pytest.mark.asyncio
+async def test_set_compass_enabled(mocker: MockerFixture, caplog):
+    mocker.patch("calypso_anemometer.core.BleakClient.connect", AsyncMock(return_value=None))
+    mocker.patch("calypso_anemometer.core.BleakClient.write_gatt_char", AsyncMock(return_value=None))
+
+    spy = mocker.spy(calypso_anemometer.core.BleakClient, "write_gatt_char")
+    async with CalypsoDeviceApi(ble_address="bar") as calypso:
+        await calypso.set_compass(CalypsoDeviceCompassStatus.ON)
+
+    assert spy.mock_calls == [call("0000a003-0000-1000-8000-00805f9b34fb", data=b"\x01", response=True)]
+
+    assert "Connecting to device at 'bar' with adapter 'hci0'" in caplog.messages
+    assert "Setting compass status to 1" in caplog.messages
+    assert "Disconnecting" in caplog.messages
+
+
+@pytest.mark.asyncio
+async def test_set_compass_disabled(mocker: MockerFixture, caplog):
+    mocker.patch("calypso_anemometer.core.BleakClient.connect", AsyncMock(return_value=None))
+    mocker.patch("calypso_anemometer.core.BleakClient.write_gatt_char", AsyncMock(return_value=None))
+
+    spy = mocker.spy(calypso_anemometer.core.BleakClient, "write_gatt_char")
+    async with CalypsoDeviceApi(ble_address="bar") as calypso:
+        await calypso.set_compass(CalypsoDeviceCompassStatus.OFF)
+
+    assert spy.mock_calls == [call("0000a003-0000-1000-8000-00805f9b34fb", data=b"\x00", response=True)]
+
+    assert "Connecting to device at 'bar' with adapter 'hci0'" in caplog.messages
+    assert "Setting compass status to 0" in caplog.messages
+    assert "Disconnecting" in caplog.messages

testing/test_fake.py

@@ -6,7 +6,7 @@
 import pytest
 
 from calypso_anemometer.fake import MAXIMUM_VALUES, MINIMUM_VALUES, CalypsoDeviceApiFake
-from calypso_anemometer.model import CalypsoDeviceDataRate, CalypsoReading
+from calypso_anemometer.model import CalypsoDeviceCompassStatus, CalypsoDeviceDataRate, CalypsoReading
 
 
 @pytest.mark.asyncio
@@ -26,6 +26,7 @@ async def test_set_datarate():
 async def test_subscribe_once():
     callback_mock = Mock()
     async with CalypsoDeviceApiFake() as fake:
+        await fake.set_compass(CalypsoDeviceCompassStatus.ON)
         await fake.set_datarate(CalypsoDeviceDataRate.HZ_8)
         await fake.subscribe_reading(callback=callback_mock, run_once=True)