Start

sternwarte-homeassistant-integration/custom_components/sternwarte/api.py

@@ -0,0 +1,109 @@
+"""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:
+            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:
+            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:
+            state.counters = [int(parts[counter_index + 1 + i] or "0") for i in range(12)]
+
+        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}&")