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         )  # Use correct ReportBaseType
 122         try:
 123             response = await self.call(request)
 124             logging.info("Send %s", Action.GetBaseReport)
 125
 126             if response.status == ReportBaseType.accepted:
 127                 logging.info("%s successful", Action.GetBaseReport)
 128             else:
 129                 logging.info("%s failed", Action.GetBaseReport)
 130
 131         except ConnectionClosed as e:
 132             logging.error("Connection closed: %s", str(e))
 133
 134         except Exception as e:
 135             logging.error("Unexpected error occurred: %s", str(e))
 136
 137         logging.info("send_get_base_report done.")
 138
 139
 140 # Function to send OCPP command
 141 async def send_ocpp_command(cp, command_name, delay=None, period=None):
 142     # If delay is not None, sleep for delay seconds
 143     if delay:
 144         await asyncio.sleep(delay)
 145
 146     # If period is not None, send command repeatedly with period interval
 147     if period:
 148
 149         async def send_command_repeatedly():
 150             while True:
 151                 command_name = await cp.receive_command()
 152                 try:
 153                     match command_name:
 154                         case Action.ClearCache:
 155                             logging.info("ClearCache parser working")
 156                             await cp.send_clear_cache()
 157                         case Action.GetBaseReport:
 158                             logging.info("GetBaseReport parser working")
 159                             await cp.send_get_base_report()
 160                         case _:
 161                             logging.warning(f"Unsupported command {command_name}")
 162                 except Exception:
 163                     logging.exception(
 164                         f"Failure while processing command {command_name}"
 165                     )
 166                 finally:
 167                     await asyncio.sleep(period)
 168
 169         timer = RepeatTimer(period, send_command_repeatedly)
 170         await timer.start()
 171         await timer.wait_closed()  # Wait for timer to finish before exiting
 172
 173     else:
 174         if command_name == "GetBaseReport":
 175             await cp.send_get_base_report()
 176
 177
 178 # Function to handle new WebSocket connections.
 179 async def on_connect(websocket, path, args):
 180     """For every new charge point that connects, create a ChargePoint instance and start
 181     listening for messages."""
 182     try:
 183         requested_protocols = websocket.request_headers["Sec-WebSocket-Protocol"]
 184     except KeyError:
 185         logging.info("Client hasn't requested any Subprotocol. Closing Connection")
 186         return await websocket.close()
 187
 188     if websocket.subprotocol:
 189         logging.info("Protocols Matched: %s", websocket.subprotocol)
 190     else:
 191         logging.warning(
 192             "Protocols Mismatched | Expected Subprotocols: %s,"
 193             " but client supports %s | Closing connection",
 194             websocket.available_subprotocols,
 195             requested_protocols,
 196         )
 197         return await websocket.close()
 198
 199     charge_point_id = path.strip("/")
 200     cp = ChargePoint(charge_point_id, websocket)
 201
 202     ChargePoints.add(cp)
 203     try:
 204         await cp.start()
 205         # Check if request argument is specified
 206         if args.request:
 207             asyncio.create_task(
 208                 send_ocpp_command(cp, args.request, args.delay, args.period)
 209             )
 210
 211     except ConnectionClosed:
 212         logging.info("ChargePoint %s closed connection", cp.id)
 213         ChargePoints.remove(cp)
 214         logging.debug("Connected ChargePoint(s): %d", len(ChargePoints))
 215
 216
 217 # Main function to start the WebSocket server.
 218 async def main():
 219     # Define argument parser
 220     parser = argparse.ArgumentParser(description="OCPP Charge Point Simulator")
 221     parser.add_argument("--request", type=str, help="OCPP 2 Command Name")
 222     parser.add_argument("--delay", type=int, help="Delay in seconds")
 223     parser.add_argument("--period", type=int, help="Period in seconds")
 224
 225     args = parser.parse_args()
 226
 227     on_connect_bound = partial(on_connect, args=args)  # Add args to on_connect
 228
 229     # Create the WebSocket server and specify the handler for new connections.
 230     server = await websockets.serve(
 231         on_connect_bound,
 232         "127.0.0.1",  # Listen on loopback.
 233         9000,  # Port number.
 234         subprotocols=["ocpp2.0", "ocpp2.0.1"],  # Specify OCPP 2.0.1 subprotocols.
 235     )
 236     logging.info("WebSocket Server Started")
 237
 238     # Wait for the server to close (runs indefinitely).
 239     await server.wait_closed()
 240
 241
 242 # Entry point of the script.
 243 if __name__ == "__main__":
 244     # Run the main function to start the server.
 245     asyncio.run(main())