life-noc/lib/life_noc_inputs.py

from __future__ import annotations

from datetime import datetime, timezone
from pathlib import Path
import yaml

INPUTS_DIR = Path("data/inputs")
DOMAINS_FILE = Path("domains.yaml")


def utc_now_iso() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def utc_today_date() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%d")


def domain_file(domain: str) -> Path:
    return INPUTS_DIR / f"{domain}.yaml"


def load_domain(domain: str) -> dict:
    path = domain_file(domain)
    if not path.exists():
        return {}
    data = yaml.safe_load(path.read_text(encoding="utf-8"))
    if data is None:
        return {}
    if not isinstance(data, dict):
        raise ValueError(f"Fichier invalide: {path}")
    return data


def save_domain(domain: str, data: dict) -> Path:
    path = domain_file(domain)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(
        yaml.safe_dump(data, sort_keys=False, allow_unicode=True),
        encoding="utf-8",
    )
    return path


def list_items(domain: str) -> list[str]:
    return list(load_domain(domain).keys())


def get_item(domain: str, item_key: str) -> dict:
    data = load_domain(domain)
    if item_key not in data:
        raise KeyError(item_key)
    item = data[item_key]
    if not isinstance(item, dict):
        raise ValueError(f"Entrée invalide pour {item_key}")
    return item


def set_item(domain: str, item_key: str, value: str, origin: str = "manual") -> dict:
    data = load_domain(domain)
    data[item_key] = {
        "value": value,
        "captured_at": utc_now_iso(),
        "origin": origin,
    }
    save_domain(domain, data)
    return data[item_key]


def set_counter_pair(domain: str, item_key: str, current_value: str, reference_value: str, origin: str = "manual") -> dict:
    data = load_domain(domain)
    data[item_key] = {
        "value": current_value,
        "reference_value": reference_value,
        "captured_at": utc_now_iso(),
        "origin": origin,
    }
    save_domain(domain, data)
    return data[item_key]


def complete_item(domain: str, item_key: str, origin: str = "manual") -> dict:
    return set_item(domain, item_key, utc_today_date(), origin=origin)


def load_domains_definition() -> dict:
    if not DOMAINS_FILE.exists():
        raise ValueError(f"Fichier introuvable: {DOMAINS_FILE}")
    data = yaml.safe_load(DOMAINS_FILE.read_text(encoding="utf-8"))
    if not isinstance(data, dict):
        raise ValueError("domains.yaml invalide")
    domains = data.get("domains")
    if not isinstance(domains, dict):
        raise ValueError("domains.yaml invalide: clé 'domains' absente ou invalide")
    return domains


def get_defined_service(domain: str, item_key: str) -> dict:
    domains = load_domains_definition()
    items = domains.get(domain)
    if not isinstance(items, list):
        raise KeyError(domain)

    for item in items:
        if isinstance(item, dict) and item.get("name") == item_key:
            return item

    raise KeyError(item_key)


def compute_state_for_item(domain: str, item_key: str) -> str:
    service = get_defined_service(domain, item_key)
    probe = service.get("probe")
    if not isinstance(probe, dict):
        return "MOCK"

    probe_type = str(probe.get("type", "")).strip()
    source = probe.get("source", {})
    thresholds = probe.get("thresholds", {})
    metric = probe.get("metric", {})
    source_type = str(source.get("type", "")).strip()
    metric_unit = str(metric.get("unit", "")).strip()

    item = get_item(domain, item_key)

    if probe_type in {"elapsed_time", "days_until_due"}:
        if source_type != "manual_date":
            return "CRITICAL"
        if metric_unit != "days":
            return "CRITICAL"

        value = str(item.get("value", "")).strip()
        dt = datetime.strptime(value, "%Y-%m-%d").replace(tzinfo=timezone.utc)
        now = datetime.now(timezone.utc)

        if probe_type == "elapsed_time":
            days = (now - dt).total_seconds() / 86400.0
            if "critical_gte" in thresholds and days >= float(thresholds["critical_gte"]):
                return "CRITICAL"
            if "warning_gte" in thresholds and days >= float(thresholds["warning_gte"]):
                return "WARNING"
            if "ok_gte" in thresholds and days >= float(thresholds["ok_gte"]):
                return "OK"
            if "unknown_lt" in thresholds and days < float(thresholds["unknown_lt"]):
                return "UNKNOWN"
            return "CRITICAL"

        days = (dt - now).total_seconds() / 86400.0
        if "critical_lt" in thresholds and days < float(thresholds["critical_lt"]):
            return "CRITICAL"
        if "warning_lt" in thresholds and days < float(thresholds["warning_lt"]):
            return "WARNING"
        if "ok_lt" in thresholds and days < float(thresholds["ok_lt"]):
            return "OK"
        if "unknown_gte" in thresholds and days >= float(thresholds["unknown_gte"]):
            return "UNKNOWN"
        return "CRITICAL"

    if probe_type == "elapsed_distance":
        if source_type != "manual_counter":
            return "CRITICAL"
        if metric_unit != "km":
            return "CRITICAL"

        current_value = float(str(item.get("value", "")).strip())
        reference_value = float(str(item.get("reference_value", "")).strip())
        distance = current_value - reference_value

        if "unknown_lt" in thresholds and distance < float(thresholds["unknown_lt"]):
            return "UNKNOWN"
        if "critical_gte" in thresholds and distance >= float(thresholds["critical_gte"]):
            return "CRITICAL"
        if "warning_gte" in thresholds and distance >= float(thresholds["warning_gte"]):
            return "WARNING"
        if "ok_gte" in thresholds and distance >= float(thresholds["ok_gte"]):
            return "OK"
        return "CRITICAL"

    if probe_type == "remaining_quantity":
        if source_type != "manual_value":
            return "CRITICAL"
        if metric_unit not in {"%", "units", "litres"}:
            return "CRITICAL"

        value = float(str(item.get("value", "")).strip())

        if "critical_lt" in thresholds and value < float(thresholds["critical_lt"]):
            return "CRITICAL"
        if "warning_lt" in thresholds and value < float(thresholds["warning_lt"]):
            return "WARNING"
        if "ok_gte" in thresholds and value >= float(thresholds["ok_gte"]):
            return "OK"
        if "unknown_gte" in thresholds and value >= float(thresholds["unknown_gte"]):
            return "UNKNOWN"
        return "CRITICAL"

    if probe_type == "current_value":
        if source_type != "manual_value":
            return "CRITICAL"
        if metric_unit not in {"%", "units", "litres", "volts", "watts"}:
            return "CRITICAL"

        value = float(str(item.get("value", "")).strip())

        if "critical_lt" in thresholds and value < float(thresholds["critical_lt"]):
            return "CRITICAL"
        if "critical_gte" in thresholds and value >= float(thresholds["critical_gte"]):
            return "CRITICAL"
        if "warning_lt" in thresholds and value < float(thresholds["warning_lt"]):
            return "WARNING"
        if "warning_gte" in thresholds and value >= float(thresholds["warning_gte"]):
            return "WARNING"
        if "ok_gte" in thresholds and value >= float(thresholds["ok_gte"]):
            return "OK"
        if "ok_lt" in thresholds and value < float(thresholds["ok_lt"]):
            return "OK"
        if "unknown_gte" in thresholds and value >= float(thresholds["unknown_gte"]):
            return "UNKNOWN"
        if "unknown_lt" in thresholds and value < float(thresholds["unknown_lt"]):
            return "UNKNOWN"
        return "CRITICAL"

    return "CRITICAL"