open-energy.at
/
Fastlane
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

„scripts/ultimo_meter_readings.py“ löschen

Browse Source
main
martin.schweitzer 5 days ago
parent 4d09ddd233
commit f4b78dd6c9
1 changed files with 0 additions and 835 deletions
Show Stats Download Patch File Download Diff File

835
scripts/ultimo_meter_readings.py
Unescape Escape View File

@ -1,835 +0,0 @@
import json
import httpx
import re
from datetime import datetime
import traceback
def main():
try:
# Check if we have enough parameters to proceed
search_term = PARAMS.get('search_term', '').strip()
sensor_id = PARAMS.get('sensor_id', '').strip()
# Phase 1: If no sensor is selected, show sensor selection
if not sensor_id and not search_term:
return {"type": "error", "message": "Bitte geben Sie entweder einen Suchbegriff oder eine Sensor-ID ein."}
# Phase 2: If we have search term but no sensor_id, search for sensors
if search_term and not sensor_id:
sensors = search_sensors(search_term)
if not sensors:
return {"type": "error", "message": f"Keine Sensoren gefunden für Suchbegriff '{search_term}'."}
# Create dynamic form with found sensors
sensor_options = []
for sensor in sensors:
display_name = f"{sensor['sensorName'].strip()} (ID: {sensor['sensorId']})"
if sensor.get('sensorNameExtern') and sensor['sensorNameExtern'].strip() != '-':
display_name += f" - {sensor['sensorNameExtern'].strip()}"
if sensor['measureConcept']['name']:
display_name += f" [{sensor['measureConcept']['name'].strip()}]"
sensor_options.append({
"value": sensor['sensorId'],
"label": display_name
})
dynamic_form = {
"title": "Sensor auswählen und Ultimo-Stände eingeben",
"description": f"Gefundene Sensoren für '{search_term}'. Wählen Sie einen Sensor aus und geben Sie die Ultimo-Stände ein.",
"layout": "sections",
"fields": [
{
"name": "sensor_id",
"widget": "dropdown",
"label": "Sensor auswählen",
"options": sensor_options,
"validators": [{"type": "required"}]
},
{
"name": "variable_name",
"widget": "text_field",
"label": "Variable Name (optional)",
"hint_text": "Leer lassen für Standard-Variable",
"helper_text": "Z.B. ACTIVE_ENERGY_ABSORBED_10, ENERGY_INST_VAL"
},
{
"name": "variable_unit",
"widget": "text_field",
"label": "Einheit (optional)",
"hint_text": "Leer lassen für Standard-Einheit",
"helper_text": "Z.B. WH, KWH"
},
{
"name": "readings_input",
"widget": "text_field",
"label": "Zählerstände",
"hint_text": "Format: YYYY-MM:Wert (pro Zeile) oder YYYY-MM:Wert,YYYY-MM:Wert",
"helper_text": "Beispiel: 2024-01:1000.5 oder 2024-01:1000,2024-02:1100",
"text_field_config": {
"keyboard_type": "multiline",
"max_lines": 10,
"min_lines": 2
},
"validators": [
{"type": "required", "error_text": "Zählerstände sind erforderlich"}
]
}
],
"sections": [
{
"title": "Sensor",
"icon": "sensors",
"field_names": ["sensor_id"]
},
{
"title": "Variable & Einheit",
"icon": "settings",
"field_names": ["variable_name", "variable_unit"]
},
{
"title": "Zählerstände eingeben",
"icon": "edit",
"field_names": ["readings_input"]
}
]
}
return {"type": "form", "form_definition": dynamic_form}
# Phase 3: We have sensor_id, process the readings
if not sensor_id:
return {"type": "error", "message": "Sensor-ID ist erforderlich."}
readings_input = PARAMS.get('readings_input', '').strip()
if not readings_input:
return {"type": "error", "message": "Zählerstände sind erforderlich."}
# Parse readings input
readings = parse_readings_input(readings_input)
if not readings:
return {"type": "error", "message": "Keine gültigen Zählerstände gefunden. Format: YYYY-MM:Wert"}
# Get sensor info and available variables
sensor_info = get_sensor_info(sensor_id)
if not sensor_info:
return {"type": "error", "message": f"Sensor mit ID {sensor_id} nicht gefunden."}
# Get available variables for the sensor
available_variables = get_available_variables(sensor_id)
# Determine variable and unit to use
variable_name = PARAMS.get('variable_name', '').strip()
variable_unit = PARAMS.get('variable_unit', '').strip()
if not variable_name and available_variables:
# Use first available variable as default
variable_name = available_variables[0]['variableName'].strip()
variable_unit = available_variables[0]['unitName'].strip()
# Record the readings
result = record_ultimo_readings(sensor_id, variable_name, variable_unit, readings)
# Generate HTML report
html_content = generate_html_report(sensor_info, variable_name, variable_unit, readings, result, available_variables)
return {"type": "html", "content": html_content}
except Exception as e:
return {"type": "error", "message": f"Fehler: {str(e)}\n\nDetails: {traceback.format_exc()}"}
def search_sensors(search_term):
"""Search for sensors by meter number"""
try:
# First try sensorsForMeterNumber
query = '''
query SearchSensors($meterNumber: String!) {
sensorsForMeterNumber(meterNumber: $meterNumber) {
sensorId
sensorName
sensorNameExtern
descr
measureConcept {
id
name
descr
}
}
}
'''
with httpx.Client() as client:
response = client.post(
f"{EXTERNAL_BASE_URL}/graphql",
headers=AUTH_HEADERS,
json={"query": query, "variables": {"meterNumber": search_term}}
)
if response.status_code == 200:
data = response.json()
if 'errors' not in data:
sensors = data.get('data', {}).get('sensorsForMeterNumber', [])
if sensors:
return sensors
# If no results, try general sensor search
query = '''
query GetAllSensors {
sensors {
id
name
nameExtern
description
measureConcept {
id
name
description
}
}
}
'''
with httpx.Client() as client:
response = client.post(
f"{EXTERNAL_BASE_URL}/graphql",
headers=AUTH_HEADERS,
json={"query": query}
)
if response.status_code == 200:
data = response.json()
if 'errors' not in data:
all_sensors = data.get('data', {}).get('sensors', [])
# Filter sensors that match search term
filtered_sensors = []
for sensor in all_sensors:
if (search_term.lower() in sensor.get('name', '').lower() or
search_term.lower() in sensor.get('nameExtern', '').lower() or
search_term.lower() in sensor.get('description', '').lower()):
filtered_sensors.append({
'sensorId': sensor['id'],
'sensorName': sensor['name'],
'sensorNameExtern': sensor.get('nameExtern'),
'descr': sensor.get('description'),
'measureConcept': {
'id': sensor['measureConcept']['id'],
'name': sensor['measureConcept']['name'],
'descr': sensor['measureConcept'].get('description')
}
})
return filtered_sensors[:20] # Limit to first 20 results
return []
except Exception as e:
print(f"Error searching sensors: {e}")
return []
def get_sensor_info(sensor_id):
"""Get detailed sensor information"""
try:
query = '''
query GetSensor($id: ID!) {
sensor(id: $id) {
id
name
nameExtern
description
measureConcept {
id
name
description
}
}
}
'''
with httpx.Client() as client:
response = client.post(
f"{EXTERNAL_BASE_URL}/graphql",
headers=AUTH_HEADERS,
json={"query": query, "variables": {"id": sensor_id}}
)
if response.status_code == 200:
data = response.json()
if 'errors' not in data:
return data.get('data', {}).get('sensor')
return None
except Exception as e:
print(f"Error getting sensor info: {e}")
return None
def get_available_variables(sensor_id):
"""Get available variables and units for a sensor"""
try:
query = '''
query GetAvailableVariables($sensorId: ID!) {
availableVariableUnits(sensorId: $sensorId) {
variableUnitId
variableName
unitName
}
}
'''
with httpx.Client() as client:
response = client.post(
f"{EXTERNAL_BASE_URL}/graphql",
headers=AUTH_HEADERS,
json={"query": query, "variables": {"sensorId": sensor_id}}
)
if response.status_code == 200:
data = response.json()
if 'errors' not in data:
return data.get('data', {}).get('availableVariableUnits', [])
return []
except Exception as e:
print(f"Error getting available variables: {e}")
return []
def parse_readings_input(readings_input):
"""Parse the readings input string into a list of readings"""
readings = []
# Support both line-by-line and comma-separated format
lines = readings_input.replace(',', '\n').split('\n')
for line in lines:
line = line.strip()
if not line:
continue
# Expected format: YYYY-MM:value
if ':' not in line:
continue
parts = line.split(':', 1)
if len(parts) != 2:
continue
month_str = parts[0].strip()
value_str = parts[1].strip()
# Validate month format YYYY-MM
if not re.match(r'^\d{4}-\d{2}$', month_str):
continue
try:
value = float(value_str)
readings.append({
'month': month_str,
'meterValue': value
})
except ValueError:
continue
# Sort readings by month
readings.sort(key=lambda x: x['month'])
return readings
def record_ultimo_readings(sensor_id, variable_name, variable_unit, readings):
"""Record ultimo readings using GraphQL mutation"""
try:
mutation = '''
mutation RecordUltimoReadings($input: UltimoReadingsInput!) {
recordUltimoReadings(input: $input) {
success
errors {
code
message
details
}
created {
id
moment
value
meterValue
}
}
}
'''
input_data = {
'sensorId': sensor_id,
'readings': readings
}
if variable_name:
input_data['variableName'] = variable_name
if variable_unit:
input_data['variableUnit'] = variable_unit
with httpx.Client() as client:
response = client.post(
f"{EXTERNAL_BASE_URL}/graphql",
headers=AUTH_HEADERS,
json={"query": mutation, "variables": {"input": input_data}}
)
if response.status_code == 200:
data = response.json()
if 'errors' in data:
return {'success': False, 'errors': [{'message': str(data['errors'])}], 'created': []}
return data.get('data', {}).get('recordUltimoReadings', {'success': False, 'errors': [], 'created': []})
else:
return {'success': False, 'errors': [{'message': f'HTTP Error {response.status_code}'}], 'created': []}
except Exception as e:
return {'success': False, 'errors': [{'message': str(e)}], 'created': []}
def generate_html_report(sensor_info, variable_name, variable_unit, readings, result, available_variables):
"""Generate HTML report for the ultimo recording result"""
sensor_display = sensor_info.get('name', '').strip() if sensor_info else 'Unbekannt'
if sensor_info and sensor_info.get('nameExtern') and sensor_info['nameExtern'].strip() != '-':
sensor_display += f" ({sensor_info['nameExtern'].strip()})"
measure_concept_name = ''
if sensor_info and sensor_info.get('measureConcept', {}).get('name'):
measure_concept_name = sensor_info['measureConcept']['name'].strip()
# Status styling
if result['success']:
status_class = 'success'
status_text = 'Erfolgreich'
status_icon = '✅'
else:
status_class = 'error'
status_text = 'Fehler'
status_icon = '❌'
html_content = f'''
<!DOCTYPE html>
<html>
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Ultimo-Zählerstand Eingabe</title>
<style>
body {{
font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif;
margin: 0;
padding: 20px;
background-color: #f5f5f5;
line-height: 1.6;
}}
.container {{
max-width: 1200px;
margin: 0 auto;
background: white;
border-radius: 12px;
box-shadow: 0 4px 6px rgba(0,0,0,0.1);
overflow: hidden;
}}
.header {{
background: linear-gradient(135deg, var(--color-primary, #2196F3), #1976D2);
color: white;
padding: 30px;
text-align: center;
}}
.header h1 {{
margin: 0;
font-size: 2.2em;
font-weight: 300;
}}
.content {{
padding: 30px;
}}
.status-card {{
border-radius: 8px;
padding: 20px;
margin-bottom: 25px;
border-left: 5px solid;
}}
.status-card.success {{
background-color: #E8F5E8;
border-left-color: var(--color-success, #4CAF50);
color: #2E7D32;
}}
.status-card.error {{
background-color: #FFEBEE;
border-left-color: var(--color-danger, #F44336);
color: #C62828;
}}
.status-title {{
font-size: 1.3em;
font-weight: bold;
margin-bottom: 10px;
}}
.info-grid {{
display: grid;
grid-template-columns: repeat(auto-fit, minmax(300px, 1fr));
gap: 20px;
margin-bottom: 25px;
}}
.info-card {{
background: #f9f9f9;
border-radius: 8px;
padding: 20px;
border: 1px solid #e0e0e0;
}}
.info-card h3 {{
margin: 0 0 15px 0;
color: var(--color-primary, #2196F3);
font-size: 1.1em;
}}
.info-item {{
margin-bottom: 8px;
}}
.info-label {{
font-weight: bold;
display: inline-block;
width: 120px;
}}
.readings-table {{
width: 100%;
border-collapse: collapse;
margin-top: 20px;
background: white;
border-radius: 8px;
overflow: hidden;
box-shadow: 0 2px 4px rgba(0,0,0,0.1);
}}
.readings-table th {{
background: var(--color-primary, #2196F3);
color: white;
padding: 12px;
text-align: left;
font-weight: 600;
}}
.readings-table td {{
padding: 12px;
border-bottom: 1px solid #e0e0e0;
}}
.readings-table tr:nth-child(even) {{
background-color: #f9f9f9;
}}
.readings-table tr:hover {{
background-color: #f0f8ff;
}}
.created-badge {{
background-color: var(--color-success, #4CAF50);
color: white;
padding: 4px 8px;
border-radius: 4px;
font-size: 0.85em;
}}
.failed-badge {{
background-color: var(--color-danger, #F44336);
color: white;
padding: 4px 8px;
border-radius: 4px;
font-size: 0.85em;
}}
.error-list {{
background: #ffebee;
border-radius: 8px;
padding: 15px;
margin-top: 15px;
}}
.error-item {{
background: white;
border-left: 4px solid var(--color-danger, #F44336);
padding: 10px;
margin-bottom: 10px;
border-radius: 4px;
}}
.error-code {{
font-weight: bold;
color: var(--color-danger, #F44336);
}}
.summary-stats {{
display: grid;
grid-template-columns: repeat(auto-fit, minmax(150px, 1fr));
gap: 15px;
margin-top: 25px;
}}
.stat-card {{
text-align: center;
background: linear-gradient(135deg, #f8f9fa, #e9ecef);
border-radius: 8px;
padding: 20px;
border: 1px solid #dee2e6;
}}
.stat-number {{
font-size: 2em;
font-weight: bold;
margin-bottom: 5px;
}}
.stat-label {{
color: #666;
font-size: 0.9em;
}}
.variables-list {{
max-height: 200px;
overflow-y: auto;
border: 1px solid #e0e0e0;
border-radius: 4px;
padding: 10px;
background: white;
}}
.variable-item {{
padding: 5px;
border-bottom: 1px solid #f0f0f0;
font-family: monospace;
font-size: 0.9em;
}}
.variable-item:last-child {{
border-bottom: none;
}}
</style>
</head>
<body>
<div class="container">
<div class="header">
<h1>{status_icon} Ultimo-Zählerstand Eingabe</h1>
</div>
<div class="content">
<div class="status-card {status_class}">
<div class="status-title">{status_icon} Status: {status_text}</div>
'''
if result['success']:
html_content += f'''
<p>Ultimo-Zählerstände wurden erfolgreich eingegeben.</p>
<p><strong>Anzahl erfolgreich erstellt:</strong> {len(result.get('created', []))}</p>
'''
else:
html_content += f'''
<p>Bei der Eingabe der Ultimo-Zählerstände sind Fehler aufgetreten.</p>
'''
html_content += '''
</div>
<div class="info-grid">
<div class="info-card">
<h3>📊 Sensor Information</h3>
'''
if sensor_info:
html_content += f'''
<div class="info-item">
<span class="info-label">Sensor-ID:</span>
{sensor_info['id']}
</div>
<div class="info-item">
<span class="info-label">Name:</span>
{sensor_display}
</div>
'''
if measure_concept_name:
html_content += f'''
<div class="info-item">
<span class="info-label">Messkonzept:</span>
{measure_concept_name}
</div>
'''
html_content += f'''
</div>
<div class="info-card">
<h3>⚙️ Variable & Einheit</h3>
<div class="info-item">
<span class="info-label">Variable:</span>
{variable_name if variable_name else 'Standard'}
</div>
<div class="info-item">
<span class="info-label">Einheit:</span>
{variable_unit if variable_unit else 'Standard'}
</div>
</div>
</div>
'''
# Show readings table
html_content += '''
<h3>📋 Eingegeben Zählerstände</h3>
<table class="readings-table">
<thead>
<tr>
<th>Monat</th>
<th>Zählerstand</th>
<th>Status</th>
</tr>
</thead>
<tbody>
'''
created_ids = [obs['id'] for obs in result.get('created', [])]
for reading in readings:
month = reading['month']
value = reading['meterValue']
# Check if this reading was successfully created
is_created = any(obs for obs in result.get('created', []) if month in obs.get('moment', ''))
status_badge = '<span class="created-badge">Erstellt</span>' if is_created else '<span class="failed-badge">Fehler</span>'
html_content += f'''
<tr>
<td>{month}</td>
<td>{value:,.2f}</td>
<td>{status_badge}</td>
</tr>
'''
html_content += '''
</tbody>
</table>
'''
# Show errors if any
if result.get('errors'):
html_content += '''
<h3>⚠️ Fehler</h3>
<div class="error-list">
'''
for error in result['errors']:
html_content += f'''
<div class="error-item">
<div class="error-code">{error.get('code', 'ERROR')}</div>
<div>{error.get('message', 'Unbekannter Fehler')}</div>
'''
if error.get('details'):
html_content += f'<div><small>Details: {error["details"]}</small></div>'
html_content += '</div>'
html_content += '''
</div>
'''
# Show created observations details if any
if result.get('created'):
html_content += '''
<h3>✅ Erfolgreich erstellte Beobachtungen</h3>
<table class="readings-table">
<thead>
<tr>
<th>ID</th>
<th>Zeitpunkt</th>
<th>Wert</th>
<th>Zählerstand</th>
</tr>
</thead>
<tbody>
'''
for obs in result['created']:
try:
moment = datetime.fromisoformat(obs['moment'].replace('Z', '+00:00'))
moment_str = moment.strftime('%d.%m.%Y %H:%M')
except:
moment_str = obs['moment']
html_content += f'''
<tr>
<td>{obs['id']}</td>
<td>{moment_str}</td>
<td>{obs.get('value', 0):,.2f}</td>
<td>{obs.get('meterValue', 0):,.2f}</td>
</tr>
'''
html_content += '''
</tbody>
</table>
'''
# Summary statistics
total_readings = len(readings)
successful_readings = len(result.get('created', []))
failed_readings = total_readings - successful_readings
html_content += f'''
<div class="summary-stats">
<div class="stat-card">
<div class="stat-number">{total_readings}</div>
<div class="stat-label">Gesamt eingegeben</div>
</div>
<div class="stat-card">
<div class="stat-number" style="color: var(--color-success, #4CAF50);">{successful_readings}</div>
<div class="stat-label">Erfolgreich</div>
</div>
<div class="stat-card">
<div class="stat-number" style="color: var(--color-danger, #F44336);">{failed_readings}</div>
<div class="stat-label">Fehlgeschlagen</div>
</div>
</div>
'''
# Show available variables
if available_variables:
html_content += '''
<h3>🔧 Verfügbare Variablen für diesen Sensor</h3>
<div class="variables-list">
'''
for var in available_variables[:20]: # Limit to first 20
var_name = var['variableName'].strip()
unit_name = var['unitName'].strip()
html_content += f'''
<div class="variable-item">{var_name} ({unit_name})</div>
'''
if len(available_variables) > 20:
html_content += f'<div class="variable-item">... und {len(available_variables) - 20} weitere</div>'
html_content += '''
</div>
'''
html_content += '''
</div>
</div>
</body>
</html>
'''
return html_content
# Execute main function
result = main()
Write Preview
Loading…
Cancel
Save