108 lines
3.7 KiB
Python
108 lines
3.7 KiB
Python
"""HTTP API protocol collector."""
|
|
from typing import Optional
|
|
|
|
import httpx
|
|
|
|
from app.collectors.base import BaseCollector
|
|
|
|
|
|
class HttpCollector(BaseCollector):
|
|
"""Collect data by polling HTTP API endpoints.
|
|
|
|
connection_params example:
|
|
{
|
|
"url": "http://api.example.com/device/data",
|
|
"method": "GET",
|
|
"headers": {"X-API-Key": "abc123"},
|
|
"auth": {"type": "basic", "username": "user", "password": "pass"},
|
|
"data_mapping": {
|
|
"active_power": {"key": "data.power", "unit": "kW"},
|
|
"voltage": {"key": "data.voltage", "unit": "V"}
|
|
},
|
|
"timeout": 10
|
|
}
|
|
"""
|
|
|
|
def __init__(self, device_id, device_code, connection_params, collect_interval=15):
|
|
super().__init__(device_id, device_code, connection_params, collect_interval)
|
|
self._url = connection_params.get("url", "")
|
|
self._method = connection_params.get("method", "GET").upper()
|
|
self._headers = connection_params.get("headers", {})
|
|
self._auth_config = connection_params.get("auth", {})
|
|
self._data_mapping = connection_params.get("data_mapping", {})
|
|
self._timeout = connection_params.get("timeout", 10)
|
|
self._client: Optional[httpx.AsyncClient] = None
|
|
|
|
async def connect(self):
|
|
auth = None
|
|
auth_type = self._auth_config.get("type", "")
|
|
if auth_type == "basic":
|
|
auth = httpx.BasicAuth(
|
|
self._auth_config.get("username", ""),
|
|
self._auth_config.get("password", ""),
|
|
)
|
|
|
|
headers = dict(self._headers)
|
|
if auth_type == "token":
|
|
token = self._auth_config.get("token", "")
|
|
headers["Authorization"] = f"Bearer {token}"
|
|
|
|
self._client = httpx.AsyncClient(
|
|
headers=headers,
|
|
auth=auth,
|
|
timeout=self._timeout,
|
|
)
|
|
# Verify connectivity with a test request
|
|
response = await self._client.request(self._method, self._url)
|
|
response.raise_for_status()
|
|
|
|
async def disconnect(self):
|
|
if self._client:
|
|
await self._client.aclose()
|
|
self._client = None
|
|
|
|
async def collect(self) -> dict:
|
|
if not self._client:
|
|
raise ConnectionError("HTTP client not initialized")
|
|
|
|
response = await self._client.request(self._method, self._url)
|
|
response.raise_for_status()
|
|
payload = response.json()
|
|
|
|
return self._parse_response(payload)
|
|
|
|
def _parse_response(self, payload: dict) -> dict:
|
|
"""Parse HTTP JSON response into data points.
|
|
|
|
Supports dotted key paths like "data.power" to navigate nested JSON.
|
|
"""
|
|
data = {}
|
|
if self._data_mapping:
|
|
for data_type, mapping in self._data_mapping.items():
|
|
key_path = mapping.get("key", data_type)
|
|
unit = mapping.get("unit", "")
|
|
value = self._resolve_path(payload, key_path)
|
|
if value is not None:
|
|
try:
|
|
data[data_type] = (float(value), unit)
|
|
except (TypeError, ValueError):
|
|
pass
|
|
else:
|
|
# Auto-detect numeric fields at top level
|
|
for key, value in payload.items():
|
|
if isinstance(value, (int, float)):
|
|
data[key] = (float(value), "")
|
|
return data
|
|
|
|
@staticmethod
|
|
def _resolve_path(obj: dict, path: str):
|
|
"""Resolve a dotted path like 'data.power' in a nested dict."""
|
|
parts = path.split(".")
|
|
current = obj
|
|
for part in parts:
|
|
if isinstance(current, dict) and part in current:
|
|
current = current[part]
|
|
else:
|
|
return None
|
|
return current
|