111 lines
4.3 KiB
Python
111 lines
4.3 KiB
Python
"""API client for the Sternwarte Web-IO."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
import asyncio
|
|
import time
|
|
|
|
from .models import SternwarteState
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class SternwarteApi:
|
|
"""Thin Web-IO client."""
|
|
|
|
host: str
|
|
port: int
|
|
password: str
|
|
|
|
def _url(self, path: str) -> str:
|
|
separator = "&" if "?" in path else "?"
|
|
nonce = time.monotonic_ns()
|
|
return f"http://{self.host}:{self.port}{path}{separator}_ts={nonce}"
|
|
|
|
async def _get_text(self, path: str) -> str:
|
|
request_path = self._url(path).split(f"{self.host}:{self.port}", 1)[-1]
|
|
request = (
|
|
f"GET {request_path} HTTP/1.0\r\n"
|
|
f"Host: {self.host}\r\n"
|
|
"Connection: close\r\n"
|
|
"\r\n"
|
|
).encode("ascii")
|
|
|
|
last_error: Exception | None = None
|
|
for attempt in range(2):
|
|
try:
|
|
reader, writer = await asyncio.wait_for(
|
|
asyncio.open_connection(self.host, self.port),
|
|
timeout=5,
|
|
)
|
|
try:
|
|
writer.write(request)
|
|
await asyncio.wait_for(writer.drain(), timeout=5)
|
|
raw = await asyncio.wait_for(reader.read(4096), timeout=5)
|
|
content_length = None
|
|
if b"\r\n\r\n" in raw:
|
|
headers, initial_body = raw.split(b"\r\n\r\n", 1)
|
|
for header_line in headers.decode("iso-8859-1", errors="ignore").split("\r\n"):
|
|
if header_line.lower().startswith("content-length:"):
|
|
try:
|
|
content_length = int(header_line.split(":", 1)[1].strip())
|
|
except ValueError:
|
|
content_length = None
|
|
break
|
|
else:
|
|
initial_body = raw
|
|
|
|
if content_length is not None:
|
|
remaining = max(0, content_length - len(initial_body))
|
|
body = initial_body + (
|
|
await asyncio.wait_for(reader.readexactly(remaining), timeout=5)
|
|
if remaining
|
|
else b""
|
|
)
|
|
else:
|
|
body = initial_body
|
|
finally:
|
|
writer.close()
|
|
await writer.wait_closed()
|
|
|
|
return body.decode("utf-8", errors="ignore").replace("\x00", "").strip()
|
|
except (OSError, asyncio.TimeoutError) as err:
|
|
last_error = err
|
|
if attempt == 1:
|
|
raise
|
|
await asyncio.sleep(0.25)
|
|
raise last_error if last_error is not None else RuntimeError("Unbekannter Web-IO-Fehler")
|
|
|
|
async def fetch_state(self) -> SternwarteState:
|
|
"""Fetch and parse the current Web-IO state."""
|
|
raw = await self._get_text(f"/allout?PW={self.password}&")
|
|
parts = raw.split(";")
|
|
|
|
state = SternwarteState()
|
|
|
|
out_index = parts.index("output") if "output" in parts else -1
|
|
if out_index != -1 and out_index + 1 < len(parts):
|
|
value = int(parts[out_index + 1] or "0", 16)
|
|
state.outputs = [bool(value & (1 << i)) for i in range(12)]
|
|
|
|
in_index = parts.index("input") if "input" in parts else -1
|
|
if in_index != -1 and in_index + 1 < len(parts):
|
|
value = int(parts[in_index + 1] or "0", 16)
|
|
state.inputs = [bool(value & (1 << i)) for i in range(12)]
|
|
|
|
counter_index = parts.index("counter") if "counter" in parts else -1
|
|
if counter_index != -1:
|
|
for i in range(min(12, len(parts) - counter_index - 1)):
|
|
state.counters[i] = int(parts[counter_index + 1 + i] or "0")
|
|
|
|
return state
|
|
|
|
async def set_output(self, index: int, enabled: bool) -> None:
|
|
"""Set a digital output."""
|
|
state = "ON" if enabled else "OFF"
|
|
await self._get_text(f"/outputaccess{index}?PW={self.password}&State={state}&")
|
|
|
|
async def clear_counter(self, index: int) -> None:
|
|
"""Clear a counter."""
|
|
await self._get_text(f"/counterclear{index}?PW={self.password}&")
|