Utility modules (pyobs.modules.utils)

Utilities TODO: write doc

AutonomousWarning

class AutonomousWarning(warn_sound: str, warn_interval: float = 1, start_sound: Optional[str] = None, started_sound: Optional[str] = None, stop_sound: Optional[str] = None, stopped_sound: Optional[str] = None, player: str = 'mpg123', trigger_file: Optional[str] = None, **kwargs: Any)

Bases: pyobs.modules.Module

A module that can plays a warning sound while an IAutonomous module is running.

Initialize a new warning.

Parameters:
  • warn_sound – Name of file to play.

  • warn_interval – Interval in seconds between sounds.

  • start_sound – Sound to play when starting systems.

  • started_sound – Sound to play when systems started.

  • stop_sound – Sound to play when stopping systems.

  • stopped_sound – Sound to play when systems stopped.

  • trigger_file – File, which triggers to switch on-off and vice versa, when created. Will be deleted afterwards.

FluentLogger

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

Bases: pyobs.modules.Module

Log to fluentd server.

Initialize a new logger.

Parameters:
  • hostname – Hostname of server.

  • port – Port of server.

async open() None[source]

Open module.

Kiosk

class Kiosk(camera: Union[pyobs.interfaces.ICamera, str], port: int = 37077, **kwargs: Any)

Bases: pyobs.modules.Module, pyobs.interfaces.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: aiohttp.web_request.Request) aiohttp.web_response.Response[source]

Handles access to /* and returns a specified image.

Parameters:

request – Request to respond to.

Returns:

Response containing image.

async is_running(**kwargs: Any) bool[source]

Whether kiosk mode is running.

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.

Telegram

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

Bases: pyobs.modules.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: pyobs.modules.Module, pyobs.interfaces.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[source]

Whether a service is running.

async open() None[source]

Open module.

async start(**kwargs: Any) None[source]

Starts a service.

async stop(**kwargs: Any) None[source]

Stops a service.