"""Persist batched probe summaries and apply device health transitions.""" import logging import time from collections import defaultdict from dataclasses import dataclass from datetime import datetime from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.config import Settings, settings from app.models.device import Device, DeviceMonitoringPolicy from app.models.ping_record import PingRecord from app.services.fping_runner import ProbeResult, run_fping_count from app.services.state_machine import StateDecision, evaluate_health logger = logging.getLogger("pingwatch.pinger") @dataclass(frozen=True) class DeviceStateChange: """A persisted device health transition produced by one probe round.""" device: Device old_status: str new_status: str consecutive_failures: int event_type: str | None = None reason: str = "" class Pinger: """Run multi-packet fping probes and update monitored device health.""" def __init__(self, runtime_settings: Settings | None = None): self._settings = runtime_settings or settings self._round_num = 0 async def run_one_round( self, db: AsyncSession, ) -> list[DeviceStateChange]: """Persist one summary per enabled device and commit state changes once.""" self._round_num += 1 round_num = self._round_num query_result = await db.execute( select(Device).where(Device.is_enabled.is_(True)) ) devices = list(query_result.scalars().all()) if not devices: logger.info("[Round %s] no enabled devices", round_num) return [] started_at = time.monotonic() policies = { device.id: DeviceMonitoringPolicy.from_device( device, self._settings, ) for device in devices } probe_results = await self._probe_by_packet_count(devices, policies) observed_at = datetime.now() records_by_device: dict[int, PingRecord] = {} for device in devices: policy = policies[device.id] probe = probe_results.get( (device.id, policy.probe_packets_per_round) ) if probe is None: probe = ProbeResult( ip=device.ip, sent_count=policy.probe_packets_per_round, received_count=0, average_rtt_ms=None, is_valid=False, failure_reason="missing runner result", ) record = self._to_record( device=device, probe=probe, round_num=round_num, observed_at=observed_at, ) records_by_device[device.id] = record db.add(record) await db.flush() changes: list[DeviceStateChange] = [] for device in devices: record = records_by_device[device.id] device.last_ping_time = observed_at if not record.is_valid: continue policy = policies[device.id] recent = await self._load_recent_valid( db, device.id, policy, ) decision = evaluate_health( device.current_status or "unknown", recent, policy, ) self._update_observation_fields(device, recent, observed_at) change = self._apply_decision( device, decision, observed_at, ) if change is not None: changes.append(change) # The scheduler records transition events and commits the complete # probe/state/outbox unit. A direct caller retains the same ownership. await db.flush() elapsed = time.monotonic() - started_at logger.info( "[Round %s] persisted %s device summaries with %s transitions " "in %.2fs", round_num, len(devices), len(changes), elapsed, ) return changes async def _probe_by_packet_count( self, devices: list[Device], policies: dict[int, DeviceMonitoringPolicy], ) -> dict[tuple[int, int], ProbeResult]: """Run one argv-safe batch for each distinct packet-count policy.""" groups: dict[int, list[Device]] = defaultdict(list) for device in devices: groups[policies[device.id].probe_packets_per_round].append(device) results: dict[tuple[int, int], ProbeResult] = {} timeout_ms = int(self._settings.PING_TIMEOUT_SECONDS * 1000) for packet_count, grouped_devices in groups.items(): unique_ips = list(dict.fromkeys( device.ip for device in grouped_devices )) batch = await run_fping_count( unique_ips, packets_per_round=packet_count, timeout_ms=timeout_ms, executable=self._settings.FPING_PATH, ) for device in grouped_devices: probe = batch.get(device.ip) if probe is not None: results[(device.id, packet_count)] = probe return results @staticmethod def _to_record( device: Device, probe: ProbeResult, round_num: int, observed_at: datetime, ) -> PingRecord: """Convert a runner aggregate into the persisted compatibility model.""" return PingRecord( device_id=device.id, is_alive=probe.is_valid and probe.received_count > 0, response_time_ms=probe.average_rtt_ms, round_num=round_num, created_at=observed_at, sent_count=probe.sent_count, received_count=probe.received_count, packet_loss_percent=( probe.packet_loss_percent if probe.is_valid else None ), average_rtt_ms=probe.average_rtt_ms, is_valid=probe.is_valid, failure_reason=probe.failure_reason, ) @staticmethod async def _load_recent_valid( db: AsyncSession, device_id: int, policy: DeviceMonitoringPolicy, ) -> list[PingRecord]: """Load enough valid summaries for every configured transition rule.""" history_size = max( policy.offline_consecutive_rounds, policy.degraded_window_rounds, policy.recovery_consecutive_clean_rounds, ) query_result = await db.execute( select(PingRecord) .where(PingRecord.device_id == device_id) .where(PingRecord.is_valid.is_(True)) .order_by(PingRecord.created_at.desc(), PingRecord.id.desc()) .limit(history_size) ) newest_first = list(query_result.scalars().all()) return list(reversed(newest_first)) @staticmethod def _update_observation_fields( device: Device, recent: list[PingRecord], observed_at: datetime, ) -> None: """Maintain compatibility timestamps and the full-loss streak counter.""" current = recent[-1] if current.received_count > 0: device.last_online_time = observed_at full_loss_count = 0 for record in reversed(recent): if record.sent_count and record.received_count == 0: full_loss_count += 1 else: break device.consecutive_failures = full_loss_count @staticmethod def _apply_decision( device: Device, decision: StateDecision, observed_at: datetime, ) -> DeviceStateChange | None: """Apply one pure decision and materialize its transition value.""" old_status = device.current_status or "unknown" if decision.next_status == old_status: return None device.current_status = decision.next_status if decision.next_status == "offline": device.last_offline_time = observed_at if decision.next_status == "online": device.last_online_time = observed_at return DeviceStateChange( device=device, old_status=old_status, new_status=decision.next_status, consecutive_failures=device.consecutive_failures or 0, event_type=decision.event_type, reason=decision.reason, )