9756538f16
Organizes 11 projects for Cerbo GX/Venus OS into a single repository: - axiom-nmea: Raymarine LightHouse protocol decoder - dbus-generator-ramp: Generator current ramp controller - dbus-lightning: Blitzortung lightning monitor - dbus-meteoblue-forecast: Meteoblue weather forecast - dbus-no-foreign-land: noforeignland.com tracking - dbus-tides: Tide prediction from depth + harmonics - dbus-vrm-history: VRM cloud history proxy - dbus-windy-station: Windy.com weather upload - mfd-custom-app: MFD app deployment package - venus-html5-app: Custom Victron HTML5 app fork - watermaker: Watermaker PLC control UI Adds root README, .gitignore, project template, and per-project .gitignore files. Sensitive config files excluded via .gitignore with .example templates provided. Made-with: Cursor
93 lines
3.2 KiB
Python
93 lines
3.2 KiB
Python
"""
|
|
VRM API client using urllib (available on Venus OS without pip).
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
from urllib.request import Request, urlopen
|
|
from urllib.error import HTTPError, URLError
|
|
|
|
logger = logging.getLogger('VrmClient')
|
|
|
|
|
|
class VrmClient:
|
|
BASE_URL = "https://vrmapi.victronenergy.com/v2"
|
|
|
|
def __init__(self, token, site_id=None):
|
|
self.token = token
|
|
self.site_id = site_id
|
|
|
|
def _request(self, path, params=None):
|
|
url = f"{self.BASE_URL}{path}"
|
|
if params:
|
|
query = "&".join(f"{k}={v}" for k, v in params.items())
|
|
url = f"{url}?{query}"
|
|
|
|
req = Request(url)
|
|
req.add_header("x-authorization", f"Token {self.token}")
|
|
req.add_header("Content-Type", "application/json")
|
|
|
|
try:
|
|
with urlopen(req, timeout=30) as resp:
|
|
return json.loads(resp.read().decode("utf-8"))
|
|
except HTTPError as e:
|
|
body = e.read().decode("utf-8", errors="replace")
|
|
logger.error(f"VRM API HTTP {e.code}: {body[:200]}")
|
|
raise
|
|
except URLError as e:
|
|
logger.error(f"VRM API connection error: {e.reason}")
|
|
raise
|
|
|
|
def discover_site_id(self, portal_id):
|
|
"""Auto-discover numeric siteId from the GX device portal ID.
|
|
|
|
1. GET /users/me -> idUser
|
|
2. GET /users/{idUser}/installations -> find matching identifier
|
|
"""
|
|
me = self._request("/users/me")
|
|
id_user = me.get("idUser") or me.get("user", {}).get("idUser")
|
|
if not id_user:
|
|
raise ValueError("Could not determine idUser from VRM API")
|
|
|
|
installs = self._request(f"/users/{id_user}/installations")
|
|
records = installs.get("records", [])
|
|
|
|
for inst in records:
|
|
if inst.get("identifier") == portal_id:
|
|
logger.info(f"Found VRM site: {inst.get('name')} (id={inst.get('idSite')})")
|
|
return inst["idSite"]
|
|
|
|
available = [f"{r.get('name')}({r.get('identifier')})" for r in records[:5]]
|
|
raise ValueError(
|
|
f"No installation matches portal ID '{portal_id}'. "
|
|
f"Available: {', '.join(available)}"
|
|
)
|
|
|
|
def get_stats(self, stat_type, attribute_codes, start, end, interval=None):
|
|
"""GET /installations/{siteId}/stats
|
|
|
|
stat_type: 'live_feed' or 'kwh'
|
|
attribute_codes: list of VRM attribute codes (e.g. ['bs'] for battery SOC)
|
|
"""
|
|
if not self.site_id:
|
|
raise ValueError("site_id not configured")
|
|
|
|
params = {"type": stat_type, "start": str(start), "end": str(end)}
|
|
for i, code in enumerate(attribute_codes):
|
|
params[f"attributeCodes[{i}]"] = code
|
|
if interval:
|
|
params["interval"] = interval
|
|
|
|
return self._request(f"/installations/{self.site_id}/stats", params)
|
|
|
|
def get_widget(self, widget_name, instance=None):
|
|
"""GET /installations/{siteId}/widgets/{widgetName}"""
|
|
if not self.site_id:
|
|
raise ValueError("site_id not configured")
|
|
|
|
params = {}
|
|
if instance is not None:
|
|
params["instance"] = str(instance)
|
|
|
|
return self._request(f"/installations/{self.site_id}/widgets/{widget_name}", params)
|