"""
cap_adapter — CAPAdapter Protocol, FileDropCAPAdapter, NoOpCAPAdapter.
Concrete live adapters (MaximoCAPAdapter, SAPPMCAPAdapter) live in separate
files and implement the same Protocol. See CAP_INTEGRATION_GUIDE.md for the
implementation skeleton.
"""
from __future__ import annotations
import json
import os
import re
import tempfile
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional, Protocol
[docs]
JsonDict = Dict[str, Any]
[docs]
def _utcnow_iso() -> str:
return datetime.now(timezone.utc).isoformat()
[docs]
def _safe_export_filename(export_id: str) -> str:
"""Return a filesystem-safe stem for CAP export files."""
normalized = str(export_id or "unknown").replace("::", "_").replace(" ", "_")
# Windows-invalid chars: <>:"/\|?* ; keep output deterministic across OSes.
return re.sub(r'[<>:"/\\|?*]', "_", normalized)
# ---------------------------------------------------------------------------
# Submission receipt
# ---------------------------------------------------------------------------
@dataclass
[docs]
class CAPSubmissionReceipt:
"""
Returned by every CAPAdapter.submit() call.
Attributes
----------
receipt_id:
Unique ID for this submission attempt.
submitted_at:
ISO-8601 UTC timestamp.
adapter:
Class name of the adapter that produced this receipt.
export_id:
``export_id`` from the CAPExportPackage that was submitted.
cr_numbers:
CMMS-assigned corrective action / work order numbers.
Empty list for adapters that do not receive synchronous confirmation
(e.g. ``FileDropCAPAdapter``).
status:
``"submitted"`` — accepted by CMMS synchronously.
``"pending"`` — written to drop zone; CMMS import job pending.
``"partial"`` — some records submitted, some failed.
``"failed"`` — no records submitted.
``"noop"`` — NoOpCAPAdapter; nothing was done.
errors:
List of per-record error dicts when status is partial or failed.
notes:
Free-text adapter notes (e.g. path of the written file).
"""
[docs]
cr_numbers: List[str] = field(default_factory=list)
[docs]
status: str = "pending"
[docs]
errors: List[JsonDict] = field(default_factory=list)
[docs]
notes: Optional[str] = None
[docs]
def to_dict(self) -> JsonDict:
"""
Return the receipt as a plain JSON-serializable dict.
Returns
-------
dict
Keys mirror the dataclass attributes; suitable for persisting
alongside the CAPExportPackage as the ``cap_submission_receipt``
artifact.
"""
return {
"receipt_id": self.receipt_id,
"submitted_at": self.submitted_at,
"adapter": self.adapter,
"export_id": self.export_id,
"cr_numbers": self.cr_numbers,
"status": self.status,
"errors": self.errors,
"notes": self.notes,
}
# ---------------------------------------------------------------------------
# Protocol
# ---------------------------------------------------------------------------
[docs]
class CAPAdapter(Protocol):
"""
Protocol for CAP submission adapters.
All implementations must be idempotent with respect to
``package["export_id"]``: submitting the same package twice must not
create duplicate CMMS records.
"""
[docs]
def submit(self, package: JsonDict) -> CAPSubmissionReceipt:
"""
Submit a CAPExportPackage to the target CMMS (or file drop zone).
Parameters
----------
package:
Dict conforming to ``schemas/cap_export_package.json``.
Returns
-------
CAPSubmissionReceipt
"""
...
# ---------------------------------------------------------------------------
# FileDropCAPAdapter
# ---------------------------------------------------------------------------
[docs]
class FileDropCAPAdapter:
"""
Writes the CAPExportPackage as a JSON file to a watched directory.
This is the default adapter for dev and pre-live deployments. The plant's
CMMS import job polls the directory and ingests the file asynchronously.
Returns a receipt with ``status="pending"`` and an empty ``cr_numbers``
list — CMMS record numbers are not available at submission time.
Parameters
----------
drop_dir:
Directory where export files are written. Created if absent.
"""
def __init__(self, drop_dir: str | Path) -> None:
[docs]
self.drop_dir = Path(drop_dir)
self.drop_dir.mkdir(parents=True, exist_ok=True)
[docs]
def submit(self, package: JsonDict) -> CAPSubmissionReceipt:
"""
Write the package to the drop zone as JSON and return a receipt.
The file is written atomically (to a hidden ``*.tmp`` sibling, then
``os.replace``d into place) so a CMMS import job polling the directory
never observes a partially written ``cap_export_*.json`` file.
Parameters
----------
package:
Dict conforming to ``schemas/cap_export_package.json``.
Returns
-------
CAPSubmissionReceipt
``status="pending"`` with an empty ``cr_numbers`` list; ``notes``
holds the written file path.
"""
export_id = package.get("export_id") or "unknown"
submitted_at = _utcnow_iso()
receipt_id = f"RCPT::{export_id}::{submitted_at}"
safe_name = _safe_export_filename(export_id)
file_path = self.drop_dir / f"cap_export_{safe_name}.json"
self._atomic_write_json(file_path, package)
return CAPSubmissionReceipt(
receipt_id=receipt_id,
submitted_at=submitted_at,
adapter=self.__class__.__name__,
export_id=export_id,
cr_numbers=[],
status="pending",
notes=str(file_path),
)
@staticmethod
[docs]
def _atomic_write_json(file_path: Path, package: JsonDict) -> None:
"""Write ``package`` as JSON to ``file_path`` via a temp file + os.replace."""
data = json.dumps(package, indent=2, default=str)
fd, tmp = tempfile.mkstemp(
dir=str(file_path.parent), prefix=f".{file_path.stem}.", suffix=".tmp"
)
try:
with os.fdopen(fd, "w", encoding="utf-8") as fh:
fh.write(data)
os.replace(tmp, file_path)
except BaseException:
if os.path.exists(tmp):
os.unlink(tmp)
raise
# ---------------------------------------------------------------------------
# NoOpCAPAdapter
# ---------------------------------------------------------------------------
[docs]
class NoOpCAPAdapter:
"""
Silently discards the package. Used in unit tests and CI.
Returns a receipt with ``status="noop"`` and an empty ``cr_numbers`` list.
Makes no file I/O and has no external dependencies.
"""
[docs]
def submit(self, package: JsonDict) -> CAPSubmissionReceipt:
"""
Discard the package and return a ``status="noop"`` receipt.
Parameters
----------
package:
Dict conforming to ``schemas/cap_export_package.json``. Only
``export_id`` is read; nothing is written.
Returns
-------
CAPSubmissionReceipt
``status="noop"`` with an empty ``cr_numbers`` list.
"""
export_id = package.get("export_id") or "unknown"
submitted_at = _utcnow_iso()
return CAPSubmissionReceipt(
receipt_id=f"RCPT::NOOP::{submitted_at}",
submitted_at=submitted_at,
adapter=self.__class__.__name__,
export_id=export_id,
cr_numbers=[],
status="noop",
notes="NoOpCAPAdapter — package discarded.",
)