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 GetBaseReport")
 103         return ocpp.v201.call_result.GetBaseReport(status="Accepted")
 104
 105     # Request handlers to emit OCPP messages.
 106     async def send_clear_cache(self):
 107         request = ocpp.v201.call.ClearCache()
 108         response = await self.call(request)
 109
 110         if response.status == ClearCacheStatusType.accepted:
 111             logging.info("%s successful", Action.ClearCache)
 112         else:
 113             logging.info("%s failed", Action.ClearCache)
 114
 115     async def send_get_base_report(self):
 116         logging.info("Executing send_get_base_report...")
 117         request = ocpp.v201.call.GetBaseReport(
 118             reportBase=ReportBaseType.ConfigurationInventory
 119         )  # Use correct ReportBaseType
 120         try:
 121             response = await self.call(request)
 122             logging.info("Send GetBaseReport")
 123
 124             if (
 125                 response.status == "Accepted"
 126             ):  # Adjust depending on the structure of your response
 127                 logging.info("Send GetBaseReport successful")
 128             else:
 129                 logging.info("Send GetBaseReport failed")
 130         except Exception as e:
 131             logging.error(f"Send GetBaseReport failed: {str(e)}")
 132         logging.info("send_get_base_report done.")
 133
 134
 135 # Define argument parser
 136 parser = argparse.ArgumentParser(description="OCPP Charge Point Simulator")
 137 parser.add_argument("--request", type=str, help="OCPP 2 Command Name")
 138 parser.add_argument("--delay", type=int, help="Delay in seconds")
 139 parser.add_argument("--period", type=int, help="Period in seconds")
 140
 141 args = parser.parse_args()
 142
 143
 144 # Function to send OCPP command
 145 async def send_ocpp_command(cp, command_name, delay=None, period=None):
 146     # If delay is not None, sleep for delay seconds
 147     if delay:
 148         await asyncio.sleep(delay)
 149
 150     # If period is not None, send command repeatedly with period interval
 151     if period:
 152         while True:
 153             if command_name == "GetBaseReport":
 154                 logging.info("GetBaseReport parser working")
 155                 await cp.send_get_base_report()
 156
 157             await asyncio.sleep(period)
 158     else:
 159         if command_name == "GetBaseReport":
 160             await cp.send_get_base_report()
 161
 162
 163 # Function to handle new WebSocket connections.
 164 async def on_connect(websocket, path):
 165     """For every new charge point that connects, create a ChargePoint instance and start
 166     listening for messages."""
 167     try:
 168         requested_protocols = websocket.request_headers["Sec-WebSocket-Protocol"]
 169     except KeyError:
 170         logging.info("Client hasn't requested any Subprotocol. Closing Connection")
 171         return await websocket.close()
 172
 173     if websocket.subprotocol:
 174         logging.info("Protocols Matched: %s", websocket.subprotocol)
 175     else:
 176         logging.warning(
 177             "Protocols Mismatched | Expected Subprotocols: %s,"
 178             " but client supports %s | Closing connection",
 179             websocket.available_subprotocols,
 180             requested_protocols,
 181         )
 182         return await websocket.close()
 183
 184     charge_point_id = path.strip("/")
 185     cp = ChargePoint(charge_point_id, websocket)
 186
 187     # Check if request argument is specified
 188     if args.request:
 189         asyncio.create_task(
 190             send_ocpp_command(cp, args.request, args.delay, args.period)
 191         )
 192
 193     # Start the ChargePoint instance to listen for incoming messages.
 194     await cp.start()
 195
 196     ChargePoints.add(cp)
 197     try:
 198         await cp.start()
 199     except ConnectionClosed:
 200         logging.info("ChargePoint %s closed connection", cp.id)
 201         ChargePoints.remove(cp)
 202         logging.debug("Connected ChargePoint(s): %d", len(ChargePoints))
 203
 204
 205 # Main function to start the WebSocket server.
 206 async def main():
 207     # Create the WebSocket server and specify the handler for new connections.
 208     server = await websockets.serve(
 209         on_connect,
 210         "127.0.0.1",  # Listen on loopback.
 211         9000,  # Port number.
 212         subprotocols=["ocpp2.0", "ocpp2.0.1"],  # Specify OCPP 2.0.1 subprotocols.
 213     )
 214     logging.info("WebSocket Server Started")
 215
 216     # Wait for the server to close (runs indefinitely).
 217     await server.wait_closed()
 218
 219
 220 # Entry point of the script.
 221 if __name__ == "__main__":
 222     # Run the main function to start the server.
 223     asyncio.run(main())