Skip to content

Ports (Protocols)

The protocol contracts that decouple the core from infrastructure adapters. Concrete adapters live in haclient.infra and implement these interfaces.

ports

Ports — protocol definitions decoupling the core from infrastructure.

The core services and domain code depend only on these protocols. Concrete implementations (e.g. aiohttp-based REST and WebSocket clients) live under haclient.infra and are wired in by HAClient. This is the hexagonal boundary: infrastructure adapters implement these ports; nothing in the core layer imports from haclient.infra.

Notes

The protocols intentionally describe a minimal surface. Adapters may expose additional helpers, but the core promises only to use what is declared here.

EventHandler module-attribute 

EventHandler = Callable[[dict[str, Any]], Awaitable[None] | None]

Callable invoked with a single Home Assistant event payload.

DisconnectListener module-attribute 

DisconnectListener = Callable[[], Awaitable[None] | None]

Callable invoked when the WebSocket connection drops.

ReconnectListener module-attribute 

ReconnectListener = Callable[[], Awaitable[None] | None]

Callable invoked after a successful WebSocket reconnect.

RestPort

Bases: Protocol

REST transport contract used by the core.

Implementations talk to the Home Assistant /api/ HTTP endpoints.

Source code in src/haclient/ports.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
@runtime_checkable
class RestPort(Protocol):
    """REST transport contract used by the core.

    Implementations talk to the Home Assistant ``/api/`` HTTP endpoints.
    """

    @property
    def base_url(self) -> str:
        """Return the configured Home Assistant base URL."""
        ...

    async def get_states(self) -> list[dict[str, Any]]:
        """Return all current entity states."""
        ...

    async def get_state(self, entity_id: str) -> dict[str, Any] | None:
        """Return a single entity state, or ``None`` if not found."""
        ...

    async def call_service(
        self,
        domain: str,
        service: str,
        data: dict[str, Any] | None = None,
    ) -> list[dict[str, Any]]:
        """Invoke a Home Assistant service via the REST API."""
        ...

    async def close(self) -> None:
        """Release any resources owned by the adapter."""
        ...

base_url property 

base_url: str

Return the configured Home Assistant base URL.

get_states async 

get_states() -> list[dict[str, Any]]

Return all current entity states.

Source code in src/haclient/ports.py 
44
45
46
async def get_states(self) -> list[dict[str, Any]]:
    """Return all current entity states."""
    ...

get_state async 

get_state(entity_id: str) -> dict[str, Any] | None

Return a single entity state, or None if not found.

Source code in src/haclient/ports.py 
48
49
50
async def get_state(self, entity_id: str) -> dict[str, Any] | None:
    """Return a single entity state, or ``None`` if not found."""
    ...

call_service async 

call_service(domain: str, service: str, data: dict[str, Any] | None = None) -> list[dict[str, Any]]

Invoke a Home Assistant service via the REST API.

Source code in src/haclient/ports.py 
52
53
54
55
56
57
58
59
async def call_service(
    self,
    domain: str,
    service: str,
    data: dict[str, Any] | None = None,
) -> list[dict[str, Any]]:
    """Invoke a Home Assistant service via the REST API."""
    ...

close async 

close() -> None

Release any resources owned by the adapter.

Source code in src/haclient/ports.py 
61
62
63
async def close(self) -> None:
    """Release any resources owned by the adapter."""
    ...

WebSocketPort

Bases: Protocol

WebSocket transport contract used by the core.

Implementations talk to the Home Assistant /api/websocket endpoint and handle authentication, reconnect, and event subscription.

Source code in src/haclient/ports.py 
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
@runtime_checkable
class WebSocketPort(Protocol):
    """WebSocket transport contract used by the core.

    Implementations talk to the Home Assistant ``/api/websocket`` endpoint
    and handle authentication, reconnect, and event subscription.
    """

    @property
    def connected(self) -> bool:
        """``True`` while the underlying socket is open."""
        ...

    async def connect(self) -> None:
        """Open the connection and authenticate."""
        ...

    async def close(self) -> None:
        """Close the connection and stop background tasks."""
        ...

    async def send_command(
        self,
        payload: dict[str, Any],
        *,
        timeout: float | None = None,
    ) -> Any:
        """Send a command and await its response."""
        ...

    async def subscribe_events(
        self,
        handler: EventHandler,
        event_type: str | None = None,
    ) -> int:
        """Subscribe to a Home Assistant event stream."""
        ...

    async def unsubscribe(self, subscription_id: int) -> None:
        """Cancel a previously registered subscription."""
        ...

    def on_disconnect(self, handler: DisconnectListener) -> DisconnectListener:
        """Register a listener for connection drops."""
        ...

    def on_reconnect(self, handler: ReconnectListener) -> ReconnectListener:
        """Register a listener for successful reconnections."""
        ...

connected property 

connected: bool

True while the underlying socket is open.

connect async 

connect() -> None

Open the connection and authenticate.

Source code in src/haclient/ports.py 
79
80
81
async def connect(self) -> None:
    """Open the connection and authenticate."""
    ...

close async 

close() -> None

Close the connection and stop background tasks.

Source code in src/haclient/ports.py 
83
84
85
async def close(self) -> None:
    """Close the connection and stop background tasks."""
    ...

send_command async 

send_command(payload: dict[str, Any], *, timeout: float | None = None) -> Any

Send a command and await its response.

Source code in src/haclient/ports.py 
87
88
89
90
91
92
93
94
async def send_command(
    self,
    payload: dict[str, Any],
    *,
    timeout: float | None = None,
) -> Any:
    """Send a command and await its response."""
    ...

subscribe_events async 

subscribe_events(handler: EventHandler, event_type: str | None = None) -> int

Subscribe to a Home Assistant event stream.

Source code in src/haclient/ports.py 
 96
 97
 98
 99
100
101
102
async def subscribe_events(
    self,
    handler: EventHandler,
    event_type: str | None = None,
) -> int:
    """Subscribe to a Home Assistant event stream."""
    ...

unsubscribe async 

unsubscribe(subscription_id: int) -> None

Cancel a previously registered subscription.

Source code in src/haclient/ports.py 
104
105
106
async def unsubscribe(self, subscription_id: int) -> None:
    """Cancel a previously registered subscription."""
    ...

on_disconnect 

on_disconnect(handler: DisconnectListener) -> DisconnectListener

Register a listener for connection drops.

Source code in src/haclient/ports.py 
108
109
110
def on_disconnect(self, handler: DisconnectListener) -> DisconnectListener:
    """Register a listener for connection drops."""
    ...

on_reconnect 

on_reconnect(handler: ReconnectListener) -> ReconnectListener

Register a listener for successful reconnections.

Source code in src/haclient/ports.py 
112
113
114
def on_reconnect(self, handler: ReconnectListener) -> ReconnectListener:
    """Register a listener for successful reconnections."""
    ...

Clock

Bases: Protocol

Loop / scheduling contract used by entities and event dispatchers.

Decouples scheduling of background coroutines from the concrete event loop, which makes it easy to substitute during testing.

Source code in src/haclient/ports.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
@runtime_checkable
class Clock(Protocol):
    """Loop / scheduling contract used by entities and event dispatchers.

    Decouples scheduling of background coroutines from the concrete event
    loop, which makes it easy to substitute during testing.
    """

    def loop(self) -> asyncio.AbstractEventLoop | None:
        """Return the running event loop, or ``None`` if unavailable."""
        ...

    def schedule(self, coro: Awaitable[Any]) -> None:
        """Run *coro* on the loop without blocking the caller."""
        ...

loop 

loop() -> asyncio.AbstractEventLoop | None

Return the running event loop, or None if unavailable.

Source code in src/haclient/ports.py 
125
126
127
def loop(self) -> asyncio.AbstractEventLoop | None:
    """Return the running event loop, or ``None`` if unavailable."""
    ...

schedule 

schedule(coro: Awaitable[Any]) -> None

Run coro on the loop without blocking the caller.

Source code in src/haclient/ports.py 
129
130
131
def schedule(self, coro: Awaitable[Any]) -> None:
    """Run *coro* on the loop without blocking the caller."""
    ...