test: migrate ocpp server to poetry to ease usage
[e-mobility-charging-stations-simulator.git] / tests / ocpp-server / server.py
1 import asyncio
2 import logging
3 import websockets
4 from datetime import datetime, timezone
5
6 from ocpp.routing import on
7 from ocpp.v201 import ChargePoint as cp
8 from ocpp.v201 import call_result
9 from ocpp.v201.enums import RegistrationStatusType, GenericDeviceModelStatusType
10
11 # Setting up the logging configuration to display debug level messages.
12 logging.basicConfig(level=logging.DEBUG)
13
14 # Define a ChargePoint class inheriting from the OCPP 2.0.1 ChargePoint class.
15 class ChargePoint(cp):
16 # Define a handler for the BootNotification message.
17 @on('BootNotification')
18 async def on_boot_notification(self, charging_station, reason, **kwargs):
19 logging.info("Received BootNotification")
20 # Create and return a BootNotification response with the current time,
21 # an interval of 10 seconds, and an accepted status.
22 return call_result.BootNotification(
23 current_time = datetime.now(timezone.utc).isoformat(),
24 interval=10,
25 status=RegistrationStatusType.accepted
26 )
27
28 # Define a handler for the GetBaseReport message.
29 @on('GetBaseReport')
30 async def on_get_base_report(self, request_id, report_base, **kwargs):
31 try:
32 logging.info(f"Received GetBaseReport request with RequestId: {request_id} and ReportBase: {report_base}")
33
34 # Create a mock response for demonstration purposes, indicating the report is accepted.
35 response = call_result.GetBaseReport(
36 status=GenericDeviceModelStatusType.accepted
37 )
38
39 logging.info(f"Sending GetBaseReport response: {response}")
40 return response
41 except Exception as e:
42 # Log any errors that occur while handling the GetBaseReport request.
43 logging.error(f"Error handling GetBaseReport request: {e}", exc_info=True)
44 # Return a rejected status in case of error.
45 return call_result.GetBaseReport(
46 status=GenericDeviceModelStatusType.rejected
47 )
48
49 # Function to handle new WebSocket connections.
50 async def on_connect(websocket, path):
51 """ For every new charge point that connects, create a ChargePoint instance and start
52 listening for messages."""
53 try:
54 requested_protocols = websocket.request_headers['Sec-WebSocket-Protocol']
55 except KeyError:
56 logging.info("Client hasn't requested any Subprotocol. Closing Connection")
57 return await websocket.close()
58
59 if websocket.subprotocol:
60 logging.info("Protocols Matched: %s", websocket.subprotocol)
61 else:
62 logging.warning('Protocols Mismatched | Expected Subprotocols: %s,'
63 ' but client supports %s | Closing connection',
64 websocket.available_subprotocols,
65 requested_protocols)
66 return await websocket.close()
67
68 charge_point_id = path.strip('/')
69 cp = ChargePoint(charge_point_id, websocket)
70
71 # Start the ChargePoint instance to listen for incoming messages.
72 await cp.start()
73
74 # Main function to start the WebSocket server.
75 async def main():
76 # Create the WebSocket server and specify the handler for new connections.
77 server = await websockets.serve(
78 on_connect,
79 '127.0.0.1', # Listen on loopback.
80 9000, # Port number.
81 subprotocols=['ocpp2.0', 'ocpp2.0.1'] # Specify OCPP 2.0.1 subprotocols.
82 )
83 logging.info("WebSocket Server Started")
84 # Wait for the server to close (runs indefinitely).
85 await server.wait_closed()
86
87 # Entry point of the script.
88 if __name__ == '__main__':
89 # Run the main function to start the server.
90 asyncio.run(main())