Building after poetry update

demos/python/sdk_wireless_camera_control/docs/conf.py

@@ -8,7 +8,6 @@
 
 from sphinx.ext.intersphinx import missing_reference
 
-import open_gopro
 import open_gopro.models
 
 gopro = WirelessGoPro(enable_wifi=False)
@@ -41,8 +40,8 @@
 autodoc_default_options = {
     "members": True,
 }
-# https://autodoc-pydantic.readthedocs.io/en/stable/users/installation.html#configuration
-autodoc_pydantic_model_show_json = True
+# https://autodoc-pydantic.readthedocs.io/en/stable/users/configuration.html
+autodoc_pydantic_model_show_json = False
 autodoc_pydantic_settings_show_json = False
 intersphinx_mapping = {
     "python": ("https://docs.python.org/3", None),

demos/python/sdk_wireless_camera_control/open_gopro/api/ble_commands.py

@@ -1082,9 +1082,7 @@ def __init__(self, communicator: GoProBle):
         """File type of photo output"""
 
         self.video_duration: BleSetting[Params.VideoDuration] = BleSetting[Params.VideoDuration](
-            communicator,
-            SettingId.VIDEO_DURATION,
-            Params.VideoDuration
+            communicator, SettingId.VIDEO_DURATION, Params.VideoDuration
         )
         """If set, a video will automatically be stopped after recording for this long."""
 

demos/python/sdk_wireless_camera_control/open_gopro/api/http_commands.py

@@ -711,7 +711,6 @@ def __init__(self, communicator: GoProHttp):
 
         self.hindsight: HttpSetting[Params.Hindsight] = HttpSetting[Params.Hindsight](
             communicator,
-
             SettingId.HINDSIGHT,
         )
         """Hindsight time / disable"""
@@ -735,8 +734,7 @@ def __init__(self, communicator: GoProHttp):
         """File type of photo output"""
 
         self.video_duration: HttpSetting[Params.VideoDuration] = HttpSetting[Params.VideoDuration](
-            communicator,
-            SettingId.VIDEO_DURATION
+            communicator, SettingId.VIDEO_DURATION
         )
         """If set, a video will automatically be stopped after recording for this long."""
 

demos/python/sdk_wireless_camera_control/open_gopro/demos/log_battery.py

@@ -1,33 +1,157 @@
+# log_battery.py/Open GoPro, Version 2.0 (C) Copyright 2021 GoPro, Inc. (http://gopro.com/OpenGoPro).
+# This copyright was auto-generated on Wed, Sep  1, 2021  5:05:45 PM
+
+"""Example to continuously read the battery (with no Wifi connection)"""
+
+import argparse
 import asyncio
+import csv
+import logging
+from dataclasses import dataclass
+from datetime import datetime
 from pathlib import Path
+from typing import Optional
+
+from rich.console import Console
+
+from open_gopro import WirelessGoPro, types
+from open_gopro.constants import StatusId
+from open_gopro.logger import set_stream_logging_level, setup_logging
+from open_gopro.util import add_cli_args_and_parse, ainput
+
+console = Console()
+
+last_percentage = 0
+last_bars = 0
+
+
+@dataclass
+class Sample:
+    """Simple class to store battery samples"""
+
+    index: int
+    percentage: int
+    bars: int
+
+    def __post_init__(self) -> None:
+        self.time = datetime.now()
+
+    def __str__(self) -> str:
+        return f"Index {self.index} @ time {self.time.strftime('%H:%M:%S')} --> bars: {self.bars}, percentage: {self.percentage}"
+
 
-from open_gopro import WirelessGoPro
-from open_gopro.logger import setup_logging
+SAMPLE_INDEX = 0
+SAMPLES: list[Sample] = []
 
 
-async def main() -> None:
-    logger = setup_logging(__name__, Path("demo.log"))
+def dump_results_as_csv(location: Path) -> None:
+    """Write all of the samples to a csv file
 
-    gopro: WirelessGoPro | None = None
-    async with WirelessGoPro(enable_wifi=False) as gopro:
-        logger.critical("Set AP Mode on")
-        await gopro.ble_command.enable_wifi_ap(enable=True)
+    Args:
+        location (Path): File to write to
+    """
+    console.print(f"Dumping results as CSV to {location}")
+    with open(location, mode="w") as f:
+        w = csv.writer(f, delimiter=",", quotechar='"', quoting=csv.QUOTE_MINIMAL)
+        w.writerow(["index", "time", "percentage", "bars"])
+        initial_time = SAMPLES[0].time
+        for s in SAMPLES:
+            w.writerow([s.index, (s.time - initial_time).seconds, s.percentage, s.bars])
 
-        logger.critical("Wait 10 seconds")
-        await asyncio.sleep(10)
 
-        logger.critical("Initiate AP scan")
-        await gopro.ble_command.scan_wifi_networks()
+async def process_battery_notifications(update: types.UpdateType, value: int) -> None:
+    """Handle asynchronous battery update notifications
 
-        logger.critical("Wait 5 minutes before disconnecting")
-        await asyncio.sleep(300)
+    Args:
+        update (types.UpdateType): type of update
+        value (int): value of update
+    """
 
+    global last_percentage
+    global last_bars
+
+    if update == StatusId.INT_BATT_PER:
+        last_percentage = value
+    elif update == StatusId.BATT_LEVEL:
+        last_bars = value
+
+    # Append and print sample
+    global SAMPLE_INDEX
+    SAMPLES.append(Sample(index=SAMPLE_INDEX, percentage=last_percentage, bars=last_bars))
+    console.print(str(SAMPLES[-1]))
+    SAMPLE_INDEX += 1
+
+
+async def main(args: argparse.Namespace) -> None:
+    logger = setup_logging(__name__, args.log)
+
+    gopro: Optional[WirelessGoPro] = None
+    try:
+        async with WirelessGoPro(args.identifier, enable_wifi=False) as gopro:
+            set_stream_logging_level(logging.ERROR)
+
+            async def log_battery() -> None:
+                global SAMPLE_INDEX
+                if args.poll:
+                    with console.status("[bold green]Polling the battery until it dies..."):
+                        while True:
+                            SAMPLES.append(
+                                Sample(
+                                    index=SAMPLE_INDEX,
+                                    percentage=(await gopro.ble_status.int_batt_per.get_value()).data,
+                                    bars=(await gopro.ble_status.batt_level.get_value()).data,
+                                )
+                            )
+                            console.print(str(SAMPLES[-1]))
+                            SAMPLE_INDEX += 1
+                            await asyncio.sleep(args.poll)
+                else:  # Not polling. Set up notifications
+                    global last_bars
+                    global last_percentage
+
+                    console.print("Configuring battery notifications...")
+                    # Enable notifications of the relevant battery statuses. Also store initial values.
+                    last_bars = (
+                        await gopro.ble_status.batt_level.register_value_update(process_battery_notifications)
+                    ).data
+                    last_percentage = (
+                        await gopro.ble_status.int_batt_per.register_value_update(process_battery_notifications)
+                    ).data
+                    # Append initial sample
+                    SAMPLES.append(Sample(index=SAMPLE_INDEX, percentage=last_percentage, bars=last_bars))
+                    SAMPLE_INDEX += 1
+                    console.print(str(SAMPLES[-1]))
+                    console.print("[bold green]Receiving battery notifications until it dies...")
+
+            asyncio.create_task(log_battery())
+            await ainput("[purple]Press enter to exit.", console.print)
+            console.print("Exiting...")
+
+    except KeyboardInterrupt:
+        logger.warning("Received keyboard interrupt. Shutting down...")
+    if SAMPLES:
+        csv_location = Path(args.log.parent) / "battery_results.csv"
+        dump_results_as_csv(csv_location)
     if gopro:
         await gopro.close()
 
 
+def parse_arguments() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Connect to the GoPro via BLE only and continuously read the battery (either by polling or notifications)."
+    )
+    parser.add_argument(
+        "-p",
+        "--poll",
+        type=int,
+        help="Set to poll the battery at a given interval. If not set, battery level will be notified instead. Defaults to notifications.",
+        default=None,
+    )
+    return add_cli_args_and_parse(parser, wifi=False)
+
+
 def entrypoint() -> None:
-    asyncio.run(main())
+    asyncio.run(main(parse_arguments()))
 
 
 if __name__ == "__main__":

demos/python/sdk_wireless_camera_control/open_gopro/demos/video.py

@@ -36,7 +36,7 @@ async def main(args: argparse.Namespace) -> None:
             # assert (await gopro.http_command.set_shutter(shutter=Params.Toggle.ENABLE)).ok
             # await asyncio.sleep(args.record_time)
             # assert (await gopro.http_command.set_shutter(shutter=Params.Toggle.DISABLE)).ok
-            assert (await gopro.http_setting.video_duration.set(Params.VideoDuration.DUR_15_SECONDS))
+            assert await gopro.http_setting.video_duration.set(Params.VideoDuration.DUR_15_SECONDS)
             assert (await gopro.http_command.set_shutter(shutter=Params.Toggle.ENABLE)).ok
 
             # Get the media list after
@@ -58,13 +58,7 @@ async def main(args: argparse.Namespace) -> None:
 
 def parse_arguments() -> argparse.Namespace:
     parser = argparse.ArgumentParser(description="Connect to a GoPro camera, take a video, then download it.")
-    parser.add_argument(
-        "-r",
-        "--record_time",
-        type=float,
-        help="How long to record for",
-        default=2.0
-    )
+    parser.add_argument("-r", "--record_time", type=float, help="How long to record for", default=2.0)
     parser.add_argument(
         "-o",
         "--output",