ocpp_client.py

import asyncio
import logging
from datetime import datetime
import nest_asyncio
import websockets
from ocpp.routing import on, after
from ocpp.v16 import call, call_result
from ocpp.v16 import ChargePoint as Cp
from ocpp.v16.enums import (
    Action,
    AvailabilityStatus,
    ChargingProfileStatus,
    ChargePointStatus,
    ClearChargingProfileStatus,
    ConfigurationStatus,
    DataTransferStatus,
    ReservationStatus,
    ResetStatus,
    TriggerMessageStatus,
    UnlockStatus,
)

logging.basicConfig(level=logging.INFO, format="[%(asctime)s] %(levelname)s: %(message)s")


def ts() -> str:
    """ISO-8601 UTC timestamp."""
    return datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S") + "Z"


class ChargePoint(Cp):
    """
    Demo OCPP 1.6J Charge Point
    - Handles calls from CSMS (@on).
    - Triggers additional actions (@after) following certain calls.
    - The CP side sends active calls using self.call(...).
    """

    def __init__(self, id, connection):
        super().__init__(id, connection)
        self.tx_id: int | None = None

    # ---------------------------
    # CSMS -> CP 
    # ---------------------------

    @on(Action.GetConfiguration)
    async def on_get_configuration(self, **kwargs):
        return call_result.GetConfigurationPayload(configuration_key=[], unknown_key=["unknown_key"])

    @on(Action.ChangeConfiguration)
    async def on_change_configuration(self, key: str = "", value: str = "", **kwargs):
        return call_result.ChangeConfigurationPayload(status=ConfigurationStatus.accepted)

    @on(Action.ChangeAvailability)
    async def on_change_availability(self, **kwargs):
        return call_result.ChangeAvailabilityPayload(status=AvailabilityStatus.accepted)

    @on(Action.UnlockConnector)
    async def on_unlock_connector(self, **kwargs):
        return call_result.UnlockConnectorPayload(status=UnlockStatus.unlocked)

    @on(Action.Reset)
    async def on_reset(self, **kwargs):
        return call_result.ResetPayload(status=ResetStatus.accepted)

    @on(Action.SetChargingProfile)
    async def on_set_charging_profile(self, **kwargs):
        return call_result.SetChargingProfilePayload(status=ChargingProfileStatus.accepted)

    @on(Action.ClearChargingProfile)
    async def on_clear_charging_profile(self, **kwargs):
        return call_result.ClearChargingProfilePayload(status=ClearChargingProfileStatus.accepted)

    @on(Action.TriggerMessage)
    async def on_trigger_message(self, **kwargs):
        return call_result.TriggerMessagePayload(status=TriggerMessageStatus.accepted)

    @on(Action.SendLocalList)
    async def on_send_local_list(self, **kwargs):
        return call_result.SendLocalListPayload(status="Accepted")

    @on(Action.ReserveNow)
    async def on_reserve_now(self, **kwargs):
        return call_result.ReserveNowPayload(status=ReservationStatus.accepted)

    @on(Action.UpdateFirmware)
    async def on_update_firmware(self, **kwargs):
        return call_result.UpdateFirmwarePayload()

    @on(Action.GetDiagnostics)
    async def on_get_diagnostics(self, **kwargs):
        return call_result.GetDiagnosticsPayload(file_name="diag.log")

    @on(Action.DataTransfer)
    async def on_data_transfer(self, **kwargs):
        return call_result.DataTransferPayload(status=DataTransferStatus.accepted, data="OK")

    # ---------------------------
    # RemoteStart / RemoteStop
    # ---------------------------

    @on(Action.RemoteStartTransaction)
    async def on_remote_start(self, id_tag: str | None = None, connector_id: int | None = None, **kwargs):
        # When CSMS sends ‘RemoteStartTransaction’, the first response is:
        return call_result.RemoteStartTransactionPayload(status="Accepted")

    @after(Action.RemoteStartTransaction)
    async def after_remote_start(self, id_tag: str | None = None, connector_id: int | None = None, **kwargs):
        # Then the CP side simulates preparation + StartTransaction + MeterValues.
        connector_id = connector_id or 1
        id_tag = id_tag or "TEST_TAG"

        # PREPARING
        await self.call(
            call.StatusNotificationPayload(
                connector_id=connector_id,
                error_code="NoError",
                status=ChargePointStatus.preparing,
                timestamp=ts(),
                vendor_id="TestVendor",
                info="Demo CP",
            )
        )

        # StartTransaction
        start_conf = await self.call(
            call.StartTransactionPayload(
                connector_id=connector_id,
                id_tag=id_tag,
                meter_start=0,
                timestamp=ts(),
            )
        )
        self.tx_id = start_conf.transaction_id

        # Örnek DataTransfer
        await self.call(
            call.DataTransferPayload(
                vendor_id="TestVendor",
                message_id="EVCCID",
                data="SXWZAPYW;BURAK",
            )
        )

        # CHARGING
        await self.call(
            call.StatusNotificationPayload(
                connector_id=connector_id,
                error_code="NoError",
                status=ChargePointStatus.charging,
                timestamp=ts(),
                vendor_id="TestVendor",
                info="Demo CP",
            )
        )

        # Birkaç örnek MeterValues
        async def send_mv(soc_percent: int, energy_Wh: str):
            await self.call(
                call.MeterValuesPayload(
                    connector_id=connector_id,
                    transaction_id=self.tx_id,
                    meter_value=[
                        {
                            "timestamp": ts(),
                            "sampledValue": [
                                {
                                    "value": "230.0",
                                    "context": "Sample.Periodic",
                                    "format": "Raw",
                                    "measurand": "Voltage",
                                    "location": "Inlet",
                                    "unit": "V",
                                },
                                {
                                    "value": "16.0",
                                    "context": "Sample.Periodic",
                                    "format": "Raw",
                                    "measurand": "Current.Import",
                                    "location": "Outlet",
                                    "unit": "A",
                                },
                                {
                                    "value": "13680.00",
                                    "context": "Sample.Periodic",
                                    "format": "Raw",
                                    "measurand": "Power.Active.Import",
                                    "location": "Outlet",
                                    "unit": "W",
                                },
                                {
                                    "value": energy_Wh,
                                    "context": "Sample.Periodic",
                                    "format": "Raw",
                                    "measurand": "Energy.Active.Import.Register",
                                    "location": "Outlet",
                                    "unit": "Wh",
                                },
                                {
                                    "value": str(soc_percent),
                                    "unit": "Percent",
                                    "measurand": "SoC",
                                },
                            ],
                        }
                    ],
                )
            )

        await asyncio.sleep(1)
        await send_mv(34, "315.0")
        await asyncio.sleep(5)
        await send_mv(35, "0")
        await asyncio.sleep(10)
        await send_mv(36, "0")
        await asyncio.sleep(10)
        await send_mv(38, "0")

        # StopTransaction
        await self.call(
            call.StopTransactionPayload(
                meter_stop=15000,
                timestamp=ts(),
                transaction_id=self.tx_id,
            )
        )

        # AVAILABLE
        await asyncio.sleep(1)
        await self.call(
            call.StatusNotificationPayload(
                connector_id=connector_id,
                error_code="NoError",
                status=ChargePointStatus.available,
                timestamp=ts(),
                vendor_id="TestVendor",
                info="Demo CP",
            )
        )

    @on(Action.RemoteStopTransaction)
    async def on_remote_stop(self, transaction_id: int | None = None, **kwargs):
        return call_result.RemoteStopTransactionPayload(status="Accepted")

    @after(Action.RemoteStopTransaction)
    async def after_remote_stop(self, transaction_id: int | None = None, **kwargs):
        # Finishing -> StopTransaction -> Available
        await self.call(
            call.StatusNotificationPayload(
                connector_id=1,
                error_code="NoError",
                status=ChargePointStatus.finishing,
                timestamp=ts(),
                vendor_id="TestVendor",
                info="Demo CP",
            )
        )
        await self.call(
            call.StopTransactionPayload(
                meter_stop=15000,
                timestamp=ts(),
                transaction_id=transaction_id or self.tx_id,
            )
        )
        await asyncio.sleep(1)
        await self.call(
            call.StatusNotificationPayload(
                connector_id=1,
                error_code="NoError",
                status=ChargePointStatus.available,
                timestamp=ts(),
                vendor_id="TestVendor",
                info="Demo CP",
            )
        )

    # ---------------------------
    # CP -> CSMS
    # ---------------------------

    async def send_boot_notification(self):
        await self.call(
            call.BootNotificationPayload(
                charge_point_model="TestModel",
                charge_point_vendor="TestVendor",
                charge_point_serial_number="1234",
                firmware_version="OCPP-1.6J-Demo",
                meter_type="TestMeter",
                meter_serial_number="0",
            )
        )

    async def heartbeat(self):
        await self.call(call.HeartbeatPayload())


async def main():
    # websocket URL of the CSMS
    ws_url = "ws://localhost:8000/OCPP/1/CP2"
    async with websockets.connect(ws_url, subprotocols=["ocpp1.6"]) as ws:
        charge_point = ChargePoint("CP1", ws)
        await asyncio.gather(
            charge_point.start(),
            charge_point.send_boot_notification(),
            # charge_point.heartbeat(),
        )


if __name__ == "__main__":
    # asyncio.run() is used when running this example with Python >= 3.7v
    nest_asyncio.apply()
    asyncio.run(main())