sigen

Async Python client for the Sigenergy cloud API. Provides three independent interfaces:

• Sigen — End-user API client (OAuth2 login, energy flow, modes, smart loads)
• NorthboundClient — Developer/Northbound REST API (key-based auth, mode switching, onboard/offboard)
• SigenMQTT — Real-time MQTT telemetry and battery command dispatch

Replaces the unmaintained PyPI sigen package (v3.0.2) with async/await, proper error handling, and MQTT + Northbound support.

Installation

pip install -e /path/to/sigen-api

Requires Python 3.10+. Dependencies: aiohttp, pycryptodome, aiomqtt.

Sigen Client (End-User API)

OAuth2 authentication with username/password. For interactive use and energy monitoring.

from sigen import Sigen

api = Sigen("user@example.com", "password", region="eu")
await api.async_initialize()

# Energy flow
flow = await api.get_energy_flow()

# Operational modes
mode = await api.get_operational_mode()       # current mode label
modes = await api.get_operational_modes()      # all available modes
await api.set_operational_mode(mode=1, profile_id=-1)

# Dynamic mode methods (created during init from available modes)
await api.set_operational_mode_tou()

# Smart loads
loads = await api.get_smart_loads()
await api.set_smart_load_state(load_path=123, state=1)  # on
await api.set_smart_load_state(load_path=123, state=0)  # off

# Dynamic smart load methods (created during init)
await api.enable_smart_load_heat_pump()
await api.disable_smart_load_heat_pump()

# Northbound via Sigen client
await api.init_northbound()
mode = await api.nb_query_mode()
await api.nb_switch_mode(mode=8)

NorthboundClient (Developer API)

REST API for system control. Two auth methods: user/password or app key/secret.

from sigen import NorthboundClient

# Key-based auth (used by VPP optimizer and collector)
nb = await NorthboundClient.from_app_key(
    base_url="https://api-eu.sigencloud.com/",
    app_key="your-app-key",
    app_secret="your-app-secret",
)

# Or user/password auth
nb = NorthboundClient("https://api-eu.sigencloud.com/", "user", "password")
await nb.login()

# System management
await nb.onboard(["system-id-1", "system-id-2"])
await nb.offboard(["system-id-1"])

# Mode control
result = await nb.query_mode("system-id")       # GET settings
await nb.switch_mode("system-id", mode=8)        # PUT settings

Northbound Modes

Mode	Value	Description
NBMode.MSC	0	Maximum Self-Consumption
NBMode.FFG	5	Fully Feed-in to Grid
NBMode.VPP	6	VPP
NBMode.NBI	8	North Bound Interface (required for battery commands)

from_app_key() Instantiation

NorthboundClient.from_app_key() uses cls.__new__() to bypass __init__ — key-based auth doesn't need username/password. The resulting instance has username=None, password=None, and stores _app_key/_app_secret for token refresh.

SigenMQTT (Real-Time Telemetry)

MQTT client for real-time telemetry subscription and battery command dispatch.

from sigen import SigenMQTT, TelemetryData

mqtt = SigenMQTT(
    app_key="your-app-key",
    app_secret="your-app-secret",
    app_identifier="your-app-identifier",
    system_ids=["system-id-1", "system-id-2"],
    ca_cert_path="/path/to/ca.pem",
    # Optional:
    client_cert_path="/path/to/client.pem",   # for mTLS
    client_key_path="/path/to/client.key",
    broker="mqtt-eu.sigencloud.com",           # default
    port=8883,                                 # default (TLS)
    base_url="https://api-eu.sigencloud.com/", # for token auth
)

# Listen for telemetry (blocks indefinitely, auto-reconnects)
async def on_telemetry(data: TelemetryData):
    print(f"PV={data.pv_power_kw}kW SOC={data.soc_percent}%")

await mqtt.listen(
    on_telemetry=on_telemetry,
    on_system=None,    # optional: system change events
    on_alarm=None,     # optional: alarm events
)

TelemetryData

Parsed telemetry with power values in kW:

Field	Type	Sign Convention
pv_power_kw	float	Always >= 0
battery_power_kw	float	Positive = discharging, negative = charging (inverted from Sigen's convention)
grid_power_kw	float	Positive = importing, negative = exporting
load_power_kw	float	Always >= 0
soc_percent	float	0-100
system_id	str	Source system
timestamp	str	From Sigen
raw	dict	Original values dict

Battery Commands

Send time-ahead battery schedules via MQTT (max 24 commands per batch):

commands = [
    {
        "systemId": "system-id",
        "startTime": 1709283600,      # unix timestamp
        "duration": 60,                # minutes
        "activeMode": "discharge",     # or "charge", "selfConsumption"
        "chargingPower": 5.0,          # kW
        "maxSellPower": 12.0,          # connection limit
        "dischargePriorityType": "BATTERY",
    },
]
await mqtt.send_battery_commands(commands)

MQTT Topics

Topic	Direction	Purpose
openapi/subscription/period	Publish	Subscribe to periodic telemetry
openapi/subscription/change	Publish	Subscribe to system changes
openapi/subscription/alarm	Publish	Subscribe to alarms
openapi/period/{appKey}/{systemId}	Subscribe	5-min telemetry data
openapi/change/{appKey}/{systemId}	Subscribe	System state changes
openapi/alarm/{appKey}/{systemId}	Subscribe	Alarm events
openapi/instruction/command	Publish	Battery charge/discharge commands

Connection Behavior

• Authenticates via NorthboundClient.from_app_key() before connecting
• Auto-onboards system IDs on connect
• Auto-reconnects on connection loss (30s backoff for MQTT errors, 60s for unexpected errors)
• Token refresh handled automatically with 1-hour safety margin
• Logs raw telemetry field names on first message for diagnostics
• TLS required; relaxes VERIFY_X509_STRICT for Sigen's CA cert (lacks keyUsage extension)

Regions

Region	Base URL
eu	https://api-eu.sigencloud.com/
cn	https://api-cn.sigencloud.com/
apac	https://api-apac.sigencloud.com/
us	https://api-us.sigencloud.com/

Project Structure

src/sigen/
├── __init__.py          # Public exports
├── client.py            # Sigen end-user client (orchestrates all below)
├── auth.py              # AES password encryption, OAuth2 TokenManager
├── northbound.py        # NorthboundClient: REST API for developer access
├── mqtt.py              # SigenMQTT: telemetry + battery commands
├── constants.py         # Region URLs, AES keys, OAuth constants, NBMode
├── exceptions.py        # SigenError → SigenAuthError, SigenAPIError
├── station.py           # Station info endpoint
├── energy.py            # Energy flow endpoint
├── modes.py             # Operational mode endpoints + dynamic method creation
└── smart_loads.py       # Smart load list, details, consumption, control

Error Handling

SigenError (base)
├── SigenAuthError          # Bad credentials, login failures
│   └── SigenTokenExpiredError  # Token expired and refresh failed
└── SigenAPIError           # Non-auth API failures (includes status_code, response_body)

Gotchas

• Battery sign inversion: Sigen sends +charge/-discharge; TelemetryData.from_mqtt_payload() inverts to +discharge/-charge to match VPP convention
• Wildcard MQTT subscriptions blocked: Sigen broker requires exact topic subscriptions per system ID
• app_key vs app_identifier: app_key is used as MQTT username and in topic paths; app_identifier is not used for MQTT
• Double-encoded JSON: Northbound login response has data as a JSON string inside JSON — parsed with json.loads(data_str)
• Sigen CA cert: Missing keyUsage extension, requires relaxed OpenSSL 3.4+ verification
• No historical API: Sigen provides real-time data only; use Pstryk for historical meter data