Source code for pumaguard.presets

"""
The presets for each model.
"""

import copy
import logging
import os
from pathlib import (
    Path,
)
from typing import (
    Tuple,
)

import yaml

logger = logging.getLogger("PumaGuard")


[docs] def get_xdg_config_home() -> Path: """ Get the XDG config home directory according to XDG Base Directory spec. Returns: Path to XDG_CONFIG_HOME (defaults to ~/.config if not set) """ xdg_config = os.environ.get("XDG_CONFIG_HOME") if xdg_config: return Path(xdg_config) return Path.home() / ".config"
[docs] def get_xdg_data_home() -> Path: """ Get the XDG data home directory. Returns: Path to XDG_DATA_HOME (defaults to ~/.local/share if not set) """ xdg_data = os.environ.get("XDG_DATA_HOME") if xdg_data: return Path(xdg_data) return Path.home() / ".local" / "share"
[docs] def get_xdg_cache_home() -> Path: """ Get the XDG cache home directory. Returns: Path to XDG_CACHE_HOME (defaults to ~/.cache if not set) """ xdg_cache = os.environ.get("XDG_CACHE_HOME") if xdg_cache: return Path(xdg_cache) return Path.home() / ".cache"
[docs] def get_default_settings_file() -> str: """ 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 """ # Check if running as snap (strict confinement requires snap path) snap_user_data = os.environ.get("SNAP_USER_DATA") if snap_user_data: snap_config_dir = Path(snap_user_data) / "pumaguard" snap_settings_file = snap_config_dir / "pumaguard-settings.yaml" # If snap settings file exists, use it if snap_settings_file.exists(): return str(snap_settings_file) # Create snap config directory if needed and return path snap_config_dir.mkdir(parents=True, exist_ok=True) return str(snap_settings_file) # XDG compliant location xdg_config_dir = get_xdg_config_home() / "pumaguard" xdg_settings_file = xdg_config_dir / "pumaguard-settings.yaml" # If the XDG file exists, use it if xdg_settings_file.exists(): return str(xdg_settings_file) xdg_config_dir.mkdir(parents=True, exist_ok=True) return str(xdg_settings_file)
[docs] class PresetError(Exception): """ Docstring for PresetError """
# pylint: disable=too-many-public-methods
[docs] class Settings: """ Base class for Presets """ _base_output_directory: str = "" _model_file: str = "" def __init__(self): self.settings_file = get_default_settings_file() self.yolo_min_size = 0.02 self.yolo_conf_thresh = 0.15 self.yolo_max_dets = 2 self.yolo_model_filename = "yolov8s_101425.pt" self.classifier_model_filename = "colorbw_111325.h5" self.puma_threshold = 0.5 self.base_output_directory = os.path.join( os.path.dirname(__file__), "../pumaguard-models" ) self.sound_path = os.path.join( os.path.dirname(__file__), "../pumaguard-sounds" ) self.deterrent_sound_files = ["deterrent_puma.mp3"] self.verification_path = "data/stable/stable_test" self.notebook_number = 1 self.file_stabilization_extra_wait = 1 self.epochs = 300 self.image_dimensions: tuple[int, int] = (128, 128) self.lion_directories: list[str] = [] self.validation_lion_directories: list[str] = [] self.model_function_name = "xception" self.model_version = "undefined" self.play_sound = True self.volume = 80 # Volume level 0-100 for ALSA playback self.print_download_progress = True self.cameras: list[dict[str, str]] = [] # List of detected cameras self.plugs: list[dict[str, str]] = [] # List of detected plugs self.device_history: dict[str, dict[str, str]] = ( {} ) # Device history by MAC self.no_lion_directories: list[str] = [] self.validation_no_lion_directories: list[str] = [] self.with_augmentation = False # Camera heartbeat monitoring settings self.camera_heartbeat_enabled = True self.camera_heartbeat_interval = 60 # Check interval in seconds self.camera_heartbeat_method = "tcp" # "icmp", "tcp", or "both" self.camera_heartbeat_tcp_port = 80 # TCP port to check self.camera_heartbeat_tcp_timeout = 3 # TCP timeout in seconds self.camera_heartbeat_icmp_timeout = 2 # ICMP timeout in seconds # Device auto-removal settings (applies to both cameras and plugs) self.device_auto_remove_enabled = True # Enabled by default self.device_auto_remove_hours = 24 # Remove after 24 hours # WiFi client networks for the USB wireless adapter (wifi1). # Each entry is a dict with keys: ssid (str), psk (str, may be # empty for open networks), priority (int, higher = preferred). # Credentials are written at runtime by the PumaGuard wifi API # to /etc/wpa_supplicant/wpa_supplicant-wifi1.conf. self.wifi_networks: list[dict[str, str | int]] = [] # Plug heartbeat monitoring settings self.plug_heartbeat_enabled = True self.plug_heartbeat_interval = 60 # Check interval in seconds self.plug_heartbeat_timeout = 5 # HTTP timeout in seconds # Classification product directories (XDG data location by default) data_root = get_xdg_data_home() / "pumaguard" self.classification_root_dir = str(data_root / "classified") self.classified_puma_dir = str( Path(self.classification_root_dir) / "puma" ) self.classified_other_dir = str( Path(self.classification_root_dir) / "other" ) self.intermediate_dir = str( Path(self.classification_root_dir) / "intermediate" ) self.intermediate_puma_dir = str( Path(self.classification_root_dir) / "intermediate-puma" ) self.intermediate_other_dir = str( Path(self.classification_root_dir) / "intermediate-other" ) # Default watch directory (incoming images) self.default_watch_dir = str(data_root / "watch") # Ensure directories exist for d in [ self.classification_root_dir, self.classified_puma_dir, self.classified_other_dir, self.intermediate_dir, self.intermediate_puma_dir, self.intermediate_other_dir, self.default_watch_dir, ]: try: Path(d).mkdir(parents=True, exist_ok=True) except OSError as exc: # pragma: no cover (rare failure) logger.error("Could not create directory %s: %s", d, exc)
[docs] def load(self, filename: str): """ Load settings from YAML file. """ logger.info("loading settings from %s", filename) # Update settings_file to the file we're loading from self.settings_file = filename try: with open(filename, encoding="utf-8") as fd: settings = yaml.safe_load(fd) except FileNotFoundError: logger.error( "Could not open settings (%s), using defaults", filename ) return except yaml.constructor.ConstructorError as e: error_msg = str(e) if "python/tuple" in error_msg: raise PresetError( f"{error_msg}\n\n" + "Your settings file contains Python-specific tuple " + "formatting that is no longer supported.\n" + f"Please update {filename} to use YAML list syntax.\n" + "For example, change:\n" + " image-dimensions: !!python/tuple [512, 512]\n" + "to:\n" + " image-dimensions:\n" + " - 512\n" + " - 512\n" + "Or delete the file to use defaults." ) from e raise PresetError(error_msg) from e self.yolo_min_size = settings.get("YOLO-min-size", 0.02) self.yolo_conf_thresh = settings.get("YOLO-conf-thresh", 0.15) self.yolo_max_dets = settings.get("YOLO-max-dets", 12) self.yolo_model_filename = settings.get( "YOLO-model-filename", "yolov8s_101425.pt" ) self.classifier_model_filename = settings.get( "classifier-model-filename", "colorbw_111325.h5" ) self.puma_threshold = settings.get("puma-threshold", 0.5) self.sound_path = settings.get( "sound-path", os.path.dirname(__file__) + "../pumaguard-sounds" ) # Support both old single file (string) and new multiple files (list) deterrent_sound = settings.get("deterrent-sound-files", None) if deterrent_sound is None: # Backwards compatibility: check for old single file setting deterrent_sound = settings.get( "deterrent-sound-file", "cougar_call.mp3" ) if isinstance(deterrent_sound, str): self.deterrent_sound_files = ( [deterrent_sound] if deterrent_sound else ["cougar_call.mp3"] ) elif isinstance(deterrent_sound, list): self.deterrent_sound_files = ( deterrent_sound if deterrent_sound else ["cougar_call.mp3"] ) else: self.deterrent_sound_files = ["cougar_call.mp3"] self.volume = settings.get("volume", 80) self.notebook_number = settings.get("notebook", 1) self.epochs = settings.get("epochs", 1) dimensions = settings.get("image-dimensions", [0, 0]) if ( not isinstance(dimensions, list) or len(dimensions) != 2 or not all(isinstance(d, int) for d in dimensions) ): raise ValueError( "expected image-dimensions to be a list of two integers" ) self.image_dimensions = tuple(dimensions) self.model_version = settings.get("model-version", "undefined") self.model_function_name = settings.get("model-function", "undefined") self.base_output_directory = settings.get( "base-output-directory", "undefined" ) self.verification_path = settings.get( "verification-path", "data/stable/stable_test" ) lions = settings.get("lion-directories", ["undefined"]) if not isinstance(lions, list) or not all( isinstance(p, str) for p in lions ): raise ValueError("expected lion-directories to be a list of paths") self.lion_directories = lions no_lions = settings.get("no-lion-directories", ["undefined"]) if not isinstance(no_lions, list) or not all( isinstance(p, str) for p in no_lions ): raise ValueError( "expected no-lion-directories to be a list of paths" ) self.no_lion_directories = no_lions validation_lions = settings.get("validation-lion-directories", []) if not isinstance(validation_lions, list) or not all( isinstance(p, str) for p in validation_lions ): raise ValueError( "expected validation-lion-directories to be a list of paths" ) self.validation_lion_directories = validation_lions validation_no_lions = settings.get( "validation-no-lion-directories", [] ) if not isinstance(validation_no_lions, list) or not all( isinstance(p, str) for p in validation_no_lions ): raise ValueError( "expected validation-no-lion-directories to be a list of paths" ) self.validation_no_lion_directories = validation_no_lions self.with_augmentation = settings.get("with-augmentation", False) self.file_stabilization_extra_wait = settings.get( "file-stabilization-extra-wait", 1 ) self.play_sound = settings.get("play-sound", True) self.volume = settings.get("volume", 80) self.print_download_progress = settings.get( "print-download-progress", True ) self.cameras = settings.get("cameras", []) self.plugs = settings.get("plugs", []) self.device_history = settings.get("device-history", {}) self.wifi_networks = settings.get("wifi-networks", []) # Load camera heartbeat settings self.camera_heartbeat_enabled = settings.get( "camera-heartbeat-enabled", True ) self.camera_heartbeat_interval = settings.get( "camera-heartbeat-interval", 60 ) self.camera_heartbeat_method = settings.get( "camera-heartbeat-method", "tcp" ) self.camera_heartbeat_tcp_port = settings.get( "camera-heartbeat-tcp-port", 80 ) self.camera_heartbeat_tcp_timeout = settings.get( "camera-heartbeat-tcp-timeout", 3 ) self.camera_heartbeat_icmp_timeout = settings.get( "camera-heartbeat-icmp-timeout", 2 ) # Load device auto-removal settings (backward compatible) # Check new generic names first, fall back to old camera-specific names self.device_auto_remove_enabled = settings.get( "device-auto-remove-enabled", settings.get("camera-auto-remove-enabled", True), ) self.device_auto_remove_hours = settings.get( "device-auto-remove-hours", settings.get("camera-auto-remove-hours", 24), ) # Load plug heartbeat settings self.plug_heartbeat_enabled = settings.get( "plug-heartbeat-enabled", True ) self.plug_heartbeat_interval = settings.get( "plug-heartbeat-interval", 60 ) self.plug_heartbeat_timeout = settings.get("plug-heartbeat-timeout", 5)
[docs] def save(self): """ Write presets to settings file. """ settings_dict = dict(self) with open(self.settings_file, "w", encoding="utf-8") as f: yaml.dump(settings_dict, f, default_flow_style=False) logger.info("Settings saved to %s", self.settings_file)
def _relative_paths(self, base: str, paths: list[str]) -> list[str]: """ The directories relative to a base path. """ return [os.path.relpath(path, start=base) for path in paths] def __iter__(self): """ Serialize this class. """ # pylint: disable=line-too-long yield from { "YOLO-min-size": self.yolo_min_size, "YOLO-conf-thresh": self.yolo_conf_thresh, "YOLO-max-dets": self.yolo_max_dets, "YOLO-model-filename": self.yolo_model_filename, "classifier-model-filename": self.classifier_model_filename, "puma-threshold": self.puma_threshold, "sound-path": self.sound_path, "deterrent-sound-files": self.deterrent_sound_files, "play-sound": self.play_sound, "volume": self.volume, "file-stabilization-extra-wait": self.file_stabilization_extra_wait, "epochs": self.epochs, "image-dimensions": list(self.image_dimensions), "lion-directories": self.lion_directories, "validation-lion-directories": self.validation_lion_directories, "model-function": self.model_function_name, "model-version": self.model_version, "no-lion-directories": self.no_lion_directories, "validation-no-lion-directories": self.validation_no_lion_directories, "notebook": self.notebook_number, "verification-path": self.verification_path, "with-augmentation": self.with_augmentation, "cameras": self.cameras, "plugs": self.plugs, "device-history": self.device_history, "wifi-networks": self.wifi_networks, "camera-heartbeat-enabled": self.camera_heartbeat_enabled, "camera-heartbeat-interval": self.camera_heartbeat_interval, "camera-heartbeat-method": self.camera_heartbeat_method, "camera-heartbeat-tcp-port": self.camera_heartbeat_tcp_port, "camera-heartbeat-tcp-timeout": self.camera_heartbeat_tcp_timeout, "camera-heartbeat-icmp-timeout": self.camera_heartbeat_icmp_timeout, "device-auto-remove-enabled": self.device_auto_remove_enabled, "device-auto-remove-hours": self.device_auto_remove_hours, "plug-heartbeat-enabled": self.plug_heartbeat_enabled, "plug-heartbeat-interval": self.plug_heartbeat_interval, "plug-heartbeat-timeout": self.plug_heartbeat_timeout, }.items() def __str__(self): """ Serialize this class. """ return yaml.dump(dict(self), indent=2) @property def yolo_min_size(self) -> float: """ Get the YOLO min-size. """ return self._yolo_min_size @yolo_min_size.setter def yolo_min_size(self, yolo_min_size: float): """ Set the YOLO min-size. """ if not isinstance(yolo_min_size, float): raise TypeError( "yolo_min_size needs to be a floating point number" ) if yolo_min_size <= 0 or yolo_min_size > 1: raise ValueError("yolo_min_size needs to be between (0, 1]") self._yolo_min_size = yolo_min_size @property def yolo_conf_thresh(self) -> float: """ Get the YOLO conf-thresh. """ return self._yolo_conf_thresh @yolo_conf_thresh.setter def yolo_conf_thresh(self, yolo_conf_thresh: float): """ Set the YOLO conf-thresh. """ if not isinstance(yolo_conf_thresh, float): raise TypeError( "yolo_conf_thresh needs to be a floating point number" ) if yolo_conf_thresh <= 0 or yolo_conf_thresh > 1: raise ValueError("yolo_conf_thresh needs to be between (0, 1]") self._yolo_conf_thresh = yolo_conf_thresh @property def yolo_max_dets(self) -> int: """ Get the YOLO max-dets. """ return self._yolo_max_dets @yolo_max_dets.setter def yolo_max_dets(self, yolo_max_dets: int): """ Set the YOLO max-dets. """ if not isinstance(yolo_max_dets, int): raise TypeError("yolo_max_dets needs to be a integer") if yolo_max_dets <= 0 or yolo_max_dets > 20: raise ValueError("yolo_max_dets needs to be between (0, 20]") self._yolo_max_dets = yolo_max_dets @property def yolo_model_filename(self) -> str: """ Get the YOLO model filename. """ return self._yolo_model_filename @yolo_model_filename.setter def yolo_model_filename(self, yolo_model_filename: str): """ Set the YOLO model filename. """ if not isinstance(yolo_model_filename, str): raise TypeError("yolo_model_filename needs to be a string") self._yolo_model_filename = yolo_model_filename @property def classifier_model_filename(self) -> str: """ Get the classifier model filename. """ return self._classifier_model_filename @classifier_model_filename.setter def classifier_model_filename(self, classifier_model_filename: str): """ Set the classifier model filename. """ if not isinstance(classifier_model_filename, str): raise TypeError("classifier_model_filename needs to be a string") self._classifier_model_filename = classifier_model_filename @property def puma_threshold(self) -> float: """ Get the puma classification threshold. """ return self._puma_threshold @puma_threshold.setter def puma_threshold(self, puma_threshold: float): """ Set the puma classification threshold. """ if not isinstance(puma_threshold, (float, int)): raise TypeError( "puma_threshold needs to be a floating point number" ) if puma_threshold <= 0 or puma_threshold > 1: raise ValueError("puma_threshold needs to be between (0, 1]") self._puma_threshold = float(puma_threshold) @property def base_output_directory(self) -> str: """ Get the base_output_directory. """ return self._base_output_directory @base_output_directory.setter def base_output_directory(self, path: str): """ Set the base_output_directory. """ self._base_output_directory = path @property def verification_path(self) -> str: """ Get the verification path. """ return self._verification_path @verification_path.setter def verification_path(self, path: str): """ Set the verification path. """ self._verification_path = path @property def notebook_number(self) -> int: """ Get notebook number. """ return ( self._notebook_number if hasattr(self, "_notebook_number") else 0 ) @notebook_number.setter def notebook_number(self, notebook: int): """ Set the notebook number. """ if notebook < 1: raise ValueError( f"notebook can not be zero or negative ({notebook})" ) self._notebook_number = notebook @property def file_stabilization_extra_wait(self) -> float: """ Get extra wait. """ return ( self._file_stabilization_extra_wait if hasattr(self, "_file_stabilization_extra_wait") else 0 ) @file_stabilization_extra_wait.setter def file_stabilization_extra_wait(self, extra_wait: float): """ Set the extra wait. """ if extra_wait < 0: raise ValueError(f"extra_wait can not be negative ({extra_wait})") self._file_stabilization_extra_wait = extra_wait @property def model_version(self) -> str: """ Get the model version name. """ return self._model_version @model_version.setter def model_version(self, model_version: str): """ Set the model version name. """ self._model_version = model_version @property def sound_path(self): """ Get the sound path. """ return self._sound_path @sound_path.setter def sound_path(self, sound_path: str): """ Set the sound path. """ self._sound_path = sound_path @property def deterrent_sound_files(self): """ Get the list of deterrent sound files. """ return self._deterrent_sound_files @deterrent_sound_files.setter def deterrent_sound_files(self, sound_files: list[str]): """ Set the list of deterrent sound files. At least one sound file must be provided. """ if not isinstance(sound_files, list): raise TypeError("deterrent_sound_files must be a list") if not sound_files: raise ValueError("At least one sound file must be provided") if not all(isinstance(f, str) for f in sound_files): raise TypeError("All sound files must be strings") self._deterrent_sound_files = sound_files @property def model_file(self): """ Get the location of the model file. """ if self._model_file != "": return self._model_file return os.path.realpath( f"{self.base_output_directory}/" f"model_weights_{self.notebook_number}" f"_{self.model_version}" f"_{self.image_dimensions[0]}" f"_{self.image_dimensions[1]}" ".keras" ) @model_file.setter def model_file(self, filename: str): """ Set the location of the model file. """ self._model_file = filename @property def history_file(self): """ Get the history file. """ return os.path.realpath( f"{self.base_output_directory}/" f"model_history_{self.notebook_number}" f"_{self.model_version}" f"_{self.image_dimensions[0]}" f"_{self.image_dimensions[1]}" ".pickle" ) @property def settings_file(self) -> str: """ Get the settings file. """ return self._settings_file @settings_file.setter def settings_file(self, filename: str): """ Set the settings file. """ self._settings_file = filename @property def number_color_channels(self) -> int: """ The number of color channels. """ return self._number_color_channels @number_color_channels.setter def number_color_channels(self, channels: int): """ Set the number of color channels. """ if channels not in [1, 3]: raise ValueError(f"illegal number of color channels ({channels})") self._number_color_channels = channels @property def image_dimensions(self) -> Tuple[int, int]: """ Get the image dimensions. """ return self._image_dimensions @image_dimensions.setter def image_dimensions(self, dimensions: Tuple[int, int]): """ Set the image dimensions. """ if not ( isinstance(dimensions, tuple) and len(dimensions) == 2 and all(isinstance(dim, int) for dim in dimensions) ): raise TypeError("image dimensions needs to be a tuple") if not all(x > 0 for x in dimensions): raise ValueError("image dimensions need to be positive") self._image_dimensions = copy.deepcopy(dimensions) @property def epochs(self) -> int: """ The number of epochs. """ return self._epochs @epochs.setter def epochs(self, epochs: int): """ Set the number of epochs. """ if not isinstance(epochs, int): raise TypeError("epochs must be int") if epochs < 1: raise ValueError("epochs needs to be a positive integer") self._epochs = epochs @property def lion_directories(self) -> list[str]: """ The directories containing lion images. """ return self._lion_directories @lion_directories.setter def lion_directories(self, lions: list[str]): """ Set the lion directories. """ self._lion_directories = copy.deepcopy(lions) @property def validation_lion_directories(self) -> list[str]: """ The directories containing lion images for validation. """ return self._validation_lion_directories @validation_lion_directories.setter def validation_lion_directories(self, lions: list[str]): """ Set the lion directories for validation. """ self._validation_lion_directories = copy.deepcopy(lions) @property def no_lion_directories(self) -> list[str]: """ The directories containing no_lion images. """ return self._no_lion_directories @no_lion_directories.setter def no_lion_directories(self, no_lions: list[str]): """ Set the no_lion directories. """ self._no_lion_directories = copy.deepcopy(no_lions) @property def validation_no_lion_directories(self) -> list[str]: """ The directories containing no_lion images for validation. """ return self._validation_no_lion_directories @validation_no_lion_directories.setter def validation_no_lion_directories(self, no_lions: list[str]): """ Set the no_lion directories for validation. """ self._validation_no_lion_directories = copy.deepcopy(no_lions) @property def model_function_name(self) -> str: """ Get the model function name. """ return self._model_function_name @model_function_name.setter def model_function_name(self, name: str): """ Set the model function name. """ self._model_function_name = name @property def with_augmentation(self) -> bool: """ Get whether to augment training data. """ return self._with_augmentation @with_augmentation.setter def with_augmentation(self, with_augmentation: bool): """ Set whether to use augment training data. """ self._with_augmentation = with_augmentation @property def play_sound(self) -> bool: """ Get play-sound. """ return self._play_sound @play_sound.setter def play_sound(self, play_sound: bool): """ Set play-sound. """ if not isinstance(play_sound, bool): raise TypeError("play_sound needs to be a bool") self._play_sound = play_sound @property def volume(self) -> int: """ Get volume level (0-100). """ return self._volume @volume.setter def volume(self, volume: int): """ Set volume level (0-100). """ if not isinstance(volume, int): raise TypeError("volume needs to be an int") if volume < 0 or volume > 100: raise ValueError("volume must be between 0 and 100") self._volume = volume