Support for Muse Athena

MuseLSL2/__main__.py

@@ -1,8 +1,6 @@
 import argparse
 import sys
 
-from .cli import CLI
-
 
 def main():
     parser = argparse.ArgumentParser(
@@ -18,6 +16,8 @@ def main():
     # exclude the rest of the args too, or validation will fail
     args = parser.parse_args(sys.argv[1:2])
 
+    from .cli import CLI
+
     if not hasattr(CLI, args.command):
         print("Incorrect usage. See help below.")
         parser.print_help()

MuseLSL2/backends.py

@@ -74,9 +74,29 @@ def char_write_handle(
             )
         )
 
+    def char_write_uuid(self, uuid, value, wait_for_response=True, timeout=30):
+        _wait(self._client.write_gatt_char(uuid, bytearray(value), wait_for_response))
+
     def subscribe(self, uuid, callback=None, indication=False, wait_for_response=True):
         def wrap(gatt_characteristic, data):
             value_handle = gatt_characteristic.handle + 1
             callback(value_handle, data)
 
         _wait(self._client.start_notify(uuid, wrap))
+
+    def has_characteristic(self, uuid: str) -> bool:
+        services = self._client.services
+        if services is None:
+            # Ensure services are resolved
+            services = _wait(self._client.get_services())
+        for service in services:
+            for char in service.characteristics:
+                if str(char.uuid).lower() == uuid.lower():
+                    return True
+        return False
+
+    def get_services(self):
+        services = self._client.services
+        if services is None:
+            services = _wait(self._client.get_services())
+        return services

MuseLSL2/cli.py

@@ -14,7 +14,9 @@ def find(self):
         find_devices(max_duration=10, verbose=True)
 
     def stream(self):
-        parser = argparse.ArgumentParser(description="Start an LSL stream from Muse headset.")
+        parser = argparse.ArgumentParser(
+            description="Start an LSL stream from Muse headset."
+        )
         parser.add_argument(
             "-a",
             "--address",
@@ -51,8 +53,8 @@ def stream(self):
             type=str,
             help="Select preset which dictates data channels to be streamed. Default is p50, but can also be 'none'",
         )
-
         args = parser.parse_args(sys.argv[2:])
+
         from .stream import stream
 
         stream(args.address, args.ppg, args.acc, args.gyro, args.preset)
@@ -61,3 +63,25 @@ def view(self):
         from .view import view
 
         view()
+
+    def inspect(self):
+        parser = argparse.ArgumentParser(
+            description="Inspect BLE GATT services/characteristics for a device."
+        )
+        parser.add_argument("--address", "-a", required=True, help="Device MAC address")
+        args = parser.parse_args(sys.argv[2:])
+        from .backends import BleakBackend
+
+        adapter = BleakBackend()
+        dev = adapter.connect(args.address)
+        try:
+            services = dev.get_services()
+            for svc in services:
+                print(f"Service {svc.uuid} ({getattr(svc, 'description', '')})")
+                for ch in svc.characteristics:
+                    props = ",".join(sorted(getattr(ch, "properties", [])))
+                    print(
+                        f"  Char {ch.uuid} handle={ch.handle} props=[{props}] desc={getattr(ch, 'description', '')}"
+                    )
+        finally:
+            dev.disconnect()

MuseLSL2/find.py

@@ -5,7 +5,8 @@ def find_devices(max_duration=10, verbose=True):
     adapter = BleakBackend()
 
     adapter.start()
-    print(f"Searching for Muses (max. {max_duration} seconds)...")
+    if verbose:
+        print(f"Searching for Muses (max. {max_duration} seconds)...")
     devices = adapter.scan(timeout=max_duration)  # Muse scan timeout
     adapter.stop()
     muses = [d for d in devices if d["name"] and "Muse" in d["name"]]