366 lines
14 KiB
Python
Executable File
366 lines
14 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
SteVe Transaction Report Generator
|
|
Genereert gedetailleerde rapporten van laadtransacties uit de SteVe database
|
|
"""
|
|
|
|
import mysql.connector
|
|
from datetime import datetime, timedelta
|
|
import argparse
|
|
import sys
|
|
from typing import List, Dict, Optional
|
|
|
|
class SteVeReporter:
|
|
def __init__(self, host: str, database: str, user: str, password: str, port: int = 3306):
|
|
"""Initialiseer database connectie"""
|
|
self.config = {
|
|
'host': host,
|
|
'database': database,
|
|
'user': user,
|
|
'password': password,
|
|
'port': port
|
|
}
|
|
self.conn = None
|
|
self.cursor = None
|
|
|
|
def connect(self):
|
|
"""Maak verbinding met de database"""
|
|
try:
|
|
self.conn = mysql.connector.connect(**self.config)
|
|
self.cursor = self.conn.cursor(dictionary=True)
|
|
print(f"✓ Verbonden met database {self.config['database']} op {self.config['host']}")
|
|
except mysql.connector.Error as err:
|
|
print(f"✗ Database fout: {err}")
|
|
sys.exit(1)
|
|
|
|
def disconnect(self):
|
|
"""Sluit database verbinding"""
|
|
if self.cursor:
|
|
self.cursor.close()
|
|
if self.conn:
|
|
self.conn.close()
|
|
|
|
def get_transactions(self, limit: Optional[int] = None, transaction_id: Optional[int] = None) -> List[Dict]:
|
|
"""Haal transacties op uit de database"""
|
|
if transaction_id:
|
|
query = """
|
|
SELECT
|
|
t.transaction_pk,
|
|
t.id_tag,
|
|
t.start_timestamp,
|
|
t.stop_timestamp,
|
|
t.start_value,
|
|
t.stop_value,
|
|
t.stop_reason,
|
|
cb.charge_box_id,
|
|
cb.description as charge_box_description,
|
|
c.connector_id
|
|
FROM transaction t
|
|
JOIN connector c ON t.connector_pk = c.connector_pk
|
|
JOIN charge_box cb ON c.charge_box_id = cb.charge_box_id
|
|
WHERE t.transaction_pk = %s
|
|
"""
|
|
self.cursor.execute(query, (transaction_id,))
|
|
else:
|
|
query = """
|
|
SELECT
|
|
t.transaction_pk,
|
|
t.id_tag,
|
|
t.start_timestamp,
|
|
t.stop_timestamp,
|
|
t.start_value,
|
|
t.stop_value,
|
|
t.stop_reason,
|
|
cb.charge_box_id,
|
|
cb.description as charge_box_description,
|
|
c.connector_id
|
|
FROM transaction t
|
|
JOIN connector c ON t.connector_pk = c.connector_pk
|
|
JOIN charge_box cb ON c.charge_box_id = cb.charge_box_id
|
|
ORDER BY t.start_timestamp DESC
|
|
"""
|
|
if limit:
|
|
query += f" LIMIT {limit}"
|
|
self.cursor.execute(query)
|
|
|
|
return self.cursor.fetchall()
|
|
|
|
def get_meter_values(self, transaction_pk: int) -> List[Dict]:
|
|
"""Haal alle meterwaarden op voor een specifieke transactie"""
|
|
query = """
|
|
SELECT
|
|
value_timestamp,
|
|
value,
|
|
reading_context,
|
|
measurand,
|
|
location,
|
|
unit
|
|
FROM connector_meter_value
|
|
WHERE transaction_pk = %s
|
|
ORDER BY value_timestamp ASC
|
|
"""
|
|
self.cursor.execute(query, (transaction_pk,))
|
|
return self.cursor.fetchall()
|
|
|
|
def get_user_info(self, id_tag: str) -> Optional[Dict]:
|
|
"""Haal gebruikersinformatie op"""
|
|
query = """
|
|
SELECT u.first_name, u.last_name, u.e_mail
|
|
FROM user u
|
|
JOIN user_ocpp_tag uot ON u.user_pk = uot.user_pk
|
|
JOIN ocpp_tag ot ON uot.ocpp_tag_pk = ot.ocpp_tag_pk
|
|
WHERE ot.id_tag = %s
|
|
"""
|
|
self.cursor.execute(query, (id_tag,))
|
|
return self.cursor.fetchone()
|
|
|
|
def analyze_hourly_consumption(self, meter_values: List[Dict]) -> List[Dict]:
|
|
"""Analyseer verbruik per uur"""
|
|
if not meter_values:
|
|
return []
|
|
|
|
hourly_data = {}
|
|
|
|
for mv in meter_values:
|
|
timestamp = mv['value_timestamp']
|
|
value = float(mv['value'])
|
|
hour_key = timestamp.strftime('%Y-%m-%d %H:00')
|
|
|
|
if hour_key not in hourly_data:
|
|
hourly_data[hour_key] = {
|
|
'hour': hour_key,
|
|
'start_value': value,
|
|
'end_value': value,
|
|
'start_time': timestamp,
|
|
'end_time': timestamp
|
|
}
|
|
else:
|
|
hourly_data[hour_key]['end_value'] = value
|
|
hourly_data[hour_key]['end_time'] = timestamp
|
|
|
|
# Bereken consumptie per uur
|
|
result = []
|
|
for hour_key in sorted(hourly_data.keys()):
|
|
hour_data = hourly_data[hour_key]
|
|
consumption = hour_data['end_value'] - hour_data['start_value']
|
|
result.append({
|
|
'hour': hour_key,
|
|
'start_value': hour_data['start_value'],
|
|
'end_value': hour_data['end_value'],
|
|
'consumption': consumption,
|
|
'start_time': hour_data['start_time'],
|
|
'end_time': hour_data['end_time']
|
|
})
|
|
|
|
return result
|
|
|
|
def print_transaction_report(self, transaction: Dict, detailed: bool = True):
|
|
"""Print een geformatteerd rapport van een transactie"""
|
|
print("\n" + "=" * 80)
|
|
print(f"LAADSESSIE RAPPORT - Transactie #{transaction['transaction_pk']}")
|
|
print("=" * 80)
|
|
|
|
# Basis informatie
|
|
print(f"Laadpaal: {transaction['charge_box_id']}")
|
|
if transaction['charge_box_description']:
|
|
print(f" {transaction['charge_box_description']}")
|
|
print(f"Connector: {transaction['connector_id']}")
|
|
print(f"RFID Tag: {transaction['id_tag']}")
|
|
|
|
# Gebruikersinformatie
|
|
user_info = self.get_user_info(transaction['id_tag'])
|
|
if user_info and user_info['first_name']:
|
|
user_name = f"{user_info['first_name']}"
|
|
if user_info['last_name']:
|
|
user_name += f" {user_info['last_name']}"
|
|
print(f"Gebruiker: {user_name}")
|
|
|
|
# Tijden
|
|
start_time = transaction['start_timestamp']
|
|
stop_time = transaction['stop_timestamp']
|
|
|
|
if start_time:
|
|
print(f"Start: {start_time.strftime('%d-%m-%Y %H:%M:%S')}")
|
|
if stop_time:
|
|
print(f"Einde: {stop_time.strftime('%d-%m-%Y %H:%M:%S')}")
|
|
if start_time:
|
|
duration = stop_time - start_time
|
|
hours = duration.total_seconds() / 3600
|
|
print(f"Duur: {int(hours)}u {int((hours % 1) * 60)}m")
|
|
|
|
# Verbruik
|
|
start_value = float(transaction['start_value']) if transaction['start_value'] else 0
|
|
stop_value = float(transaction['stop_value']) if transaction['stop_value'] else 0
|
|
total_consumption = stop_value - start_value
|
|
|
|
print(f"\nStart meterstand: {start_value:.3f} kWh")
|
|
print(f"Eind meterstand: {stop_value:.3f} kWh")
|
|
print(f"Totaal geladen: {total_consumption:.2f} kWh")
|
|
|
|
if transaction['stop_reason']:
|
|
print(f"Stop reden: {transaction['stop_reason']}")
|
|
|
|
print("=" * 80)
|
|
|
|
# Gedetailleerde analyse per uur
|
|
if detailed and stop_time:
|
|
meter_values = self.get_meter_values(transaction['transaction_pk'])
|
|
|
|
if meter_values:
|
|
hourly_data = self.analyze_hourly_consumption(meter_values)
|
|
|
|
if hourly_data:
|
|
print("\nVERBRUIK PER UUR:")
|
|
print("-" * 80)
|
|
print(f"{'Uur':<20} {'Start (kWh)':<15} {'Eind (kWh)':<15} {'Geladen (kWh)':<15}")
|
|
print("-" * 80)
|
|
|
|
total_detailed = 0
|
|
for hour in hourly_data:
|
|
hour_dt = datetime.strptime(hour['hour'], '%Y-%m-%d %H:00')
|
|
hour_display = hour_dt.strftime('%d-%m %H:%M')
|
|
|
|
print(f"{hour_display:<20} {hour['start_value']:<15.3f} "
|
|
f"{hour['end_value']:<15.3f} {hour['consumption']:<15.3f}")
|
|
total_detailed += hour['consumption']
|
|
|
|
print("-" * 80)
|
|
print(f"{'TOTAAL':<20} {hourly_data[0]['start_value']:<15.3f} "
|
|
f"{hourly_data[-1]['end_value']:<15.3f} {total_detailed:<15.3f}")
|
|
|
|
# Statistieken
|
|
active_hours = [h for h in hourly_data if h['consumption'] > 0.01]
|
|
if active_hours:
|
|
print("\n" + "=" * 80)
|
|
print("STATISTIEKEN:")
|
|
print("-" * 80)
|
|
print(f"Actieve laaduren: {len(active_hours)}")
|
|
avg_per_hour = sum(h['consumption'] for h in active_hours) / len(active_hours)
|
|
print(f"Gemiddeld per actief uur: {avg_per_hour:.2f} kWh")
|
|
|
|
max_hour = max(hourly_data, key=lambda x: x['consumption'])
|
|
if max_hour['consumption'] > 0:
|
|
max_time = datetime.strptime(max_hour['hour'], '%Y-%m-%d %H:00')
|
|
print(f"Piekuur: {max_time.strftime('%d-%m %H:00')} "
|
|
f"({max_hour['consumption']:.2f} kWh)")
|
|
|
|
# Bereken geschatte kosten (indicatief)
|
|
print(f"\nGeschatte kosten (€0.30/kWh): €{total_consumption * 0.30:.2f}")
|
|
print(f"Geschatte kosten (€0.10/kWh): €{total_consumption * 0.10:.2f}")
|
|
|
|
print("=" * 80)
|
|
|
|
def list_transactions(self, limit: int = 10):
|
|
"""Toon een overzicht van recente transacties"""
|
|
transactions = self.get_transactions(limit=limit)
|
|
|
|
if not transactions:
|
|
print("Geen transacties gevonden.")
|
|
return
|
|
|
|
print("\n" + "=" * 100)
|
|
print(f"OVERZICHT VAN LAATSTE {len(transactions)} TRANSACTIES")
|
|
print("=" * 100)
|
|
print(f"{'ID':<6} {'Start':<20} {'Duur':<12} {'kWh':<10} {'Laadpaal':<20} {'RFID':<15}")
|
|
print("-" * 100)
|
|
|
|
for t in transactions:
|
|
trans_id = t['transaction_pk']
|
|
start = t['start_timestamp'].strftime('%d-%m-%Y %H:%M') if t['start_timestamp'] else 'Onbekend'
|
|
|
|
if t['start_timestamp'] and t['stop_timestamp']:
|
|
duration = t['stop_timestamp'] - t['start_timestamp']
|
|
hours = duration.total_seconds() / 3600
|
|
duration_str = f"{int(hours)}u {int((hours % 1) * 60)}m"
|
|
else:
|
|
duration_str = "Actief" if not t['stop_timestamp'] else "?"
|
|
|
|
start_val = float(t['start_value']) if t['start_value'] else 0
|
|
stop_val = float(t['stop_value']) if t['stop_value'] else 0
|
|
kwh = stop_val - start_val
|
|
|
|
charge_box = t['charge_box_id'][:18] if t['charge_box_id'] else "?"
|
|
rfid = t['id_tag'][:13] if t['id_tag'] else "?"
|
|
|
|
print(f"{trans_id:<6} {start:<20} {duration_str:<12} {kwh:<10.2f} {charge_box:<20} {rfid:<15}")
|
|
|
|
print("=" * 100)
|
|
print(f"\nGebruik: python3 {sys.argv[0]} --transaction <ID> voor details\n")
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(
|
|
description='SteVe Transaction Report Generator',
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
epilog="""
|
|
Voorbeelden:
|
|
# Toon lijst van laatste 10 transacties
|
|
python3 steve_transaction_report.py --list
|
|
|
|
# Toon gedetailleerd rapport van transactie 1
|
|
python3 steve_transaction_report.py --transaction 1
|
|
|
|
# Toon laatste 20 transacties
|
|
python3 steve_transaction_report.py --list --limit 20
|
|
|
|
# Gebruik andere database credentials
|
|
python3 steve_transaction_report.py --host 192.168.1.100 --user steve --password geheim --transaction 1
|
|
"""
|
|
)
|
|
|
|
# Database opties
|
|
parser.add_argument('--host', default='localhost', help='Database host (default: localhost)')
|
|
parser.add_argument('--port', type=int, default=3306, help='Database poort (default: 3306)')
|
|
parser.add_argument('--database', default='stevedb', help='Database naam (default: stevedb)')
|
|
parser.add_argument('--user', default='steve', help='Database gebruiker (default: steve)')
|
|
parser.add_argument('--password', required=True, help='Database wachtwoord')
|
|
|
|
# Actie opties
|
|
parser.add_argument('--list', action='store_true', help='Toon overzicht van transacties')
|
|
parser.add_argument('--transaction', type=int, metavar='ID', help='Toon gedetailleerd rapport van transactie ID')
|
|
parser.add_argument('--limit', type=int, default=10, help='Aantal transacties bij --list (default: 10)')
|
|
parser.add_argument('--simple', action='store_true', help='Eenvoudig rapport zonder uurdetails')
|
|
|
|
args = parser.parse_args()
|
|
|
|
# Minimaal één actie nodig
|
|
if not args.list and not args.transaction:
|
|
parser.print_help()
|
|
sys.exit(1)
|
|
|
|
# Maak reporter object
|
|
reporter = SteVeReporter(
|
|
host=args.host,
|
|
database=args.database,
|
|
user=args.user,
|
|
password=args.password,
|
|
port=args.port
|
|
)
|
|
|
|
try:
|
|
reporter.connect()
|
|
|
|
if args.list:
|
|
reporter.list_transactions(limit=args.limit)
|
|
|
|
if args.transaction:
|
|
transactions = reporter.get_transactions(transaction_id=args.transaction)
|
|
if transactions:
|
|
reporter.print_transaction_report(transactions[0], detailed=not args.simple)
|
|
else:
|
|
print(f"✗ Transactie {args.transaction} niet gevonden")
|
|
sys.exit(1)
|
|
|
|
except KeyboardInterrupt:
|
|
print("\n\nAfgebroken door gebruiker")
|
|
sys.exit(0)
|
|
except Exception as e:
|
|
print(f"✗ Fout: {e}")
|
|
sys.exit(1)
|
|
finally:
|
|
reporter.disconnect()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|