# The connector contract

Connectors integrate devices and services through typed messages.
This reference covers the contract the engine supports; start with
[Writing a connector](../code/connectors.md) for a working package.

## Contract version

```python
from fluksio.flow.connector import CONTRACT_VERSION, ConnectorNode


class MyConnector(ConnectorNode):
    contract = CONTRACT_VERSION
    title = "My connector"
    description = "Reads a device."
```

Declare `contract` on the class itself: inheritance does not count.
The current contract version is `1`. A mismatch is logged and the connector
is skipped. Breaking changes bump the version; compatible additions do not.

`title` defaults to the entry point name; `description` may be empty.

## Packaging and discovery

Register the class in your package's `pyproject.toml`:

```toml
[project.entry-points."fluksio.node_types"]
my_device = "fluksio_connector_mydevice:MyConnector"
```

The entry point name is the stored node type. Names already registered,
including built-in types, are skipped. The editor shows the package name
and version from its installed metadata.

Discovery runs at engine startup. Install connectors in the engine's
environment and restart after installation or code changes.

## Parameters

Subclass `ConnectorNode.Params` to keep `poll_interval` and add your settings:

```python
from pydantic import Field


class Params(ConnectorNode.Params):
    host: str = "localhost"
    api_key: str | None = Field(
        default=None, json_schema_extra={"x-secret": True}
    )
```

The node panel builds a form from this model's JSON schema: switches for
booleans, inputs for numbers and strings, and JSON fields for objects and
arrays. Parsed settings are available as `self.config`.

Mark credentials with `x-secret`. The editor stores a reference such as
`{"$secret": "device-key"}`, and the engine resolves it while building the node.

## Ports and messages

The flow author declares ports as `MessageSpec` values. Each has a message
name, a `DType`, and optional delivery settings.

- `poll()` returns values keyed by **output port**.
- `write()` receives values keyed by **input port**.
- The base class maps local port names to message names and validates types.
- Unknown output keys fail the reading. Return only declared ports.
- Port `interval` limits are enforced by the engine; the connector controls
  how often it reads the device.

For devices with variable readings, filter to the configured outputs:

```python
declared = {spec.port for spec in self.output_ports if spec.name}
return {port: value for port, value in reading.items() if port in declared}
```

Log missing device fields once so a typo does not leave a port silently empty.

## Polling

```python
async def poll(self):
    return {"reading": await self._read_device()}
```

Set `poll_interval` in seconds; `0` disables polling. Return `None` when
there is nothing new. The base loop publishes only changed values and retries
a failed reading or publication on the next tick. It reports connection health
around each poll.

`poll()` runs on the event loop. Use `asyncio.to_thread` for blocking reads.
The base loop already calls `inject` on a worker thread.

For pushed readings, leave `poll_interval=0`, manage the subscription in
`start()` and `stop()`, and call `self.inject({...})` when values arrive.
Offload `inject` too when calling it from the event loop.

## Writing

```python
def write(self, **ports):
    self._device.set(ports["level"])
    return None
```

Settings stay on `self.config`; arguments contain input ports only.
Return `None`, or a dictionary keyed by output port if the device responds
with useful values.

Writes run on a scheduler thread. Return promptly or queue slow work.
Polling and writing are independent; a connector may support both.

Set `idempotent = False` for commands that must not be redelivered after a
crash. Leave `write()` unchanged on a read-only connector.

## Devices whose readings are bytes

```python
import asyncio


async def poll(self):
    jpeg = await asyncio.to_thread(self._grab)
    return {
        "frame": self.save_artifact(
            jpeg, "frame.jpg", media_type="image/jpeg", volatile=True
        )
    }
```

`save_artifact(data, name="", media_type="application/octet-stream",
volatile=False)` returns a reference for an artifact or media port.
Match the media type to an `image`, `audio`, or `video` port.

The artifact store is available after `start()`. Use `volatile=True`
for disposable live frames: they go to a bounded memory ring and expire as
new frames arrive. Use durable storage for recordings.

Local screens receive pushed live frames; portal screens fetch the bytes.
Choose a modest poll rate for remote viewing.

`self.wanted(port)` says whether anyone wants what a port publishes right
now. That is a dashboard tile on screen, a portal screen, or a node of a
running flow reading its message. Check it before fetching a frame so a camera
sends nothing while nobody watches:

```python
async def poll(self):
    if not self.wanted("frame"):
        return None
    ...
```

On a remote worker it is always true, since the worker cannot see who watches.

## Keeping state across restarts

Keep connections and device handles in node attributes. Store short strings
such as pairing tokens or pinned certificates in `self.store`:

```python
value = await asyncio.to_thread(self.store.get, "token:device")
await asyncio.to_thread(self.store.set, "token:device", "value")
await asyncio.to_thread(self.store.delete, "token:device")
```

- Keys and values are strings. `get` returns `None` for a missing key;
  `delete` returns whether it existed.
- All nodes from the connector's top-level package share the namespace.
  Include a device identifier in each key.
- Storage is in the engine's database and is available before `start()`.
- Keys allow 1–200 characters, without `/` or whitespace. Values allow
  64 KiB and each package allows 256 keys; exceeding a limit raises `ValueError`.
- Calls block on the database: use `asyncio.to_thread` on the event loop.

Superusers can list keys, delete one, or clear them all under **Stored state**
in the node panel. Values are never returned. The equivalent API paths are:

| Method | Path |
|---|---|
| `GET` | `/api/v1/connectors/{node_type}/state` |
| `DELETE` | `/api/v1/connectors/{node_type}/state/{key}` |
| `DELETE` | `/api/v1/connectors/{node_type}/state` |

Handle missing keys: clearing one asks the connector to pair or learn again.

### Pinning a device's certificate

`fluksio.flow.connector_tls` supports trusting a local device's certificate
on first contact:

```python
import httpx

from fluksio.flow.connector_tls import (
    context_for,
    pinned_certificate,
    raise_if_changed,
)

address = f"{self.config.host}:{self.config.port}"
pem = pinned_certificate(self.store, address, self.config.timeout)
try:
    with httpx.Client(verify=context_for(pem)) as client:
        response = client.get(f"https://{address}/status")
        response.raise_for_status()
except httpx.ConnectError as exc:
    raise_if_changed(exc, pem, address, self.config.timeout, "my_device")
    raise
```

Run this blocking work off the event loop. First contact pins the certificate
under `cert:<address>`; a changed certificate is refused before sending a
credential or command. `raise_if_changed` reports `CertificateChanged`
with the fingerprint and key to clear.

Pins trust the exact certificate, including after its expiry. Clearing the
key accepts the certificate presented on the next connection.

## Lifecycle and health

```python
async def start(self, app=None):
    await super().start(app)


async def stop(self, app=None):
    await super().stop(app)
```

`start` runs when a flow starts or rebuilds. `stop` runs before rebuilding
or stopping, even after a failed start; make it idempotent. Call `super()`
when overriding these hooks if you want the base polling loop.

The optional `app` is the FastAPI application, for connectors that mount
routes. Most connectors can ignore it.

Report connection changes if you manage your own subscription:

```python
self.report_health("ok")
self.report_health("degraded", "Some readings timed out")
self.report_health("down", "Connection lost")
```

Repeated statuses do not emit another event. `down` appears among flow issues
and on Home; `degraded` means the connection still works. Loading failures
are reported separately.

Keep shared flow values in messages, credentials in secret parameters, and
blocking I/O off the event loop.