tests/ocpp-server/server.py

   1 import argparse
   2 import asyncio
   3 import logging
   4 from datetime import datetime, timezone
   5 from threading import Timer
   6
   7 import ocpp.v201
   8 import websockets
   9 from ocpp.routing import on
  10 from ocpp.v201.enums import (
  11     Action,
  12     AuthorizationStatusType,
  13     ClearCacheStatusType,
  14     RegistrationStatusType,
  15     ReportBaseType,
  16     TransactionEventType,
  17 )
  18 from websockets import ConnectionClosed
  19
  20 # Setting up the logging configuration to display debug level messages.
  21 logging.basicConfig(level=logging.DEBUG)
  22
  23 ChargePoints = set()
  24
  25
  26 class RepeatTimer(Timer):
  27     """Class that inherits from the Timer class. It will run a
  28     function at regular intervals."""
  29
  30     def run(self):
  31         while not self.finished.wait(self.interval):
  32             self.function(*self.args, **self.kwargs)
  33
  34
  35 # Define a ChargePoint class inheriting from the OCPP 2.0.1 ChargePoint class.
  36 class ChargePoint(ocpp.v201.ChargePoint):
  37     # Message handlers to receive OCPP messages.
  38     @on(Action.BootNotification)
  39     async def on_boot_notification(self, charging_station, reason, **kwargs):
  40         logging.info("Received %s", Action.BootNotification)
  41         # Create and return a BootNotification response with the current time,
  42         # an interval of 60 seconds, and an accepted status.
  43         return ocpp.v201.call_result.BootNotification(
  44             current_time=datetime.now(timezone.utc).isoformat(),
  45             interval=60,
  46             status=RegistrationStatusType.accepted,
  47         )
  48
  49     @on(Action.Heartbeat)
  50     async def on_heartbeat(self, **kwargs):
  51         logging.info("Received %s", Action.Heartbeat)
  52         return ocpp.v201.call_result.Heartbeat(
  53             current_time=datetime.now(timezone.utc).isoformat()
  54         )
  55
  56     @on(Action.StatusNotification)
  57     async def on_status_notification(
  58         self, timestamp, evse_id: int, connector_id: int, connector_status, **kwargs
  59     ):
  60         logging.info("Received %s", Action.StatusNotification)
  61         return ocpp.v201.call_result.StatusNotification()
  62
  63     @on(Action.Authorize)
  64     async def on_authorize(self, id_token, **kwargs):
  65         logging.info("Received %s", Action.Authorize)
  66         return ocpp.v201.call_result.Authorize(
  67             id_token_info={"status": AuthorizationStatusType.accepted}
  68         )
  69
  70     @on(Action.TransactionEvent)
  71     async def on_transaction_event(
  72         self,
  73         event_type: TransactionEventType,
  74         timestamp,
  75         trigger_reason,
  76         seq_no: int,
  77         transaction_info,
  78         **kwargs,
  79     ):
  80         match event_type:
  81             case TransactionEventType.started:
  82                 logging.info("Received %s Started", Action.TransactionEvent)
  83                 return ocpp.v201.call_result.TransactionEvent(
  84                     id_token_info={"status": AuthorizationStatusType.accepted}
  85                 )
  86             case TransactionEventType.updated:
  87                 logging.info("Received %s Updated", Action.TransactionEvent)
  88                 return ocpp.v201.call_result.TransactionEvent(total_cost=10)
  89             case TransactionEventType.ended:
  90                 logging.info("Received %s Ended", Action.TransactionEvent)
  91                 return ocpp.v201.call_result.TransactionEvent()
  92
  93     @on(Action.MeterValues)
  94     async def on_meter_values(self, evse_id: int, meter_value, **kwargs):
  95         logging.info("Received %s", Action.MeterValues)
  96         return ocpp.v201.call_result.MeterValues()
  97
  98     @on(Action.GetBaseReport)
  99     async def on_get_base_report(
 100         self, request_id: int, report_base: ReportBaseType, **kwargs
 101     ):
 102         logging.info("Received %s", Action.GetBaseReport)
 103         return ocpp.v201.call_result.GetBaseReport(
 104             id_token_info={"status": ReportBaseType.accepted}
 105         )
 106
 107     # Request handlers to emit OCPP messages.
 108     async def send_clear_cache(self):
 109         request = ocpp.v201.call.ClearCache()
 110         response = await self.call(request)
 111
 112         if response.status == ClearCacheStatusType.accepted:
 113             logging.info("%s successful", Action.ClearCache)
 114         else:
 115             logging.info("%s failed", Action.ClearCache)
 116
 117     async def send_get_base_report(self):
 118         request = ocpp.v201.call.GetBaseReport(
 119             reportBase=ReportBaseType.ConfigurationInventory
 120         )
 121         response = await self.call(request)
 122
 123         if response.status == ReportBaseType.accepted:
 124             logging.info("%s successful", Action.GetBaseReport)
 125         else:
 126             logging.info("%s failed", Action.GetBaseReport)
 127
 128
 129 # Function to send OCPP command
 130 async def send_ocpp_command(cp, command_name, delay=None, period=None):
 131     try:
 132         match command_name:
 133             case Action.ClearCache:
 134                 logging.info("%s Send:", Action.ClearCache)
 135                 await cp.send_clear_cache()
 136             case Action.GetBaseReport:
 137                 logging.info("%s Send:", Action.GetBaseReport)
 138                 await cp.send_get_base_report()
 139     except Exception:
 140         logging.exception(
 141             f"Not supported or Failure while processing command {command_name}"
 142         )
 143
 144     if delay:
 145         await asyncio.sleep(delay)
 146
 147     if period:
 148         my_timer = RepeatTimer(
 149             period, asyncio.create_task, [cp.send_ocpp_command(command_name)]
 150         )
 151         my_timer.start()
 152
 153
 154 # Function to handle new WebSocket connections.
 155 async def on_connect(websocket, path):
 156     """For every new charge point that connects, create a ChargePoint instance and start
 157     listening for messages."""
 158     try:
 159         requested_protocols = websocket.request_headers["Sec-WebSocket-Protocol"]
 160     except KeyError:
 161         logging.info("Client hasn't requested any Subprotocol. Closing Connection")
 162         return await websocket.close()
 163
 164     if websocket.subprotocol:
 165         logging.info("Protocols Matched: %s", websocket.subprotocol)
 166     else:
 167         logging.warning(
 168             "Protocols Mismatched | Expected Subprotocols: %s,"
 169             " but client supports %s | Closing connection",
 170             websocket.available_subprotocols,
 171             requested_protocols,
 172         )
 173         return await websocket.close()
 174
 175     charge_point_id = path.strip("/")
 176     cp = ChargePoint(charge_point_id, websocket)
 177
 178     ChargePoints.add(cp)
 179     try:
 180         await cp.start()
 181
 182     except ConnectionClosed:
 183         logging.info("ChargePoint %s closed connection", cp.id)
 184         ChargePoints.remove(cp)
 185         logging.debug("Connected ChargePoint(s): %d", len(ChargePoints))
 186
 187
 188 # Main function to start the WebSocket server.
 189 async def main():
 190     # Define argument parser
 191     parser = argparse.ArgumentParser(description="OCPP2 Charge Point Simulator")
 192     parser.add_argument("--command", type=str, help="OCPP2 Command Name")
 193     parser.add_argument("--delay", type=int, help="Delay in seconds")
 194     parser.add_argument("--period", type=int, help="Period in seconds")
 195
 196     # Create the WebSocket server and specify the handler for new connections.
 197     server = await websockets.serve(
 198         on_connect,
 199         "127.0.0.1",  # Listen on loopback.
 200         9000,  # Port number.
 201         subprotocols=["ocpp2.0", "ocpp2.0.1"],  # Specify OCPP 2.0.1 subprotocols.
 202     )
 203     logging.info("WebSocket Server Started")
 204
 205     args = parser.parse_args()
 206
 207     if args.command:
 208         for cp in ChargePoints:
 209             asyncio.create_task(
 210                 send_ocpp_command(cp, args.command, args.delay, args.period)
 211             )
 212
 213     # Wait for the server to close (runs indefinitely).
 214     await server.wait_closed()
 215
 216
 217 # Entry point of the script.
 218 if __name__ == "__main__":
 219     # Run the main function to start the server.
 220     asyncio.run(main())