From 1cf39a63a961e36e6ff9405c027fe4c666f8f234 Mon Sep 17 00:00:00 2001 From: "martin.schweitzer" Date: Wed, 15 Apr 2026 08:40:07 +0000 Subject: [PATCH] deploy script: ultimo_staende_eingabe --- scripts/ultimo_staende_eingabe.py | 447 ++++++++++++++++++++++++++++++ 1 file changed, 447 insertions(+) create mode 100644 scripts/ultimo_staende_eingabe.py diff --git a/scripts/ultimo_staende_eingabe.py b/scripts/ultimo_staende_eingabe.py new file mode 100644 index 0000000..3b7c38d --- /dev/null +++ b/scripts/ultimo_staende_eingabe.py @@ -0,0 +1,447 @@ +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}

+
+ +
+ + + + + + + + + + {obs_html} + +
ZeitpunktZählerstandWert
+
+
+ """ + +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)}" + }