Messwerte Dashboard - {safe_sensor

import html import json from datetime import datetime, timedelta import httpx # === PARAMETER START === METER_NUMBER = "Any - Als Parameter" CHART_TYPE = "line" GROUP_BY = "hour" INCLUDE_STATISTICS = True # === PARAMETER ENDE === FIND_SENSORS_QUERY = """ query FindSensors($meterNumber: String!) { sensorsForMeterNumber(meterNumber: $meterNumber) { sensorId sensorName sensorNameExtern descr measureConcept { id name descr } } } """ GET_VARIABLES_QUERY = """ query GetVariables($sensorId: ID!) { availableVariableUnits(sensorId: $sensorId) { variableUnitId variableName unitName } } """ FIND_OBSERVATIONS_QUERY = """ query FindObservations($measurementConceptId: ID!, $sensorName: String, $startTime: String, $endTime: String) { findObservation( measurementConceptId: $measurementConceptId sensorName: $sensorName startTime: $startTime endTime: $endTime ) { id moment value meterValue observationVariableUnit { observationVariable { name description } unit { name description } } quality } } """ def escape_text(value, default=""): text = default if value is None else str(value) return html.escape(text, quote=True) def build_message(message, css_class): return ( f"

" f"{escape_text(message)}" "

" ) def make_graphql_request(query, variables): try: with httpx.Client() as client: response = client.post( f"{EXTERNAL_BASE_URL}/graphql", headers=AUTH_HEADERS, json={"query": query, "variables": variables}, timeout=30.0, ) response.raise_for_status() return response.json() except Exception as exc: return {"errors": [str(exc)]} def calculate_statistics(observations): if not observations: return {} values = [obs.get("value") or 0 for obs in observations] meter_values = [obs.get("meterValue") or 0 for obs in observations] return { "total_readings": len(observations), "value_min": min(values) if values else 0, "value_max": max(values) if values else 0, "value_avg": sum(values) / len(values) if values else 0, "meter_min": min(meter_values) if meter_values else 0, "meter_max": max(meter_values) if meter_values else 0, "meter_avg": sum(meter_values) / len(meter_values) if meter_values else 0, "consumption": max(meter_values) - min(meter_values) if meter_values else 0, } def format_datetime(dt_string): try: dt = datetime.fromisoformat(dt_string.replace("Z", "+00:00")) return dt.strftime("%Y-%m-%d %H:%M:%S") except Exception: return dt_string def build_report(): sensors_response = make_graphql_request( FIND_SENSORS_QUERY, {"meterNumber": METER_NUMBER}, ) if "errors" in sensors_response: return build_message( f"Fehler beim Abrufen der Sensoren: {sensors_response['errors']}", "error", ) sensors = sensors_response.get("data", {}).get("sensorsForMeterNumber", []) if not sensors: return build_message( f"Keine Sensoren fuer Zaehlernummer '{METER_NUMBER}' gefunden.", "warning", ) sensor = sensors[0] sensor_id = sensor["sensorId"] sensor_name = sensor["sensorName"] measure_concept = sensor.get("measureConcept") or {} measure_concept_id = measure_concept.get("id") measure_concept_name = measure_concept.get("name") or "Unbekannt" variables_response = make_graphql_request( GET_VARIABLES_QUERY, {"sensorId": sensor_id}, ) available_vars = variables_response.get("data", {}).get("availableVariableUnits", []) end_time = datetime.now() start_time = end_time - timedelta(days=90) observations_response = make_graphql_request( FIND_OBSERVATIONS_QUERY, { "measurementConceptId": measure_concept_id, "sensorName": sensor_name, "startTime": start_time.isoformat(), "endTime": end_time.isoformat(), }, ) if "errors" in observations_response: return build_message( f"Fehler beim Abrufen der Messwerte: {observations_response['errors']}", "error", ) observations = observations_response.get("data", {}).get("findObservation", []) stats = calculate_statistics(observations) chart_labels = [] chart_values = [] chart_meter_values = [] for obs in observations: chart_labels.append(format_datetime(obs.get("moment", ""))) chart_values.append(obs.get("value") or 0) chart_meter_values.append(obs.get("meterValue") or 0) safe_sensor_name = escape_text(sensor_name, "Unbekannt") safe_meter_number = escape_text(METER_NUMBER) safe_measure_concept_name = escape_text(measure_concept_name, "Unbekannt") result_html = f""" Messwerte Dashboard - {safe_sensor_name}

""" if INCLUDE_STATISTICS and stats: result_html += f"""

{stats['total_readings']:,}

Gesamte Messwerte

{stats['consumption']:,.2f}

Verbrauch (Differenz)

{stats['value_avg']:,.2f}

Durchschnitt Messwert

{stats['meter_max']:,.2f}

Max Zaehlerstand

""" result_html += f"""

Messwerte Verlauf ({escape_text(CHART_TYPE.title())})

Zaehlerstaende

Verfuegbare Variablen:

{variable_name} ({unit_name})

Anzahl Sensoren gefunden: {len(sensors)}

Measure Concept: {safe_measure_concept_name}

""" return result_html try: result = build_report() except Exception as exc: result = build_message(f"Allgemeiner Fehler: {exc}", "error") print(result)