#!/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, price_host: str = None, price_port: int = None, price_user: str = None, price_password: str = None): """Initialiseer database connectie""" self.config = { 'host': host, 'database': database, 'user': user, 'password': password, 'port': port } # Prijzen database config (gebruik SteVe config als fallback) self.price_config = { 'host': price_host or host, 'port': price_port or port, 'user': price_user or user, 'password': price_password or password } self.conn = None self.cursor = None self.price_conn = None self.price_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() if hasattr(self, 'price_cursor') and self.price_cursor: self.price_cursor.close() if hasattr(self, 'price_conn') and self.price_conn: self.price_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 get_price(self, timestamp: datetime, provider: str, price_db: str) -> Optional[float]: """Haal prijs op voor een specifieke timestamp en provider""" # Maak verbinding met prijzen database indien nodig if not hasattr(self, 'price_conn') or self.price_conn is None: try: # Gebruik price_config met de juiste database naam price_config = self.price_config.copy() price_config['database'] = price_db self.price_conn = mysql.connector.connect(**price_config) self.price_cursor = self.price_conn.cursor(dictionary=True) print(f"✓ Verbonden met prijzen database {price_db} op {price_config['host']}") except mysql.connector.Error as err: print(f"⚠ Waarschuwing: Kan geen verbinding maken met prijzen database: {err}") return None # Round timestamp naar het uur hour_timestamp = timestamp.replace(minute=0, second=0, microsecond=0) # Probeer eerst dynamic_price_data (voor historische en vandaag) query = """ SELECT price FROM dynamic_price_data WHERE datetime = %s AND provider_code = %s LIMIT 1 """ try: self.price_cursor.execute(query, (hour_timestamp, provider)) result = self.price_cursor.fetchone() if result: return float(result['price']) except mysql.connector.Error: pass # Zo niet, probeer dynamic_price_data_tommorow (alleen NextEnergy) if provider == 'NE': query = """ SELECT price FROM dynamic_price_data_tommorow WHERE datetime = %s LIMIT 1 """ try: self.price_cursor.execute(query, (hour_timestamp,)) result = self.price_cursor.fetchone() if result: return float(result['price']) except mysql.connector.Error: pass return None 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, provider: str = 'NE', price_db: str = 'alfen'): """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: {start_value:.3f} kWh") print(f"Einde: {stop_value:.3f} kWh") print(f"Totaal geladen: {total_consumption:.2f} kWh") if transaction['stop_reason']: print(f"Stop: {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 EN KOSTEN PER UUR:") print("-" * 80) print(f"{'Uur':<12} {'Start':<10} {'Eind':<10} {'kWh':<8} {'€/kWh':<11} {'Kosten':<10}") print("-" * 80) total_detailed = 0 total_cost = 0 hours_with_price = 0 hours_without_price = 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') # Haal prijs op voor dit uur price = self.get_price(hour_dt, provider, price_db) if price is not None: cost = hour['consumption'] * price total_cost += cost hours_with_price += 1 price_str = f"{price:.5f}" cost_str = f"€{cost:.2f}" else: hours_without_price += 1 price_str = "n.v.t." cost_str = "n.v.t." print(f"{hour_display:<12} {hour['start_value']:<10.3f} " f"{hour['end_value']:<10.3f} {hour['consumption']:<8.3f} " f"{price_str:<11} {cost_str:<10}") total_detailed += hour['consumption'] print("-" * 80) cost_total_str = f"€{total_cost:.2f}" if hours_without_price == 0 else f"€{total_cost:.2f}*" print(f"{'TOTAAL':<12} {hourly_data[0]['start_value']:<10.3f} " f"{hourly_data[-1]['end_value']:<10.3f} {total_detailed:<8.3f} " f"{'':11} {cost_total_str:<10}") # 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 uren: {len(active_hours)}") print(f"Provider: {provider}") avg_per_hour = sum(h['consumption'] for h in active_hours) / len(active_hours) print(f"Gemiddeld/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')} ({max_hour['consumption']:.2f} kWh)") # Vind duurste en goedkoopste uur hours_with_costs = [] for hour in hourly_data: if hour['consumption'] > 0.01: hour_dt = datetime.strptime(hour['hour'], '%Y-%m-%d %H:00') price = self.get_price(hour_dt, provider, price_db) if price is not None: hours_with_costs.append((hour_dt, price, hour['consumption'])) if hours_with_costs: most_expensive = max(hours_with_costs, key=lambda x: x[1]) cheapest = min(hours_with_costs, key=lambda x: x[1]) print(f"Duurste uur: {most_expensive[0].strftime('%d-%m %H:00')} " f"(€{most_expensive[1]:.5f}/kWh)") print(f"Goedkoopste uur: {cheapest[0].strftime('%d-%m %H:00')} " f"(€{cheapest[1]:.5f}/kWh)") # Totale kosten print(f"\nTOTAALE KOSTEN: {cost_total_str}") if hours_without_price > 0: print(f"* {hours_without_price} uur zonder prijsdata") # Gemiddelde prijs if hours_with_price > 0: avg_price = total_cost / total_detailed if total_detailed > 0 else 0 print(f"Gemiddelde prijs: €{avg_price:.5f}/kWh") 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" + "=" * 80) print(f"OVERZICHT VAN LAATSTE {len(transactions)} TRANSACTIES") print("=" * 80) print(f"{'ID':<4} {'Start':<17} {'Duur':<10} {'kWh':<8} {'Laadpaal':<16} {'RFID':<12}") print("-" * 80) 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'][:14] if t['charge_box_id'] else "?" rfid = t['id_tag'][:10] if t['id_tag'] else "?" print(f"{trans_id:<4} {start:<17} {duration_str:<10} {kwh:<8.2f} {charge_box:<16} {rfid:<12}") print("=" * 80) print(f"\nGebruik --transaction 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 --host 192.168.178.201 --port 3307 --password 'geheim' --list # Toon gedetailleerd rapport van transactie 1 met NextEnergy prijzen python3 steve_transaction_report.py --host 192.168.178.201 --port 3307 --password 'geheim' --transaction 1 # Gebruik aparte credentials voor prijzen database python3 steve_transaction_report.py --host 192.168.178.201 --port 3307 --password 'geheim' \ --price-host 192.168.178.201 --price-port 3307 --price-user prijzen --price-password 'geheim2' --transaction 1 # Toon rapport met andere provider (bijv. 'AA' voor AllAbout) python3 steve_transaction_report.py --host 192.168.178.201 --port 3307 --password 'geheim' --transaction 1 --provider AA # Toon laatste 20 transacties in lijst python3 steve_transaction_report.py --host 192.168.178.201 --port 3307 --password 'geheim' --list --limit 20 # Toon alle transacties gedetailleerd (laatste 10) python3 steve_transaction_report.py --host 192.168.178.201 --port 3307 --password 'geheim' """ ) # 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 compact overzicht van transacties (tabel format)') parser.add_argument('--transaction', type=int, metavar='ID', help='Toon gedetailleerd rapport van specifieke transactie ID') parser.add_argument('--limit', type=int, default=10, help='Aantal transacties (default: 10)') parser.add_argument('--simple', action='store_true', help='Eenvoudig rapport zonder uurdetails') # Prijzen opties parser.add_argument('--provider', default='NE', help='Energie provider code (default: NE voor NextEnergy)') parser.add_argument('--price-db', default='alfen', help='Database naam voor prijzen (default: alfen)') parser.add_argument('--price-host', default=None, help='Database host voor prijzen (default: zelfde als --host)') parser.add_argument('--price-port', type=int, default=None, help='Database poort voor prijzen (default: zelfde als --port)') parser.add_argument('--price-user', default=None, help='Database gebruiker voor prijzen (default: zelfde als --user)') parser.add_argument('--price-password', default=None, help='Database wachtwoord voor prijzen (default: zelfde als --password)') args = parser.parse_args() # Maak reporter object reporter = SteVeReporter( host=args.host, database=args.database, user=args.user, password=args.password, port=args.port, price_host=args.price_host, price_port=args.price_port, price_user=args.price_user, price_password=args.price_password ) try: reporter.connect() if args.transaction: # Toon specifieke transactie transactions = reporter.get_transactions(transaction_id=args.transaction) if transactions: reporter.print_transaction_report(transactions[0], detailed=not args.simple, provider=args.provider, price_db=args.price_db) else: print(f"✗ Transactie {args.transaction} niet gevonden") sys.exit(1) elif args.list: # Toon compact overzicht reporter.list_transactions(limit=args.limit) else: # Geen specifieke optie: toon alle transacties gedetailleerd transactions = reporter.get_transactions(limit=args.limit) if not transactions: print("Geen transacties gevonden.") else: print(f"\n{'=' * 80}") print(f"GEDETAILLEERDE RAPPORTEN VAN LAATSTE {len(transactions)} TRANSACTIES") print(f"{'=' * 80}") for i, trans in enumerate(transactions): if i > 0: print("\n\n") reporter.print_transaction_report(trans, detailed=not args.simple, provider=args.provider, price_db=args.price_db) 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()