feat: device registry with CRUD, state tracking, command sending, and log management

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

src/mqtt_home/device_registry.py

@@ -0,0 +1,147 @@
+import json
+import logging
+from datetime import datetime, timezone
+from typing import Optional
+
+from sqlalchemy import select, func, desc
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from mqtt_home.models import Device, DeviceLog
+from mqtt_home.schemas import DeviceCreate, DeviceUpdate
+
+logger = logging.getLogger(__name__)
+
+MAX_LOGS_PER_DEVICE = 100
+
+
+async def list_devices(db: AsyncSession) -> list[Device]:
+    result = await db.execute(select(Device).order_by(Device.created_at.desc()))
+    return list(result.scalars().all())
+
+
+async def get_device(db: AsyncSession, device_id: str) -> Optional[Device]:
+    return await db.get(Device, device_id)
+
+
+async def create_device(db: AsyncSession, data: DeviceCreate) -> Device:
+    device = Device(
+        name=data.name,
+        type=data.type,
+        protocol=data.protocol,
+        mqtt_topic=data.mqtt_topic,
+        command_topic=data.command_topic,
+    )
+    db.add(device)
+    await db.commit()
+    await db.refresh(device)
+    logger.info("Device created: %s (%s)", device.name, device.id)
+    return device
+
+
+async def update_device(db: AsyncSession, device_id: str, data: DeviceUpdate) -> Optional[Device]:
+    device = await db.get(Device, device_id)
+    if not device:
+        return None
+    update_data = data.model_dump(exclude_unset=True)
+    for key, value in update_data.items():
+        setattr(device, key, value)
+    await db.commit()
+    await db.refresh(device)
+    return device
+
+
+async def delete_device(db: AsyncSession, device_id: str) -> bool:
+    device = await db.get(Device, device_id)
+    if not device:
+        return False
+    await db.delete(device)
+    await db.commit()
+    logger.info("Device deleted: %s", device_id)
+    return True
+
+
+async def send_command(db: AsyncSession, device_id: str, payload: str, publish_fn=None) -> Optional[DeviceLog]:
+    device = await db.get(Device, device_id)
+    if not device:
+        return None
+    if not device.command_topic:
+        raise ValueError(f"Device {device_id} has no command_topic configured")
+
+    log = DeviceLog(
+        device_id=device_id,
+        direction="tx",
+        topic=device.command_topic,
+        payload=payload,
+    )
+    db.add(log)
+
+    if publish_fn:
+        await publish_fn(device.command_topic, payload)
+
+    await db.commit()
+    await db.refresh(log)
+    logger.info("Command sent to %s: %s", device.command_topic, payload)
+    return log
+
+
+async def handle_state_update(db: AsyncSession, topic: str, payload: str) -> Optional[Device]:
+    result = await db.execute(select(Device).where(Device.mqtt_topic == topic))
+    device = result.scalar_one_or_none()
+    if not device:
+        return None
+
+    device.state = payload
+    device.last_seen = datetime.now(timezone.utc)
+    device.is_online = True
+
+    log = DeviceLog(
+        device_id=device.id,
+        direction="rx",
+        topic=topic,
+        payload=payload,
+    )
+    db.add(log)
+
+    # Clean old logs, keep MAX_LOGS_PER_DEVICE
+    count_result = await db.execute(
+        select(func.count()).select_from(DeviceLog).where(DeviceLog.device_id == device.id)
+    )
+    count = count_result.scalar() or 0
+    if count >= MAX_LOGS_PER_DEVICE:
+        oldest = await db.execute(
+            select(DeviceLog).where(DeviceLog.device_id == device.id)
+            .order_by(DeviceLog.timestamp.asc())
+            .limit(count - MAX_LOGS_PER_DEVICE + 1)
+        )
+        for old_log in oldest.scalars().all():
+            await db.delete(old_log)
+
+    await db.commit()
+    await db.refresh(device)
+    return device
+
+
+async def get_device_logs(db: AsyncSession, device_id: str, limit: int = 20) -> list[DeviceLog]:
+    result = await db.execute(
+        select(DeviceLog)
+        .where(DeviceLog.device_id == device_id)
+        .order_by(desc(DeviceLog.timestamp))
+        .limit(limit)
+    )
+    return list(result.scalars().all())
+
+
+async def get_dashboard_stats(db: AsyncSession) -> dict:
+    total_result = await db.execute(select(func.count()).select_from(Device))
+    total = total_result.scalar() or 0
+    online_result = await db.execute(select(func.count()).select_from(Device).where(Device.is_online == True))
+    online = online_result.scalar() or 0
+    recent_logs_result = await db.execute(
+        select(DeviceLog).order_by(desc(DeviceLog.timestamp)).limit(10)
+    )
+    return {
+        "total_devices": total,
+        "online_devices": online,
+        "offline_devices": total - online,
+        "recent_logs": list(recent_logs_result.scalars().all()),
+    }