443 lines
15 KiB
Python
443 lines
15 KiB
Python
"""Modbus device CRUD, readings, metrics, and test-read API (M5-T05).
|
|
|
|
All endpoints are under /api/modbus, require an authenticated session, and
|
|
write endpoints (POST/PATCH/DELETE + test-read) additionally require a
|
|
non-empty X-CSRF-Token header.
|
|
|
|
Deletion safety (red-line):
|
|
DELETE /api/modbus/devices/{uuid} explicitly queries the application
|
|
layer for existing modbus_reading rows before deleting. If any exist
|
|
the endpoint returns 409 and suggests disabling the device instead.
|
|
We do NOT rely on the SQLite FK RESTRICT constraint because SQLite does
|
|
not enforce foreign-key constraints at runtime unless
|
|
PRAGMA foreign_keys=ON is set per connection (and it is not in this
|
|
project — see M5-T02 review note OBS-1).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.api.routes.api.deps import require_csrf, require_session
|
|
from app.dependencies import get_db
|
|
from app.integrations.modbus import driver as modbus_driver
|
|
from app.integrations.modbus.driver import ModbusDriverError
|
|
from app.integrations.modbus.profiles import (
|
|
ProfileNotFoundError,
|
|
decode as decode_profile,
|
|
list_profiles,
|
|
load_profile,
|
|
)
|
|
from app.models.modbus import ModbusDevice, ModbusReading
|
|
from app.schemas.modbus import (
|
|
MetricInfo,
|
|
ModbusDeviceCreate,
|
|
ModbusDeviceListResponse,
|
|
ModbusDeviceResponse,
|
|
ModbusDeviceUpdate,
|
|
ModbusLatestResponse,
|
|
ModbusMetricsResponse,
|
|
ModbusProfilesResponse,
|
|
ModbusReadingResponse,
|
|
ModbusReadingsResponse,
|
|
ModbusTestReadResponse,
|
|
ProfileSummary,
|
|
)
|
|
from app.services.auth import AuthenticatedSession
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(prefix="/api/modbus", tags=["api-modbus"])
|
|
|
|
# Maximum number of readings that can be returned per request.
|
|
_READINGS_LIMIT_MAX = 5000
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _get_device_or_404(db: Session, uuid: str) -> ModbusDevice:
|
|
"""Return the device with the given UUID or raise 404."""
|
|
device = db.execute(
|
|
select(ModbusDevice).where(ModbusDevice.uuid == uuid)
|
|
).scalar_one_or_none()
|
|
if device is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Device '{uuid}' not found.",
|
|
)
|
|
return device
|
|
|
|
|
|
def _label_from_key(key: str) -> str:
|
|
"""Derive a human-readable label from a metric key.
|
|
|
|
Example: ``"active_power"`` → ``"Active Power"``
|
|
"""
|
|
return key.replace("_", " ").title()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /api/modbus/profiles
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.get("/profiles", response_model=ModbusProfilesResponse)
|
|
def get_profiles(
|
|
_auth: AuthenticatedSession = Depends(require_session),
|
|
) -> ModbusProfilesResponse:
|
|
"""List all available Modbus YAML profiles (name + description).
|
|
|
|
Intended for the front-end's device-creation profile drop-down.
|
|
"""
|
|
pairs = list_profiles()
|
|
return ModbusProfilesResponse(
|
|
profiles=[ProfileSummary(name=name, description=desc) for name, desc in pairs]
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Device CRUD
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.get("/devices", response_model=ModbusDeviceListResponse)
|
|
def list_devices(
|
|
db: Session = Depends(get_db),
|
|
_auth: AuthenticatedSession = Depends(require_session),
|
|
) -> ModbusDeviceListResponse:
|
|
"""Return all Modbus devices (no pagination — device counts stay small)."""
|
|
devices = db.execute(select(ModbusDevice)).scalars().all()
|
|
items = [ModbusDeviceResponse.model_validate(d) for d in devices]
|
|
return ModbusDeviceListResponse(items=items, total=len(items))
|
|
|
|
|
|
@router.post(
|
|
"/devices",
|
|
response_model=ModbusDeviceResponse,
|
|
status_code=status.HTTP_201_CREATED,
|
|
)
|
|
def create_device(
|
|
body: ModbusDeviceCreate,
|
|
db: Session = Depends(get_db),
|
|
_auth: AuthenticatedSession = Depends(require_session),
|
|
_csrf: None = Depends(require_csrf),
|
|
) -> ModbusDeviceResponse:
|
|
"""Create a new Modbus device.
|
|
|
|
- Validates that the referenced ``profile`` exists; returns 422 if not.
|
|
- Returns 201 with the created device on success.
|
|
"""
|
|
# Validate profile exists (422 if not)
|
|
try:
|
|
load_profile(body.profile)
|
|
except ProfileNotFoundError:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail=f"Profile '{body.profile}' does not exist.",
|
|
)
|
|
|
|
now = datetime.now(UTC)
|
|
device = ModbusDevice(
|
|
friendly_name=body.friendly_name,
|
|
transport=body.transport,
|
|
host=body.host,
|
|
port=body.port,
|
|
unit_id=body.unit_id,
|
|
profile=body.profile,
|
|
poll_interval_s=body.poll_interval_s,
|
|
enabled=body.enabled,
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
db.add(device)
|
|
db.commit()
|
|
db.refresh(device)
|
|
logger.info("Created Modbus device %r (uuid=%s)", device.friendly_name, device.uuid)
|
|
return ModbusDeviceResponse.model_validate(device)
|
|
|
|
|
|
@router.get("/devices/{uuid}", response_model=ModbusDeviceResponse)
|
|
def get_device(
|
|
uuid: str,
|
|
db: Session = Depends(get_db),
|
|
_auth: AuthenticatedSession = Depends(require_session),
|
|
) -> ModbusDeviceResponse:
|
|
"""Return a single Modbus device by UUID."""
|
|
device = _get_device_or_404(db, uuid)
|
|
return ModbusDeviceResponse.model_validate(device)
|
|
|
|
|
|
@router.patch("/devices/{uuid}", response_model=ModbusDeviceResponse)
|
|
def patch_device(
|
|
uuid: str,
|
|
body: ModbusDeviceUpdate,
|
|
db: Session = Depends(get_db),
|
|
_auth: AuthenticatedSession = Depends(require_session),
|
|
_csrf: None = Depends(require_csrf),
|
|
) -> ModbusDeviceResponse:
|
|
"""Partially update a Modbus device (including enable/disable).
|
|
|
|
Only fields explicitly provided in the request body are updated.
|
|
Providing ``profile`` triggers a profile-existence check (422 if unknown).
|
|
"""
|
|
device = _get_device_or_404(db, uuid)
|
|
|
|
update_data = body.model_dump(exclude_none=True)
|
|
|
|
# If profile is being changed, validate it exists.
|
|
if "profile" in update_data:
|
|
try:
|
|
load_profile(update_data["profile"])
|
|
except ProfileNotFoundError:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail=f"Profile '{update_data['profile']}' does not exist.",
|
|
)
|
|
|
|
for field, value in update_data.items():
|
|
setattr(device, field, value)
|
|
|
|
device.updated_at = datetime.now(UTC)
|
|
db.commit()
|
|
db.refresh(device)
|
|
logger.info("Updated Modbus device %r (uuid=%s)", device.friendly_name, device.uuid)
|
|
return ModbusDeviceResponse.model_validate(device)
|
|
|
|
|
|
@router.delete(
|
|
"/devices/{uuid}",
|
|
status_code=status.HTTP_204_NO_CONTENT,
|
|
response_model=None,
|
|
)
|
|
def delete_device(
|
|
uuid: str,
|
|
db: Session = Depends(get_db),
|
|
_auth: AuthenticatedSession = Depends(require_session),
|
|
_csrf: None = Depends(require_csrf),
|
|
) -> None:
|
|
"""Delete a Modbus device.
|
|
|
|
Returns 409 Conflict if the device has any associated readings; use
|
|
``enabled=false`` to disable it instead.
|
|
|
|
**Application-layer safety**: the check is performed via an explicit
|
|
SELECT COUNT query — not by relying on SQLite's FK RESTRICT constraint,
|
|
which is not enforced at runtime in this project (no PRAGMA foreign_keys=ON).
|
|
"""
|
|
device = _get_device_or_404(db, uuid)
|
|
|
|
# Application-layer guard: refuse deletion if any readings exist.
|
|
# We do NOT rely on SQLite FK RESTRICT — it is not enforced at runtime
|
|
# without PRAGMA foreign_keys=ON, which is not set in this project.
|
|
reading_count: int = db.execute(
|
|
select(func.count(ModbusReading.id)).where(ModbusReading.device_id == device.id)
|
|
).scalar_one()
|
|
|
|
if reading_count > 0:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail=(
|
|
f"Device '{uuid}' has {reading_count} associated reading(s) and cannot be "
|
|
"deleted. Disable the device (set enabled=false) instead to stop polling "
|
|
"without losing historical data."
|
|
),
|
|
)
|
|
|
|
logger.info("Deleting Modbus device %r (uuid=%s)", device.friendly_name, uuid)
|
|
db.delete(device)
|
|
db.commit()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Readings endpoints
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.get("/devices/{uuid}/latest", response_model=ModbusLatestResponse)
|
|
def get_latest_reading(
|
|
uuid: str,
|
|
db: Session = Depends(get_db),
|
|
_auth: AuthenticatedSession = Depends(require_session),
|
|
) -> ModbusLatestResponse:
|
|
"""Return the most recent reading for a device.
|
|
|
|
If no readings exist yet, returns ``{"found": false, "recorded_at": null,
|
|
"payload": null}`` (200, not 404) so the front-end can distinguish
|
|
"device exists but has no data" from "device not found".
|
|
"""
|
|
device = _get_device_or_404(db, uuid)
|
|
|
|
row = db.execute(
|
|
select(ModbusReading)
|
|
.where(ModbusReading.device_id == device.id)
|
|
.order_by(ModbusReading.recorded_at.desc())
|
|
.limit(1)
|
|
).scalar_one_or_none()
|
|
|
|
if row is None:
|
|
return ModbusLatestResponse(found=False, recorded_at=None, payload=None)
|
|
|
|
return ModbusLatestResponse(
|
|
found=True,
|
|
recorded_at=row.recorded_at,
|
|
payload=row.payload,
|
|
)
|
|
|
|
|
|
@router.get("/devices/{uuid}/readings", response_model=ModbusReadingsResponse)
|
|
def get_readings(
|
|
uuid: str,
|
|
start: datetime | None = Query(default=None),
|
|
end: datetime | None = Query(default=None),
|
|
limit: int = Query(default=500, ge=1, le=_READINGS_LIMIT_MAX),
|
|
db: Session = Depends(get_db),
|
|
_auth: AuthenticatedSession = Depends(require_session),
|
|
) -> ModbusReadingsResponse:
|
|
"""Return time-range readings for a device.
|
|
|
|
When the window contains more rows than ``limit``, the **most recent** N rows
|
|
are returned (``ORDER BY recorded_at DESC LIMIT n``), then reversed to
|
|
ascending order before being sent to the client. This ensures that for long
|
|
time-range requests (e.g. 6 h / 24 h) the caller always sees the latest data
|
|
rather than the oldest segment of the window.
|
|
|
|
When the window has fewer rows than ``limit`` the full window is returned in
|
|
ascending order — behaviour is identical to a plain ascending query.
|
|
|
|
The response schema is unchanged: items are always ``recorded_at`` ascending.
|
|
|
|
The query uses the ``(device_id, recorded_at)`` composite index for
|
|
efficient time-window scans.
|
|
|
|
Query parameters:
|
|
- ``start``: inclusive lower bound (ISO8601 datetime)
|
|
- ``end``: inclusive upper bound (ISO8601 datetime)
|
|
- ``limit``: max rows to return (default 500, max {_READINGS_LIMIT_MAX})
|
|
"""
|
|
device = _get_device_or_404(db, uuid)
|
|
|
|
stmt = (
|
|
select(ModbusReading)
|
|
.where(ModbusReading.device_id == device.id)
|
|
.order_by(ModbusReading.recorded_at.desc())
|
|
.limit(limit)
|
|
)
|
|
|
|
if start is not None:
|
|
stmt = stmt.where(ModbusReading.recorded_at >= start)
|
|
if end is not None:
|
|
stmt = stmt.where(ModbusReading.recorded_at <= end)
|
|
|
|
# Fetch most-recent N rows, then reverse to restore ascending order for the response.
|
|
rows = list(reversed(db.execute(stmt).scalars().all()))
|
|
items = [ModbusReadingResponse(recorded_at=r.recorded_at, payload=r.payload) for r in rows]
|
|
return ModbusReadingsResponse(items=items)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Metrics endpoint
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.get("/devices/{uuid}/metrics", response_model=ModbusMetricsResponse)
|
|
def get_metrics(
|
|
uuid: str,
|
|
db: Session = Depends(get_db),
|
|
_auth: AuthenticatedSession = Depends(require_session),
|
|
) -> ModbusMetricsResponse:
|
|
"""Return the metric catalogue for a device's profile.
|
|
|
|
Each entry has ``key``, ``label`` (derived from key if the profile has
|
|
none — underscores → spaces, title-cased), ``unit``, and ``device_class``.
|
|
This is the authoritative metadata source for front-end card labels and
|
|
chart axis labels.
|
|
"""
|
|
device = _get_device_or_404(db, uuid)
|
|
|
|
try:
|
|
profile = load_profile(device.profile)
|
|
except ProfileNotFoundError:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail=f"Device profile '{device.profile}' could not be loaded.",
|
|
)
|
|
|
|
metrics = [
|
|
MetricInfo(
|
|
key=m.key,
|
|
label=_label_from_key(m.key),
|
|
unit=m.unit,
|
|
device_class=m.device_class,
|
|
)
|
|
for m in profile.metrics
|
|
]
|
|
|
|
return ModbusMetricsResponse(profile=profile.name, metrics=metrics)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Test-read endpoint
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.post("/devices/{uuid}/test", response_model=ModbusTestReadResponse)
|
|
def test_read(
|
|
uuid: str,
|
|
db: Session = Depends(get_db),
|
|
_auth: AuthenticatedSession = Depends(require_session),
|
|
_csrf: None = Depends(require_csrf),
|
|
) -> ModbusTestReadResponse:
|
|
"""Immediately read and decode the device once without persisting any data.
|
|
|
|
Useful for validating gateway connectivity and Modbus address settings
|
|
before relying on the background polling job.
|
|
|
|
This is a write-class endpoint (it initiates network I/O on demand) and
|
|
therefore requires a CSRF token. The response payload is never stored.
|
|
"""
|
|
device = _get_device_or_404(db, uuid)
|
|
|
|
try:
|
|
profile = load_profile(device.profile)
|
|
except ProfileNotFoundError:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
|
detail=f"Device profile '{device.profile}' could not be loaded.",
|
|
)
|
|
|
|
try:
|
|
registers = modbus_driver.read_blocks(
|
|
device.host,
|
|
device.port,
|
|
device.unit_id,
|
|
[{"start": b.start, "count": b.count} for b in profile.blocks],
|
|
)
|
|
payload: dict[str, Any] = decode_profile(profile, registers)
|
|
return ModbusTestReadResponse(ok=True, payload=payload)
|
|
|
|
except ModbusDriverError as exc:
|
|
logger.warning(
|
|
"Test read failed for device %r (uuid=%s): %s",
|
|
device.friendly_name,
|
|
uuid,
|
|
exc,
|
|
)
|
|
return ModbusTestReadResponse(ok=False, error=str(exc))
|
|
except Exception as exc: # noqa: BLE001
|
|
logger.warning(
|
|
"Unexpected error during test read for device %r (uuid=%s): %s",
|
|
device.friendly_name,
|
|
uuid,
|
|
exc,
|
|
)
|
|
return ModbusTestReadResponse(ok=False, error=f"Unexpected error: {exc}")
|