diff --git a/scripts/ultimo_zaehlerstand_eingabe.py b/scripts/ultimo_zaehlerstand_eingabe.py new file mode 100644 index 0000000..3cc8356 --- /dev/null +++ b/scripts/ultimo_zaehlerstand_eingabe.py @@ -0,0 +1,576 @@ +import httpx +import json +import re +from datetime import datetime, timezone +from typing import List, Dict, Any + +# Globals: EXTERNAL_BASE_URL, AUTH_HEADERS, PARAMS + +def make_graphql_request(query: str, variables: Dict = None) -> Dict: + """Sichere GraphQL-Anfrage mit Fehlerbehandlung""" + try: + response = httpx.post( + f"{EXTERNAL_BASE_URL}/graphql", + headers=AUTH_HEADERS, + json={"query": query, "variables": variables or {}}, + timeout=30.0 + ) + response.raise_for_status() + data = response.json() + + if "errors" in data and data["errors"]: + error_msgs = [err.get("message", "Unbekannter Fehler") for err in data["errors"]] + return {"success": False, "error": "; ".join(error_msgs)} + + return {"success": True, "data": data.get("data", {})} + except httpx.TimeoutException: + return {"success": False, "error": "Anfrage-Timeout nach 30 Sekunden"} + except httpx.HTTPStatusError as e: + return {"success": False, "error": f"HTTP-Fehler {e.response.status_code}: {e.response.text}"} + except Exception as e: + return {"success": False, "error": f"Unerwarteter Fehler: {str(e)}"} + +def search_sensors(query_text: str) -> List[Dict]: + """Sucht Sensoren nach Name oder Nummer""" + gql_query = """ + query SearchSensors { + sensors { + id + name + nameExtern + description + measureConcept { + id + name + description + } + } + } + """ + + response = make_graphql_request(gql_query) + if not response["success"]: + return [] + + sensors = response["data"].get("sensors", []) + if not query_text: + return sensors[:20] # Limit für Performance + + # Filter nach Name oder Number + filtered = [] + query_lower = query_text.lower() + for sensor in sensors: + name = (sensor.get("name") or "").strip().lower() + name_extern = (sensor.get("nameExtern") or "").strip().lower() + if query_lower in name or query_lower in name_extern: + filtered.append(sensor) + + return filtered[:10] # Top 10 Treffer + +def get_sensor_variables(sensor_id: str) -> List[Dict]: + """Holt verfügbare Variablen für einen Sensor""" + gql_query = """ + query GetSensorVariables($sensorId: ID!) { + availableVariableUnits(sensorId: $sensorId) { + variableUnitId + variableName + unitName + } + } + """ + + response = make_graphql_request(gql_query, {"sensorId": sensor_id}) + if not response["success"]: + return [] + + return response["data"].get("availableVariableUnits", []) + +def get_last_observations(sensor_id: str, variable_name: str, limit: int = 10) -> List[Dict]: + """Holt die letzten N Observations für einen Sensor""" + # Da es keine direkte Query gibt, verwenden wir findObservation mit einem weiten Zeitraum + gql_query = """ + query GetLastObservations($measureConceptId: ID!, $sensorName: String!, $variableName: String!) { + findObservation( + measurementConceptId: $measureConceptId + sensorName: $sensorName + observationVariableNamePattern: $variableName + startTime: "2020-01-01" + endTime: "2030-12-31" + ) { + id + moment + meterValue + observationVariableUnit { + observationVariable { + name + } + unit { + name + } + } + } + } + """ + + # Wir brauchen den MeasureConcept und SensorName + sensor_query = f""" + query GetSensor($sensorId: ID!) {{ + sensor(id: $sensorId) {{ + name + measureConcept {{ + id + }} + }} + }} + """ + + sensor_response = make_graphql_request(sensor_query, {"sensorId": sensor_id}) + if not sensor_response["success"] or not sensor_response["data"].get("sensor"): + return [] + + sensor = sensor_response["data"]["sensor"] + sensor_name = (sensor.get("name") or "").strip() + mc_id = sensor["measureConcept"]["id"] + + response = make_graphql_request(gql_query, { + "measureConceptId": mc_id, + "sensorName": sensor_name, + "variableName": variable_name + }) + + if not response["success"]: + return [] + + observations = response["data"].get("findObservation", []) + # Sortiere nach Zeitstempel absteigend und limitiere + sorted_obs = sorted(observations, key=lambda x: x.get("moment", ""), reverse=True) + return sorted_obs[:limit] + +def parse_ultimo_text(text: str) -> List[Dict]: + """Parst Ultimo-Text-Eingabe zu strukturierten Daten""" + lines = [line.strip() for line in text.split('\n') if line.strip()] + readings = [] + + for line in lines: + # Pattern: DD.MM.YYYY: Wert [Einheit] + # Beispiele: 31.12.2025: 60,645 MWh oder 28.2.2026: 68,771 + match = re.match(r'(\d{1,2})\.(\d{1,2})\.(\d{4})\s*:?\s*([\d.,]+)', line) + if match: + day, month, year, value_str = match.groups() + + # Wert parsen (Komma und Punkt als Dezimaltrennzeichen akzeptieren) + value_str = value_str.replace(',', '.') + try: + value = float(value_str) + # Für Ultimo nehmen wir immer Ende des Monats + date_str = f"{year}-{month.zfill(2)}" + readings.append({ + "month": date_str, + "meterValue": value + }) + except ValueError: + continue # Ungültige Zahl ignorieren + + return readings + +def record_ultimo_readings(sensor_id: str, variable_name: str, variable_unit: str, readings: List[Dict]) -> Dict: + """Führt Ultimo-Batch-Eingabe aus""" + gql_query = """ + mutation RecordUltimoReadings($input: UltimoReadingsInput!) { + recordUltimoReadings(input: $input) { + success + created { + id + moment + meterValue + observationVariableUnit { + observationVariable { + name + } + unit { + name + } + } + } + errors { + code + message + details + } + } + } + """ + + input_data = { + "sensorId": sensor_id, + "variableName": variable_name, + "variableUnit": variable_unit, + "readings": readings + } + + response = make_graphql_request(gql_query, {"input": input_data}) + if not response["success"]: + return {"success": False, "error": response["error"]} + + return response["data"].get("recordUltimoReadings", {}) + +def record_single_reading(sensor_id: str, moment: str, value: float, variable_name: str, variable_unit: str) -> Dict: + """Führt Einzelwert-Eingabe aus""" + gql_query = """ + mutation RecordMeterReading($input: MeterReadingInput!) { + recordMeterReading(input: $input) { + success + observation { + id + moment + meterValue + observationVariableUnit { + observationVariable { + name + } + unit { + name + } + } + } + errors { + code + message + details + } + } + } + """ + + input_data = { + "sensorId": sensor_id, + "moment": moment, + "value": value, + "variableName": variable_name, + "variableUnit": variable_unit + } + + response = make_graphql_request(gql_query, {"input": input_data}) + if not response["success"]: + return {"success": False, "error": response["error"]} + + return response["data"].get("recordMeterReading", {}) + +def format_datetime(dt_str: str) -> str: + """Formatiert Datetime-String für Anzeige""" + try: + dt = datetime.fromisoformat(dt_str.replace('Z', '+00:00')) + return dt.strftime('%d.%m.%Y %H:%M') + except: + return dt_str + +def generate_html_response(success: bool, result: Dict, sensor_info: Dict, last_observations: List[Dict]) -> str: + """Generiert HTML-Response""" + + # CSS Styles + styles = """ + + """ + + # Header + html = f""" + {styles} +
{created_count} Zählerstände wurden erfolgreich gespeichert:
' + html += 'Neuer Zählerstand: {moment}: {value:,.3f} {unit}
' + + html += '| Datum/Zeit | Zählerstand | Einheit | Variable |
|---|---|---|---|
| {moment} | {value:,.3f} | {unit} | {variable} |
Fehler: {message}
' + if details: + html += f'{result["error"]}
' + else: + html += 'Unbekannter Fehler aufgetreten.
' + + html += '| Datum/Zeit | Zählerstand | Einheit | Variable |
|---|---|---|---|
| {moment} | {value:,.3f} | {unit} | {variable} |
Bitte wählen Sie einen Zähler aus.
Kein Zähler gefunden für: {sensor_selection}
Bitte geben Sie Ultimo-Stände ein.
Keine gültigen Zählerstände im Text gefunden.
Erwartetes Format: DD.MM.YYYY: Wert
Bitte geben Sie Datum und Zählerstand ein.
Ungültiges Datum oder Zählerstand: {str(e)}
{str(e)}
Unerwarteter Fehler beim Ausführen des Scripts: {str(e)}