Connection module#

Connection helpers for discovering and managing wired GoPro devices.

Utilities to create connection objects for GoPro devices, scan for devices over USB/mdns, and manage open/close life-cycle of WiredConnection objects.

class pytermite.connection.WiredConnection(**kwargs: Any)[source]#

Subclass of WiredGoPro providing a cached human-readable name.

Parameters:

**kwargs – All keyword arguments are passed to the WiredGoPro constructor. The name keyword is reserved for an optional cached camera name that can be provided on initialization. If not provided, the name will be lazily loaded on first access by querying the camera’s information via the HTTP API and falling back to a name derived from the serial numbers mapping.

Attributes:

ble_command: Used to call the BLE commands
ble_setting: Used to access the BLE settings
ble_status: Used to access the BLE statuses
http_command: Used to access the USB commands
http_setting: Used to access the USB settings
identifier: Unique identifier for the connected GoPro Client
ip_address: The IP address of the GoPro device
is_ble_connected: Are we connected via BLE to the GoPro device?
is_cohn_provisioned: Is COHN currently provisioned?
is_http_connected: Are we connected via Wifi to the GoPro device?
is_open: Is this client ready for communication?
is_ready: Is gopro ready to receive commands
name: Asynchronously return the human-friendly name of the camera.
version: The Open GoPro API version of the GoPro Client

Methods

`close`()	Gracefully close the GoPro Client connection
`configure_cohn`([timeout])	Prepare Camera on the Home Network
`open`([timeout, retries])	Connect to the Wired GoPro Client and prepare it for communication
`register_update`(callback, update)	Register for callbacks when an update occurs
`unregister_update`(callback[, update])	Unregister for asynchronous update(s)

property name: str#

Asynchronously return the human-friendly name of the camera.

If the name has not been determined yet this will query the camera for its information via the HTTP API and fall back to a name derived from the serial numbers mapping.

Returns:: The camera name.
Return type:: str

async pytermite.connection.close_gopros(gopros: dict[str, WiredConnection] | set[WiredConnection]) → None[source]#

Close all provided WiredConnection objects.

Parameters:: gopros (dict[str, WiredConnection] | set[WiredConnection]) – Mapping of camera keys to WiredConnection objects to close.

async pytermite.connection.connect_gopros(gopros: dict[str, WiredConnection]) → AsyncGenerator[WiredConnection, None][source]#

Attempt to open a connection to each provided WiredConnection.

This is an async generator that yields each connected WiredConnection.

Parameters:: gopros (dict[str, WiredConnection]) – Mapping of camera keys to WiredConnection objects to connect.
Yields:: WiredConnection – Each successfully connected WiredConnection object.

pytermite.connection.create_wired_gopros(gopro_serials: dict[str, str] | set[str]) → dict[str, WiredConnection][source]#

Create WiredConnection objects for provided serial numbers.

Parameters:: gopro_serials (dict[str, str] | set[str]) – Mapping from camera name to serial number, or a set of serial numbers.
Returns:: Mapping from provided key (camera name or serial) to a WiredConnection instance.
Return type:: dict[str, WiredConnection]

async pytermite.connection.scan_for_gopros(waiting_time: int = 10) → set[str][source]#

Scan for connected GoPro devices via USB connection and return a set of serials.

The scan runs until either the user requests to stop (press Enter) or the timeout happens. Both are executed concurrently alongside the USB scanning task.

Parameters:: waiting_time (int, optional) – Maximum seconds to wait for discovery. Default is 10.
Returns:: Set of discovered device serial numbers (strings).
Return type:: set[str]

async pytermite.connection.scan_for_gopros_usb() → None[source]#: Continuously scan for GoPro devices via mDNS until interrupted.

async pytermite.connection.wait_for_user_interrupt() → None[source]#: Wait for the user to press Enter using a non-blocking stdin reader.

Notes

Uses the event loop’s add_reader API so the waiter can be cancelled immediately (the reader is removed in the finally block). This avoids the problem where awaiting a blocking input call prevents task cancellation.