Metadata-Version: 2.4
Name: helin-edge-sdk
Version: 0.0.2
Summary: Receive and respond to Helin Platform edge-module-requests.
License: MIT
License-File: LICENSE
Keywords: iot,azure-iot-edge,edge,helin,helin platform,configuration
Author: Helin
Author-email: support@helindata.com
Requires-Python: >=3.11,<4.0
Classifier: Development Status :: 4 - Beta
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: POSIX :: Linux
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Communications
Requires-Dist: paho-mqtt (>=2.1.0,<3.0.0)
Project-URL: Homepage, https://www.helindata.com
Description-Content-Type: text/markdown

# helin-edge-sdk
The Helin Edge SDK allows you to securely and reliably get and set JSON payloads via the Helin Platform's **edge-module-requests** API: `get_configuration`, `set_configuration` and `get_health` to your containers (modules) deployed on Nodes on the edge. 

For more information see https://docs.helinplatform.com

## Install

```bash
pip install helin-edge-sdk
```

## Usage

Implement the three methods and run the client:

```python
from typing import Any
from helin_edge_sdk import EdgeModuleRequestsHandler, EdgeModuleClient, ConfigurationFailedError

class MyModule(EdgeModuleRequestsHandler):
    def __init__(self) -> None:
        self.configuration = {"opcua_ip": "123.34.233.1", "opcua_port": 4840, "poll_interval_ms": 1000}

    def on_get_configuration(self, payload: dict[str, Any]) -> Any:
        return self.configuration

    def on_set_configuration(self, payload: dict[str, Any]) -> Any:
        cfg = payload.get("configuration")
        if cfg is None:
            raise ConfigurationFailedError("missing configuration", status_code=400,
                                           current_configuration=self.configuration)
        self.configuration = cfg
        return {"configuration": cfg}

    def on_get_health(self, payload: dict[str, Any]) -> Any:
        return {"metrics": [{"name": "opcua_connected", "value": 1}]}

EdgeModuleClient.from_edge_environment(MyModule()).run() #Fetches all required env variables on your provisioned Node
```

Each callback receives the **decompressed** request payload and returns a JSON-serializable value, which is wrapped as 
```{"status": 200, "response": <value>}```

compressed, and published back to the caller. 

Raise `ConfigurationFailedError` to return the configuration-failure envelope:

```{"status": <code>, "response": {"error": ..., "current_configuration": ...}}```.

## Status codes

| Handler outcome | Status |
| --- | --- |
| Returns a value | 200 |
| Raises `ValueError` | 400 |
| Raises `ConnectionError` | 400 |
| Unknown method | 400 |
| Raises `FileNotFoundError` | 404 |
| Raises `ConfigurationFailedError` | its `status_code` (default 409) |
| Raises any other exception | 500 |

## Local testing

```bash
# 1. start a plaintext broker
docker compose -f docker/docker-compose.test.yml up -d

# 2. run the module against it
python -m examples.run_local

# 3. invoke a method (simulates the Platform API)
python -m examples.publish_method get_health
python -m examples.publish_method set_configuration '{"configuration": {"opcua_ip": "10.0.0.5", "opcua_port": 4840}}'
```

## Develop

```bash
poetry install
poetry run pytest -m "not integration"   # unit tests, no broker
poetry run pytest                        # all tests (needs the broker above)
poetry run ruff check
poetry run ty check
```

## Architecture

```mermaid
flowchart TB
    subgraph platform["Cloud"]
        API["Platform API <br/> POST /api/v1/edge-module-requests/nodes/{node_id}/modules/{module_id} <br/> GET /api/v1/edge-module-requests/nodes/{node_id}/modules/{module_id} <br/> GET /api/v1/edge-module-requests/nodes/{node_id}/modules/{module_id}/health"]
    end

    subgraph hub["Azure IoT Hub (Cloud)"]
        IOTHUB
    end

    subgraph edge["Node — Edge runtime"]
        EDGEHUB["edgeHub<br/>MQTT broker :8883 (TLS)"]
        WL["Edge workload API<br/>unix:///…/workload.sock<br/>/sign · /trust-bundle"]

        subgraph mod["Module using helin-edge-sdk"]
            direction TB
            CLIENT["EdgeModuleClient<br/>from_edge_environment()"]
            LISTENER["MqttModuleListener<br/>(paho-mqtt transport)"]
            HANDLER["EdgeModuleRequestsHandler<br/>(your on_* callbacks)"]
            CLIENT --> LISTENER
            LISTENER -- "dispatch(method, payload)" --> HANDLER
            HANDLER -- "result / ConfigurationFailedError" --> LISTENER
        end
    end

    API <== "edge-module-requests<br/>(gzip payload)" ==> IOTHUB
    IOTHUB <== "cloud uplink" ==> EDGEHUB

    EDGEHUB -- "SUB - (gzip payload)" --> LISTENER
    LISTENER -- "PUB - (gzip payload)" --> EDGEHUB

    LISTENER -- "SAS token + trust bundle" --> WL
```
