Utility modules (pyobs.modules.utils)

Utilities TODO: write doc

FluentLogger

class FluentLogger(hostname: str, port: int, *args: Any, **kwargs: Any)

Bases: Module

Log to fluentd server.

Initialize a new logger.

Parameters:

hostname – Hostname of server.
port – Port of server.

async open() → None[source]: Open module.

HttpFileCache

class HttpFileCache(port: int = 37075, cache_size: int = 25, max_file_size: int = 100, token: str | None = None, **kwargs: Any)[source]

Bases: Module

A file cache based on a HTTP server.

Initializes file cache.

Parameters:

port – Port for HTTP server.
cache_size – Size of file cache, i.e. number of files to cache.
max_file_size – Maximum file size in MB.
token – Shared secret required in the “Authorization: Bearer <token>” header for download/upload access. If None (default), no auth is enforced.

async close() → None[source]: Close server

async download_handler(request: Request) → Response[source]

Handles GET access to /{filename} and returns image.

Parameters:: request – Request to respond to.
Returns:: Response containing image.

async open() → None[source]: Open server

property opened: bool: Whether the server is started.

async options_handler(request: Request) → Response[source]

Handles OPTIONS access to /{filename} for CORS preflight requests.

Parameters:: request – Request to respond to.
Returns:: Empty response with CORS headers.

async ping_handler(request: Request) → Response[source]

Handles GET access to /ping for testing connectivity.

Parameters:: request – Request to respond to.
Returns:: Response with a JSON status.

async upload_handler(request: Request) → Response[source]

Handles PUSH access to /, stores image and returns filename.

Parameters:: request – Request to respond to.
Returns:: Response containing filename.

Kiosk

class Kiosk(camera: ICamera | str, port: int = 37077, **kwargs: Any)

Bases: Module, IStartStop

A kiosk mode for a pyobs camera that takes images and published them via HTTP.

Initializes file cache.

Parameters:

camera – Camera to use for kiosk mode.
port – Port for HTTP server.

async close() → None[source]: Close server

async image_handler(request: Request) → Response[source]

Handles access to /* and returns a specified image.

Parameters:: request – Request to respond to.
Returns:: Response containing image.

async open() → None[source]: Open server

property opened: bool: Whether the server is started.

async start(**kwargs: Any) → None[source]: Start kiosk mode.

async stop(**kwargs: Any) → None[source]: Stop kiosk mode.

Matrix

class Matrix(server: str, user_id: str, password: str, room_id: str, name: str = 'Bot', log_level: str = 'WARNING', **kwargs: Any)

Bases: Module

A matrix bot.

Initialize a new bot.

Parameters:

server – Server to connect to.
user_id – ID of user to connect as.
password – Password for user.
room_id – Room ID to send messages to.
name – Name of bot.
log_level – Log level to use.

async close() → None[source]: Close module.

async open() → None[source]: Open module.

Telegram

class Telegram(token: str, password: str, allow_new_users: bool = True, **kwargs: Any)

Bases: Module

A telegram bot.

Initialize a new bot.

Parameters:

token – The telegram API token.
password – Password for users to log in.
allow_new_users – Whether new users are allowed to connect.

async close() → None[source]: Close module.

async open() → None[source]: Open module.

Trigger

class Trigger(triggers: list[dict[str, Any]], **kwargs: Any)

Bases: Module, IAutonomous

A module that can call another module’s methods when a specific event occurs.

Initialize a new trigger module.

Parameters:: triggers – List of dictionaries defining the trigger. Must contain fields for event, module and method, may contain a sender.

async open() → None[source]: Open module.

async start(**kwargs: Any) → None[source]: Starts a service.

async stop(**kwargs: Any) → None[source]: Stops a service.