"""Modbus polling service — periodic read-and-store for enabled Modbus devices. Design notes ------------ - ``poll_device`` loads the device's YAML profile, calls the driver to bulk-read all register blocks in a single TCP connection, decodes the raw registers into an engineering-value dict via the profile, and persists one ``ModbusReading`` row. It also stamps ``last_poll_at`` / ``last_poll_ok`` on the device. - All exceptions are caught and logged inside ``poll_device``; the function never re-raises. This prevents a single broken device from aborting the sweep for all other devices. - ``poll_all_enabled_devices`` respects per-device ``poll_interval_s``: it skips a device whose ``last_poll_at`` is recent enough. This lets the APScheduler job run on a short base tick (e.g. 5 s) while each device self-regulates its own cadence. - The global ``modbus_polling_enabled`` flag from ``Settings`` (merged with any DB overrides via ``build_runtime_settings``) gates the entire sweep as a no-op. T08 will expose this flag in CONFIG_FIELDS/UI; for now it is controlled by the ``MODBUS_POLLING_ENABLED`` env var (or the default True). """ from __future__ import annotations import logging from datetime import UTC, datetime from sqlalchemy import select from sqlalchemy.orm import Session from app.config import Settings from app.integrations.modbus import driver, profiles from app.models.modbus import ModbusDevice, ModbusReading from app.services.config_page import build_runtime_settings logger = logging.getLogger(__name__) # Base polling tick in seconds (used by APScheduler; each device checks its own interval). BASE_POLL_TICK_SECONDS = 5 def poll_device(session: Session, device: ModbusDevice) -> ModbusReading | None: """Read one enabled Modbus device, decode, and persist a reading row. Parameters ---------- session: Open SQLAlchemy session. The caller is responsible for session lifecycle (open / commit / close). device: ``ModbusDevice`` ORM instance to poll. Returns ------- ModbusReading | None The newly-persisted reading on success, or ``None`` on any failure. Failure is logged and the device's ``last_poll_ok`` is set to ``False``. """ now = datetime.now(UTC) try: profile = profiles.load_profile(device.profile) registers = driver.read_blocks( device.host, device.port, device.unit_id, [{"start": b.start, "count": b.count} for b in profile.blocks], ) payload = profiles.decode(profile, registers) reading = ModbusReading( device_id=device.id, recorded_at=now, payload=payload, ) session.add(reading) device.last_poll_at = now device.last_poll_ok = True session.commit() logger.debug( "Polled device %r (id=%d): %d metrics recorded at %s", device.friendly_name, device.id, len(payload), now.isoformat(), ) return reading except Exception: # noqa: BLE001 logger.exception( "Failed to poll device %r (id=%d, host=%s:%d, unit_id=%d)", device.friendly_name, device.id, device.host, device.port, device.unit_id, ) # Rollback first — if session.add(reading) already ran in the success # branch before commit() failed, that pending ModbusReading must be # discarded. Without rollback it would remain staged and could be # flushed/committed by a subsequent operation (even for a different # device), producing a spurious reading row for a poll that failed. # Rollback also expires all ORM objects; device fields are re-set below. try: session.rollback() except Exception: # noqa: BLE001 logger.exception( "session.rollback() failed for device id=%d; session may be unusable", device.id ) try: device.last_poll_at = now device.last_poll_ok = False session.commit() except Exception: # noqa: BLE001 logger.exception( "Also failed to persist last_poll_ok=False for device id=%d", device.id ) return None def poll_all_enabled_devices( session: Session, *, bootstrap_settings: Settings, ) -> None: """Sweep all enabled Modbus devices and poll each one whose interval has elapsed. Global kill-switch: if ``build_runtime_settings(...).modbus_polling_enabled`` is ``False``, this function returns immediately without touching any device. Per-device interval: a device is skipped if its ``last_poll_at`` is not yet ``poll_interval_s`` seconds in the past. This allows the APScheduler job to run on a short fixed tick while each device self-regulates its own cadence. Exceptions from individual devices are swallowed inside ``poll_device``; this function itself will not raise. Parameters ---------- session: Open SQLAlchemy session. bootstrap_settings: The bootstrap ``Settings`` instance (from ``get_settings()``), used as the base for ``build_runtime_settings`` to pick up any DB overrides. """ try: runtime_settings = build_runtime_settings(session, bootstrap_settings) except Exception: # noqa: BLE001 logger.exception("Failed to build runtime settings for Modbus poll; aborting sweep") return if not runtime_settings.modbus_polling_enabled: logger.debug("Modbus polling is disabled (modbus_polling_enabled=False); skipping sweep") return try: devices = session.execute( select(ModbusDevice).where(ModbusDevice.enabled == True) # noqa: E712 ).scalars().all() except Exception: # noqa: BLE001 logger.exception("Failed to query enabled Modbus devices; aborting sweep") return now = datetime.now(UTC) for device in devices: # Respect per-device poll interval: skip if last poll was too recent. if device.last_poll_at is not None: # SQLite may return a naive datetime despite DateTime(timezone=True) — # treat naive values as UTC so the comparison is safe. last_poll = device.last_poll_at if last_poll.tzinfo is None: last_poll = last_poll.replace(tzinfo=UTC) elapsed = (now - last_poll).total_seconds() if elapsed < device.poll_interval_s: logger.debug( "Skipping device %r (id=%d): %.1fs elapsed < %ds interval", device.friendly_name, device.id, elapsed, device.poll_interval_s, ) continue poll_device(session, device)