pumaguard package

pumaguard package#

Subpackages#

pumaguard.web_routes package

Submodules#

pumaguard.camera_heartbeat module#

Camera heartbeat monitoring for PumaGuard.

This module provides background monitoring of camera availability using ICMP ping and TCP connection checks.

class pumaguard.camera_heartbeat.CameraHeartbeat(webui: WebUI, interval: int = 60, enabled: bool = True, check_method: str = 'tcp', tcp_port: int = 80, tcp_timeout: int = 3, icmp_timeout: int = 2, status_change_callback: Callable[[str, dict], None] | None = None, auto_remove_enabled: bool = False, auto_remove_hours: int = 24)[source]#

Bases: DeviceHeartbeat

Background service to monitor camera availability via ICMP ping and TCP checks.

The heartbeat monitor runs in a separate thread and periodically checks if cameras are reachable. It updates the camera status and last_seen timestamp based on the results.

check_camera(ip_address: str) → bool[source]#

Check if a camera is reachable using the configured method.

Args:: ip_address: IP address of the camera
Returns:: True if camera is reachable, False otherwise

check_device(ip_address: str) → bool[source]#

Check if a camera is reachable.

This is the abstract method implementation that delegates to check_camera.

Args:: ip_address: IP address of the camera
Returns:: True if camera is reachable, False otherwise

pumaguard.classify module#

This script classifies images.

pumaguard.classify.configure_subparser(parser: ArgumentParser)[source]#: Parse the commandline

pumaguard.classify.main(options: Namespace, presets: Settings)[source]#: Main entry point

pumaguard.device_heartbeat module#

Base class for device heartbeat monitoring in PumaGuard.

This module provides a common abstract base class for monitoring device availability in the background. It handles thread management, status updates, persistence, and stale device removal.

class pumaguard.device_heartbeat.DeviceHeartbeat(webui: WebUI, device_type: str, interval: int = 60, enabled: bool = True, status_change_callback: Callable[[str, dict], None] | None = None, auto_remove_enabled: bool = False, auto_remove_hours: int = 24)[source]#

Bases: ABC

Abstract base class for device heartbeat monitoring.

Provides common functionality for monitoring device availability via background threads. Subclasses must implement device-specific check methods and persistence logic.

abstract check_device(ip_address: str) → bool[source]#

Check if a device is reachable.

Subclasses must implement this method with their specific connectivity check logic (HTTP, ICMP, TCP, etc.).

Args:: ip_address: IP address of the device
Returns:: True if device is reachable, False otherwise

check_now() → dict[str, bool][source]#

Immediately check all devices and return results.

This can be called manually to force a check outside the regular interval.

Returns:: Dictionary mapping MAC addresses to reachability status

start() → None[source]#: Start the heartbeat monitoring thread.

stop() → None[source]#: Stop the heartbeat monitoring thread.

pumaguard.lock_manager module#

A simple lock manager.

class pumaguard.lock_manager.PumaGuardLock(lock: allocate_lock)[source]#

Bases: object

A global lock.

acquire() → bool[source]#: Acquire the lock. Record when the acquire attempt started so we can measure how long we have been waiting or have held the lock since the call.

release()[source]#: Releases the lock.

time_waited() → float[source]#: Return the time since acquire() was called (how long we’ve been waiting/held).

pumaguard.lock_manager.acquire_lock() → PumaGuardLock[source]#

Acquire the lock.

This method will block until the lock is available.

pumaguard.lock_manager.release(lock: PumaGuardLock)[source]#: Release the lock.

pumaguard.main module#

pumaguard.model_cli module#

CLI commands for managing models.

pumaguard.model_cli.configure_subparser(parser: ArgumentParser)[source]#: Parses command line arguments.

pumaguard.model_cli.main(args: Namespace, presets: Settings)[source]#: Main entry point.

pumaguard.model_downloader module#

Model downloader utility for PumaGuard.

pumaguard.model_downloader.assemble_model_fragments(fragment_paths: list[Path], output_path: Path, expected_sha256: str | None = None) → bool[source]#

Assemble model fragments into a single file (equivalent to ‘cat file* > output’).

Args:: fragment_paths: List of paths to fragment files (in order) output_path: Path where assembled file should be written
Returns:: bool: True if assembly successful

pumaguard.model_downloader.cache_models()[source]#: Cache all available models.

pumaguard.model_downloader.clear_model_cache()[source]#: Clear all downloaded models from cache.

pumaguard.model_downloader.create_registry(models_dir: Path)[source]#

Create a new registry file in the cache directory.

This file stores the checksums of the models cached.

pumaguard.model_downloader.download_file(url: str, destination: Path, expected_sha256: str | None = None, print_progress: bool = True) → bool[source]#

Download a file from URL to destination with progress reporting.

Args:: url: URL to download from destination: Local file path to save to expected_sha256: Optional SHA256 checksum for verification
Returns:: bool: True if download and verification successful

pumaguard.model_downloader.download_model_fragments(fragment_urls: list[str], models_dir: Path, print_progress: bool = True) → list[Path][source]#

Download all fragments for a split model.

Args:: fragment_urls: List of URLs to download fragments from models_dir: Directory to store fragments
Returns:: List[Path]: Paths to downloaded fragment files

pumaguard.model_downloader.ensure_model_available(model_name: str, print_progress: bool = True) → Path[source]#

Ensure a model is available locally, downloading and assembling if necessary.

Args:: model_name: Name of the model (must be in MODEL_REGISTRY)
Returns:: Path: Path to the local model file
Raises:: ValueError: If model_name not in registry RuntimeError: If download or assembly fails

pumaguard.model_downloader.export_registry()[source]#: Export registry to standard out.

pumaguard.model_downloader.get_models_directory() → Path[source]#: Get the directory where models should be stored. Uses XDG_DATA_HOME or defaults to ~/.local/share/pumaguard/models

pumaguard.model_downloader.list_available_models() → list[str][source]#

List all available models in the registry.

Returns:: Dict: Mapping of model names to their URLs

pumaguard.model_downloader.update_model()[source]#: Update a model to cache.

pumaguard.model_downloader.verify_file_checksum(file_path: Path, expected_sha256: str) → bool[source]#: Verify file checksum.

pumaguard.plug_heartbeat module#

Plug heartbeat monitoring for PumaGuard.

This module provides background monitoring of Shelly plug availability using HTTP REST API checks.

class pumaguard.plug_heartbeat.PlugHeartbeat(webui: WebUI, interval: int = 60, enabled: bool = True, timeout: int = 5, status_change_callback: Callable[[str, dict], None] | None = None, auto_remove_enabled: bool = False, auto_remove_hours: int = 24)[source]#

Bases: DeviceHeartbeat

Background service to monitor Shelly plug availability via HTTP REST API.

The heartbeat monitor runs in a separate thread and periodically checks if plugs are reachable by querying their Shelly Gen2 REST API. It updates the plug status and last_seen timestamp based on the results.

check_device(ip_address: str) → bool[source]#

Check if a plug is reachable.

This is the abstract method implementation that delegates to check_plug.

Args:: ip_address: IP address of the plug
Returns:: True if plug is reachable, False otherwise

check_plug(ip_address: str) → bool[source]#

Check if a plug is reachable.

Args:: ip_address: IP address of the plug
Returns:: True if plug is reachable, False otherwise

pumaguard.presets module#

The presets for each model.

exception pumaguard.presets.PresetError[source]#

Bases: Exception

Docstring for PresetError

class pumaguard.presets.Settings[source]#

Bases: object

Base class for Presets

property base_output_directory: str#: Get the base_output_directory.

property classifier_model_filename: str#: Get the classifier model filename.

property deterrent_sound_files#: Get the list of deterrent sound files.

property epochs: int#: The number of epochs.

property file_stabilization_extra_wait: float#: Get extra wait.

property history_file#: Get the history file.

property image_dimensions: Tuple[int, int]#: Get the image dimensions.

property lion_directories: list[str]#: The directories containing lion images.

load(filename: str)[source]#: Load settings from YAML file.

property model_file#: Get the location of the model file.

property model_function_name: str#: Get the model function name.

property model_version: str#: Get the model version name.

property no_lion_directories: list[str]#: The directories containing no_lion images.

property notebook_number: int#: Get notebook number.

property number_color_channels: int#: The number of color channels.

property play_sound: bool#: Get play-sound.

property puma_threshold: float#: Get the puma classification threshold.

save()[source]#: Write presets to settings file.

property settings_file: str#: Get the settings file.

property sound_path#: Get the sound path.

property validation_lion_directories: list[str]#: The directories containing lion images for validation.

property validation_no_lion_directories: list[str]#: The directories containing no_lion images for validation.

property verification_path: str#: Get the verification path.

property volume: int#: Get volume level (0-100).

property with_augmentation: bool#: Get whether to augment training data.

property yolo_conf_thresh: float#: Get the YOLO conf-thresh.

property yolo_max_dets: int#: Get the YOLO max-dets.

property yolo_min_size: float#: Get the YOLO min-size.

property yolo_model_filename: str#: Get the YOLO model filename.

pumaguard.presets.get_default_settings_file() → str[source]#

Get the default settings file path using XDG standards.

Checks in order: 1. If running as snap: SNAP_USER_DATA/pumaguard/settings.yaml 2. XDG_CONFIG_HOME/pumaguard/settings.yaml (e.g., ~/.config/pumaguard/settings.yaml) 3. Current directory pumaguard-settings.yaml (for backwards compatibility)

Returns:: Path to the settings file

pumaguard.presets.get_xdg_cache_home() → Path[source]#

Get the XDG cache home directory.

Returns:: Path to XDG_CACHE_HOME (defaults to ~/.cache if not set)

pumaguard.presets.get_xdg_config_home() → Path[source]#

Get the XDG config home directory according to XDG Base Directory spec.

Returns:: Path to XDG_CONFIG_HOME (defaults to ~/.config if not set)

pumaguard.presets.get_xdg_data_home() → Path[source]#

Get the XDG data home directory.

Returns:: Path to XDG_DATA_HOME (defaults to ~/.local/share if not set)

pumaguard.server module#

pumaguard.shelly_control module#

Shared utilities for controlling Shelly smart plugs.

This module provides common functions for interacting with Shelly Gen2 devices via their RPC API, used by both the server and web routes.

pumaguard.shelly_control.get_shelly_status(ip_address: str, hostname: str = 'unknown', timeout: int = 5) → Tuple[bool, Dict | None, str | None][source]#

Get the current status of a Shelly plug using the Gen2 Switch.GetStatus API.

Args:

ip_address: IP address of the Shelly plug hostname: Hostname of the plug for logging (optional) timeout: Request timeout in seconds (default: 5)

Returns:

Tuple of (success, status_data, error_message):

success: True if operation succeeded, False
otherwise
status_data: Dict with Shelly status data if
successful, None otherwise
error_message: Error description if failed, None if
successful

Example:

>>> success, data, error = get_shelly_status(
...     "192.168.1.100", "porch-plug"
... )
>>> if success:
...     is_on = data['output']
...     print(f"Switch is: {'ON' if is_on else 'OFF'}")
... else:
...     print(f"Error: {error}")

pumaguard.shelly_control.set_shelly_switch(ip_address: str, on_state: bool, hostname: str = 'unknown', timeout: int = 5) → Tuple[bool, Dict | None, str | None][source]#

Control a Shelly plug switch using the Gen2 Switch.Set API.

Args:

ip_address: IP address of the Shelly plug on_state: True to turn on, False to turn off hostname: Hostname of the plug for logging (optional) timeout: Request timeout in seconds (default: 5)

Returns:

Tuple of (success, response_data, error_message):

success: True if operation succeeded, False otherwise
response_data: Dict with Shelly response data if
successful, None otherwise
error_message: Error description if failed, None if
successful

Example:

>>> success, data, error = set_shelly_switch(
...     "192.168.1.100", True, "porch-plug"
... )
>>> if success:
...     print(f"Was previously: {data['was_on']}")
... else:
...     print(f"Error: {error}")

pumaguard.sound module#

Sounds

pumaguard.sound.detect_volume_control() → str | None[source]#

Auto-detect the ALSA simple-mixer control to use for volume.

Runs amixer scontents once and finds all controls that advertise pvolume (playback volume) capability. From those, the first name that appears in _VOLUME_CONTROL_PRIORITY is returned. If none of the preferred names match, the first playback-capable control found is returned instead. The result is cached so that amixer is only invoked once for the lifetime of the process.

Returns:: The name of the best available playback volume control (e.g. "PCM", "Master"), or None if amixer is unavailable or no suitable control could be found.

pumaguard.sound.get_volume(control: str | None = None) → int | None[source]#

Get the current ALSA mixer volume using amixer.

Args:

control: ALSA mixer control name. When None (the default) the: control is auto-detected via detect_volume_control().

Returns:

Current volume level as an integer from 0-100, or None if it could not be determined.

pumaguard.sound.is_playing()[source]#

Check if a sound is currently playing.

Returns:: bool: True if a sound is currently playing, False otherwise

pumaguard.sound.main()[source]#: Main entry point.

pumaguard.sound.playsound(soundfile: str, volume: int = 80, blocking: bool = True)[source]#

Play a sound file with specified volume.

The volume is applied via the ALSA mixer (amixer) before playback, so the system output level is adjusted rather than using mpg123’s software scaling.

Args:: soundfile: Path to the sound file to play volume: Volume level from 0-100 (default: 80) blocking: If True, wait for sound to finish. If False, return immediately (default: True)

pumaguard.sound.reset_volume_control_cache()[source]#

Reset the cached ALSA volume control detection result.

This forces detect_volume_control() to re-probe amixer on its next call. Intended for use in tests and for situations where the audio device has changed at runtime.

pumaguard.sound.set_volume(volume: int, control: str | None = None) → bool[source]#

Set ALSA volume and return success status.

Args:: volume: Volume level from 0-100 control: ALSA mixer control name. When None (the default) the control is auto-detected via detect_volume_control().
Returns:: True when volume was successfully applied, False otherwise.

pumaguard.sound.stop_sound()[source]#

Stop any currently playing sound.

Returns:: bool: True if a sound was stopped, False if nothing was playing

pumaguard.stats module#

Module for statistics and plotting.

pumaguard.stats.plot_training_progress(filename, full_history)[source]#: Plot the training progress and store in file.

pumaguard.utils module#

Some utility functions.

pumaguard.utils.cache_model_two_stage(yolo_model_filename: str, classifier_model_filename: str, print_progress: bool = True)[source]#: Caches the model weights.

pumaguard.utils.classify_image(presets: Settings, image_path: str) → float[source]#

Classify the image and print out the result.

Args:

presets (BasePreset): An instance of the BasePreset class containing image processing settings.

model (keras.Model): A pre-trained Keras model used for image classification.

image_path (str): The file path to the image to be classified.

Returns:

float: The classification result as a float value.

Prints:

The color mode being used, the image being classified, and the time taken for classification.

pumaguard.utils.classify_image_two_stage(presets: Settings, image_path: str, print_progress: bool = True, intermediate_dir: str | None = None) → float[source]#

Classify the image using two-stage approach: YOLO detection + EfficientNet classification.

Args:: presets (Preset): An instance of the Preset class containing settings. image_path (str): The file path to the image to be classified.
Args:: presets (Preset): Settings preset. image_path (str): Path to image file. print_progress (bool): Whether to print model download progress. intermediate_dir (str | None): If provided, store visualization and CSV summaries inside this directory instead of CWD.
Returns:: float: Maximum puma probability from all detections (0.0 if no detections)

pumaguard.utils.clear_model_cache()[source]#: Clear the model cache. Use this if you need to reload models or free memory. Warning: Next classification will need to reload models from disk.

pumaguard.utils.copy_images(work_directory, lion_images, no_lion_images)[source]#: Copy images to work directory.

pumaguard.utils.get_cached_model(model_type: str, model_path: Path)[source]#

Get a cached model or load it if not in cache. Thread-safe model caching to prevent memory leaks.

Args:: model_type: Either ‘classifier’ or ‘detector’ model_path: Path to the model file
Returns:: The loaded model from cache or freshly loaded

pumaguard.utils.get_duration(start_time: datetime, end_time: datetime) → float[source]#

Get duration between start and end time in seconds.

Args:: start_time (datetime.timezone): The start time. end_time (datetime.timezone): The end time.
Returns:: float: The duration in seconds.

pumaguard.utils.get_md5(filepath: str) → str[source]#: Compute the MD5 hash for a file.

pumaguard.utils.get_sha256(filepath: str) → str[source]#: Compute the SHA-256 hash for a file.

pumaguard.utils.prepare_image(img_path: str, image_dimensions: Tuple[int, int])[source]#: Prepare the image.

pumaguard.utils.print_bash_completion(command: str, shell: str)[source]#: Print bash completion script.

pumaguard.verify module#

This script verifies models against a standard set of images.

pumaguard.verify.configure_subparser(parser: ArgumentParser)[source]#: Parse the commandline

pumaguard.verify.get_accuracy(predictions: list[tuple[str, float, int]]) → float[source]#: Get the accuracy of the model.

pumaguard.verify.get_binary_accuracy(predictions: list[tuple[str, float, int]]) → float[source]#: Get the accuracy of the model.

pumaguard.verify.get_crossentropy_loss(predictions: list[tuple[str, float, int]]) → float[source]#: Get the log-loss (crossentropy loss) of the model.

pumaguard.verify.get_mean_squared_error(predictions: list[tuple[str, float, int]]) → float[source]#: Get the mean squared error of the model.

pumaguard.verify.main(args: Namespace, presets: Settings)[source]#: Main entry point

pumaguard.verify.verify_model(presets: Settings)[source]#: Verify a model by calculating its accuracy across a standard set of images.

pumaguard.web_ui module#

Module contents#

PumaGuard

pumaguard package

Contents

pumaguard package#

Subpackages#

Submodules#

pumaguard.camera_heartbeat module#

pumaguard.classify module#

pumaguard.device_heartbeat module#

pumaguard.lock_manager module#

pumaguard.main module#

pumaguard.model_cli module#

pumaguard.model_downloader module#

pumaguard.plug_heartbeat module#

pumaguard.presets module#

pumaguard.server module#

pumaguard.shelly_control module#

pumaguard.sound module#

pumaguard.stats module#

pumaguard.utils module#

pumaguard.verify module#

pumaguard.web_ui module#

Module contents#