Zum Inhalt

API Referenz - TC100 Display

Vollständige API-Dokumentation für AWTRIX3 Display-Integration.

Übersicht

Die Display-Integration bietet zwei API-Ebenen:

  1. Python Display Service - High-Level Abstraktion
  2. AWTRIX3 HTTP API - Low-Level Direct Access

1. Python Display Service API

Initialisierung

from app.services.display_service import DisplayService

# Mit Settings
from app.config import get_settings
settings = get_settings()

display = DisplayService(
    display_ip=settings.display_ip,
    enabled=settings.display_enabled
)

# Manuell
display = DisplayService(
    display_ip="192.168.178.48",
    enabled=True
)

send_notification()

Sendet eine temporäre Notification an das Display.

Signatur

async def send_notification(
    self,
    text: str,
    color: Optional[List[int]] = None,
    icon: Optional[int] = None,
    duration: int = 15,
    **kwargs
) -> bool

Parameter

Parameter Typ Pflicht Standard Beschreibung
text str ✅ - Text auf Display
color List[int] ❌ [0,255,255] RGB-Farbe [R,G,B]
icon int ❌ 4644 AWTRIX Icon ID
duration int ❌ 15 Anzeigedauer (Sekunden)
hold bool ❌ False Bleibt bis Button-Druck
rainbow bool ❌ False Regenbogen-Effekt
pushIcon int ❌ 0 Icon-Animation (0/1/2)
repeat int ❌ 1 Wiederholungen

Rückgabewert

  • True - Erfolgreich gesendet
  • False - Fehler beim Senden

Beispiele

# Einfache Notification
await display.send_notification(
    text="⚡ Produktion gestartet"
)

# Mit Custom-Farbe
await display.send_notification(
    text="🌞 Hohe Produktion!",
    color=[255, 215, 0],  # Gold
    duration=30
)

# Permanente Notification
await display.send_notification(
    text="🚨 Wechselrichter offline!",
    color=[255, 0, 0],
    icon=29105,
    hold=True  # Bleibt bis Button-Druck
)

# Regenbogen-Effekt
await display.send_notification(
    text="🎉 System gestartet",
    rainbow=True,
    duration=10
)

send_warning()

Sendet eine severity-basierte Warnung.

Signatur

async def send_warning(
    self,
    message: str,
    severity: str = "warning",
    title: Optional[str] = None,
    icon: Optional[int] = None
) -> bool

Parameter

Parameter Typ Pflicht Standard Beschreibung
message str ✅ - Warnungs-Text
severity str ❌ "warning" info, warning, critical
title str ❌ None Optionaler Titel
icon int ❌ Auto Auto-Select basierend auf Severity

Severity-Mapping

Severity Farbe Icon Hold Verwendung
info 🟢 Grün 4644 ❌ Informationen
warning 🟠 Orange 29105 ❌ Warnungen
critical 🔴 Rot 29105 ✅ Kritische Fehler

Beispiele

# Info
await display.send_warning(
    message="System Update verfügbar",
    severity="info"
)

# Warnung
await display.send_warning(
    message="Hohe Temperatur: 85°C",
    severity="warning",
    title="Temperatur"
)

# Kritisch (bleibt stehen)
await display.send_warning(
    message="Wechselrichter 1 offline!",
    severity="critical",
    title="Kritisch"
)

update_solar_status()

Aktualisiert Solar-Status am Display.

Signatur

async def update_solar_status(
    self,
    power_w: int,
    daily_kwh: float,
    online_count: int,
    total_count: int,
    duration: int = 0
) -> bool

Parameter

Parameter Typ Pflicht Beschreibung
power_w int ✅ Aktuelle Leistung (Watt)
daily_kwh float ✅ Tagesproduktion (kWh)
online_count int ✅ Anzahl Online-WR
total_count int ✅ Anzahl Gesamt-WR
duration int ❌ Anzeigedauer (0 = permanent)

Format

⚡{power_kW}kW | ☀️{daily_kwh}kWh heute

Farbe

  • 🟢 Grün: Alle Wechselrichter online
  • 🔴 Rot: Mindestens ein WR offline

Beispiele

# Alle online
await display.update_solar_status(
    power_w=3200,        # 3.2 kW
    daily_kwh=8.5,       # 8.5 kWh
    online_count=2,
    total_count=2
)
# Zeigt: "⚡3.2kW | ☀️8.5kWh heute" (Grün)

# Ein WR offline
await display.update_solar_status(
    power_w=1600,
    daily_kwh=8.5,
    online_count=1,
    total_count=2
)
# Zeigt: "⚡1.6kW | ☀️8.5kWh heute" (Rot)

set_indicator()

Steuert die 3 Status-LEDs.

Signatur

async def set_indicator(
    self,
    index: int,
    color: Optional[List[int]] = None,
    fade: int = 0,
    blink: int = 0
) -> bool

Parameter

Parameter Typ Pflicht Standard Beschreibung
index int ✅ - Indicator 1-3
color List[int] ❌ None RGB [R,G,B] (None = aus)
fade int ❌ 0 Fade-Effekt (ms)
blink int ❌ 0 Blink-Intervall (ms)

Positionen

Display (8×32):
┌─────────────────────┐
│                   🟢 │ ← Indicator 1
│                   🟢 │ ← Indicator 2
│                   ⚫ │ ← Indicator 3
└─────────────────────┘

Beispiele

# Indicator 1: Grün, pulsierend
await display.set_indicator(
    index=1,
    color=[0, 255, 0],
    fade=1000  # 1s Fade
)

# Indicator 2: Rot, blinkend
await display.set_indicator(
    index=2,
    color=[255, 0, 0],
    blink=500  # 500ms Intervall
)

# Indicator 3: Ausschalten
await display.set_indicator(
    index=3,
    color=None  # Aus
)

# Alle 3 setzen
for i in [1, 2, 3]:
    await display.set_indicator(
        index=i,
        color=[0, 255, 0],
        fade=1000
    )

get_stats()

Ruft Display-Status ab.

Signatur

async def get_stats(self) -> Optional[dict]

Rückgabewert

{
    "bat": 74,              # Batterie %
    "bat_raw": 1836,        # Rohwert
    "type": 1,              # Display-Typ
    "lux": 145,             # Lichtsensor
    "ldr_raw": 89,          # LDR Rohwert
    "ram": 168304,          # Freier RAM
    "bri": 150,             # Helligkeit
    "temp": 16,             # Temperatur °C
    "hum": 0,               # Luftfeuchtigkeit %
    "uptime": 123456,       # Uptime (Sekunden)
    "wifi_signal": -50,     # WiFi Signal (dBm)
    "messages": 0,          # Anzahl Messages
    "version": "0.98",      # Firmware-Version
    "indicator1": true,     # Indicator 1 Status
    "indicator2": true,     # Indicator 2 Status
    "indicator3": false,    # Indicator 3 Status
    "app": "TimeApp",       # Aktive App
    "uid": "abc123",        # Geräte-ID
    "matrix": true,         # Matrix aktiv
    "ip_address": "192.168.178.48"
}

Beispiel

stats = await display.get_stats()
if stats:
    print(f"🔋 Batterie: {stats['bat']}%")
    print(f"📶 WiFi: {stats['wifi_signal']} dBm")
    print(f"🌡️ Temperatur: {stats['temp']}°C")
    print(f"💡 Helligkeit: {stats['bri']}")
    print(f"📦 Firmware: {stats['version']}")

2. Display Bridge API

Event-basierte Integration zwischen WebSocket und Display.

handle_inverter_update()

Verarbeitet Wechselrichter-Status-Updates.

Signatur

async def handle_inverter_update(data: dict) -> None

Input Data

{
    "inverter_id": 1,
    "status": "offline",  # "online" | "offline"
    "name": "Wechselrichter 1",
    "power_w": 0,
    "voltage_v": 0
}

Verhalten

  • Online → Offline: Sendet kritische Warnung
  • Offline → Online: Sendet Info-Notification
  • Tracking: Speichert letzten Status pro WR

Beispiel

from app.services.display_bridge import handle_inverter_update

# Simuliert WR Offline
await handle_inverter_update({
    "inverter_id": 1,
    "status": "offline",
    "name": "Wechselrichter 1"
})
# Display zeigt: "🚨 Wechselrichter 1 offline!" (Rot, Hold)

handle_production_data()

Verarbeitet Solar-Produktions-Daten.

Signatur

async def handle_production_data(data: dict) -> None

Input Data

{
    "power_w": 3200,
    "daily_kwh": 8.5,
    "inverters": [
        {"id": 1, "status": "online"},
        {"id": 2, "status": "online"}
    ]
}

Verhalten

  1. Aktualisiert Solar-Status auf Display
  2. Setzt Indicators für jeden WR:
  3. 🟢 Grün (pulsierend) = Online
  4. 🔴 Rot (blinkend) = Offline

Beispiel

from app.services.display_bridge import handle_production_data

await handle_production_data({
    "power_w": 3200,
    "daily_kwh": 8.5,
    "inverters": [
        {"id": 1, "status": "online"},
        {"id": 2, "status": "online"}
    ]
})
# Display zeigt:
# - "⚡3.2kW | ☀️8.5kWh heute" (Grün)
# - Indicators 1+2: Grün, pulsierend

handle_error()

Verarbeitet System-Fehler.

Signatur

async def handle_error(error: dict) -> None

Input Data

{
    "message": "Datenbankverbindung verloren",
    "severity": "critical",  # "info" | "warning" | "critical"
    "component": "Database",
    "timestamp": "2025-01-15T10:30:00Z"
}

Verhalten

Sendet send_warning() mit entsprechender Severity.

Beispiel

from app.services.display_bridge import handle_error

await handle_error({
    "message": "Datenbankverbindung verloren",
    "severity": "critical",
    "component": "Database"
})
# Display zeigt: "🚨 Datenbankverbindung verloren" (Rot, Hold)

3. AWTRIX3 HTTP API (Direct)

Base URL

http://192.168.178.48

Endpoints

POST /api/notify

Sendet eine Notification.

Request: 

curl -X POST http://192.168.178.48/api/notify \
  -H "Content-Type: application/json" \
  -d '{
    "text": "Hello World",
    "color": [255, 255, 255],
    "icon": 1234,
    "duration": 15
  }'

Response: 

{"success": true}

POST /api/custom/{name}

Erstellt/Aktualisiert Custom App.

Request: 

curl -X POST http://192.168.178.48/api/custom/solar \
  -H "Content-Type: application/json" \
  -d '{
    "text": "3.2kW",
    "color": [0, 255, 0]
  }'

Custom Apps

Custom Apps müssen in Display WebUI aktiviert werden!

POST /api/indicator{1|2|3}

Steuert Indicators.

Request: 

curl -X POST http://192.168.178.48/api/indicator1 \
  -H "Content-Type: application/json" \
  -d '{
    "color": [0, 255, 0],
    "fade": 1000
  }'

GET /api/stats

Holt Display-Status.

Request: 

curl http://192.168.178.48/api/stats

Response: 

{
  "bat": 74,
  "wifi_signal": -50,
  "temp": 16,
  "version": "0.98"
}

POST /api/power

Display ein-/ausschalten.

Request: 

# Ausschalten
curl -X POST http://192.168.178.48/api/power \
  -H "Content-Type: application/json" \
  -d '{"power": false}'

# Einschalten
curl -X POST http://192.168.178.48/api/power \
  -H "Content-Type: application/json" \
  -d '{"power": true}'

POST /api/rtttl

Spielt RTTTL-Ton.

Request: 

curl -X POST http://192.168.178.48/api/rtttl \
  -H "Content-Type: application/json" \
  -d '{
    "rtttl": "two_short:d=4,o=5,b=100:16e6,16e6"
  }'

4. Integration Beispiele

Produktions-Update Flow

# 1. WebSocket Manager erhält Daten
async def broadcast_production_data(self, data: dict):
    # An WebSocket Clients
    await self._send_message("production_data", data)

    # An Display
    try:
        from app.services.display_bridge import handle_production_data
        await handle_production_data(data)
    except Exception as e:
        logger.warning(f"Display failed: {e}")

# 2. Display Bridge verarbeitet
async def handle_production_data(data: dict):
    settings = get_settings()
    if not settings.display_enabled:
        return

    display = DisplayService(
        display_ip=settings.display_ip,
        enabled=True
    )

    # Solar Status
    await display.update_solar_status(
        power_w=data["power_w"],
        daily_kwh=data["daily_kwh"],
        online_count=sum(1 for inv in data["inverters"] if inv["status"] == "online"),
        total_count=len(data["inverters"])
    )

    # Indicators
    for i, inverter in enumerate(data["inverters"][:3], start=1):
        if inverter["status"] == "online":
            await display.set_indicator(i, color=[0, 255, 0], fade=1000)
        else:
            await display.set_indicator(i, color=[255, 0, 0], blink=500)

Error Handling

try:
    await display.send_notification(text="Test")
except Exception as e:
    logger.error(f"Display error: {e}")
    # Backend läuft weiter

Batch Operations

# Mehrere Indicators gleichzeitig
async def set_all_indicators(color: List[int]):
    tasks = [
        display.set_indicator(i, color=color, fade=1000)
        for i in [1, 2, 3]
    ]
    await asyncio.gather(*tasks)

# Verwendung
await set_all_indicators([0, 255, 0])  # Alle grün

5. Error Codes

HTTP Status Codes

Code Bedeutung Lösung
200 OK Erfolgreich
400 Bad Request JSON prüfen
404 Not Found Endpoint prüfen
500 Server Error Display neu starten
Timeout Keine Antwort IP/Netzwerk prüfen

Python Exceptions

try:
    await display.send_notification(text="Test")
except httpx.TimeoutException:
    logger.error("Display timeout - IP prüfen")
except httpx.ConnectError:
    logger.error("Display nicht erreichbar")
except Exception as e:
    logger.error(f"Unbekannter Fehler: {e}")

6. Performance

Request Timings

Operation Typ Dauer Blocking
send_notification() POST ~50ms ❌
update_solar_status() POST ~50ms ❌
set_indicator() POST ~30ms ❌
get_stats() GET ~100ms ❌

Alle Operationen sind async und blockieren Backend nicht!

Rate Limits

# Display verarbeitet ~20 Requests/Sekunde
# Empfehlung: Max 5 Updates/Sekunde

# Batch statt Loop
await asyncio.gather(
    display.set_indicator(1, [0,255,0]),
    display.set_indicator(2, [0,255,0]),
    display.set_indicator(3, [0,255,0])
)

7. Testing

Unit Tests

import pytest
from app.services.display_service import DisplayService

@pytest.mark.asyncio
async def test_send_notification():
    display = DisplayService(
        display_ip="192.168.178.48",
        enabled=True
    )
    result = await display.send_notification(text="Test")
    assert result is True

@pytest.mark.asyncio
async def test_disabled_display():
    display = DisplayService(
        display_ip="192.168.178.48",
        enabled=False
    )
    result = await display.send_notification(text="Test")
    assert result is False  # Deaktiviert

Integration Tests

# Test-Script
docker exec solarlog-backend python -c "
import asyncio
from app.services.display_service import DisplayService

async def test():
    display = DisplayService('192.168.178.48', True)

    # Test 1
    assert await display.send_notification(text='Test 1')
    await asyncio.sleep(2)

    # Test 2
    assert await display.update_solar_status(3200, 8.5, 2, 2)
    await asyncio.sleep(2)

    # Test 3
    assert await display.set_indicator(1, [0,255,0])

    print('✅ Alle Tests erfolgreich')

asyncio.run(test())
"

Zusammenfassung

High-Level (Python): - send_notification() - Temporäre Nachrichten - send_warning() - Severity-basierte Warnungen - update_solar_status() - Live Solar-Daten - set_indicator() - Status-LEDs - get_stats() - Display-Status

Event-Based (Bridge): - handle_inverter_update() - WR Status - handle_production_data() - Solar Produktion - handle_error() - System Fehler

Low-Level (HTTP): - /api/notify - Notifications - /api/custom/{name} - Custom Apps - /api/indicator{1-3} - Indicators - /api/stats - Status - /api/power - Ein/Aus

Nächste Schritte: