+ async def _send_get_base_report(self):
+ request = ocpp.v201.call.GetBaseReport(
+ request_id=1, report_base=ReportBaseType.full_inventory
+ )
+ response = await self.call(request)
+
+ if response.status == GenericDeviceModelStatusType.accepted:
+ logging.info("%s successful", Action.GetBaseReport)
+ else:
+ logging.info("%s failed", Action.GetBaseReport)
+
+ async def _send_command(self, command_name: Action):
+ logging.debug("Sending OCPP command %s", command_name)
+ match command_name:
+ case Action.ClearCache:
+ await self._send_clear_cache()
+ case Action.GetBaseReport:
+ await self._send_get_base_report()
+ case _:
+ logging.info(f"Not supported command {command_name}")
+
+ async def send_command(
+ self, command_name: Action, delay: Optional[float], period: Optional[float]
+ ):
+ try:
+ if delay and not self._command_timer:
+ self._command_timer = Timer(
+ delay,
+ False,
+ self._send_command,
+ [command_name],
+ )
+ if period and not self._command_timer:
+ self._command_timer = Timer(
+ period,
+ True,
+ self._send_command,
+ [command_name],
+ )
+ except ConnectionClosed:
+ self.handle_connection_closed()
+
+ def handle_connection_closed(self):
+ logging.info("ChargePoint %s closed connection", self.id)
+ if self._command_timer:
+ self._command_timer.cancel()
+ ChargePoints.remove(self)
+ logging.debug("Connected ChargePoint(s): %d", len(ChargePoints))
+