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

Open module.

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

Close server

async image_handler(request: Request) → Response

Handles access to /* and returns a specified image.

Parameters:

request – Request to respond to.

Returns:

Response containing image.

async is_running(**kwargs: Any) → bool

Whether kiosk mode is running.

async open() → None

Open server

property opened: bool

Whether the server is started.

async start(**kwargs: Any) → None

Start kiosk mode.

async stop(**kwargs: Any) → None

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

Close module.

async open() → None

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

Close module.

async open() → None

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 is_running(**kwargs: Any) → bool

Whether a service is running.

async open() → None

Open module.

async start(**kwargs: Any) → None

Starts a service.

async stop(**kwargs: Any) → None

Stops a service.