3 from datetime
import datetime
, timezone
4 from threading
import Timer
8 from ocpp
.routing
import on
9 from ocpp
.v201
.enums
import (
11 AuthorizationStatusType
,
13 RegistrationStatusType
,
16 from websockets
import ConnectionClosed
18 # Setting up the logging configuration to display debug level messages.
19 logging
.basicConfig(level
=logging
.DEBUG
)
24 class RepeatTimer(Timer
):
25 """Class that inherits from the Timer class. It will run a
26 function at regular intervals."""
29 while not self
.finished
.wait(self
.interval
):
30 self
.function(*self
.args
, **self
.kwargs
)
33 # Define a ChargePoint class inheriting from the OCPP 2.0.1 ChargePoint class.
34 class ChargePoint(ocpp
.v201
.ChargePoint
):
35 def __init__(self
, charge_point_id
, connection
):
36 super().__init
__(self
, charge_point_id
, connection
)
37 self
._ws
_ping
_timer
= RepeatTimer(60, self
.web_socket_ping
)
38 self
._ws
_ping
_timer
.start()
41 self
._ws
_ping
_timer
.cancel()
43 def web_socket_ping(self
):
45 self
._connection
.ping()
46 except ConnectionClosed
:
47 ChargePoints
.remove(self
)
50 # Message handlers to receive OCPP messages.
51 @on(Action
.BootNotification
)
52 async def on_boot_notification(self
, charging_station
, reason
, **kwargs
):
53 logging
.info("Received BootNotification")
54 # Create and return a BootNotification response with the current time,
55 # an interval of 60 seconds, and an accepted status.
56 return ocpp
.v201
.call_result
.BootNotification(
57 current_time
=datetime
.now(timezone
.utc
).isoformat(),
59 status
=RegistrationStatusType
.accepted
,
63 async def on_heartbeat(self
, **kwargs
):
64 logging
.info("Received Heartbeat")
65 return ocpp
.v201
.call_result
.Heartbeat(
66 current_time
=datetime
.now(timezone
.utc
).isoformat()
69 @on(Action
.StatusNotification
)
70 async def on_status_notification(
71 self
, timestamp
, evse_id
: int, connector_id
: int, connector_status
, **kwargs
73 logging
.info("Received StatusNotification")
74 return ocpp
.v201
.call_result
.StatusNotification()
77 async def on_authorize(self
, id_token
, **kwargs
):
78 logging
.info("Received Authorize")
79 return ocpp
.v201
.call_result
.Authorize(
80 id_token_info
={"status": AuthorizationStatusType
.accepted
}
83 @on(Action
.TransactionEvent
)
84 async def on_transaction_event(
86 event_type
: TransactionEventType
,
94 case TransactionEventType
.started
:
95 logging
.info("Received TransactionEvent Started")
96 return ocpp
.v201
.call_result
.TransactionEvent(
97 id_token_info
={"status": AuthorizationStatusType
.accepted
}
99 case TransactionEventType
.updated
:
100 logging
.info("Received TransactionEvent Updated")
101 return ocpp
.v201
.call_result
.TransactionEvent(total_cost
=10)
102 case TransactionEventType
.ended
:
103 logging
.info("Received TransactionEvent Ended")
104 return ocpp
.v201
.call_result
.TransactionEvent()
106 @on(Action
.MeterValues
)
107 async def on_meter_values(self
, evse_id
: int, meter_value
, **kwargs
):
108 logging
.info("Received MeterValues")
109 return ocpp
.v201
.call_result
.MeterValues()
111 # Request handlers to emit OCPP messages.
112 async def send_clear_cache(self
):
113 request
= ocpp
.v201
.call
.ClearCache()
114 response
= await self
.call(request
)
116 if response
.status
== ClearCacheStatusType
.accepted
:
117 logging
.info("Cache clearing successful")
119 logging
.info("Cache clearing failed")
122 # Function to handle new WebSocket connections.
123 async def on_connect(websocket
, path
):
124 """For every new charge point that connects, create a ChargePoint instance and start
125 listening for messages."""
127 requested_protocols
= websocket
.request_headers
["Sec-WebSocket-Protocol"]
129 logging
.info("Client hasn't requested any Subprotocol. Closing Connection")
130 return await websocket
.close()
132 if websocket
.subprotocol
:
133 logging
.info("Protocols Matched: %s", websocket
.subprotocol
)
136 "Protocols Mismatched | Expected Subprotocols: %s,"
137 " but client supports %s | Closing connection",
138 websocket
.available_subprotocols
,
141 return await websocket
.close()
143 charge_point_id
= path
.strip("/")
144 cp
= ChargePoint(charge_point_id
, websocket
)
147 # Start the ChargePoint instance to listen for incoming messages.
151 # Main function to start the WebSocket server.
153 # Create the WebSocket server and specify the handler for new connections.
154 server
= await websockets
.serve(
156 "127.0.0.1", # Listen on loopback.
158 subprotocols
=["ocpp2.0", "ocpp2.0.1"], # Specify OCPP 2.0.1 subprotocols.
160 logging
.info("WebSocket Server Started")
161 # Wait for the server to close (runs indefinitely).
162 await server
.wait_closed()
165 # Entry point of the script.
166 if __name__
== "__main__":
167 # Run the main function to start the server.