4 from datetime
import datetime
, timezone
5 from functools
import partial
6 from typing
import Optional
10 from ocpp
.routing
import on
11 from ocpp
.v201
.enums
import (
13 AuthorizationStatusType
,
15 GenericDeviceModelStatusType
,
16 RegistrationStatusType
,
20 from websockets
import ConnectionClosed
22 from timer
import Timer
24 # Setting up the logging configuration to display debug level messages.
25 logging
.basicConfig(level
=logging
.DEBUG
)
30 # Define a ChargePoint class inheriting from the OCPP 2.0.1 ChargePoint class.
31 class ChargePoint(ocpp
.v201
.ChargePoint
):
32 _command_timer
: Optional
[Timer
]
34 def __init__(self
, connection
):
35 super().__init
__(connection
.path
.strip("/"), connection
)
36 self
._command
_timer
= None
38 # Message handlers to receive OCPP messages.
39 @on(Action
.BootNotification
)
40 async def on_boot_notification(self
, charging_station
, reason
, **kwargs
):
41 logging
.info("Received %s", Action
.BootNotification
)
42 # Create and return a BootNotification response with the current time,
43 # an interval of 60 seconds, and an accepted status.
44 return ocpp
.v201
.call_result
.BootNotification(
45 current_time
=datetime
.now(timezone
.utc
).isoformat(),
47 status
=RegistrationStatusType
.accepted
,
51 async def on_heartbeat(self
, **kwargs
):
52 logging
.info("Received %s", Action
.Heartbeat
)
53 return ocpp
.v201
.call_result
.Heartbeat(
54 current_time
=datetime
.now(timezone
.utc
).isoformat()
57 @on(Action
.StatusNotification
)
58 async def on_status_notification(
59 self
, timestamp
, evse_id
: int, connector_id
: int, connector_status
, **kwargs
61 logging
.info("Received %s", Action
.StatusNotification
)
62 return ocpp
.v201
.call_result
.StatusNotification()
65 async def on_authorize(self
, id_token
, **kwargs
):
66 logging
.info("Received %s", Action
.Authorize
)
67 return ocpp
.v201
.call_result
.Authorize(
68 id_token_info
={"status": AuthorizationStatusType
.accepted
}
71 @on(Action
.TransactionEvent
)
72 async def on_transaction_event(
74 event_type
: TransactionEventType
,
82 case TransactionEventType
.started
:
83 logging
.info("Received %s Started", Action
.TransactionEvent
)
84 return ocpp
.v201
.call_result
.TransactionEvent(
85 id_token_info
={"status": AuthorizationStatusType
.accepted
}
87 case TransactionEventType
.updated
:
88 logging
.info("Received %s Updated", Action
.TransactionEvent
)
89 return ocpp
.v201
.call_result
.TransactionEvent(total_cost
=10)
90 case TransactionEventType
.ended
:
91 logging
.info("Received %s Ended", Action
.TransactionEvent
)
92 return ocpp
.v201
.call_result
.TransactionEvent()
94 @on(Action
.MeterValues
)
95 async def on_meter_values(self
, evse_id
: int, meter_value
, **kwargs
):
96 logging
.info("Received %s", Action
.MeterValues
)
97 return ocpp
.v201
.call_result
.MeterValues()
99 # Request handlers to emit OCPP messages.
100 async def _send_clear_cache(self
):
101 request
= ocpp
.v201
.call
.ClearCache()
102 response
= await self
.call(request
)
104 if response
.status
== ClearCacheStatusType
.accepted
:
105 logging
.info("%s successful", Action
.ClearCache
)
107 logging
.info("%s failed", Action
.ClearCache
)
109 async def _send_get_base_report(self
):
110 request
= ocpp
.v201
.call
.GetBaseReport(
111 request_id
=1, report_base
=ReportBaseType
.full_inventory
113 response
= await self
.call(request
)
115 if response
.status
== GenericDeviceModelStatusType
.accepted
:
116 logging
.info("%s successful", Action
.GetBaseReport
)
118 logging
.info("%s failed", Action
.GetBaseReport
)
120 async def _send_command(self
, command_name
: Action
):
121 logging
.debug("Sending OCPP command %s", command_name
)
123 case Action
.ClearCache
:
124 await self
._send
_clear
_cache
()
125 case Action
.GetBaseReport
:
126 await self
._send
_get
_base
_report
()
128 logging
.info(f
"Not supported command {command_name}")
130 async def send_command(
131 self
, command_name
: Action
, delay
: Optional
[float], period
: Optional
[float]
133 if not delay
and not period
:
134 raise ValueError("Either delay or period must be defined")
135 if delay
and delay
<= 0:
136 raise ValueError("Delay must be a positive number")
137 if period
and period
<= 0:
138 raise ValueError("Period must be a positive number")
140 if delay
and not self
._command
_timer
:
141 self
._command
_timer
= Timer(
147 if period
and not self
._command
_timer
:
148 self
._command
_timer
= Timer(
154 except ConnectionClosed
:
155 self
.handle_connection_closed()
157 def handle_connection_closed(self
):
158 logging
.info("ChargePoint %s closed connection", self
.id)
159 if self
._command
_timer
:
160 self
._command
_timer
.cancel()
161 ChargePoints
.remove(self
)
162 logging
.debug("Connected ChargePoint(s): %d", len(ChargePoints
))
165 # Function to handle new WebSocket connections.
166 async def on_connect(
168 command_name
: Optional
[Action
],
169 delay
: Optional
[float],
170 period
: Optional
[float],
172 """For every new charge point that connects, create a ChargePoint instance and start
173 listening for messages.
176 requested_protocols
= websocket
.request_headers
["Sec-WebSocket-Protocol"]
178 logging
.info("Client hasn't requested any Subprotocol. Closing Connection")
179 return await websocket
.close()
181 if websocket
.subprotocol
:
182 logging
.info("Protocols Matched: %s", websocket
.subprotocol
)
185 "Protocols Mismatched | Expected Subprotocols: %s,"
186 " but client supports %s | Closing connection",
187 websocket
.available_subprotocols
,
190 return await websocket
.close()
192 cp
= ChargePoint(websocket
)
194 await cp
.send_command(command_name
, delay
, period
)
200 except ConnectionClosed
:
201 cp
.handle_connection_closed()
204 # Main function to start the WebSocket server.
206 parser
= argparse
.ArgumentParser(description
="OCPP2 Server")
207 parser
.add_argument("-c", "--command", type=Action
, help="OCPP2 Command Name")
208 parser
.add_argument("-d", "--delay", type=float, help="Delay in seconds")
209 parser
.add_argument("-p", "--period", type=float, help="Period in seconds")
211 args
= parser
.parse_args()
213 # Create the WebSocket server and specify the handler for new connections.
214 server
= await websockets
.serve(
216 on_connect
, command_name
=args
.command
, delay
=args
.delay
, period
=args
.period
218 "127.0.0.1", # Listen on loopback.
220 subprotocols
=["ocpp2.0", "ocpp2.0.1"], # Specify OCPP 2.0.1 subprotocols.
222 logging
.info("WebSocket Server Started")
224 # Wait for the server to close (runs indefinitely).
225 await server
.wait_closed()
228 # Entry point of the script.
229 if __name__
== "__main__":
230 # Run the main function to start the server.