Connection Management¶

BLE Connection Manager¶

gopro_sdk.connection.ble_manager.BleConnectionManager ¶

BleConnectionManager(
    target: str, timeout_config: TimeoutConfig
)

BLE connection manager.

Responsibilities: - Establish and disconnect BLE connections - Accumulate BLE response fragments - Handle BLE notification callbacks

Initialize BLE connection manager.

Parameters:

Name	Type	Description	Default
`target`	`str`	Last four digits of camera serial number	required
`timeout_config`	`TimeoutConfig`	Timeout configuration	required

Attributes¶

is_connected `property` ¶

is_connected: bool

Whether BLE is connected.

Functions¶

connect `async` ¶

connect() -> None

Establish BLE connection.

Implementation based on official Tutorial, directly using bleak's BleakScanner and BleakClient.

Raises:

Type	Description
`BleConnectionError`	BLE connection failed

disconnect `async` ¶

disconnect() -> None

Disconnect BLE connection.

wait_for_response `async` ¶

wait_for_response(timeout: float | None = None) -> bytes

Wait for BLE response.

Parameters:

Name	Type	Description	Default
`timeout`	`float \| None`	Timeout in seconds, None means use default timeout	`None`

Returns:

Type	Description
`bytes`	Response data

Raises:

Type	Description
`BleConnectionError`	Wait timeout

write `async` ¶

write(uuid: str, data: bytes) -> None

Write BLE data (automatic fragmentation).

Parameters:

Name	Type	Description	Default
`uuid`	`str`	Characteristic UUID string (standard format: 8-4-4-4-12)	required
`data`	`bytes`	Data to write	required

Raises:

Type	Description
`BleConnectionError`	Write failed

clear_response_queue ¶

clear_response_queue() -> None

Clear response queue.

HTTP Connection Manager¶

gopro_sdk.connection.http_manager.HttpConnectionManager ¶

HttpConnectionManager(
    target: str,
    timeout_config: TimeoutConfig,
    credentials: CohnCredentials | None = None,
)

HTTP/COHN connection manager.

Responsibilities: - Establish and disconnect HTTP sessions - Configure SSL context (support COHN self-signed certificates) - Send HTTP requests and handle responses

Initialize HTTP connection manager.

Parameters:

Name	Type	Description	Default
`target`	`str`	Last four digits of camera serial number	required
`timeout_config`	`TimeoutConfig`	Timeout configuration	required
`credentials`	`CohnCredentials \| None`	COHN credentials (optional, can be set later)	`None`

Attributes¶

is_connected `property` ¶

is_connected: bool

Whether HTTP is connected.

base_url `property` ¶

base_url: str

HTTP base URL.

Multi-camera scenarios must use COHN mode (each camera has independent IP).

Returns:

Type	Description
`str`	COHN mode: https://{ip_address} (default port 443)

Raises:

Type	Description
`HttpConnectionError`	COHN credentials not configured

Functions¶

set_credentials ¶

set_credentials(credentials: CohnCredentials) -> None

Set COHN credentials.

Parameters:

Name	Type	Description	Default
`credentials`	`CohnCredentials`	COHN credentials	required

connect `async` ¶

connect() -> None

Establish HTTP connection (COHN mode).

Multi-camera scenarios must use COHN mode: HTTPS + SSL + authentication

Raises:

Type	Description
`HttpConnectionError`	HTTP connection failed or COHN credentials not configured

quick_connectivity_check `async` ¶

quick_connectivity_check() -> bool

Quickly check if IP is reachable (without waiting too long).

Returns:

Type	Description
`bool`	True if connection is possible, False otherwise

disconnect `async` ¶

disconnect() -> None

Disconnect HTTP connection.

get ¶

get(
    endpoint: str, params: dict[str, Any] | None = None
) -> _AutoConnectContext

Send GET request (returns async context manager).

Note: - This method is synchronous, returns an async context manager - Will automatically establish HTTPS connection on first call - Usage: async with self.get(...) as resp:

Parameters:

Name	Type	Description	Default
`endpoint`	`str`	API endpoint (relative path, e.g., "gopro/camera/state")	required
`params`	`dict[str, Any] \| None`	Query parameters	`None`

Returns:

Type	Description
`_AutoConnectContext`	Async context manager for HTTP response (_AutoConnectContext)

Raises:

Type	Description
`HttpConnectionError`	Request failed

Usage example

async with self.get("gopro/camera/state") as resp: data = await resp.json()

put ¶

put(endpoint: str, data: Any = None) -> _AutoConnectContext

Send PUT request (returns async context manager).

Note: - This method is synchronous, returns an async context manager - Will automatically establish HTTPS connection on first call - Usage: async with self.put(...) as resp:

Parameters:

Name	Type	Description	Default
`endpoint`	`str`	API endpoint	required
`data`	`Any`	Request data	`None`

Returns:

Type	Description
`_AutoConnectContext`	Async context manager for HTTP response (_AutoConnectContext)

Raises:

Type	Description
`HttpConnectionError`	Request failed

Usage example

async with self.put("gopro/camera/setting", data=...) as resp: result = await resp.json()

download `async` ¶

download(
    endpoint: str,
    destination: str,
    chunk_size: int = 8192,
    progress_callback: Callable[[int, int], None]
    | None = None,
) -> int

Download file.

Parameters:

Name	Type	Description	Default
`endpoint`	`str`	API endpoint	required
`destination`	`str`	Destination file path	required
`chunk_size`	`int`	Chunk size	`8192`
`progress_callback`	`Callable[[int, int], None] \| None`	Progress callback function (downloaded: int, total: int) -> None	`None`

Returns:

Type	Description
`int`	Number of bytes downloaded

Raises:

Type	Description
`HttpConnectionError`	Download failed

Health Check Mixin¶

gopro_sdk.connection.health_check.HealthCheckMixin ¶

Health check and auto-reconnect Mixin.

Provides connection health check and auto-reconnect functionality. Requires subclasses to have ble and http attributes.

Functions¶

is_healthy `async` ¶

is_healthy() -> bool

Check connection health status.

Returns:

Type	Description
`bool`	True if connection is healthy, False otherwise

reconnect `async` ¶

reconnect() -> bool

Smart reconnect.

Returns:

Type	Description
`bool`	True if reconnect successful, False otherwise

set_auto_reconnect ¶

set_auto_reconnect(enabled: bool) -> None

Set whether to enable auto-reconnect.

Parameters:

Name	Type	Description	Default
`enabled`	`bool`	True to enable, False to disable	required

set_max_reconnect_attempts ¶

set_max_reconnect_attempts(attempts: int) -> None

Set maximum reconnect attempt count.

Parameters:

Name	Type	Description	Default
`attempts`	`int`	Maximum attempt count (>= 1)	required

get_health_stats ¶

get_health_stats() -> dict[str, any]

Get health statistics.

Returns:

Type	Description
`dict[str, any]`	Health statistics dictionary

BLE Scanner¶

gopro_sdk.connection.ble_scanner.BleScanner ¶

BLE scanner

Used to discover nearby GoPro camera devices.

Functions¶

scan_devices_stream `async` `staticmethod` ¶

scan_devices_stream(
    duration: float = 8.0,
    idle_timeout: float = 2.0,
    target_count: int | None = None,
) -> AsyncIterator[list[dict[str, Any]]]

Stream scan for GoPro devices

Parameters:

Name	Type	Description	Default
`duration`	`float`	Maximum scan time (seconds)	`8.0`
`idle_timeout`	`float`	Idle timeout (seconds), ends early if no new devices after this time	`2.0`
`target_count`	`int \| None`	Target device count, ends early when reached	`None`

Yields:

Type	Description
`AsyncIterator[list[dict[str, Any]]]`	Device list, each device is a {"name": str, "address": str} dictionary

Usage example:

async for devices in BleScanner.scan_devices_stream(duration=8.0):
    for dev in devices:
        print(f"Discovered: {dev['name']}")

scan_devices `async` `staticmethod` ¶

scan_devices(duration: float = 8.0) -> list[dict[str, Any]]

One-time scan for GoPro devices

Parameters:

Name	Type	Description	Default
`duration`	`float`	Scan time (seconds)	`8.0`

Returns:

Type	Description
`list[dict[str, Any]]`	Device list, each device is a {"name": str, "address": str} dictionary

Usage example:

devices = await BleScanner.scan_devices(duration=5.0)
for dev in devices:
    print(f"Discovered: {dev['name']}")

scan_serials_stream `async` `staticmethod` ¶

scan_serials_stream(
    duration: float = 8.0,
    idle_timeout: float = 2.0,
    target_count: int | None = None,
) -> AsyncIterator[str]

Stream scan for GoPro serial numbers

Parameters:

Name	Type	Description	Default
`duration`	`float`	Maximum scan time (seconds)	`8.0`
`idle_timeout`	`float`	Idle timeout (seconds)	`2.0`
`target_count`	`int \| None`	Target count	`None`

Yields:

Type	Description
`AsyncIterator[str]`	Camera serial number (extracted from device name)

Usage example:

async for serial in BleScanner.scan_serials_stream(duration=8.0):
    print(f"Discovered serial number: {serial}")

scan_serials `async` `staticmethod` ¶

scan_serials(duration: float = 8.0) -> list[str]

One-time scan for GoPro serial numbers

Parameters:

Name	Type	Description	Default
`duration`	`float`	Scan time (seconds)	`8.0`

Returns:

Type	Description
`list[str]`	Serial number list

Usage example:

serials = await BleScanner.scan_serials(duration=5.0)
print(f"Discovered {len(serials)} cameras")

Connection Management¶

BLE Connection Manager¶

gopro_sdk.connection.ble_manager.BleConnectionManager ¶

Attributes¶

is_connected property ¶

Functions¶

connect async ¶

disconnect async ¶

wait_for_response async ¶

write async ¶

clear_response_queue ¶

HTTP Connection Manager¶

gopro_sdk.connection.http_manager.HttpConnectionManager ¶

Attributes¶

is_connected property ¶

base_url property ¶

Functions¶

set_credentials ¶

connect async ¶

quick_connectivity_check async ¶

disconnect async ¶

get ¶

put ¶

download async ¶

Health Check Mixin¶

gopro_sdk.connection.health_check.HealthCheckMixin ¶

Functions¶

is_healthy async ¶

reconnect async ¶

set_auto_reconnect ¶

set_max_reconnect_attempts ¶

get_health_stats ¶

BLE Scanner¶

gopro_sdk.connection.ble_scanner.BleScanner ¶

Functions¶

scan_devices_stream async staticmethod ¶

scan_devices async staticmethod ¶

scan_serials_stream async staticmethod ¶

scan_serials async staticmethod ¶

is_connected `property` ¶

connect `async` ¶

disconnect `async` ¶

wait_for_response `async` ¶

write `async` ¶

is_connected `property` ¶

base_url `property` ¶

connect `async` ¶

quick_connectivity_check `async` ¶

disconnect `async` ¶

download `async` ¶

is_healthy `async` ¶

reconnect `async` ¶

scan_devices_stream `async` `staticmethod` ¶

scan_devices `async` `staticmethod` ¶

scan_serials_stream `async` `staticmethod` ¶

scan_serials `async` `staticmethod` ¶