289 lines
10 KiB
Python
289 lines
10 KiB
Python
"""Client TCP asynchrone, strictement en lecture seule, pour le proxy Arkteos."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import inspect
|
|
from collections.abc import Awaitable, Callable, Sequence
|
|
from contextlib import suppress
|
|
from typing import Final, TypeAlias
|
|
|
|
from .frame_extractor import FrameExtractor
|
|
from .parser import FilterResult, FrameError, ParsedFrame, filter_values, parse_frame
|
|
|
|
|
|
DEFAULT_PROXY_PORT: Final = 9641
|
|
DEFAULT_RECONNECT_DELAYS: Final[tuple[float, ...]] = (1.0, 2.0, 5.0, 10.0, 30.0)
|
|
|
|
DataCallback: TypeAlias = Callable[
|
|
[str, dict[str, int | float | str], dict[str, int | float]], object
|
|
]
|
|
AvailabilityCallback: TypeAlias = Callable[[bool], object]
|
|
SleepCallable: TypeAlias = Callable[[float], Awaitable[None]]
|
|
|
|
|
|
class ArkteosClient:
|
|
"""Lit le proxy Arkteos, extrait les trames et diffuse les données filtrées."""
|
|
|
|
def __init__(
|
|
self,
|
|
host: str,
|
|
port: int = DEFAULT_PROXY_PORT,
|
|
*,
|
|
read_size: int = 1024,
|
|
frame_timeout: float = 30.0,
|
|
reconnect_delays: Sequence[float] = DEFAULT_RECONNECT_DELAYS,
|
|
max_buffer_size: int = 4096,
|
|
sleep: SleepCallable = asyncio.sleep,
|
|
) -> None:
|
|
if not host:
|
|
raise ValueError("host ne peut pas être vide")
|
|
if not 1 <= port <= 65535:
|
|
raise ValueError("port doit être compris entre 1 et 65535")
|
|
if read_size < 1:
|
|
raise ValueError("read_size doit être positif")
|
|
if frame_timeout <= 0:
|
|
raise ValueError("frame_timeout doit être positif")
|
|
if not reconnect_delays or any(delay < 0 for delay in reconnect_delays):
|
|
raise ValueError("reconnect_delays doit contenir des délais positifs ou nuls")
|
|
|
|
self.host = host
|
|
self.port = port
|
|
self.read_size = read_size
|
|
self.frame_timeout = frame_timeout
|
|
self.reconnect_delays = tuple(reconnect_delays)
|
|
|
|
self.running = False
|
|
self.connected = False
|
|
self.available = False
|
|
self.last_valid_frame: ParsedFrame | None = None
|
|
self.last_frame_type: str | None = None
|
|
self.latest_frigo_data: dict[str, int | float | str] | None = None
|
|
self.latest_regulation_data: dict[str, int | float | str] | None = None
|
|
self.latest_metadata: dict[str, int | float | str] | None = None
|
|
self.frames_received = 0
|
|
self.frames_rejected = 0
|
|
self.bytes_received = 0
|
|
self.connection_attempts = 0
|
|
self.reconnect_count = 0
|
|
self.last_error: Exception | None = None
|
|
|
|
self._extractor = FrameExtractor(max_buffer_size=max_buffer_size)
|
|
self._sleep = sleep
|
|
self._task: asyncio.Task[None] | None = None
|
|
self._writer: asyncio.StreamWriter | None = None
|
|
self._connected_event = asyncio.Event()
|
|
self._data_callbacks: list[DataCallback] = []
|
|
self._availability_callbacks: list[AvailabilityCallback] = []
|
|
|
|
async def start(self) -> None:
|
|
"""Démarre la tâche de connexion si le client n'est pas déjà actif."""
|
|
|
|
if self.running:
|
|
return
|
|
|
|
self.running = True
|
|
self._task = asyncio.create_task(self._run(), name="arkteos-client")
|
|
|
|
async def stop(self) -> None:
|
|
"""Arrête le client, ferme le flux et attend toutes ses tâches."""
|
|
|
|
self.running = False
|
|
self._set_connected(False)
|
|
await self._set_available(False)
|
|
await self._close_writer()
|
|
|
|
task = self._task
|
|
self._task = None
|
|
if task is not None and task is not asyncio.current_task() and not task.done():
|
|
task.cancel()
|
|
with suppress(asyncio.CancelledError):
|
|
await task
|
|
|
|
async def wait_until_connected(self, timeout: float | None = None) -> bool:
|
|
"""Attend une connexion établie, avec délai optionnel."""
|
|
|
|
if self.connected:
|
|
return True
|
|
try:
|
|
if timeout is None:
|
|
await self._connected_event.wait()
|
|
else:
|
|
await asyncio.wait_for(self._connected_event.wait(), timeout)
|
|
except TimeoutError:
|
|
return False
|
|
return self.connected
|
|
|
|
def add_data_callback(self, callback: DataCallback) -> None:
|
|
"""Ajoute un callback de données sans le dupliquer."""
|
|
|
|
if callback not in self._data_callbacks:
|
|
self._data_callbacks.append(callback)
|
|
|
|
def remove_data_callback(self, callback: DataCallback) -> None:
|
|
"""Retire un callback de données s'il est enregistré."""
|
|
|
|
with suppress(ValueError):
|
|
self._data_callbacks.remove(callback)
|
|
|
|
def add_availability_callback(self, callback: AvailabilityCallback) -> None:
|
|
"""Ajoute un callback de disponibilité sans le dupliquer."""
|
|
|
|
if callback not in self._availability_callbacks:
|
|
self._availability_callbacks.append(callback)
|
|
|
|
def remove_availability_callback(self, callback: AvailabilityCallback) -> None:
|
|
"""Retire un callback de disponibilité s'il est enregistré."""
|
|
|
|
with suppress(ValueError):
|
|
self._availability_callbacks.remove(callback)
|
|
|
|
async def _run(self) -> None:
|
|
delay_index = 0
|
|
try:
|
|
while self.running:
|
|
self.connection_attempts += 1
|
|
try:
|
|
reader, writer = await asyncio.open_connection(self.host, self.port)
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception as error:
|
|
self.last_error = error
|
|
self._set_connected(False)
|
|
await self._set_available(False)
|
|
if not await self._wait_before_reconnect(delay_index):
|
|
break
|
|
delay_index = min(delay_index + 1, len(self.reconnect_delays) - 1)
|
|
continue
|
|
|
|
self._writer = writer
|
|
self._set_connected(True)
|
|
received_valid_frame = await self._read_connection(reader)
|
|
if received_valid_frame:
|
|
delay_index = 0
|
|
|
|
self._set_connected(False)
|
|
await self._set_available(False)
|
|
await self._close_writer()
|
|
if not self.running:
|
|
break
|
|
if not await self._wait_before_reconnect(delay_index):
|
|
break
|
|
delay_index = min(delay_index + 1, len(self.reconnect_delays) - 1)
|
|
except asyncio.CancelledError:
|
|
raise
|
|
finally:
|
|
self._set_connected(False)
|
|
await self._set_available(False)
|
|
await self._close_writer()
|
|
|
|
async def _wait_before_reconnect(self, delay_index: int) -> bool:
|
|
if not self.running:
|
|
return False
|
|
self.reconnect_count += 1
|
|
await self._sleep(self.reconnect_delays[delay_index])
|
|
return self.running
|
|
|
|
async def _read_connection(self, reader: asyncio.StreamReader) -> bool:
|
|
received_valid_frame = False
|
|
loop = asyncio.get_running_loop()
|
|
deadline = loop.time() + self.frame_timeout
|
|
|
|
while self.running:
|
|
remaining = deadline - loop.time()
|
|
if remaining <= 0:
|
|
await self._set_available(False)
|
|
deadline = loop.time() + self.frame_timeout
|
|
remaining = self.frame_timeout
|
|
|
|
try:
|
|
data = await asyncio.wait_for(reader.read(self.read_size), remaining)
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except TimeoutError:
|
|
await self._set_available(False)
|
|
deadline = loop.time() + self.frame_timeout
|
|
continue
|
|
except Exception as error:
|
|
self.last_error = error
|
|
return received_valid_frame
|
|
|
|
if not data:
|
|
return received_valid_frame
|
|
|
|
self.bytes_received += len(data)
|
|
try:
|
|
raw_frames = self._extractor.feed(data)
|
|
except Exception as error:
|
|
self.frames_rejected += 1
|
|
self.last_error = error
|
|
continue
|
|
|
|
for raw_frame in raw_frames:
|
|
try:
|
|
frame = parse_frame(raw_frame)
|
|
filtered = filter_values(frame)
|
|
except FrameError as error:
|
|
self.frames_rejected += 1
|
|
self.last_error = error
|
|
continue
|
|
|
|
received_valid_frame = True
|
|
deadline = loop.time() + self.frame_timeout
|
|
await self._accept_frame(frame, filtered)
|
|
|
|
return received_valid_frame
|
|
|
|
async def _accept_frame(self, frame: ParsedFrame, filtered: FilterResult) -> None:
|
|
accepted = dict(filtered.accepted)
|
|
rejected = dict(filtered.rejected)
|
|
frame_type = accepted["frame_type"]
|
|
assert isinstance(frame_type, str)
|
|
|
|
self.frames_received += 1
|
|
self.last_valid_frame = frame
|
|
self.last_frame_type = frame_type
|
|
if frame_type == "frigo":
|
|
self.latest_frigo_data = accepted
|
|
elif frame_type == "regulation":
|
|
self.latest_regulation_data = accepted
|
|
else:
|
|
self.latest_metadata = accepted
|
|
|
|
await self._set_available(True)
|
|
for callback in tuple(self._data_callbacks):
|
|
await self._invoke_callback(callback, frame_type, dict(accepted), dict(rejected))
|
|
|
|
async def _set_available(self, value: bool) -> None:
|
|
if self.available == value:
|
|
return
|
|
self.available = value
|
|
for callback in tuple(self._availability_callbacks):
|
|
await self._invoke_callback(callback, value)
|
|
|
|
async def _invoke_callback(self, callback: Callable[..., object], *args: object) -> None:
|
|
try:
|
|
result = callback(*args)
|
|
if inspect.isawaitable(result):
|
|
await result
|
|
except Exception as error:
|
|
self.last_error = error
|
|
|
|
def _set_connected(self, value: bool) -> None:
|
|
self.connected = value
|
|
if value:
|
|
self._connected_event.set()
|
|
else:
|
|
self._connected_event.clear()
|
|
|
|
async def _close_writer(self) -> None:
|
|
writer = self._writer
|
|
self._writer = None
|
|
if writer is None:
|
|
return
|
|
writer.close()
|
|
try:
|
|
await writer.wait_closed()
|
|
except Exception as error:
|
|
self.last_error = error
|