kfj001
b60b32830f
AgateClient takes address= default; attach_local accepts address/host/gateway_ip per call with join control. An explicit address skips the SoftAP radio join by default (join=True forces it). derive_hotspot_credentials and join_device_hotspot accept a gateway_ip override. battery/power host defaults now use DEVICE_GATEWAY_IP. Verified live at 192.168.1.142:9000.
110 lines
4.7 KiB
Python
110 lines
4.7 KiB
Python
"""
|
|
Agate Client - Main interface for AGATE device operations.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Optional
|
|
|
|
from .bluetooth import BluetoothManager
|
|
from .wifi import DEVICE_GATEWAY_IP, HotspotProfile, WifiBackend, WifiManager
|
|
|
|
|
|
class AgateClient:
|
|
"""Main client for FranklinWH AGATE device operations.
|
|
|
|
:param device_id: optional default device identifier.
|
|
:param wifi_backend: platform backend used for hotspot joins.
|
|
:param address: optional default aGate address (IP or hostname).
|
|
Overrides the ``10.100.1.1`` SoftAP default for direct-connect
|
|
calls — e.g. the home-LAN address (``192.168.1.x``) once the unit
|
|
is on the router. Per-call ``address``/``host``/``gateway_ip``
|
|
arguments take precedence over this default.
|
|
"""
|
|
|
|
def __init__(self, device_id: Optional[str] = None,
|
|
wifi_backend: Optional[WifiBackend] = None,
|
|
address: Optional[str] = None) -> None:
|
|
self.device_id = device_id
|
|
self.bluetooth_manager: Optional[BluetoothManager] = None
|
|
self.wifi_manager = WifiManager(backend=wifi_backend)
|
|
self.address = address
|
|
|
|
# -- device hotspot (local connect) -------------------------------------
|
|
def join_hotspot(self, sn: str, gateway_ip: Optional[str] = None,
|
|
**kwargs: Any) -> HotspotProfile:
|
|
"""Join the device SoftAP (derives creds, joins, verifies).
|
|
|
|
``gateway_ip`` overrides the address recorded on the profile
|
|
without changing the radio join.
|
|
"""
|
|
return self.wifi_manager.join_device_hotspot(
|
|
sn, gateway_ip=gateway_ip or self.address, **kwargs)
|
|
|
|
def leave_hotspot(self) -> None:
|
|
"""Disconnect from the device hotspot."""
|
|
self.wifi_manager.leave_device_hotspot()
|
|
|
|
def attach_local(self, sn: str, *, verify_join: bool = True,
|
|
settle_timeout: float = 30.0, poll_interval: float = 1.0,
|
|
address: Optional[str] = None,
|
|
host: Optional[str] = None,
|
|
gateway_ip: Optional[str] = None,
|
|
join: Optional[bool] = None,
|
|
**local_kwargs: Any):
|
|
"""Join the device hotspot, then return a ``LocalDeviceClient``.
|
|
|
|
Without an explicit address this keeps the existing behaviour:
|
|
join the SoftAP and aim at the profile gateway
|
|
(``https://10.100.1.1:18000`` by default).
|
|
|
|
Pass ``address`` (``host``/``gateway_ip`` are aliases) to reach the
|
|
aGate elsewhere — e.g. its home-LAN IP once it is on the router.
|
|
An explicit address defaults to *skipping* the hotspot radio join
|
|
(pass ``join=True`` to force it); ``join=False`` never touches the
|
|
radio. The client-level ``address`` default applies when no
|
|
per-call address is given.
|
|
|
|
``verify_join`` gates the association check (TLS verification of the
|
|
device API is a separate concern — pass ``verify=`` in
|
|
``local_kwargs``). Raises
|
|
:class:`~franklinwh_agate_client.wifi.HotspotJoinError` if the
|
|
association never verifies. (``LocalDeviceClient`` imports
|
|
``requests``; install it before calling this on a cloud-less host.)
|
|
|
|
Extra keyword arguments are passed through to ``LocalDeviceClient``
|
|
(``port``, ``scheme``, ``verify``, ``socket_port``, ``key_radix``,
|
|
``timeout``, ``extra_headers``, ``session``).
|
|
"""
|
|
explicit = address if address is not None else (
|
|
host if host is not None else gateway_ip)
|
|
target = explicit if explicit is not None else self.address
|
|
do_join = (target is None) if join is None else join
|
|
if do_join:
|
|
profile = self.wifi_manager.join_device_hotspot(
|
|
sn, verify=verify_join, settle_timeout=settle_timeout,
|
|
poll_interval=poll_interval,
|
|
gateway_ip=target,
|
|
)
|
|
target = profile.gateway_ip
|
|
from .device_local import LocalDeviceClient
|
|
|
|
return LocalDeviceClient(sn=sn, host=target or DEVICE_GATEWAY_IP,
|
|
**local_kwargs)
|
|
|
|
# -- cloud / stub operations (pending RE) -------------------------------
|
|
def query_agate_info(self, device_id: Optional[str] = None):
|
|
"""Query AGATE device information."""
|
|
# Reverse engineered operation: queryAgateInfo
|
|
pass
|
|
|
|
def query_device_info_http(self, device_id: Optional[str] = None):
|
|
"""Query device info via HTTP."""
|
|
# Reverse engineered operation: queryDeviceInfoHttp
|
|
pass
|
|
|
|
def query_modes_by_http(self, device_id: Optional[str] = None):
|
|
"""Query operating modes via HTTP."""
|
|
# Reverse engineered operation: queryModesByHttp
|
|
pass
|