Skip to content

Manager

System-level services: CoreManager (plugin-to-plugin RPC, FFmpeg, signed requests), DeviceManager (cameras, discovered cameras), DownloadManager (one-shot download tokens).

camera_ui_sdk.manager

CoreManagerEvent

Bases: TypedDict

Core manager event payload.

Emitted when a core system event occurs (e.g. cloud account changes, remote-server availability, plugin lifecycle changes). Subscribe via coreManager.onEvent to react to system-level state changes.

type instance-attribute 

type: str

Event type identifier (e.g. cloudAccountChanged).

data instance-attribute 

data: Any

Event-specific data payload. Shape depends on the event type.

CoreManager

Bases: Protocol

Core manager interface for system-level operations.

Provides access to cross-cutting services like the FFmpeg binary path, server addresses, HMAC signing for cloud requests, inter-plugin lookup, and a stream of core system events.

Accessed via api.coreManager in plugins.

Example 
ffmpeg_path = await api.coreManager.getFFmpegPath()
addresses = await api.coreManager.getServerAddresses()


def on_event(e):
    if e["type"] == "cloudAccountChanged":
        print("Cloud account state:", e["data"])


api.coreManager.onEvent.subscribe(on_event)

onEvent property 

onEvent: Observable[CoreManagerEvent]

Observable for core manager events (e.g. cloud account changes).

Example 
api.coreManager.onEvent.subscribe(lambda e: print(e["type"], e["data"]))

connectToPlugin async 

connectToPlugin(pluginName: str) -> BasePlugin | None

Connect to another plugin by name.

Parameters:

Name Type Description Default
pluginName str

Name of the plugin to connect to

 required

Returns:

Type Description
BasePlugin | None

Plugin instance or None if not found. Cast to specific interface as needed.

getFFmpegPath async 

getFFmpegPath() -> str

Get the FFmpeg executable path.

Returns:

Type Description
str

Path to FFmpeg binary

getServerAddresses async 

getServerAddresses() -> list[str]

Get server addresses (IP addresses the server is listening on).

Returns:

Type Description
list[str]

List of server addresses

getCloudServerId async 

getCloudServerId() -> str

Get the cloud server identity this server is registered as.

Returns the cloud server_id from the active cloud pairing, or an empty string when the server is not connected to the cloud.

Returns:

Type Description
str

Cloud server id, or an empty string if not paired

getPluginsByInterface async 

getPluginsByInterface(interfaceName: PluginInterface) -> list[PluginInfo]

Get all active plugins that implement a specific interface.

Parameters:

Name Type Description Default
interfaceName PluginInterface

Plugin interface name (e.g., 'ClipDetection')

 required

Returns:

Type Description
list[PluginInfo]

List of plugin info dicts with id, name, contract

DeviceManager

Bases: Protocol

Device manager interface for camera operations. Provides methods to get cameras and push discovered cameras.

Accessed via api.deviceManager in plugins.

Example 
# Get a camera by ID or name
camera = await api.deviceManager.getCamera("Front Door")

# Push discovered cameras (for cloud-based discovery)
discovered = await fetch_cameras_from_cloud()
await api.deviceManager.pushDiscoveredCameras(discovered)

pushDiscoveredCameras async 

pushDiscoveredCameras(cameras: list[DiscoveredCamera]) -> None

Push discovered cameras to the backend. Use this when cameras are discovered asynchronously (e.g., after cloud login). Cameras will be immediately visible in the UI for adoption.

Parameters:

Name Type Description Default
cameras list[DiscoveredCamera]

List of discovered cameras to push

 required

getCamera async 

getCamera(cameraIdOrName: str) -> CameraDevice | None

Get a camera by ID or name.

Parameters:

Name Type Description Default
cameraIdOrName str

Camera ID or name

 required

Returns:

Type Description
CameraDevice | None

Camera device or None if not found

DiscoveredCamera

Bases: TypedDict

Discovered camera from a discovery provider.

Represents a camera found during network scanning or cloud lookup that can be adopted into the system. Push these via deviceManager.pushDiscoveredCameras so the user can pick them in the UI without waiting for the next discovery poll.

id instance-attribute 

id: str

Unique, stable identifier for this discovered camera (used for deduplication).

name instance-attribute 

name: str

Display name shown in the UI adoption list.

manufacturer instance-attribute 

manufacturer: NotRequired[str]

Camera manufacturer label (optional).

model instance-attribute 

model: NotRequired[str]

Camera model label (optional).

CreateDownloadOptions

Bases: TypedDict

Options for creating a download.

filePath instance-attribute 

filePath: str

Absolute path to the file on disk.

filename instance-attribute 

filename: NotRequired[str]

Filename for Content-Disposition header.

mimeType instance-attribute 

mimeType: NotRequired[str]

MIME type for Content-Type header.

ttlMs instance-attribute 

ttlMs: NotRequired[int]

Time-to-live in milliseconds.

cleanup instance-attribute 

cleanup: NotRequired[Literal['never', 'on-expiry', 'on-download']]

When the file on disk is deleted (registry always expires at TTL).

  • never (default): file persists; caller manages it.
  • on-expiry: deleted at TTL. Can be fetched N times during the window — correct mode for notification images that fan out to multiple devices/recipients.
  • on-download: deleted after first successful download OR on TTL, whichever first. One-shot mode for things like backup exports.

CreateStreamDownloadOptions

Bases: CreateDownloadOptions

Options for creating a streaming download (progressive file tailing).

markerPath instance-attribute 

markerPath: str

Path to a marker file that signals export is still in progress.

DownloadToken

Bases: TypedDict

Token returned after registering a download.

token instance-attribute 

token: str

Unique download token.

url instance-attribute 

url: str

In-app, same-origin URL: /api/download/<token>. Use for callers already authenticated against this server.

publicUrl instance-attribute 

publicUrl: str

Externally-reachable, session-less URL the server publishes for out-of-band fetchers (push-notification image attachments, FCM / APNs payloads, share recipients). Shape: <externalUrl>/api/download/<token> — the token in the URL is the auth. Empty string when the server has no external URL configured (LAN-only deployments); fall back to url for in-app callers.

expiresAt instance-attribute 

expiresAt: int

Unix timestamp (ms) when the token expires.

DownloadManager

Bases: Protocol

Download manager interface for token-based file downloads.

Allows plugins to register files for HTTP download via a token URL. No JWT authentication is needed — the token itself is the auth.

Accessed via api.downloadManager in plugins.

Example 
result = await api.downloadManager.createDownload(
    {
        "filePath": "/tmp/export.mp4",
        "filename": "recording.mp4",
        "mimeType": "video/mp4",
        "ttlMs": 600000,
        "cleanup": "on-download",
    }
)
token, url = result["token"], result["url"]

createDownload async 

createDownload(options: CreateDownloadOptions) -> DownloadToken

Register a file for download and get a token-based URL.

Parameters:

Name Type Description Default
options CreateDownloadOptions

Download options

 required

Returns:

Type Description
DownloadToken

Token, URL, and expiry information

createStreamDownload async 

createStreamDownload(options: CreateStreamDownloadOptions) -> DownloadToken

Register a streaming file for progressive download.

The file is tailed during writing; the marker file signals completion.

Parameters:

Name Type Description Default
options CreateStreamDownloadOptions

Streaming download options (includes markerPath)

 required

Returns:

Type Description
DownloadToken

Token, URL, and expiry information

deleteDownload async 

deleteDownload(token: str) -> None

Remove a download token and optionally delete the file.

Parameters:

Name Type Description Default
token str

The download token to remove

 required

NotificationManager

Bases: Protocol

Notification manager interface for publishing notifications into the host.

Plugins call publish to ask the host to fan a Notification out to every installed Notifier-plugin and the in-app UI. The host applies user settings (master toggle, per-source toggle, quiet hours) and the publishing plugin's declared capabilities; calls from plugins without :attr:PluginCapability.PublishNotifications are silently dropped.

Accessed via api.notificationManager in plugins.

Example 
await api.notificationManager.publish(
    {
        "title": "Camera offline",
        "body": "Front Door stopped recording",
        "severity": Severity.Warn,
        "deepLink": "/cameras/front-door",
        "data": {"cameraId": "front-door"},
    }
)

publish async 

publish(notification: Notification) -> None

Send a notification to the host for fan-out to every installed Notifier-plugin and the in-app UI.

Resolves once the publish was handed to the transport. Downstream delivery is async and failures there never propagate back here.

Parameters:

Name Type Description Default
notification Notification

Notification payload to publish.

 required