import asyncio import os import json import mysql.connector import smtplib from email.message import EmailMessage from datetime import datetime, timedelta from azure.servicebus.aio import ServiceBusClient NAMESPACE_CONNECTION_STR = os.environ.get("NAMESPACE_CONNECTION_STR", "Endpoint=sb://charge-management.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=/lHc1UQJpSLycsjnjIBIXxwDy5IeaH49u+ASbOvhm2M=") SUBSCRIPTION_NAME = os.environ.get("SUBSCRIPTION_NAME", "charge-notification-walle") TOPIC_START_NAME = os.environ.get("TOPIC_START_NAME", "start-charging-walle") TOPIC_STOP_NAME = os.environ.get("TOPIC_STOP_NAME", "stop-charging-walle") TOPIC_PROGRESS_NAME = os.environ.get("TOPIC_PROGRESS_NAME", "progress-charging-walle") # Mail server configuration MAIL_SERVER = 'smtp.ionos.de' MAIL_PORT = 587 MAIL_USE_TLS = True MAIL_USE_SSL = False MAIL_USERNAME = 'login@wall-e.dev' MAIL_PASSWORD = 'wall2000e!mail' MAIL_DEFAULT_SENDER = 'WALL-E Charging Portal ' MAIL_RECIPIENTS = ['seb@neleso.com', 'henry.grafvonhelldorff@wall-e.works'] # Counters for processed messages processed_count = 0 signed_data_count = 0 counter_lock = asyncio.Lock() # CREATE TABLE `op_server_session_log` ( # `id` int NOT NULL, # `device_id` varchar(80) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NOT NULL, # `transaction_id` varchar(255) NOT NULL DEFAULT '', # `messageType` varchar(80) NOT NULL, # `json` longtext NOT NULL, # `ts` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP # ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci; # INSERT INTO `op_server_session_log` (`id`, `device_id`, `transaction_id`, `messageType`, `json`, `ts`) VALUES # (1, 'TACW1144122G0894', '', 'Start', '{\"id\":\"01817a59-fc2c-4b7c-b0f9-882458db4500\",\"domain\":\"ocpp\",\"deviceId\":\"TACW1144122G0894\",\"connector_id\":1,\"register\":\"Energy.Active.Import.Register\",\"value\":\"3639389\",\"unit\":\"Wh\",\"location\":null,\"context\":null,\"timestamp\":\"2024-11-14T17:29:52Z\",\"phases\":null,\"session\":{\"Transaction.Id\":1731605392,\"ID.Tag\":\"TACW1144122G0894\",\"Energy.Meter.Start\":3639389,\"Energy.Meter.Stop\":null,\"Reason.Session\":null,\"unit\":\"Wh\",\"Data.Transfer.Value\":\"{}\",\"Transaction.Data\":null}}', '2024-11-14 19:53:58'); def _send_email(processed, signed): """Send summary email with the given counts.""" msg = EmailMessage() msg['Subject'] = 'Service Bus Processing Summary' msg['From'] = MAIL_DEFAULT_SENDER msg['To'] = ', '.join(MAIL_RECIPIENTS) msg.set_content( f"Processed messages since last email: {processed}\n" f"Messages with 'signedData': {signed}" ) with smtplib.SMTP(MAIL_SERVER, MAIL_PORT) as server: if MAIL_USE_TLS: server.starttls() if MAIL_USERNAME and MAIL_PASSWORD: server.login(MAIL_USERNAME, MAIL_PASSWORD) server.send_message(msg) def _seconds_until_next_run(): now = datetime.now() # Run every six hours at 3, 9, 15 and 21 o'clock schedule_hours = [3, 9, 15, 21] for hour in schedule_hours: target = now.replace(hour=hour, minute=0, second=0, microsecond=0) if target > now: return (target - now).total_seconds() # Next day at 3:00 next_day = now + timedelta(days=1) target = next_day.replace(hour=3, minute=0, second=0, microsecond=0) return (target - now).total_seconds() async def send_periodic_email(): global processed_count, signed_data_count while True: await asyncio.sleep(_seconds_until_next_run()) async with counter_lock: processed = processed_count signed = signed_data_count processed_count = 0 signed_data_count = 0 await asyncio.to_thread(_send_email, processed, signed) def update_db_fleet(query, args=(), one=False): mydb = mysql.connector.connect( host="212.227.9.218", user="walle", password="zm0dem123", collation="utf8mb4_general_ci", database="walle_db", charset="utf8mb4" ) cur = mydb.cursor() cur.execute(query, args) mydb.commit() mydb.close() return None async def start_transaction(): global processed_count, signed_data_count while True: try: async with ServiceBusClient.from_connection_string( conn_str=NAMESPACE_CONNECTION_STR, logging_enable=True) as servicebus_client: receiver = servicebus_client.get_subscription_receiver(topic_name=TOPIC_START_NAME, subscription_name=SUBSCRIPTION_NAME) async with receiver: print("Start receiving messages from " + SUBSCRIPTION_NAME + " for topic: " + TOPIC_START_NAME) async for msg in receiver: try: print("Start: " + str(msg)) messageType = "Start" data_dict = json.loads(str(msg)) device_id = data_dict.get("deviceId") # Access Transaction.Id safely transaction_id = data_dict.get("session", {}).get("Transaction.Id", None) myt = "" # Print the result if transaction_id is not None: print(f"Transaction.Id: {transaction_id}") myt = str(transaction_id) else: print("Transaction.Id not found or is None") mysequel = "INSERT INTO op_server_session_log (device_id, messageType, json, transaction_id) VALUES (%s, %s, %s, %s)" update_db_fleet(mysequel, (device_id, messageType, str(msg), myt,)) async with counter_lock: processed_count += 1 if 'signedData' in str(msg): signed_data_count += 1 except Exception as e: print(f"Error processing start message: {e}") finally: try: # important: complete the message so that the message is removed from the subscription await receiver.complete_message(msg) except Exception as e: print(f"Error completing start message: {e}") except Exception as e: print(f"Error in start_transaction: {e}") await asyncio.sleep(5) async def stop_transaction(): global processed_count, signed_data_count while True: try: async with ServiceBusClient.from_connection_string( conn_str=NAMESPACE_CONNECTION_STR, logging_enable=True) as servicebus_client: receiver = servicebus_client.get_subscription_receiver(topic_name=TOPIC_STOP_NAME, subscription_name=SUBSCRIPTION_NAME) async with receiver: print("Start receiving messages from " + SUBSCRIPTION_NAME + " for topic: " + TOPIC_STOP_NAME) async for msg in receiver: try: print("Stop: " + str(msg)) messageType = "Stop" data_dict = json.loads(str(msg)) device_id = data_dict.get("deviceId") # Access Transaction.Id safely transaction_id = data_dict.get("session", {}).get("Transaction.Id", None) myt = "" # Print the result if transaction_id is not None: print(f"Transaction.Id: {transaction_id}") myt = str(transaction_id) else: print("Transaction.Id not found or is None") mysequel = "INSERT INTO op_server_session_log (device_id, messageType, json, transaction_id) VALUES (%s, %s, %s, %s)" update_db_fleet(mysequel, (device_id, messageType, str(msg),myt,)) async with counter_lock: processed_count += 1 if 'signedData' in str(msg): signed_data_count += 1 except Exception as e: print(f"Error processing stop message: {e}") finally: try: # important: complete the message so that the message is removed from the subscription await receiver.complete_message(msg) except Exception as e: print(f"Error completing stop message: {e}") except Exception as e: print(f"Error in stop_transaction: {e}") await asyncio.sleep(5) async def progress_transaction(): global processed_count, signed_data_count while True: try: async with ServiceBusClient.from_connection_string( conn_str=NAMESPACE_CONNECTION_STR, logging_enable=True) as servicebus_client: receiver = servicebus_client.get_subscription_receiver(topic_name=TOPIC_PROGRESS_NAME, subscription_name=SUBSCRIPTION_NAME) async with receiver: print("Start receiving messages from " + SUBSCRIPTION_NAME + " for topic: " + TOPIC_PROGRESS_NAME) async for msg in receiver: try: print("Progress: " + str(msg)) messageType = "Progress" data_dict = json.loads(str(msg)) device_id = data_dict.get("deviceId") # Access Transaction.Id safely transaction_id = data_dict.get("session", {}).get("Transaction.Id", None) myt = "" # Print the result if transaction_id is not None: print(f"Transaction.Id: {transaction_id}") myt = str(transaction_id) # Safely access the value of the "register" key register_value = data_dict.get("value", None) if register_value is not None: print(f"Register: {register_value} - insert.") mysequel = "INSERT INTO op_server_session_log (device_id, messageType, json, transaction_id) VALUES (%s, %s, %s, %s)" update_db_fleet(mysequel, (device_id, messageType, str(msg),myt,)) update_db_fleet("DELETE FROM c4_session_metadata WHERE m_key = 'LMR_" + str(transaction_id) + "'") current_year_month = str(datetime.now().strftime("%Y-%m")) istr = "INSERT INTO c4_session_metadata (tenant_id, site_id, m_key, month, m_val) VALUES ('0', '0', 'LMR_" + str(myt) + "', '" + current_year_month + "', '" + str(register_value) + "')" update_db_fleet(istr) else: print("Register key not found.") else: print("Transaction.Id not found or is None") mysequel = "INSERT INTO op_server_session_log (device_id, messageType, json, transaction_id) VALUES (%s, %s, %s, %s)" update_db_fleet(mysequel, (device_id, messageType, str(msg),myt,)) async with counter_lock: processed_count += 1 if 'signedData' in str(msg): signed_data_count += 1 except Exception as e: print(f"Error processing progress message: {e}") finally: try: # important: complete the message so that the message is removed from the subscription await receiver.complete_message(msg) except Exception as e: print(f"Error completing progress message: {e}") except Exception as e: print(f"Error in progress_transaction: {e}") await asyncio.sleep(5) async def main(): await asyncio.gather( start_transaction(), stop_transaction(), progress_transaction(), send_periodic_email(), ) # await asyncio.gather(start_transaction(), stop_transaction()) current_year_month = str(datetime.now().strftime("%Y-%m"))