tests/ocpp-server/server.py

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