COHN + Hero 12 v1.30 support (#193)

demos/python/sdk_wireless_camera_control/docs/api.rst

@@ -81,6 +81,8 @@ GoPro Enum
 
 .. autoclass:: open_gopro.enum.GoProEnum
 
+.. autoclass:: open_gopro.enum.GoProIntEnum
+
 BLE Setting
 ^^^^^^^^^^^
 

demos/python/sdk_wireless_camera_control/docs/changelog.rst

@@ -9,6 +9,12 @@ All notable changes to this project will be documented in this file.
 The format is based on `Keep a Changelog <https://keepachangelog.com/en/1.0.0/>`_,
 and this project adheres to `Semantic Versioning <https://semver.org/spec/v2.0.0.html>`_.
 
+Unreleased
+----------
+* Add alpha support for COHN (Camera-on-the-Home-Network)
+    * A real implementation is going to require a major rearchitecture to dynamically add connection types.
+* Remove TKinter GUI. Will be replaced with Textual TUI in the future
+
 0.14.1 (September-21-2022)
 --------------------------
 * Fix BLE notifications not being routed correctly

demos/python/sdk_wireless_camera_control/docs/quickstart.rst

@@ -227,35 +227,3 @@ For more information, do:
     --identifier IDENTIFIER
                             Last 4 digits of GoPro serial number, which is the last 4 digits of the default camera SSID. If not used, first
                             discovered GoPro will be connected to
-
-API GUI Demo
--------------
-
-.. warning::
-    This is a work in progress and some complex responses are not yet easily viewed.
-
-This is a GUI which allows the user to connect a camera and send any command, view status / setting
-updates, view a video stream, and log sent / received messages. It can be started with:
-
-.. code-block:: console
-
-    $ gopro-gui
-
-This will launch a camera chooser screen where the user can either manually enter a camera to connect to
-or automatically connect to the first found camera. Once connected, the GUI will appear. Usages is as follows:
-
-- Choose a command from the Command Pallette on the left
-
-  - Note that besides supporting all of the commands from the Open GoPro API, there is also a "Compound" commands
-    section which contains commands that combine API functionality. One of these, for example, is Livestream
-    which will connect Wifi, configure and start livestreaming.
-- Once chosen, enter the desired parameters in the entry form at the top middle
-- In the same entry form, click the button to send the command
-- The sent command and received response will be logged in the log in the bottom middle as well as any
-  asynchronously received messages.
-- Any log messages with a down arrow can be expanded to view their details
-- Any received statuses, settings, and setting capabilities will be updated in the pane at the top right.
-
-  - The most recently received updates will be highlighted in blue
-- A network stream can be started using the video pane in the bottom right. This will automatically get started
-  after sending the Livestream command

demos/python/sdk_wireless_camera_control/noxfile.py

@@ -56,7 +56,7 @@ def tests(session) -> None:
         "coverage[toml]",
         "requests-mock",
     )
-    session.run("pytest", "tests/unit", "--cov-fail-under=65")
+    session.run("pytest", "tests", "--cov-fail-under=65")
 
 
 @session(python=SUPPORTED_VERSIONS[-1])

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

@@ -161,7 +161,7 @@ async def enable_wifi_ap(self, *, enable: bool) -> GoProResp[None]:
         """
 
     @ble_write_command(GoProUUIDs.CQ_COMMAND, CmdId.LOAD_PRESET_GROUP, Int16ub)
-    async def load_preset_group(self, *, group: proto.EnumPresetGroup) -> GoProResp[None]:
+    async def load_preset_group(self, *, group: proto.EnumPresetGroup.ValueType) -> GoProResp[None]:
         """Load a Preset Group.
 
         Once complete, the most recently used preset in this group will be active.
@@ -453,7 +453,9 @@ async def unregister_for_all_capabilities(self, callback: types.UpdateCb) -> GoP
         request_proto=proto.RequestSetCameraControlStatus,
         response_proto=proto.ResponseGeneric,
     )
-    async def set_camera_control(self, *, camera_control_status: proto.EnumCameraControlStatus) -> GoProResp[None]:
+    async def set_camera_control(
+        self, *, camera_control_status: proto.EnumCameraControlStatus.ValueType
+    ) -> GoProResp[None]:
         """Tell the camera that the app (i.e. External Control) wishes to claim control of the camera.
 
         Args:
@@ -494,8 +496,8 @@ async def set_turbo_mode(self, *, mode: Params.Toggle) -> GoProResp[None]:
     async def get_preset_status(
         self,
         *,
-        register: list[proto.EnumRegisterPresetStatus] | None = None,
-        unregister: list[proto.EnumRegisterPresetStatus] | None = None,
+        register: list[proto.EnumRegisterPresetStatus.ValueType] | None = None,
+        unregister: list[proto.EnumRegisterPresetStatus.ValueType] | None = None,
     ) -> GoProResp[proto.NotifyPresetStatus]:
         """Get information about what Preset Groups and Presets the camera supports in its current state
 
@@ -609,11 +611,11 @@ async def set_livestream_mode(
         self,
         *,
         url: str,
-        window_size: proto.EnumWindowSize,
+        window_size: proto.EnumWindowSize.ValueType,
         minimum_bitrate: int,
         maximum_bitrate: int,
         starting_bitrate: int,
-        lens: proto.EnumLens,
+        lens: proto.EnumLens.ValueType,
         certs: list[Path] | None = None,
     ) -> GoProResp[None]:
         """Initiate livestream to any site that accepts an RTMP URL and simultaneously encode to camera.
@@ -661,8 +663,8 @@ async def set_livestream_mode(
     async def register_livestream_status(
         self,
         *,
-        register: list[proto.EnumRegisterLiveStreamStatus] | None = None,
-        unregister: list[proto.EnumRegisterLiveStreamStatus] | None = None,
+        register: list[proto.EnumRegisterLiveStreamStatus.ValueType] | None = None,
+        unregister: list[proto.EnumRegisterLiveStreamStatus.ValueType] | None = None,
     ) -> GoProResp[proto.NotifyLiveStreamStatus]:
         """Register / unregister to receive asynchronous livestream statuses
 
@@ -677,6 +679,108 @@ async def register_livestream_status(
         """
         return {"register_live_stream_status": register or [], "unregister_live_stream_status": unregister or []}  # type: ignore
 
+    @ble_proto_command(
+        uuid=GoProUUIDs.CQ_COMMAND,
+        feature_id=FeatureId.COMMAND,
+        action_id=ActionId.RELEASE_NETWORK,
+        response_action_id=ActionId.RELEASE_NETWORK_RSP,
+        request_proto=proto.RequestReleaseNetwork,
+        response_proto=proto.ResponseGeneric,
+    )
+    async def release_network(self) -> GoProResp[None]:
+        """Disconnect the camera Wifi network in STA mode so that it returns to AP mode.
+
+        Returns:
+            GoProResp: status of release request
+        """
+
+    @ble_proto_command(
+        uuid=GoProUUIDs.CQ_QUERY,
+        feature_id=FeatureId.QUERY,
+        action_id=ActionId.REQUEST_GET_COHN_STATUS,
+        response_action_id=ActionId.RESPONSE_GET_COHN_STATUS,
+        request_proto=proto.RequestGetCOHNStatus,
+        response_proto=proto.NotifyCOHNStatus,
+    )
+    async def cohn_get_status(self, *, register: bool) -> GoProResp[proto.NotifyCOHNStatus]:
+        """Get (and optionally register for) the current COHN status
+
+        Args:
+            register (bool): whether or not to register
+
+        Returns:
+            GoProResp[proto.NotifyCOHNStatus]: current COHN status
+        """
+        return {"register_cohn_status": int(register)}  # type: ignore
+
+    @ble_proto_command(
+        uuid=GoProUUIDs.CQ_COMMAND,
+        feature_id=FeatureId.COMMAND,
+        action_id=ActionId.REQUEST_CREATE_COHN_CERT,
+        response_action_id=ActionId.RESPONSE_CREATE_COHN_CERT,
+        request_proto=proto.RequestCreateCOHNCert,
+        response_proto=proto.ResponseGeneric,
+    )
+    async def cohn_create_certificate(self, *, override: bool = False) -> GoProResp[None]:
+        """Create an SSL certificate on the camera to use for COHN
+
+        Args:
+            override (bool): Should the current cert be overwritten?. Defaults to True.
+
+        Returns:
+            GoProResp[None]: certificate creation status
+        """
+        return {"override": int(override)}  # type: ignore
+
+    @ble_proto_command(
+        uuid=GoProUUIDs.CQ_COMMAND,
+        feature_id=FeatureId.COMMAND,
+        action_id=ActionId.REQUEST_CLEAR_COHN_CERT,
+        response_action_id=ActionId.RESPONSE_CLEAR_COHN_CERT,
+        request_proto=proto.RequestClearCOHNCert,
+        response_proto=proto.ResponseGeneric,
+    )
+    async def cohn_clear_certificate(self) -> GoProResp[None]:
+        """Clear the current SSL certificate on the camera that is used for COHN
+
+        Returns:
+            GoProResp[None]: was the clear successful?
+        """
+
+    @ble_proto_command(
+        uuid=GoProUUIDs.CQ_QUERY,
+        feature_id=FeatureId.QUERY,
+        action_id=ActionId.REQUEST_GET_COHN_CERT,
+        response_action_id=ActionId.RESPONSE_GET_COHN_CERT,
+        request_proto=proto.RequestCOHNCert,
+        response_proto=proto.ResponseCOHNCert,
+    )
+    async def cohn_get_certificate(self) -> GoProResp[proto.ResponseCOHNCert]:
+        """Get the current SSL certificate that the camera is using for COHN.
+
+        Returns:
+            GoProResp[proto.ResponseCOHNCert]: the certificate
+        """
+
+    @ble_proto_command(
+        uuid=GoProUUIDs.CQ_COMMAND,
+        feature_id=FeatureId.COMMAND,
+        action_id=ActionId.REQUEST_COHN_SETTING,
+        response_action_id=ActionId.RESPONSE_COHN_SETTING,
+        request_proto=proto.RequestSetCOHNSetting,
+        response_proto=proto.ResponseGeneric,
+    )
+    async def cohn_set_setting(self, *, mode: Params.Toggle) -> GoProResp[None]:
+        """Set a COHN specific setting.
+
+        Args:
+            mode (open_gopro.api.params.Toggle): should camera auto connect to home network?
+
+        Returns:
+            GoProResp[None]: status of set
+        """
+        return {"cohn_active": mode}  # type: ignore
+
 
 class BleSettings(BleMessages[BleSetting, SettingId]):
     # pylint: disable=missing-class-docstring, unused-argument

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

@@ -38,7 +38,7 @@
     SettingId,
     StatusId,
 )
-from open_gopro.enum import GoProEnum
+from open_gopro.enum import GoProIntEnum
 from open_gopro.logger import Logger
 from open_gopro.models.general import HttpInvalidSettingResponse
 from open_gopro.models.response import GlobalParsers, GoProResp
@@ -50,7 +50,7 @@
 ValueType = TypeVar("ValueType")
 IdType = TypeVar("IdType")
 
-QueryParserType = Union[construct.Construct, type[GoProEnum], BytesParserBuilder]
+QueryParserType = Union[construct.Construct, type[GoProIntEnum], BytesParserBuilder]
 
 
 ######################################################## BLE #################################################
@@ -142,7 +142,7 @@ async def __call__(self, __communicator__: GoProBle, **kwargs: Any) -> GoProResp
         response = await __communicator__._send_ble_message(
             self._uuid, data, self._identifier, rules=self._evaluate_rules(**kwargs)
         )
-        logger.info(Logger.build_log_rx_str(response))
+        # logger.info(Logger.build_log_rx_str(response))
         return response
 
     def __str__(self) -> str:
@@ -284,7 +284,7 @@ async def __call__(self, __communicator__: GoProBle, **kwargs: Any) -> GoProResp
         data = self.build_data(**kwargs)
         # Allow exception to pass through if protobuf not completely initialized
         response = await __communicator__._send_ble_message(self._uuid, data, self.response_action_id)
-        logger.info(Logger.build_log_rx_str(response))
+        # logger.info(Logger.build_log_rx_str(response))
         return response
 
     def __str__(self) -> str:
@@ -465,7 +465,7 @@ def __init__(self, communicator: GoProBle, identifier: SettingId, parser_builder
             parser.byte_json_adapter = ByteParserBuilders.Construct(parser_builder)
         elif isinstance(parser_builder, BytesParserBuilder):
             parser.byte_json_adapter = parser_builder
-        elif issubclass(parser_builder, GoProEnum):
+        elif issubclass(parser_builder, GoProIntEnum):
             parser.byte_json_adapter = ByteParserBuilders.GoProEnum(parser_builder)
         else:
             raise TypeError(f"Unexpected {parser_builder=}")
@@ -661,7 +661,7 @@ def __init__(self, communicator: GoProBle, identifier: StatusId, parser: QueryPa
             parser_builder.byte_json_adapter = ByteParserBuilders.Construct(parser)
         elif isinstance(parser, BytesParserBuilder):
             parser_builder.byte_json_adapter = parser
-        elif issubclass(parser, GoProEnum):
+        elif issubclass(parser, GoProIntEnum):
             parser_builder.byte_json_adapter = ByteParserBuilders.GoProEnum(parser)
         else:
             raise TypeError(f"Unexpected {parser_builder=}")
@@ -697,7 +697,7 @@ async def _send_query(self, response_id: QueryCmdId) -> GoProResp:
         data = self._build_cmd(response_id)
         logger.info(Logger.build_log_tx_str(pretty_print(self._as_dict(f"{response_id.name}.{str(self._identifier)}"))))
         response = await self._communicator._send_ble_message(self.UUID, data, response_id)
-        logger.info(Logger.build_log_rx_str(response))
+        # logger.info(Logger.build_log_rx_str(response))
         return response
 
     def _as_dict(  # pylint: disable = arguments-differ

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

@@ -159,7 +159,7 @@ async def load_preset(self, *, preset: int) -> GoProResp[None]:
         return {"id": preset}  # type: ignore
 
     @http_get_json_command(endpoint="gopro/camera/presets/set_group", arguments=["id"])
-    async def load_preset_group(self, *, group: proto.EnumPresetGroup) -> GoProResp[None]:
+    async def load_preset_group(self, *, group: proto.EnumPresetGroup.ValueType) -> GoProResp[None]:
         """Set the active preset group.
 
         The most recently used Preset in this group will be set.
@@ -330,14 +330,16 @@ async def webcam_preview(self) -> GoProResp[WebcamResponse]:
 
     @http_get_json_command(
         endpoint="gopro/webcam/start",
-        arguments=["res", "fov"],
+        arguments=["res", "fov", "port", "protocol"],
         parser=Parser(json_parser=JsonParsers.PydanticAdapter(WebcamResponse)),
     )
     async def webcam_start(
         self,
         *,
         resolution: Params.WebcamResolution | None = None,
         fov: Params.WebcamFOV | None = None,
+        port: int | None = None,
+        protocol: Params.WebcamProtocol | None = None,
     ) -> GoProResp[WebcamResponse]:
         """Start the webcam.
 
@@ -346,11 +348,14 @@ async def webcam_start(
                 camera default will be used.
             fov (Optional[open_gopro.api.params.WebcamFOV]): field of view to use. If not set, camera
                 default will be used.
+            port (Optional[int]): port to use for streaming. If not set, camera default of 8554 will be used.
+            protocol (Optional[open_gopro.api.params.WebcamProtocol]): streaming protocol to use. If not set, camera
+                default of TS will be used.
 
         Returns:
             GoProResp: command status
         """
-        return {"res": resolution, "fov": fov}  # type: ignore
+        return {"res": resolution, "fov": fov, "port": port, "protocol": protocol}  # type: ignore
 
     @http_get_json_command(
         endpoint="gopro/webcam/stop",