pi-pm3/app/backend/services/hardware_service.py

"""Hardware service for plugin hardware access.

Provides controlled access to hardware interfaces (I2C, SPI, GPIO, Serial)
with permission checking and logging.
"""
import logging
from typing import Optional, Any

logger = logging.getLogger(__name__)


class HardwareService:
    """Service for controlled hardware access.

    Provides safe wrappers for hardware interfaces that:
    - Log all access attempts
    - Handle import errors gracefully (for non-Pi systems)
    - Return None if hardware is unavailable

    Permission checking is done by PluginBase before calling these methods.
    """

    # Track which plugins have accessed which hardware
    _access_log: dict[str, list[str]] = {}

    @classmethod
    def _log_access(cls, plugin_id: str, hardware_type: str) -> None:
        """Log hardware access for auditing.

        Args:
            plugin_id: Plugin requesting access
            hardware_type: Type of hardware (i2c, gpio, spi, serial)
        """
        if plugin_id not in cls._access_log:
            cls._access_log[plugin_id] = []
        if hardware_type not in cls._access_log[plugin_id]:
            cls._access_log[plugin_id].append(hardware_type)
            logger.info(f"Plugin '{plugin_id}' accessed {hardware_type}")

    @classmethod
    def get_access_log(cls) -> dict[str, list[str]]:
        """Get the hardware access log.

        Returns:
            Dictionary mapping plugin IDs to list of accessed hardware types
        """
        return cls._access_log.copy()

    @staticmethod
    def get_i2c_bus(plugin_id: str, bus: int = 1) -> Optional[Any]:
        """Get I2C bus for a plugin.

        Args:
            plugin_id: Plugin requesting access (for logging)
            bus: I2C bus number (default: 1 for Raspberry Pi)

        Returns:
            SMBus instance or None if unavailable
        """
        HardwareService._log_access(plugin_id, "i2c")

        try:
            from smbus2 import SMBus
            return SMBus(bus)
        except ImportError:
            logger.warning("smbus2 not available - I2C access disabled")
            return None
        except Exception as e:
            logger.error(f"Failed to open I2C bus {bus}: {e}")
            return None

    @staticmethod
    def get_gpio(plugin_id: str) -> Optional[Any]:
        """Get GPIO access for a plugin.

        Returns the RPi.GPIO module if available.
        Plugin is responsible for setup/cleanup.

        Args:
            plugin_id: Plugin requesting access (for logging)

        Returns:
            GPIO module or None if unavailable
        """
        HardwareService._log_access(plugin_id, "gpio")

        try:
            import RPi.GPIO as GPIO
            return GPIO
        except ImportError:
            # Try gpiozero as fallback
            try:
                import gpiozero
                logger.info("Using gpiozero instead of RPi.GPIO")
                return gpiozero
            except ImportError:
                logger.warning("No GPIO library available (RPi.GPIO or gpiozero)")
                return None
        except Exception as e:
            logger.error(f"Failed to access GPIO: {e}")
            return None

    @staticmethod
    def get_spi(plugin_id: str, bus: int = 0, device: int = 0) -> Optional[Any]:
        """Get SPI device for a plugin.

        Args:
            plugin_id: Plugin requesting access (for logging)
            bus: SPI bus number (default: 0)
            device: SPI device/chip select (default: 0)

        Returns:
            SpiDev instance or None if unavailable
        """
        HardwareService._log_access(plugin_id, "spi")

        try:
            import spidev
            spi = spidev.SpiDev()
            spi.open(bus, device)
            return spi
        except ImportError:
            logger.warning("spidev not available - SPI access disabled")
            return None
        except Exception as e:
            logger.error(f"Failed to open SPI bus {bus} device {device}: {e}")
            return None

    @staticmethod
    def get_serial(
        plugin_id: str,
        port: str,
        baudrate: int = 9600,
        timeout: float = 1.0
    ) -> Optional[Any]:
        """Get serial port for a plugin.

        Args:
            plugin_id: Plugin requesting access (for logging)
            port: Serial port path (e.g., '/dev/ttyUSB0')
            baudrate: Baud rate (default: 9600)
            timeout: Read timeout in seconds (default: 1.0)

        Returns:
            Serial instance or None if unavailable
        """
        HardwareService._log_access(plugin_id, "serial")

        try:
            import serial
            return serial.Serial(port, baudrate, timeout=timeout)
        except ImportError:
            logger.warning("pyserial not available - serial access disabled")
            return None
        except Exception as e:
            logger.error(f"Failed to open serial port {port}: {e}")
            return None

    @staticmethod
    def is_hardware_available(hardware_type: str) -> bool:
        """Check if a hardware type is available on this system.

        Args:
            hardware_type: One of 'i2c', 'gpio', 'spi', 'serial'

        Returns:
            True if the hardware library is importable
        """
        try:
            if hardware_type == "i2c":
                from smbus2 import SMBus
                return True
            elif hardware_type == "gpio":
                try:
                    import RPi.GPIO
                    return True
                except ImportError:
                    import gpiozero
                    return True
            elif hardware_type == "spi":
                import spidev
                return True
            elif hardware_type == "serial":
                import serial
                return True
            else:
                return False
        except ImportError:
            return False

    @staticmethod
    def get_available_hardware() -> list[str]:
        """Get list of available hardware types.

        Returns:
            List of available hardware type names
        """
        available = []
        for hw_type in ["i2c", "gpio", "spi", "serial"]:
            if HardwareService.is_hardware_available(hw_type):
                available.append(hw_type)
        return available