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 typing import Optional
   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     GenericDeviceModelStatusType,
  16     RegistrationStatusType,
  17     ReportBaseType,
  18     TransactionEventType,
  19 )
  20 from websockets import ConnectionClosed
  21
  22 from timer import Timer
  23
  24 # Setting up the logging configuration to display debug level messages.
  25 logging.basicConfig(level=logging.DEBUG)
  26
  27 ChargePoints = set()
  28
  29
  30 # Define a ChargePoint class inheriting from the OCPP 2.0.1 ChargePoint class.
  31 class ChargePoint(ocpp.v201.ChargePoint):
  32     _command_timer: Optional[Timer]
  33
  34     def __init__(self, connection):
  35         super().__init__(connection.path.strip("/"), connection)
  36         self._command_timer = None
  37
  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     # Request handlers to emit OCPP messages.
 100     async def _send_clear_cache(self):
 101         request = ocpp.v201.call.ClearCache()
 102         response = await self.call(request)
 103
 104         if response.status == ClearCacheStatusType.accepted:
 105             logging.info("%s successful", Action.ClearCache)
 106         else:
 107             logging.info("%s failed", Action.ClearCache)
 108
 109     async def _send_get_base_report(self):
 110         request = ocpp.v201.call.GetBaseReport(
 111             request_id=1, report_base=ReportBaseType.full_inventory
 112         )
 113         response = await self.call(request)
 114
 115         if response.status == GenericDeviceModelStatusType.accepted:
 116             logging.info("%s successful", Action.GetBaseReport)
 117         else:
 118             logging.info("%s failed", Action.GetBaseReport)
 119
 120     async def _send_command(self, command_name: Action):
 121         logging.debug("Sending OCPP command %s", command_name)
 122         match command_name:
 123             case Action.ClearCache:
 124                 await self._send_clear_cache()
 125             case Action.GetBaseReport:
 126                 await self._send_get_base_report()
 127             case _:
 128                 logging.info(f"Not supported command {command_name}")
 129
 130     async def send_command(
 131         self, command_name: Action, delay: Optional[float], period: Optional[float]
 132     ):
 133         try:
 134             if delay and not self._command_timer:
 135                 self._command_timer = Timer(
 136                     delay,
 137                     False,
 138                     self._send_command,
 139                     [command_name],
 140                 )
 141             if period and not self._command_timer:
 142                 self._command_timer = Timer(
 143                     period,
 144                     True,
 145                     self._send_command,
 146                     [command_name],
 147                 )
 148         except ConnectionClosed:
 149             self.handle_connection_closed()
 150
 151     def handle_connection_closed(self):
 152         logging.info("ChargePoint %s closed connection", self.id)
 153         if self._command_timer:
 154             self._command_timer.cancel()
 155         ChargePoints.remove(self)
 156         logging.debug("Connected ChargePoint(s): %d", len(ChargePoints))
 157
 158
 159 # Function to handle new WebSocket connections.
 160 async def on_connect(
 161     websocket,
 162     command_name: Optional[Action],
 163     delay: Optional[float],
 164     period: Optional[float],
 165 ):
 166     """For every new charge point that connects, create a ChargePoint instance and start
 167     listening for messages.
 168     """
 169     try:
 170         requested_protocols = websocket.request_headers["Sec-WebSocket-Protocol"]
 171     except KeyError:
 172         logging.info("Client hasn't requested any Subprotocol. Closing Connection")
 173         return await websocket.close()
 174
 175     if websocket.subprotocol:
 176         logging.info("Protocols Matched: %s", websocket.subprotocol)
 177     else:
 178         logging.warning(
 179             "Protocols Mismatched | Expected Subprotocols: %s,"
 180             " but client supports %s | Closing connection",
 181             websocket.available_subprotocols,
 182             requested_protocols,
 183         )
 184         return await websocket.close()
 185
 186     cp = ChargePoint(websocket)
 187     if command_name:
 188         await cp.send_command(command_name, delay, period)
 189
 190     ChargePoints.add(cp)
 191
 192     try:
 193         await cp.start()
 194     except ConnectionClosed:
 195         cp.handle_connection_closed()
 196
 197
 198 def check_positive_number(value: Optional[float]):
 199     try:
 200         value = float(value)
 201     except ValueError:
 202         raise argparse.ArgumentTypeError("must be a number") from None
 203     if value <= 0:
 204         raise argparse.ArgumentTypeError("must be a positive number")
 205     return value
 206
 207
 208 # Main function to start the WebSocket server.
 209 async def main():
 210     parser = argparse.ArgumentParser(description="OCPP2 Server")
 211     parser.add_argument("-c", "--command", type=Action, help="command name")
 212     group = parser.add_mutually_exclusive_group()
 213     group.add_argument(
 214         "-d",
 215         "--delay",
 216         type=check_positive_number,
 217         help="delay in seconds",
 218     )
 219     group.add_argument(
 220         "-p",
 221         "--period",
 222         type=check_positive_number,
 223         help="period in seconds",
 224     )
 225     group.required = parser.parse_known_args()[0].command is not None
 226
 227     args = parser.parse_args()
 228
 229     # Create the WebSocket server and specify the handler for new connections.
 230     server = await websockets.serve(
 231         partial(
 232             on_connect, command_name=args.command, delay=args.delay, period=args.period
 233         ),
 234         "127.0.0.1",  # Listen on loopback.
 235         9000,  # Port number.
 236         subprotocols=["ocpp2.0", "ocpp2.0.1"],  # Specify OCPP 2.0.1 subprotocols.
 237     )
 238     logging.info("WebSocket Server Started")
 239
 240     # Wait for the server to close (runs indefinitely).
 241     await server.wait_closed()
 242
 243
 244 # Entry point of the script.
 245 if __name__ == "__main__":
 246     # Run the main function to start the server.
 247     asyncio.run(main())