Read and decode device status bytes: mode, rate, compass

CHANGES.rst

@@ -11,6 +11,7 @@ in progress
 - Implement client interface as context manager
 - Increase default timeout values to 15 seconds
 - Rework device info acquisition
+- Read and decode device status bytes: mode, rate, compass
 
 
 2022-xx-xx 0.0.0

calypso_anemometer/cli.py

@@ -1,12 +1,11 @@
-import json
 import logging
 import os
 import sys
 
 import click
 
 from calypso_anemometer.core import CalypsoDeviceApi
-from calypso_anemometer.util import make_sync, setup_logging
+from calypso_anemometer.util import make_sync, setup_logging, to_json
 
 logger = logging.getLogger(__name__)
 
@@ -35,8 +34,10 @@ async def info(ctx):
     try:
         async with CalypsoDeviceApi(ble_address=os.getenv("CALYPSO_ADDRESS")) as calypso:
             device_info = await calypso.get_info()
-            print(json.dumps(device_info.asdict(), indent=2))
     except Exception as ex:
+            print(to_json(device_info))
+            device_status = await calypso.get_status()
+            print(to_json(device_status.aslabeldict()))
         logger.error(ex)
         sys.exit(1)
 

calypso_anemometer/core.py

@@ -16,7 +16,14 @@
 from bleak import BleakClient, BleakError, BleakScanner
 
 from calypso_anemometer.exception import BluetoothAdapterError, BluetoothConversationError, BluetoothDiscoveryError
-from calypso_anemometer.model import BleCharSpec, CalypsoDeviceInfo
+from calypso_anemometer.model import (
+    BleCharSpec,
+    CalypsoDeviceCompassStatus,
+    CalypsoDeviceDataRate,
+    CalypsoDeviceInfo,
+    CalypsoDeviceMode,
+    CalypsoDeviceStatus,
+)
 
 # Configuration section.
 DISCOVERY_TIMEOUT = 15.0
@@ -39,6 +46,11 @@ class CalypsoDeviceApi:
         BleCharSpec(uuid="00002a26-0000-1000-8000-00805f9b34fb", name="firmware_revision"),
         BleCharSpec(uuid="00002a28-0000-1000-8000-00805f9b34fb", name="software_revision"),
     ]
+    DEVICE_STATUS_CHARACTERISTICS = [
+        BleCharSpec(uuid="0000a001-0000-1000-8000-00805f9b34fb", name="mode", decoder=CalypsoDeviceMode),
+        BleCharSpec(uuid="0000a002-0000-1000-8000-00805f9b34fb", name="rate", decoder=CalypsoDeviceDataRate),
+        BleCharSpec(uuid="0000a003-0000-1000-8000-00805f9b34fb", name="compass", decoder=CalypsoDeviceCompassStatus),
+    ]
 
     def __init__(self, ble_address: Optional[str] = None):
         self.ble_address = ble_address
@@ -148,6 +160,16 @@ async def get_info(self) -> CalypsoDeviceInfo:
             data[charspec.name] = value
         return CalypsoDeviceInfo(ble_address=self.ble_address, **data)
 
+    async def get_status(self) -> CalypsoDeviceStatus:
+        logger.info("Getting status information")
+        status = CalypsoDeviceStatus()
+        for charspec in self.DEVICE_STATUS_CHARACTERISTICS:
+            rawvalue: bytearray = await self.client.read_gatt_char(charspec.uuid)
+            value: int = rawvalue[0]
+            outcome = charspec.decoder(value)
+            setattr(status, charspec.name, outcome)
+        return status
+
     async def read_characteristic(self, characteristic_id: str) -> str:
         char = await self.client.read_gatt_char(characteristic_id)
         return char.decode()

calypso_anemometer/model.py

@@ -1,11 +1,13 @@
 import dataclasses
-from typing import Optional
+from enum import Enum, IntEnum
+from typing import Optional, Union
 
 
 @dataclasses.dataclass
 class BleCharSpec:
     name: str
     uuid: str
+    decoder: Optional[Union[Enum]] = None
 
 
 @dataclasses.dataclass
@@ -20,3 +22,34 @@ class CalypsoDeviceInfo:
 
     def asdict(self):
         return dataclasses.asdict(self)
+
+
+class CalypsoDeviceMode(IntEnum):
+    SLEEP_MODE = 0x00
+    LOW_POWER = 0x01
+    NORMAL_MODE = 0x02
+
+
+class CalypsoDeviceDataRate(IntEnum):
+    HZ_1 = 0x01
+    HZ_4 = 0x04
+    HZ_8 = 0x08
+
+
+class CalypsoDeviceCompassStatus(IntEnum):
+    OFF = 0x00
+    ON = 0x01
+
+
+@dataclasses.dataclass
+class CalypsoDeviceStatus:
+    mode: Optional[CalypsoDeviceMode] = None
+    rate: Optional[CalypsoDeviceDataRate] = None
+    compass: Optional[CalypsoDeviceCompassStatus] = None
+
+    def aslabeldict(self):
+        return {
+            "mode": self.mode.name,
+            "rate": self.rate.name,
+            "compass": self.compass.name,
+        }

calypso_anemometer/util.py

@@ -1,5 +1,7 @@
 import asyncio
+import dataclasses
 import functools
+import json
 import logging
 import sys
 
@@ -21,3 +23,28 @@ def wrapper(*args, **kwargs):
         return asyncio.run(func(*args, **kwargs))
 
     return wrapper
+
+
+class JsonEncoderPlus(json.JSONEncoder):
+    """
+    JSON encoder with support for serializing Enums and Data Classes.
+
+    - https://docs.python.org/3/library/json.html#json.JSONEncoder
+    - https://docs.python.org/3/library/enum.html
+    - https://docs.python.org/3/library/dataclasses.html
+    """
+
+    def default(self, obj):
+        if dataclasses.is_dataclass(obj):
+            return dataclasses.asdict(obj)
+        return json.JSONEncoder.default(self, obj)
+
+
+def to_json(obj, pretty=True):
+    """
+    Serialize any object to JSON by using a custom encoder.
+    """
+    kwargs = {}
+    if pretty:
+        kwargs["indent"] = 2
+    return json.dumps(obj, cls=JsonEncoderPlus, **kwargs)