Seeking Best Practices

[hhuseyinpay]

Hello, I'm new to Python and the OCPP library. I'm currently working on a project where I'm using the OCPP library to manage charge points. I've written some code(inspired by #263), but I'm not sure if I'm following the best practices for Python and OCPP.
Here's a brief overview of my project:

• I'm using FastAPI to create a WebSocket endpoint for each charge point.
• I've created a CentralSystem class that manages the charge points. Each charge point is an instance of the ChargePointV16 class.
• I'm doing database queries in async action functions

Here's a snippet of my code:

import logging
import asyncio
from asyncio import Task
from typing import Optional, Dict

from ocpp.routing import on
from ocpp.v16 import call, call_result, ChargePoint as CPv16
from ocpp.v16.datatypes import IdTagInfo
from ocpp.v16.enums import Action, RegistrationStatus, AuthorizationStatus, ChargePointErrorCode, ChargePointStatus, \
    RemoteStartStopStatus

from fastapi import FastAPI, WebSocket, HTTPException
from starlette import status

from socket_adapter import SocketAdapter
from utils import get_current_time

logging.basicConfig(level=logging.DEBUG)

app = FastAPI()


class ChargePointV16(CPv16):
    @on(Action.BootNotification)
    async def on_boot_notification(self, charge_point_vendor: str, charge_point_model: str, **kwargs):
        # database queries etc.
        await asyncio.sleep(1)

        return call_result.BootNotificationPayload(
            current_time=get_current_time(),
            interval=10,
            status=RegistrationStatus.accepted,
        )

    @on(Action.StatusNotification)
    async def on_status_notification(self, connector_id: int,
                                     error_code: ChargePointErrorCode,
                                     status: ChargePointStatus,
                                     timestamp: Optional[str] = None,
                                     info: Optional[str] = None,
                                     vendor_id: Optional[str] = None,
                                     vendor_error_code: Optional[str] = None,
                                     ):
        # database queries etc.
        await asyncio.sleep(0.1)  # simulate db query

        return call_result.StatusNotificationPayload()

    @on(Action.Heartbeat)
    def on_heartbeat(self):
        return call_result.HeartbeatPayload(get_current_time())

    @on(Action.StartTransaction)
    async def start_transaction(self, connector_id: int,
                                id_tag: str,
                                meter_start: int,
                                timestamp: str):
        # database queries etc.
        await asyncio.sleep(0.1)  # simulate db query
        # TODO: check authorization

        tid = 1
        id_tag = IdTagInfo(status=AuthorizationStatus.accepted)

        return call_result.StartTransactionPayload(transaction_id=tid, id_tag_info=id_tag)

    @on(Action.StopTransaction)
    async def stop_transaction(self, **kwargs):
        # database queries etc.
        await asyncio.sleep(0.1)  # simulate db query

        return call_result.StopTransactionPayload(id_tag_info=IdTagInfo(status=AuthorizationStatus.accepted))

    async def remote_start_transaction(self, id_tag: str,
                                       connector_id: Optional[int] = None,
                                       charging_profile: Optional[Dict] = None, ) -> bool:
        # database queries etc.
        # TODO: check cp status
        await asyncio.sleep(0.1)  # simulate db query

        request = call.RemoteStartTransactionPayload(id_tag, connector_id, charging_profile)
        response = await self.call(request)
        if response.status == RemoteStartStopStatus.accepted:
            return True
        return False


class CentralSystem:
    def __init__(self):
        self._chargers: Dict[str, ChargePointV16] = {}
        self._tasks: Dict[str, Task] = {}

    def register_charger(self, client_id: str, cp: ChargePointV16) -> Task:
        task = asyncio.create_task(self.start_charger(client_id, cp))
        self._chargers[client_id] = cp
        self._tasks[client_id] = task
        return task

    async def start_charger(self, client_id: str, cp: ChargePointV16):
        try:
            await cp.start()
        except Exception as error:
            print(f"Charger {cp.id} disconnected: {error}")
        finally:
            del self._chargers[client_id]
            del self._tasks[client_id]

    def charger_exists(self, client_id: str):
        return client_id in self._chargers

    def get_charger(self, client_id: str):
        return self._chargers[client_id]


app.csms = CentralSystem()


@app.websocket("/cp/{cp_id}")
async def websocket_endpoint(websocket: WebSocket, cp_id: str):
    if app.csms.charger_exists(cp_id):
        await websocket.close()
        return
    await websocket.accept(subprotocol='ocpp1.6')
    print(f'base_url: {websocket.url.hostname}')  # TODO: hostname e göre tenant belirleme yapılacak
    cp = ChargePointV16(cp_id, SocketAdapter(websocket))
    await app.csms.register_charger(cp_id, cp)


@app.post("/transaction/{charge_point_id}")
@app.post("/transaction/{charge_point_id}/{connector_id}")
async def transaction(charge_point_id: str, connector_id: int = 1):
    cp = app.csms.get_charger(charge_point_id)
    if cp is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
                            detail=f"Charger not found")
    id_tag = "123456"
    # TODO: get id_tag from db
    await asyncio.sleep(0.1)  # simulate db query

    ok = await cp.remote_start_transaction(id_tag, connector_id)
    if not ok:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST,
                            detail=f"Transaction failed")
    return {"message": "Transaction started"}

I would appreciate it if you could review my code and provide any feedback or suggestions for improvement. Specifically, I'm interested in knowing if I'm using the OCPP library correctly and if there are any Python best practices that I should be following. Thank you for your time and assistance.