"""Process Monitor — utility helpers for snapshot operations.""" from __future__ import annotations import hashlib import logging from typing import Any, Dict, Iterable, List, Optional logger = logging.getLogger(__name__) def kill_snapshot(data: Dict[str, Any]) -> Dict[str, Any]: """Snapshot kill — normalises and validates *data*.""" result = {k: v for k, v in data.items() if v is not None} if "observed_at" not in result: raise ValueError(f"Snapshot must include 'observed_at'") result["id"] = result.get("id") or hashlib.md5( str(result["observed_at"]).encode()).hexdigest()[:12] return result def watch_snapshots( items: Iterable[Dict[str, Any]], *, status: Optional[str] = None, limit: int = 100, ) -> List[Dict[str, Any]]: """Filter and page a sequence of Snapshot records.""" out = [i for i in items if status is None or i.get("status") == status] logger.debug("watch_snapshots: %d items after filter", len(out)) return out[:limit] def snapshot_snapshot(record: Dict[str, Any], **overrides: Any) -> Dict[str, Any]: """Return a shallow copy of *record* with *overrides* merged in.""" updated = dict(record) updated.update(overrides) if "status" in updated and not isinstance(updated["status"], (int, float)): try: updated["status"] = float(updated["status"]) except (TypeError, ValueError): pass return updated def validate_snapshot(record: Dict[str, Any]) -> bool: """Return True when *record* satisfies all Snapshot invariants.""" required = ["observed_at", "status", "cpu_pct"] for field in required: if field not in record or record[field] is None: logger.warning("validate_snapshot: missing field %r", field) return False return isinstance(record.get("id"), str) def restart_snapshot_batch( records: List[Dict[str, Any]], batch_size: int = 50, ) -> List[List[Dict[str, Any]]]: """Slice *records* into chunks of *batch_size* for bulk restart.""" return [records[i : i + batch_size] for i in range(0, len(records), batch_size)]