"""Philips Hue Bridge Pro client with HomeAware movement detection integration.
HomeAware uses RF signal strength analysis of Zigbee network to detect movement
without cameras, microphones, or PIR sensors - providing privacy-focused motion detection.
"""
import asyncio
from dataclasses import dataclass
from typing import Any
import structlog
from phue import Bridge
from ..utils.response_builders import (
build_robotics_error_response,
build_success_response,
)
logger = structlog.get_logger(__name__)
@dataclass
class MovementEvent:
"""HomeAware movement detection event."""
timestamp: float
location: str
confidence: float
duration: float | None = None
intensity: float | None = None
sensor_id: str | None = None
@dataclass
class HomeAwareSensor:
"""HomeAware sensor information."""
sensor_id: str
name: str
location: str
sensitivity: float
enabled: bool
last_movement: float | None = None
movement_count: int = 0
class HueBridgeProClient:
"""Philips Hue Bridge Pro client with HomeAware integration."""
def __init__(self, bridge_ip: str, api_key: str | None = None):
"""Initialize Hue Bridge Pro client.
Args:
bridge_ip: IP address of the Hue Bridge Pro
api_key: API key for bridge authentication (optional, will be generated if needed)
"""
self.bridge_ip = bridge_ip
self.api_key = api_key
self.bridge: Bridge | None = None
self._connected = False
self.homeaware_sensors: dict[str, HomeAwareSensor] = {}
self.movement_events: list[MovementEvent] = []
self._last_update = 0
async def connect(self) -> bool:
"""Connect to Hue Bridge Pro and initialize HomeAware sensors."""
try:
logger.info("Connecting to Hue Bridge Pro", bridge_ip=self.bridge_ip)
# Create bridge connection (phue handles API key automatically)
self.bridge = Bridge(self.bridge_ip)
# First connection attempt - this will prompt for bridge button press if needed
self.bridge.connect()
self._connected = True
# Get API key from bridge object
self.api_key = getattr(self.bridge, "username", None)
# Discover HomeAware sensors
await self._discover_homeaware_sensors()
logger.info("Connected to Hue Bridge Pro", sensors=len(self.homeaware_sensors))
return True
except Exception as e:
logger.error(
"Failed to connect to Hue Bridge Pro", bridge_ip=self.bridge_ip, error=str(e)
)
return False
async def _discover_homeaware_sensors(self):
"""Discover and initialize HomeAware sensors."""
if not self.bridge:
return
try:
# Get all sensors from bridge
sensors = self.bridge.get_sensor_objects()
for sensor_id, sensor_obj in sensors.items():
sensor_data = sensor_obj.__dict__
# Check if this is a HomeAware sensor
if self._is_homeaware_sensor(sensor_data):
homeaware_sensor = HomeAwareSensor(
sensor_id=str(sensor_id),
name=sensor_data.get("name", f"Sensor {sensor_id}"),
location=self._extract_location(sensor_data),
sensitivity=self._extract_sensitivity(sensor_data),
enabled=sensor_data.get("config", {}).get("on", True),
)
self.homeaware_sensors[str(sensor_id)] = homeaware_sensor
logger.info(
"Discovered HomeAware sensor",
sensor_id=sensor_id,
name=homeaware_sensor.name,
)
except Exception as e:
logger.error("Failed to discover HomeAware sensors", error=str(e))
def _is_homeaware_sensor(self, sensor_data: dict[str, Any]) -> bool:
"""Check if sensor data represents a HomeAware sensor."""
# HomeAware sensors have specific characteristics
sensor_type = sensor_data.get("type", "")
model_id = sensor_data.get("modelid", "")
# HomeAware sensors typically have specific identifiers
return (
"HomeAware" in sensor_type
or "homeaware" in model_id.lower()
or "movement" in sensor_type.lower()
)
def _extract_location(self, sensor_data: dict[str, Any]) -> str:
"""Extract location information from sensor data."""
# Try different location fields
location = sensor_data.get("config", {}).get("location", "")
if not location:
location = sensor_data.get("name", "Unknown")
return location
def _extract_sensitivity(self, sensor_data: dict[str, Any]) -> float:
"""Extract sensitivity setting from sensor data."""
config = sensor_data.get("config", {})
sensitivity = config.get("sensitivity", 1.0)
return float(sensitivity)
async def get_movement_events(
self, since_timestamp: float | None = None
) -> list[MovementEvent]:
"""Get recent movement detection events.
Args:
since_timestamp: Only return events after this timestamp
Returns:
List of movement events
"""
if not self.bridge:
return []
try:
# Get updated sensor data
sensors = self.bridge.get_sensor_objects()
events = []
for sensor_id, sensor_obj in sensors.items():
sensor_data = sensor_obj.__dict__
if self._is_homeaware_sensor(sensor_data):
# Check for movement detection
state = sensor_data.get("state", {})
if state.get("presence", False):
# Movement detected
event = MovementEvent(
timestamp=asyncio.get_event_loop().time(),
location=self._extract_location(sensor_data),
confidence=state.get("presence_confidence", 1.0),
sensor_id=str(sensor_id),
)
events.append(event)
# Update sensor tracking
if str(sensor_id) in self.homeaware_sensors:
sensor = self.homeaware_sensors[str(sensor_id)]
sensor.last_movement = event.timestamp
sensor.movement_count += 1
# Filter by timestamp if requested
if since_timestamp:
events = [e for e in events if e.timestamp > since_timestamp]
self.movement_events.extend(events)
return events
except Exception as e:
logger.error("Failed to get movement events", error=str(e))
return []
async def get_sensor_status(self) -> dict[str, Any]:
"""Get comprehensive status of all HomeAware sensors."""
if not self.bridge:
return {"error": "Not connected to Hue Bridge"}
try:
sensors_status = {}
sensors = self.bridge.get_sensor_objects()
for sensor_id, sensor_obj in sensors.items():
sensor_data = sensor_obj.__dict__
if self._is_homeaware_sensor(sensor_data):
status = {
"name": sensor_data.get("name", f"Sensor {sensor_id}"),
"location": self._extract_location(sensor_data),
"enabled": sensor_data.get("config", {}).get("on", True),
"sensitivity": self._extract_sensitivity(sensor_data),
"battery": sensor_data.get("config", {}).get("battery", None),
"reachable": sensor_data.get("config", {}).get("reachable", True),
"presence": sensor_data.get("state", {}).get("presence", False),
"last_updated": sensor_data.get("state", {}).get("lastupdated", None),
}
sensors_status[str(sensor_id)] = status
return {
"bridge_connected": True,
"sensors": sensors_status,
"total_sensors": len(sensors_status),
"timestamp": asyncio.get_event_loop().time(),
}
except Exception as e:
logger.error("Failed to get sensor status", error=str(e))
return {"error": str(e), "bridge_connected": False}
async def configure_sensor(
self, sensor_id: str, sensitivity: float | None = None, enabled: bool | None = None
) -> bool:
"""Configure HomeAware sensor settings.
Args:
sensor_id: Sensor identifier
sensitivity: Movement detection sensitivity (0.0-1.0)
enabled: Whether sensor should be enabled
Returns:
Success status
"""
if not self.bridge:
return False
try:
# Update local sensor tracking
if sensor_id in self.homeaware_sensors:
sensor = self.homeaware_sensors[sensor_id]
if sensitivity is not None:
sensor.sensitivity = sensitivity
if enabled is not None:
sensor.enabled = enabled
# Note: Actual bridge configuration would require additional API calls
# This is a placeholder for future implementation
logger.info(
"Sensor configuration updated",
sensor_id=sensor_id,
sensitivity=sensitivity,
enabled=enabled,
)
return True
except Exception as e:
logger.error("Failed to configure sensor", sensor_id=sensor_id, error=str(e))
return False
async def get_movement_zones(self) -> dict[str, Any]:
"""Get movement detection zones and coverage areas."""
# HomeAware provides room-level detection rather than specific zones
# This returns logical zones based on sensor locations
zones = {}
for sensor_id, sensor in self.homeaware_sensors.items():
if sensor.enabled:
zones[sensor.location] = {
"sensor_id": sensor_id,
"coverage_area": "room_level", # HomeAware covers entire rooms
"detection_method": "rf_signal_analysis",
"last_activity": sensor.last_movement,
"activity_count": sensor.movement_count,
}
return {
"zones": zones,
"detection_type": "rf_based_movement",
"coverage": "room_level",
"privacy_level": "high", # No cameras, microphones, or PIR
}
# Global client instance
_hue_client: HueBridgeProClient | None = None
async def get_hue_client(
bridge_ip: str = "192.168.1.100", api_key: str | None = None
) -> HueBridgeProClient:
"""Get or create Hue Bridge Pro client instance."""
global _hue_client
if _hue_client is None:
_hue_client = HueBridgeProClient(bridge_ip, api_key)
await _hue_client.connect()
return _hue_client
async def hue_get_movement_events(robot_id: str = "hue_bridge") -> dict[str, Any]:
"""Get HomeAware movement detection events."""
try:
# Get client (will be created if needed)
client = await get_hue_client()
events = await client.get_movement_events()
return build_success_response(
operation="hue_get_movement_events",
summary=f"Hue HomeAware detected {len(events)} movement events",
result={
"robot_id": robot_id,
"movement_events": [
{
"timestamp": event.timestamp,
"location": event.location,
"confidence": event.confidence,
"sensor_id": event.sensor_id,
}
for event in events
],
"total_events": len(events),
},
)
except Exception as e:
return build_robotics_error_response(
error=f"Failed to get Hue movement events: {str(e)}",
sensor_type="hue_homeaware",
robot_id=robot_id,
)
async def hue_get_sensor_status(robot_id: str = "hue_bridge") -> dict[str, Any]:
"""Get HomeAware sensor status."""
try:
client = await get_hue_client()
status = await client.get_sensor_status()
return build_success_response(
operation="hue_get_sensor_status",
summary=f"Hue HomeAware status retrieved ({status.get('total_sensors', 0)} sensors)",
result={"robot_id": robot_id, "status": status},
)
except Exception as e:
return build_robotics_error_response(
error=f"Failed to get Hue sensor status: {str(e)}",
sensor_type="hue_homeaware",
robot_id=robot_id,
)
async def hue_get_movement_zones(robot_id: str = "hue_bridge") -> dict[str, Any]:
"""Get movement detection zones."""
try:
client = await get_hue_client()
zones = await client.get_movement_zones()
return build_success_response(
operation="hue_get_movement_zones",
summary=f"Hue HomeAware zones retrieved ({len(zones.get('zones', {}))} zones)",
result={"robot_id": robot_id, "zones": zones},
)
except Exception as e:
return build_robotics_error_response(
error=f"Failed to get Hue movement zones: {str(e)}",
sensor_type="hue_homeaware",
robot_id=robot_id,
)
