import json import httpx from datetime import datetime import traceback # GraphQL Queries FIND_SENSORS_QUERY = """ query FindSensors($meterNumber: String!) { sensorsForMeterNumber(meterNumber: $meterNumber) { sensorId sensorName sensorNameExtern descr measureConcept { id name descr } } } """ GET_VARIABLE_UNITS_QUERY = """ query GetVariableUnits($sensorId: ID!) { availableVariableUnits(sensorId: $sensorId) { variableUnitId variableName unitName } } """ GET_LAST_OBSERVATIONS_QUERY = """ query GetLastObservations($measurementConceptId: ID!, $sensorName: String, $observationVariableNamePattern: String, $startTime: String, $endTime: String) { findObservation( measurementConceptId: $measurementConceptId sensorName: $sensorName observationVariableNamePattern: $observationVariableNamePattern startTime: $startTime endTime: $endTime ) { id moment value meterValue observationVariableUnit { observationVariable { name } unit { name } } } } """ RECORD_ULTIMO_MUTATION = """ mutation RecordUltimoReadings($input: UltimoReadingsInput!) { recordUltimoReadings(input: $input) { success created { id moment value meterValue observationVariableUnit { observationVariable { name } unit { name } } } errors { code message details } } } """ def execute_graphql(query, variables=None): """Execute GraphQL query/mutation""" try: response = httpx.post( f"{EXTERNAL_BASE_URL}/graphql", headers=AUTH_HEADERS, json={"query": query, "variables": variables or {}}, timeout=30 ) response.raise_for_status() data = response.json() if "errors" in data: raise Exception(f"GraphQL Error: {data['errors']}") return data["data"] except Exception as e: raise Exception(f"GraphQL Request failed: {str(e)}") def format_datetime(dt_str): """Format datetime string for display""" try: dt = datetime.fromisoformat(dt_str.replace('T', ' ')) return dt.strftime("%d.%m.%Y %H:%M") except: return dt_str def format_number(num): """Format number with German locale""" return f"{num:,.2f}".replace(',', ' ').replace('.', ',').replace(' ', '.') # Get parameters sensor_search = PARAMS.get("sensor_search", "") variable_name = PARAMS.get("variable_name", "ACTIVE_ENERGY_ABSORBED_10") readings_data = PARAMS.get("readings_data", "") html_content = """ Ultimo-Zählerstände Eingabe

🔌 Ultimo-Zählerstände Eingabe

""" try: # Step 1: Find sensors if not sensor_search: html_content += """

⚠️ Keine Zählernummer angegeben

Bitte geben Sie eine Zählernummer ein, um fortzufahren.

""" else: # Search for sensors sensor_data = execute_graphql(FIND_SENSORS_QUERY, {"meterNumber": sensor_search}) sensors = sensor_data.get("sensorsForMeterNumber", []) if not sensors: html_content += f"""

❌ Keine Sensoren gefunden

Für die Zählernummer {sensor_search} wurden keine Sensoren gefunden.

""" else: # Display found sensors html_content += f"""

📋 Gefundene Sensoren für '{sensor_search}'

{len(sensors)} Sensor(en) gefunden

""" for sensor in sensors: html_content += f"""

{sensor['sensorName'].strip()}

ID: {sensor['sensorId']}

Extern: {sensor.get('sensorNameExtern', 'N/A')}

Beschreibung: {sensor.get('descr', 'N/A')}

Messkonzept: {sensor['measureConcept']['name'].strip()}

""" html_content += "
" # Use first sensor if multiple found selected_sensor = sensors[0] sensor_id = selected_sensor['sensorId'] if len(sensors) > 1: html_content += f"""

⚠️ Mehrere Sensoren gefunden

Es wurden {len(sensors)} Sensoren gefunden. Verwende ersten Sensor: {selected_sensor['sensorName'].strip()}

""" # Step 2: Parse readings data if not readings_data: html_content += """

⚠️ Keine Ultimo-Daten angegeben

Bitte geben Sie die Ultimo-Stände im JSON-Format ein.

""" else: try: readings_list = json.loads(readings_data) if not isinstance(readings_list, list): raise ValueError("Daten müssen ein JSON-Array sein") # Validate readings format for reading in readings_list: if not isinstance(reading, dict): raise ValueError("Jeder Eintrag muss ein Objekt sein") if 'month' not in reading or 'meterValue' not in reading: raise ValueError("Jeder Eintrag muss 'month' und 'meterValue' enthalten") html_content += f"""

✅ Ultimo-Daten erfolgreich geparst

{len(readings_list)} Ultimo-Stände zur Verarbeitung bereit

""" # Step 3: Execute Ultimo mutation mutation_input = { "sensorId": sensor_id, "variableName": variable_name, "variableUnit": "WH", "readings": readings_list } ultimo_result = execute_graphql(RECORD_ULTIMO_MUTATION, {"input": mutation_input}) ultimo_data = ultimo_result.get("recordUltimoReadings") if ultimo_data["success"]: created_observations = ultimo_data["created"] html_content += f"""

🎉 Ultimo-Stände erfolgreich eingetragen

{len(created_observations)} neue Observations wurden erstellt

""" # Show statistics if created_observations: latest_observation = max(created_observations, key=lambda x: x['moment']) earliest_observation = min(created_observations, key=lambda x: x['moment']) html_content += f"""
{len(created_observations)}
Einträge erstellt
{format_number(latest_observation['meterValue'])}
Neuester Zählerstand
{format_datetime(latest_observation['moment'])}
Letzter Zeitstempel
""" # Show created observations table html_content += """

📊 Neu erstellte Zählerstände

""" # Sort by moment descending sorted_observations = sorted(created_observations, key=lambda x: x['moment'], reverse=True) for obs in sorted_observations: html_content += f""" """ html_content += "
Zeitstempel Zählerstand Wert Variable Einheit
{format_datetime(obs['moment'])} {format_number(obs['meterValue'])} {format_number(obs['value'])} {obs['observationVariableUnit']['observationVariable']['name'].strip()} {obs['observationVariableUnit']['unit']['name'].strip()}
" # Step 4: Show last 10 observations for context try: # Get measure concept ID from sensor data measure_concept_id = selected_sensor['measureConcept']['id'] sensor_name = selected_sensor['sensorName'].strip() # Get historical observations historical_data = execute_graphql(GET_LAST_OBSERVATIONS_QUERY, { "measurementConceptId": measure_concept_id, "sensorName": sensor_name, "observationVariableNamePattern": variable_name, "startTime": "2020-01-01T00:00:00", "endTime": "2030-12-31T23:59:59" }) observations = historical_data.get("findObservation", []) if observations: # Sort by moment descending and take first 10 (most recent) recent_observations = sorted(observations, key=lambda x: x['moment'], reverse=True)[:10] html_content += f"""

📈 Letzte 10 Zählerstände (inkl. neue Einträge)

""" created_ids = {obs['id'] for obs in created_observations} for obs in recent_observations: is_new = obs['id'] in created_ids row_class = 'current-value' if is_new else '' new_badge = '🆕 ' if is_new else '' html_content += f""" """ html_content += "
Zeitstempel Zählerstand Wert Variable Einheit
{new_badge}{format_datetime(obs['moment'])} {format_number(obs['meterValue'])} {format_number(obs['value'])} {obs['observationVariableUnit']['observationVariable']['name'].strip()} {obs['observationVariableUnit']['unit']['name'].strip()}
" except Exception as e: html_content += f"""

⚠️ Historische Daten nicht verfügbar

Fehler beim Abrufen der historischen Zählerstände: {str(e)}

""" else: # Handle errors errors = ultimo_data.get("errors", []) html_content += f"""

❌ Fehler beim Eintragen der Ultimo-Stände

Die Ultimo-Stände konnten nicht eingetragen werden:

""" for error in errors: html_content += f"""
Fehlercode: {error['code']}
Nachricht: {error['message']}
Details: {error.get('details', 'Keine weiteren Details')}
""" html_content += "
" # Show detailed error explanation html_content += """

💡 Mögliche Lösungsansätze

  • DUPLICATE_DAY: Für den angegebenen Tag existiert bereits ein Zählerstand
  • MISSING_GAP: Es fehlen Ultimo-Stände für bestimmte Monate
  • NO_INITIAL_READING: Kein Anfangswert vorhanden - bitte zuerst initialisieren
  • INVALID_SEQUENCE: Die Zählerstände sind nicht chronologisch sortiert
""" except json.JSONDecodeError as e: html_content += f"""

❌ JSON-Format Fehler

Die eingegebenen Ultimo-Daten haben kein gültiges JSON-Format:

{str(e)}

Beispiel für korrektes Format:

[{{"month": "2024-01", "meterValue": 12345.5}}, {{"month": "2024-02", "meterValue": 12567.8}}]
""" except ValueError as e: html_content += f"""

❌ Datenformat Fehler

Die Struktur der Ultimo-Daten ist ungültig:

{str(e)}
""" except Exception as e: html_content += f"""

❌ Unbekannter Fehler

Ein unerwarteter Fehler ist aufgetreten:

{str(e)}
""" except Exception as e: html_content += f"""

❌ System-Fehler

Ein kritischer Fehler ist aufgetreten:

{str(e)} {traceback.format_exc()}
""" html_content += """
""" result = { "type": "html", "content": html_content }