tests/ocpp-server/server.py

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