modbus-server-nd/http_value_provider.py

"""HTTP realtime value provider."""

import requests

from value_provider import ValueProvider


class HttpValueProvider(ValueProvider):
    def __init__(self, url: str, timeout_seconds: int):
        self.url = url
        self.timeout_seconds = timeout_seconds

    def fetch_values(self, point_ids: list[str]) -> dict[str, object]:
        response = requests.post(
            self.url,
            json={"point_ids": point_ids},
            timeout=self.timeout_seconds,
        )
        response.raise_for_status()
        payload = response.json()
        if payload.get("state") != 0:
            raise RuntimeError(f"realtime api failed: {payload}")
        return {item["point_id"]: item.get("value") for item in payload.get("data", [])}