import httpx
import json
from datetime import datetime, date
from typing import List, Dict, Any, Optional
# Globals werden zur Laufzeit bereitgestellt
# EXTERNAL_BASE_URL: str
# AUTH_HEADERS: dict
# PARAMS: dict
def execute_graphql(query: str, variables: Optional[Dict] = None) -> Dict[str, Any]:
"""Führt eine GraphQL-Abfrage aus"""
try:
response = httpx.post(
f"{EXTERNAL_BASE_URL}/graphql",
json={"query": query, "variables": variables or {}},
headers=AUTH_HEADERS,
timeout=30
)
response.raise_for_status()
return response.json()
except Exception as e:
return {"errors": [{"message": f"GraphQL-Fehler: {str(e)}"}]}
def get_sensors_for_meter(meter_number: str) -> List[Dict]:
"""Sucht Sensoren für eine Zählernummer"""
query = """
query GetSensorsForMeter($meterNumber: String!) {
sensorsForMeterNumber(meterNumber: $meterNumber) {
sensorId
sensorName
sensorNameExtern
descr
measureConcept {
id
name
descr
}
}
}
"""
response = execute_graphql(query, {"meterNumber": meter_number})
if "errors" in response:
return []
return response.get("data", {}).get("sensorsForMeterNumber", [])
def get_available_variables(sensor_id: str) -> List[Dict]:
"""Holt verfügbare Variablen/Units für einen Sensor"""
query = """
query GetAvailableVariables($sensorId: ID!) {
availableVariableUnits(sensorId: $sensorId) {
variableUnitId
variableName
unitName
}
}
"""
response = execute_graphql(query, {"sensorId": sensor_id})
if "errors" in response:
return []
return response.get("data", {}).get("availableVariableUnits", [])
def record_ultimo_readings(sensor_id: str, variable_name: str, variable_unit: str, readings: List[Dict]) -> Dict:
"""Trägt Ultimo-Stände ein"""
mutation = """
mutation RecordUltimoReadings($input: UltimoReadingsInput!) {
recordUltimoReadings(input: $input) {
success
created {
id
moment
value
meterValue
}
errors {
code
message
details
}
}
}
"""
input_data = {
"sensorId": sensor_id,
"variableName": variable_name.strip(),
"variableUnit": variable_unit.strip(),
"readings": readings
}
response = execute_graphql(mutation, {"input": input_data})
if "errors" in response:
return {"success": False, "errors": response["errors"]}
return response.get("data", {}).get("recordUltimoReadings", {"success": False, "errors": [{"message": "Unbekannter Fehler"}]})
def parse_month_readings(month_data: str) -> List[Dict]:
"""Parst Monatsdaten im Format 'YYYY-MM: Wert'"""
readings = []
lines = [line.strip() for line in month_data.strip().split('\n') if line.strip()]
for line in lines:
if ':' not in line:
continue
try:
month_str, value_str = line.split(':', 1)
month = month_str.strip()
value = float(value_str.strip().replace(',', '.'))
# Validiere Monatsformat
datetime.strptime(month + '-01', '%Y-%m-%d')
readings.append({
"month": month,
"meterValue": value
})
except (ValueError, AttributeError) as e:
continue
return readings
def render_success_html(created_observations: List[Dict], sensor_name: str, variable_info: str) -> str:
"""Rendert Erfolgsmeldung als HTML"""
obs_count = len(created_observations)
obs_html = ""
for obs in created_observations:
moment = obs.get("moment", "")
meter_value = obs.get("meterValue", 0)
value = obs.get("value", 0)
obs_html += f"""
| {moment} |
{meter_value:.2f} |
{value:.2f} |
"""
return f"""
✅
Ultimo-Stände erfolgreich eingetragen
Details
Sensor: {sensor_name}
Variable/Einheit: {variable_info}
Anzahl Messungen: {obs_count}
| Zeitpunkt |
Zählerstand |
Wert |
{obs_html}
"""
def render_error_html(errors: List[Dict]) -> str:
"""Rendert Fehlermeldungen als HTML"""
error_html = ""
for error in errors:
code = error.get("code", "UNKNOWN")
message = error.get("message", "Unbekannter Fehler")
details = error.get("details", "")
error_html += f"""
{code}: {message}
{f'
{details}' if details else ''}
"""
return f"""
❌
Fehler beim Eintragen der Ultimo-Stände
{error_html}
"""
# Hauptlogik
try:
meter_number = PARAMS.get("meter_number", "").strip()
sensor_id = PARAMS.get("sensor_id", "").strip()
if meter_number and not sensor_id:
# PHASE 1: Sensoren suchen und Auswahlformular generieren
sensors = get_sensors_for_meter(meter_number)
if not sensors:
result = {
"type": "html",
"content": f"""
⚠️ Keine Sensoren gefunden
Für die Zählernummer \"{meter_number}\" wurden keine Sensoren gefunden.
"""
}
else:
# Sensor-Optionen sammeln
sensor_options = []
variable_options_by_sensor = {}
for sensor in sensors:
s_id = sensor.get("sensorId")
s_name = sensor.get("sensorName", "").strip()
s_extern = sensor.get("sensorNameExtern") or ""
mc_name = sensor.get("measureConcept", {}).get("name", "").strip()
display_name = f"{s_name}"
if s_extern.strip():
display_name += f" ({s_extern.strip()})"
if mc_name:
display_name += f" - {mc_name}"
sensor_options.append({
"value": s_id,
"label": display_name
})
# Verfügbare Variablen für diesen Sensor sammeln
variables = get_available_variables(s_id)
var_opts = []
for var in variables:
var_name = var.get("variableName", "").strip()
unit_name = var.get("unitName", "").strip()
combined_value = f"{var_name}|{unit_name}"
var_opts.append({
"value": combined_value,
"label": f"{var_name} ({unit_name})"
})
variable_options_by_sensor[s_id] = var_opts
# Da wir nur eine statische Form unterstützen, nehmen wir die Variablen des ersten Sensors
# In einer echten Anwendung würde man hier eine dynamischere Lösung wählen
first_sensor_vars = variable_options_by_sensor.get(sensors[0].get("sensorId"), [])
form_definition = {
"title": "Ultimo-Stände Eingabe",
"description": f"Sensoren für Zählernummer {meter_number} gefunden. Wählen Sie Sensor und Eingabemodus.",
"fields": [
{
"name": "sensor_id",
"widget": "dropdown",
"label": "Sensor",
"options": sensor_options,
"validators": [{"type": "required", "error_text": "Sensor auswählen"}]
},
{
"name": "variable_unit",
"widget": "dropdown",
"label": "Variable und Einheit",
"options": first_sensor_vars,
"validators": [{"type": "required", "error_text": "Variable/Einheit auswählen"}],
"helper_text": "Hinweis: Die Variablen werden für den ersten Sensor angezeigt. Nach Sensorauswahl evtl. nochmal aktualisieren."
},
{
"name": "input_mode",
"widget": "segmented_control",
"label": "Eingabemodus",
"options": [
{"value": "single", "label": "Einzelner Monat"},
{"value": "batch", "label": "Mehrere Monate"}
],
"initial_value": "single"
},
{
"name": "single_month",
"widget": "text_field",
"label": "Monat (YYYY-MM)",
"hint_text": "z.B. 2024-12",
"conditional": {
"field_name": "input_mode",
"operator": "equals",
"value": "single",
"action": "show"
},
"validators": [
{"type": "match", "value": "^\\d{4}-\\d{2}$", "error_text": "Format: YYYY-MM"}
]
},
{
"name": "single_value",
"widget": "text_field",
"label": "Zählerstand",
"text_field_config": {"keyboard_type": "number"},
"conditional": {
"field_name": "input_mode",
"operator": "equals",
"value": "single",
"action": "show"
},
"validators": [
{"type": "numeric", "error_text": "Numerischer Wert erforderlich"}
]
},
{
"name": "batch_data",
"widget": "text_field",
"label": "Monatsdaten",
"hint_text": "Eine Zeile pro Monat im Format:\n2024-10: 1500.5\n2024-11: 1650.2\n2024-12: 1800.0",
"text_field_config": {
"max_lines": 10,
"keyboard_type": "multiline"
},
"conditional": {
"field_name": "input_mode",
"operator": "equals",
"value": "batch",
"action": "show"
},
"validators": [
{"type": "required", "error_text": "Monatsdaten erforderlich"}
]
}
],
"submit_label": "Ultimo-Stände eintragen"
}
result = {
"type": "form",
"form_definition": form_definition
}
elif sensor_id:
# PHASE 2: Ultimo-Stände eintragen
variable_unit = PARAMS.get("variable_unit", "")
input_mode = PARAMS.get("input_mode", "single")
if not variable_unit or "|" not in variable_unit:
result = {
"type": "error",
"message": "Variable und Einheit müssen ausgewählt werden."
}
else:
variable_name, unit_name = variable_unit.split("|", 1)
# Readings sammeln
readings = []
if input_mode == "single":
single_month = PARAMS.get("single_month", "").strip()
single_value = PARAMS.get("single_value", "")
if not single_month or not single_value:
result = {
"type": "error",
"message": "Monat und Zählerstand müssen angegeben werden."
}
else:
try:
value = float(single_value.replace(",", "."))
readings = [{
"month": single_month,
"meterValue": value
}]
except ValueError:
result = {
"type": "error",
"message": "Ungültiger Zählerstand. Bitte numerischen Wert eingeben."
}
else: # batch
batch_data = PARAMS.get("batch_data", "").strip()
if not batch_data:
result = {
"type": "error",
"message": "Batch-Daten müssen angegeben werden."
}
else:
readings = parse_month_readings(batch_data)
if not readings:
result = {
"type": "error",
"message": "Keine gültigen Monatsdaten gefunden. Format: YYYY-MM: Wert"
}
# Wenn wir Readings haben, eintragen
if "type" not in result and readings:
# Sensor-Name für Display holen
sensor_name = "Unbekannt"
for sensor in get_sensors_for_meter(meter_number):
if sensor.get("sensorId") == sensor_id:
sensor_name = sensor.get("sensorName", "").strip()
break
# Ultimo-Stände eintragen
ultimo_result = record_ultimo_readings(sensor_id, variable_name, unit_name, readings)
if ultimo_result.get("success"):
created = ultimo_result.get("created", [])
variable_info = f"{variable_name} ({unit_name})"
result = {
"type": "html",
"content": render_success_html(created, sensor_name, variable_info)
}
else:
errors = ultimo_result.get("errors", [])
result = {
"type": "html",
"content": render_error_html(errors)
}
else:
result = {
"type": "error",
"message": "Zählernummer muss angegeben werden."
}
except Exception as e:
result = {
"type": "error",
"message": f"Unerwarteter Fehler: {str(e)}"
}