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]
134 if delay
and not self
._command
_timer
:
135 self
._command
_timer
= Timer(
141 if period
and not self
._command
_timer
:
142 self
._command
_timer
= Timer(
148 except ConnectionClosed
:
149 self
.handle_connection_closed()
151 def handle_connection_closed(self
):
152 logging
.info("ChargePoint %s closed connection", self
.id)
153 if self
._command
_timer
:
154 self
._command
_timer
.cancel()
155 ChargePoints
.remove(self
)
156 logging
.debug("Connected ChargePoint(s): %d", len(ChargePoints
))
159 # Function to handle new WebSocket connections.
160 async def on_connect(
162 command_name
: Optional
[Action
],
163 delay
: Optional
[float],
164 period
: Optional
[float],
166 """For every new charge point that connects, create a ChargePoint instance and start
167 listening for messages.
170 requested_protocols
= websocket
.request_headers
["Sec-WebSocket-Protocol"]
172 logging
.info("Client hasn't requested any Subprotocol. Closing Connection")
173 return await websocket
.close()
175 if websocket
.subprotocol
:
176 logging
.info("Protocols Matched: %s", websocket
.subprotocol
)
179 "Protocols Mismatched | Expected Subprotocols: %s,"
180 " but client supports %s | Closing connection",
181 websocket
.available_subprotocols
,
184 return await websocket
.close()
186 cp
= ChargePoint(websocket
)
188 await cp
.send_command(command_name
, delay
, period
)
194 except ConnectionClosed
:
195 cp
.handle_connection_closed()
198 def check_positive_number(value
: Optional
[float]):
202 raise argparse
.ArgumentTypeError("must be a number") from None
204 raise argparse
.ArgumentTypeError("must be a positive number")
208 # Main function to start the WebSocket server.
210 parser
= argparse
.ArgumentParser(description
="OCPP2 Server")
211 parser
.add_argument("-c", "--command", type=Action
, help="command name")
212 group
= parser
.add_mutually_exclusive_group()
216 type=check_positive_number
,
217 help="delay in seconds",
222 type=check_positive_number
,
223 help="period in seconds",
225 group
.required
= parser
.parse_known_args()[0].command
is not None
227 args
= parser
.parse_args()
229 # Create the WebSocket server and specify the handler for new connections.
230 server
= await websockets
.serve(
232 on_connect
, command_name
=args
.command
, delay
=args
.delay
, period
=args
.period
234 "127.0.0.1", # Listen on loopback.
236 subprotocols
=["ocpp2.0", "ocpp2.0.1"], # Specify OCPP 2.0.1 subprotocols.
238 logging
.info("WebSocket Server Started")
240 # Wait for the server to close (runs indefinitely).
241 await server
.wait_closed()
244 # Entry point of the script.
245 if __name__
== "__main__":
246 # Run the main function to start the server.