#!/usr/bin/env python3 """ SteVe Monthly Declaration Report Generator Genereert maandelijkse declaratie rapporten van laadtransacties """ import mysql.connector from datetime import datetime, timedelta import argparse import sys import io import os from typing import List, Dict, Optional import calendar from collections import defaultdict os.environ['PYTHONIOENCODING'] = 'utf-8' # Forceer UTF-8 encoding voor stdout/stderr op Windows if sys.platform == 'win32': sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8') sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8') class MonthlyDeclarationReporter: 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_monthly_transactions(self, year: int, month: int, user_filter: Optional[str] = None) -> List[Dict]: """Haal alle transacties op voor een specifieke maand""" # Bereken eerste en laatste dag van de maand first_day = datetime(year, month, 1) last_day = datetime(year, month, calendar.monthrange(year, month)[1], 23, 59, 59) 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.start_timestamp >= %s AND t.start_timestamp <= %s AND t.stop_timestamp IS NOT NULL """ params = [first_day, last_day] if user_filter: query += " AND t.id_tag = %s" params.append(user_filter) query += " ORDER BY t.start_timestamp ASC" self.cursor.execute(query, params) 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: 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) except mysql.connector.Error: return None # Round timestamp naar het uur hour_timestamp = timestamp.replace(minute=0, second=0, microsecond=0) # Probeer eerst dynamic_price_data 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 calculate_transaction_cost(self, transaction: Dict, provider: str, price_db: str) -> Dict: """Bereken de kosten voor een transactie""" start_time = transaction['start_timestamp'] stop_time = transaction['stop_timestamp'] start_val = float(transaction['start_value']) if transaction['start_value'] else 0 stop_val = float(transaction['stop_value']) if transaction['stop_value'] else 0 total_kwh = (stop_val - start_val) / 1000 # Omzetten naar kWh # Bereken kosten per uur current_hour = start_time.replace(minute=0, second=0, microsecond=0) total_cost = 0 hours_with_price = 0 hours_without_price = 0 while current_hour <= stop_time: # Bereken verbruik in dit uur hour_start = max(current_hour, start_time) hour_end = min(current_hour + timedelta(hours=1), stop_time) if hour_end > hour_start: # Bereken fractie van het uur hour_fraction = (hour_end - hour_start).total_seconds() / 3600 hour_kwh = total_kwh * hour_fraction * 3600 / (stop_time - start_time).total_seconds() # Haal prijs op price = self.get_price(current_hour, provider, price_db) if price is not None: hour_cost = hour_kwh * price total_cost += hour_cost hours_with_price += 1 else: hours_without_price += 1 current_hour += timedelta(hours=1) return { 'total_kwh': total_kwh, 'total_cost': total_cost, 'hours_with_price': hours_with_price, 'hours_without_price': hours_without_price, 'avg_price': total_cost / total_kwh if total_kwh > 0 and hours_with_price > 0 else 0 } def generate_monthly_report(self, year: int, month: int, provider: str = 'NE', price_db: str = 'alfen', user_filter: Optional[str] = None, format_type: str = 'detailed'): """Genereer maandelijks declaratie rapport""" transactions = self.get_monthly_transactions(year, month, user_filter) if not transactions: print(f"Geen transacties gevonden voor {calendar.month_name[month]} {year}") return # Header print("=" * 100) print(f"DECLARATIE RAPPORT LAADTRANSACTIES") print(f"Periode: {calendar.month_name[month]} {year}") print("=" * 100) # Groepeer transacties per gebruiker user_transactions = defaultdict(list) for trans in transactions: user_transactions[trans['id_tag']].append(trans) # Totalen voor het hele rapport grand_total_sessions = 0 grand_total_kwh = 0 grand_total_cost = 0 grand_total_duration = timedelta() # Print per gebruiker for id_tag, user_trans in sorted(user_transactions.items()): # Gebruikersinfo user_info = self.get_user_info(id_tag) if user_info: print(f"\nGebruiker: {user_info['first_name']} {user_info['last_name']}") print(f"Email: {user_info['e_mail']}") print(f"RFID Tag: {id_tag}") print("-" * 100) if format_type == 'detailed': # Gedetailleerde tabel print(f"{'Datum':<12} {'Start':<6} {'Einde':<6} {'Duur':<8} " f"{'kWh':>8} {'Kosten':>10} {'Gem. prijs':>12} {'Laadpaal':<20}") print("-" * 100) user_total_sessions = 0 user_total_kwh = 0 user_total_cost = 0 user_total_duration = timedelta() for trans in sorted(user_trans, key=lambda x: x['start_timestamp']): # Bereken kosten cost_info = self.calculate_transaction_cost(trans, provider, price_db) # Bereken duur duration = trans['stop_timestamp'] - trans['start_timestamp'] hours = int(duration.total_seconds() // 3600) minutes = int((duration.total_seconds() % 3600) // 60) duration_str = f"{hours}:{minutes:02d}" # Update totalen user_total_sessions += 1 user_total_kwh += cost_info['total_kwh'] user_total_cost += cost_info['total_cost'] user_total_duration += duration if format_type == 'detailed': # Print transactie regel print(f"{trans['start_timestamp'].strftime('%d-%m-%Y'):<12} " f"{trans['start_timestamp'].strftime('%H:%M'):<6} " f"{trans['stop_timestamp'].strftime('%H:%M'):<6} " f"{duration_str:<8} " f"{cost_info['total_kwh']:>8.2f} " f"€{cost_info['total_cost']:>9.2f} " f"€{cost_info['avg_price']:>11.4f} " f"{trans['charge_box_id'][:19]:<20}") if cost_info['hours_without_price'] > 0: print(f" * Waarschuwing: {cost_info['hours_without_price']} uur zonder prijsdata") # Gebruiker totalen print("-" * 100) total_hours = int(user_total_duration.total_seconds() // 3600) total_minutes = int((user_total_duration.total_seconds() % 3600) // 60) print(f"Totaal gebruiker: {user_total_sessions} sessies | " f"{total_hours}:{total_minutes:02d} uur | " f"{user_total_kwh:.2f} kWh | " f"€ {user_total_cost:.2f}") if user_total_kwh > 0: avg_price_user = user_total_cost / user_total_kwh print(f"Gemiddelde prijs: € {avg_price_user:.4f} per kWh") # Update grand totalen grand_total_sessions += user_total_sessions grand_total_kwh += user_total_kwh grand_total_cost += user_total_cost grand_total_duration += user_total_duration # Grand totalen print("\n" + "=" * 100) print("TOTAAL OVERZICHT") print("=" * 100) total_hours = int(grand_total_duration.total_seconds() // 3600) total_minutes = int((grand_total_duration.total_seconds() % 3600) // 60) print(f"Aantal gebruikers: {len(user_transactions)}") print(f"Totaal aantal sessies: {grand_total_sessions}") print(f"Totale laadtijd: {total_hours}:{total_minutes:02d} uur") print(f"Totaal verbruik: {grand_total_kwh:.2f} kWh") print(f"TOTALE KOSTEN: € {grand_total_cost:.2f}") if grand_total_kwh > 0: avg_price_total = grand_total_cost / grand_total_kwh print(f"Gemiddelde prijs: € {avg_price_total:.4f} per kWh") # Statistieken print(f"\nGemiddeld per sessie:") if grand_total_sessions > 0: avg_kwh = grand_total_kwh / grand_total_sessions avg_cost = grand_total_cost / grand_total_sessions avg_duration = grand_total_duration / grand_total_sessions avg_hours = int(avg_duration.total_seconds() // 3600) avg_minutes = int((avg_duration.total_seconds() % 3600) // 60) print(f" Verbruik: {avg_kwh:.2f} kWh") print(f" Kosten: € {avg_cost:.2f}") print(f" Duur: {avg_hours}:{avg_minutes:02d} uur") print("\n" + "=" * 100) print(f"Rapport gegenereerd op: {datetime.now().strftime('%d-%m-%Y %H:%M')}") print(f"Energieprovider: {provider}") print("=" * 100) def main(): parser = argparse.ArgumentParser( description='SteVe Monthly Declaration Report Generator', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Voorbeelden: # Genereer rapport voor oktober 2024 python3 steve_monthly_declaration.py --host 192.168.178.201 --port 3307 --password 'geheim' --month 10 --year 2024 # Genereer rapport voor specifieke gebruiker in november 2024 python3 steve_monthly_declaration.py --host 192.168.178.201 --port 3307 --password 'geheim' --month 11 --year 2024 --user 04A2CBA2C43C80 # Gebruik aparte credentials voor prijzen database python3 steve_monthly_declaration.py --host 192.168.178.201 --port 3307 --password 'geheim' \\ --price-user alfen_user --price-password 'geheim2' --month 10 --year 2024 # Genereer beknopt rapport (zonder details per transactie) python3 steve_monthly_declaration.py --host 192.168.178.201 --port 3307 --password 'geheim' --month 10 --year 2024 --summary """ ) # 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') # Maand/jaar selectie parser.add_argument('--month', type=int, required=True, choices=range(1, 13), help='Maand (1-12)') parser.add_argument('--year', type=int, required=True, help='Jaar (bijv. 2024)') # Filter opties parser.add_argument('--rfid', dest='user_filter', help='Filter op specifieke RFID tag/gebruiker') parser.add_argument('--summary', action='store_true', help='Toon alleen samenvattingen (geen transactie details)') # 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 = MonthlyDeclarationReporter( 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() # Bepaal format type format_type = 'summary' if args.summary else 'detailed' # Genereer rapport reporter.generate_monthly_report( year=args.year, month=args.month, provider=args.provider, price_db=args.price_db, user_filter=args.user_filter, format_type=format_type ) except KeyboardInterrupt: print("\n\nAfgebroken door gebruiker") sys.exit(0) except Exception as e: print(f"✗ Fout: {e}") import traceback traceback.print_exc() sys.exit(1) finally: reporter.disconnect() if __name__ == '__main__': main()