server.py

#!/usr/bin/env python3
import asyncio
import json
import logging
from ocpp.v16.enums import AvailabilityType 
import signal
from ocpp.routing import on
from datetime import datetime, timezone
from typing import Dict, Any, Optional

import websockets
from websockets.exceptions import ConnectionClosedError, ConnectionClosedOK
from websockets.server import WebSocketServerProtocol

from aiohttp import web

from ocpp.v16 import ChargePoint as CP
from ocpp.v16 import call, call_result
from ocpp.v16.enums import (
    RegistrationStatus, Action, ResetType, AvailabilityType, MessageTrigger
)
from ocpp.v16.enums import AuthorizationStatus
from ocpp.v16.datatypes import IdTagInfo
from collections import deque
import asyncio
import contextlib
# -------------------------
# Konfig
# -------------------------
OCPP_HOST = "0.0.0.0"
OCPP_PORT = 8001
OCPP_PATH_PREFIX = "/ocpp"   # ws://<ip>:8001/ocpp/<CPID>

HTTP_HOST = "0.0.0.0"
HTTP_PORT = 8080

# Ping/pong ve read timeouts
WS_PING_INTERVAL = 30
WS_PING_TIMEOUT = 20

# Son 500 log satırı için ring buffer + canlı yayın kuyruğu
LOG_BUFFER = deque(maxlen=500)
LOG_QUEUES: "set[asyncio.Queue[str]]" = set()

# -------------------------
# Logging
# -------------------------
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(message)s"
)

class _MemAndSSEHandler(logging.Handler):
    def emit(self, record):
        try:
            line = self.format(record)
        except Exception:
            line = str(record)
        LOG_BUFFER.append(line)
        # Bağlı SSE dinleyicilerine ilet
        for q in list(LOG_QUEUES):
            try:
                q.put_nowait(line)
            except Exception:
                pass

# Format aynı kalsın; hem dosyaya hem SSE handler'a yazalım
fmt = logging.Formatter("%(asctime)s | %(levelname)s | %(message)s")
mem_handler = _MemAndSSEHandler()
mem_handler.setFormatter(fmt)
logging.getLogger().addHandler(mem_handler)

# İsteğe bağlı: dosyaya da yaz (tail için)
from logging.handlers import RotatingFileHandler
file_handler = RotatingFileHandler("/opt/ocpp-min/ocpp.log", maxBytes=1_000_000, backupCount=3)
file_handler.setFormatter(fmt)
logging.getLogger().addHandler(file_handler)

log = logging.getLogger("ocpp-min")

# -------------------------
# Global Durum
# -------------------------
CPS: Dict[str, "CentralSystemCP"] = {}
STATE: Dict[str, Dict[str, Any]] = {}

# -------------------------
# Yardımcılar
# -------------------------
def to_jsonable(obj: Any) -> Any:
    if isinstance(obj, datetime):
        return obj.astimezone(timezone.utc).isoformat()
    if isinstance(obj, dict):
        return {k: to_jsonable(v) for k, v in obj.items()}
    if isinstance(obj, list):
        return [to_jsonable(v) for v in obj]
    return obj

def enum_ci(enum_cls, value, default=None):
    """Enum üyesini name/value case-insensitive eşleştir."""
    if isinstance(value, enum_cls):
        return value
    s = str(value)
    for m in enum_cls:
        if m.name.lower() == s.lower() or str(m.value).lower() == s.lower():
            return m
    return default or list(enum_cls)[0]

# -------------------------
# OCPP Tarafı
# -------------------------
class CentralSystemCP(CP):
    def __init__(self, id: str, connection: WebSocketServerProtocol):
        super().__init__(id, connection)
        self.id = id
        self._call_lock = asyncio.Lock()
        self._tuned = False

    async def safe_call(self, req, timeout: float = 10.0):
        try:
            async with self._call_lock:
                return await asyncio.wait_for(self.call(req), timeout=timeout)
        except asyncio.TimeoutError:
            log.warning("[%s] %s timed out; continuing.", self.id, type(req).__name__)
            return None
        except Exception:
            log.exception("[%s] call error on %s", self.id, type(req).__name__)
            return None

    # ------ Cihaz -> Merkez Handlers ------
    @on(Action.BootNotification)
    async def on_boot(self, charge_point_vendor: str, charge_point_model: str, **kwargs):
        extra = {
            # ocpp 0.17 handler'lara snake_case geçirir
            "charge_point_serial_number": kwargs.get("charge_point_serial_number"),
            "meter_type": kwargs.get("meter_type"),
            "meter_serial_number": kwargs.get("meter_serial_number"),
            "iccid": kwargs.get("iccid"),
            "charge_box_serial_number": kwargs.get("charge_box_serial_number"),
            "firmware_version": kwargs.get("firmware_version"),
            "imsi": kwargs.get("imsi"),
        }
        log.info("[%s] BootNotification: model=%s, vendor=%s, extra=%s",
                 self.id, charge_point_model, charge_point_vendor, extra)

        s = STATE.setdefault(self.id, {})
        s["last_boot"] = {
            "vendor": charge_point_vendor,
            "model": charge_point_model,
            "extra": extra,
            "at": datetime.now(timezone.utc),
        }

        if not self._tuned:
            asyncio.create_task(self._initial_tuning())
            self._tuned = True

        return call_result.BootNotificationPayload(
            current_time=datetime.now(timezone.utc).isoformat(),
            interval=60,
            status=RegistrationStatus.accepted
        )

    @on(Action.StatusNotification)
    async def on_status(self, connector_id: int, status: str, **kwargs):
        vendor_id = kwargs.get("vendorId", "")
        err = kwargs.get("errorCode", "")
        info = kwargs.get("info", "")
        ts = kwargs.get("timestamp")
        log.info("[%s] StatusNotification: connector=%s, status=%s, error=%s, extra=%s",
                 self.id, connector_id, status, err,
                 {"info": info, "vendor_id": vendor_id, "timestamp": ts})

        s = STATE.setdefault(self.id, {})
        connmap = s.setdefault("status", {})
        connmap[str(connector_id)] = status
        s["last_seen"] = datetime.now(timezone.utc)

        # Bazı idempotent ayarları push edelim
        # asyncio.create_task(self._initial_tuning())
        return call_result.StatusNotificationPayload()

    async def _initial_tuning(self):
        # AuthorizeRemoteTxRequests -> true
        resp = await self.safe_call(call.ChangeConfigurationPayload(
            key="AuthorizeRemoteTxRequests", value="true"
        ))
        if resp:
            log.info("[%s] ChangeConfiguration(AuthorizeRemoteTxRequests=true) result: %s", self.id, resp.status)

        # AuthorizationRequired -> false (desteklemeyebilir)
        resp = await self.safe_call(call.ChangeConfigurationPayload(
            key="AuthorizationRequired", value="false"
        ))
        if resp:
            log.info("[%s] ChangeConfiguration(AuthorizationRequired=false) result: %s", self.id, resp.status)

        # LocalAuthorizeOffline -> true
        resp = await self.safe_call(call.ChangeConfigurationPayload(
            key="LocalAuthorizeOffline", value="true"
        ))
        if resp:
            log.info("[%s] ChangeConfiguration(LocalAuthorizeOffline=true) result: %s", self.id, resp.status)

        # StopTransactionOnInvalidId -> false
        resp = await self.safe_call(call.ChangeConfigurationPayload(
            key="StopTransactionOnInvalidId", value="false"
        ))
        if resp:
            log.info("[%s] ChangeConfiguration(StopTransactionOnInvalidId=false) result: %s", self.id, resp.status)

        # HeartbeatInterval -> 60
        resp = await self.safe_call(call.ChangeConfigurationPayload(
            key="HeartbeatInterval", value="60"
        ))
        if resp:
            log.info("[%s] ChangeConfiguration(HeartbeatInterval=60) result: %s", self.id, resp.status)

        # MeterValuesSampledData
        resp = await self.safe_call(call.ChangeConfigurationPayload(
            key="MeterValuesSampledData",
            value="Energy.Active.Import.Register,Power.Active.Import,Current.Import,Voltage"
        ))
        if resp:
            log.info("[%s] ChangeConfiguration(MeterValuesSampledData) result: %s", self.id, resp.status)

        # MeterValueSampleInterval -> 15
        resp = await self.safe_call(call.ChangeConfigurationPayload(
            key="MeterValueSampleInterval", value="15"
        ))
        if resp:
            log.info("[%s] ChangeConfiguration(MeterValueSampleInterval=15) result: %s", self.id, resp.status)

    @on(Action.Heartbeat)
    async def on_heartbeat(self):
        log.info("[%s] Heartbeat", self.id)
        s = STATE.setdefault(self.id, {})
        s["last_seen"] = datetime.now(timezone.utc)
        return call_result.HeartbeatPayload(current_time=datetime.now(timezone.utc).isoformat())
    @on(Action.MeterValues)
    async def on_meter_values(self, connector_id: int, meter_value: list, **kwargs):
        tx_id = kwargs.get("transactionId")
        values = []
        for mv in meter_value:
            values.append({
                "timestamp": mv.get("timestamp"),
                "sampled_value": mv.get("sampled_value", [])
            })
        log.info("[%s] MeterValues: connector=%s, values=%s, extra=%s",
                 self.id, connector_id, values,
                 {"transaction_id": tx_id} if tx_id else {})
        s = STATE.setdefault(self.id, {})
        s["last_meter"] = {
            "connector_id": connector_id,
            "values": values,
            "tx_id": tx_id,
            "at": datetime.now(timezone.utc)
        }
        s["last_seen"] = datetime.now(timezone.utc)
        return call_result.MeterValuesPayload()

    @on(Action.StartTransaction)
    async def on_start_tx(self, connector_id: int, id_tag: str, meter_start: int, timestamp: str, **kwargs):
        log.info("[%s] StartTransaction: conn=%s, id_tag=%s, meter_start=%s, ts=%s",
                 self.id, connector_id, id_tag, meter_start, timestamp)
        s = STATE.setdefault(self.id, {})
        s["tx_active"] = True
        s["tx"] = {"id_tag": id_tag, "connector_id": connector_id, "start": timestamp, "meter_start": meter_start}
        return call_result.StartTransactionPayload(
            transaction_id=1,
            id_tag_info=IdTagInfo(status=AuthorizationStatus.accepted)
        )

    @on(Action.StopTransaction)
    async def on_stop_tx(self, transaction_id: int, meter_stop: int, timestamp: str, **kwargs):
        reason = kwargs.get("reason")
        id_tag = kwargs.get("idTag")
        log.info("[%s] StopTransaction: tx=%s, meter_stop=%s, reason=%s, ts=%s",
                 self.id, transaction_id, meter_stop, reason, timestamp)
        s = STATE.setdefault(self.id, {})
        s["tx_active"] = False
        s["tx"] = {**s.get("tx", {}), "stop": timestamp, "meter_stop": meter_stop, "reason": reason, "id_tag": id_tag}
        return call_result.StopTransactionPayload(
            id_tag_info=IdTagInfo(status=AuthorizationStatus.accepted)
        )


# -------------------------
# WS Sunucu (OCPP)
# -------------------------
async def ws_handler(websocket: WebSocketServerProtocol):
    path = websocket.path or ""
    parts = [p for p in path.split("/") if p]
    if len(parts) != 2 or f"/{parts[0]}" != OCPP_PATH_PREFIX:
        log.info("connection rejected (400 Bad Request)")
        await websocket.close(code=1008, reason="Bad path")
        return

    cpid = parts[1]
    log.info("Incoming connection from CPID=%s", cpid)

    cp = CentralSystemCP(cpid, websocket)
    CPS[cpid] = cp

    try:
        await cp.start()
    except (ConnectionClosedError, ConnectionClosedOK):
        pass
    except Exception:
        log.exception("connection handler failed")
    finally:
        log.info("connection closed")

# -------------------------
# HTTP API
# -------------------------

async def _read_body(request: web.Request) -> dict:
    try:
        if request.content_type == "application/json":
            data = await request.json()
        else:
            data = await request.post()
    except Exception:
        data = {}
    # her iki anahtar varyantını da destekle
    if "cpid" not in data and "cp_id" in data:
        data["cpid"] = data["cp_id"]
    return dict(data)

routes = web.RouteTableDef()

@routes.get("/")
async def http_dashboard(request: web.Request):
    # Tek dosyalık, basit bir HTML
    html = """
    <!doctype html>
<html lang="tr">
<head>
  <meta charset="utf-8" />
  <title>OCPP Mini Panel</title>
  <meta name="viewport" content="width=device-width, initial-scale=1" />
  <style>
    :root { font-family: system-ui, Arial, sans-serif; }
    body { margin: 0; background:#0b1220; color:#e7ecf3; }
    header { padding:12px 16px; background:#111a2e; border-bottom:1px solid #1c2a4a; }
    h1 { margin:0; font-size:18px; }
    main { display:grid; gap:16px; grid-template-columns: 360px 1fr; padding:16px; }
    .card { background:#0d162b; border:1px solid #1c2a4a; border-radius:10px; padding:14px; }
    .row { display:flex; gap:8px; align-items:center; margin:8px 0; flex-wrap:wrap; }
    label { font-size:12px; opacity:.8; }
    input, select, button, textarea {
      background:#0b1220; color:#e7ecf3; border:1px solid #1c2a4a; border-radius:8px; 
      padding:8px 10px; outline:none;
    }
    input, select { width: 100%; }
    button { cursor:pointer; }
    button.primary { background:#2f6feb; border-color:#2f6feb; }
    button:disabled { opacity:.5; cursor:not-allowed; }
    .grid-2 { display:grid; gap:8px; grid-template-columns: 1fr 1fr; }
    #logs { height: 300px; width:100%; resize: vertical; font-family: ui-monospace, SFMono-Regular, Menlo, Consolas, monospace; }
    .muted { opacity:.75; }
    .ok { color:#41d392; }
    .bad { color:#ff7b7b; }
    code { background:#091026; padding:2px 6px; border-radius:6px; border:1px solid #1c2a4a;}
  </style>
</head>
<body>
  <header>
    <h1>OCPP Kontrol Paneli</h1>
  </header>

  <main>
    <!-- Sol sütun -->
    <section class="card">
      <h3>Bağlantı / Durum</h3>
      <div class="row">
        <label>CPID</label>
        <input id="cpid" placeholder="örn: 261265" />
      </div>
      <div class="row grid-2">
        <div>
          <label>Seçili konektör</label>
          <select id="connector">
            <option value="0">0</option>
            <option value="1" selected>1</option>
            <option value="2">2</option>
          </select>
        </div>
        <div>
          <label>IdTag</label>
          <input id="idtag" value="TEST1234" />
        </div>
      </div>

      <div class="row">
        <button id="btn-refresh">Durumu Yenile</button>
      </div>

      <div class="row">
        <div>
          <div class="muted">Sunucu:</div>
          <div id="srv-ocpp" class="muted">-</div>
          <div id="srv-http" class="muted">-</div>
          <div id="srv-now" class="muted">-</div>
        </div>
      </div>

      <hr style="border-color:#1c2a4a; opacity:.4">

      <div>
        <div class="muted">Cihaz Durumları</div>
        <div id="cp-summary">“durum” yükleniyor…</div>
        <pre id="cp-status" style="white-space:pre-wrap"></pre>
      </div>
    </section>

    <!-- Sağ sütun -->
    <section class="card">
      <h3>Operasyonlar</h3>
      <div class="row grid-2">
        <button id="op-boot">Trigger Boot</button>
        <button id="op-reset-hard">Reset (Hard)</button>
      </div>
      <div class="row grid-2">
        <button id="op-available">ChangeAvailability: Operative</button>
        <button id="op-unavailable">ChangeAvailability: Inoperative</button>
      </div>
      <div class="row grid-2">
        <button id="op-start" class="primary">Remote Start</button>
        <button id="op-stop">Remote Stop (tx=1)</button>
      </div>
      <div class="row">
        <button id="op-unlock">Unlock Connector</button>
      </div>

      <h3 style="margin-top:18px">Canlı Log</h3>
      <textarea id="logs" readonly placeholder="SSE bağlanıyor…"></textarea>
    </section>
  </main>

  <script>
    // ---- Yardımcılar
    const $ = (id) => document.getElementById(id);

    function qs(data) {
      // x-www-form-urlencoded body hazırla
      return new URLSearchParams(Object.entries(data).filter(([_,v]) => v !== undefined && v !== null));
    }

    async function jget(url) {
      try {
        const res = await fetch(url);
        try { return await res.json(); }
        catch { return { raw: await res.text(), status: res.status }; }
      } catch (err) {
        return { error: String(err) };
      }
    }

    async function jpost(url, data) {
      try {
        const res = await fetch(url, {
          method: 'POST',
          headers: { 'Content-Type': 'application/x-www-form-urlencoded' },
          body: qs(data)
        });
        try { return await res.json(); }
        catch { return { raw: await res.text(), status: res.status }; }
      } catch (err) {
        return { error: String(err) };
      }
    }

    function renderStatus(s) {
      if (!s || !s.charge_points) return;
      const cpid = $('cpid').value.trim();
      const cp = s.charge_points[cpid];
      $('srv-ocpp').textContent = "OCPP: " + (s.server?.ocpp ?? "-");
      $('srv-http').textContent = "HTTP: " + (s.server?.http ?? "-");
      $('srv-now').textContent  = "Now:  " + (s.server?.now ?? "-");

      if (!cp) {
        $('cp-summary').innerHTML = `<span class="bad">CP ${cpid || '(boş)'} bağlı değil</span>`;
        $('cp-status').textContent = "";
        setButtons(false);
        return;
      }
      const last = cp.last_seen || "-";
      const statuses = cp.status || {};
      const lines = Object.entries(statuses)
  .map(([k, v]) => `Connector ${k}: ${v}`)
  .join("\\n");
      $('cp-summary').innerHTML = `<span class="ok">Bağlı</span> • last_seen: <code>${last}</code>`;
      $('cp-status').textContent = lines || "(durum yok)";
      setButtons(true);
    }

    function setButtons(enabled) {
      for (const id of [
        "op-boot","op-reset-hard","op-available","op-unavailable",
        "op-start","op-stop","op-unlock"
      ]) {
        $(id).disabled = !enabled;
      }
    }

    // ---- Durum anlık yenileme
    async function refreshStatus() {
      const s = await jget('/status');
      renderStatus(s);
    }

    // İlk değerleri doldur (localStorage isteğe bağlı)
    const savedCpid = localStorage.getItem('cpid');
    if (savedCpid) $('cpid').value = savedCpid;
    $('cpid').addEventListener('input', e => localStorage.setItem('cpid', e.target.value.trim()));

    $('btn-refresh').addEventListener('click', refreshStatus);
    refreshStatus();
    setInterval(refreshStatus, 3000);

    // ---- Operasyon butonları
    $('op-boot').addEventListener('click', async () => {
      const cpid = $('cpid').value.trim();
      const r = await jpost('/trigger_boot', { cpid });
      appendLog('Trigger Boot → ' + JSON.stringify(r));
      refreshStatus();
    });

    $('op-reset-hard').addEventListener('click', async () => {
      const cpid = $('cpid').value.trim();
      const r = await jpost('/reset', { cpid, type: 'Hard' });
      appendLog('Reset(Hard) → ' + JSON.stringify(r));
    });

    $('op-available').addEventListener('click', async () => {
      const cpid = $('cpid').value.trim();
      const connector_id = $('connector').value;
      const r = await jpost('/availability', { cpid, connector_id, type: 'Operative' });
      appendLog('Availability(Operative) → ' + JSON.stringify(r));
    });

    $('op-unavailable').addEventListener('click', async () => {
      const cpid = $('cpid').value.trim();
      const connector_id = $('connector').value;
      const r = await jpost('/availability', { cpid, connector_id, type: 'Inoperative' });
      appendLog('Availability(Inoperative) → ' + JSON.stringify(r));
    });

    $('op-start').addEventListener('click', async () => {
      const cpid = $('cpid').value.trim();
      const connector_id = $('connector').value;
      const id_tag = $('idtag').value.trim() || 'TEST1234';
      const r = await jpost('/start', { cpid, connector_id, id_tag });
      appendLog('RemoteStart → ' + JSON.stringify(r));
    });

    $('op-stop').addEventListener('click', async () => {
      const cpid = $('cpid').value.trim();
      const r = await jpost('/stop', { cpid, transaction_id: 1 });
      appendLog('RemoteStop(tx=1) → ' + JSON.stringify(r));
    });

    $('op-unlock').addEventListener('click', async () => {
      const cpid = $('cpid').value.trim();
      const connector_id = $('connector').value;
      const r = await jpost('/unlock', { cpid, connector_id });
      appendLog('UnlockConnector → ' + JSON.stringify(r));
    });

    // ---- Log akışı (SSE)
    const logsEl = $('logs');
    function appendLog(line) {
      logsEl.value += line + "\\n";
      logsEl.scrollTop = logsEl.scrollHeight;
    }

    try {
      const es = new EventSource('/logs/stream');
      es.onmessage = (e) => { appendLog(e.data); };
      es.onerror = () => { appendLog("[SSE] bağlantı hatası. Yeniden deniyor...\\n"); };
    } catch (err) {
      appendLog("[SSE] Hata: " + err + "\\n");
    }
  </script>
</body>
</html>
    """.strip()
    return web.Response(text=html, content_type="text/html")

async def _data(request: web.Request) -> dict:
    # JSON geldiyse JSON, yoksa form-data döndür
    try:
        return await request.json()
    except Exception:
        return dict(await request.post())

def _extract_cpid(d: dict) -> str:
    return str(d.get("cpid") or d.get("cp_id") or d.get("cpId") or d.get("id") or "").strip()

@routes.get("/status")
async def http_status(request: web.Request):
    body = {
        "server": {
            "ocpp": f"ws://{OCPP_HOST}:{OCPP_PORT}{OCPP_PATH_PREFIX}/<CPID>",
            "http": f"http://{HTTP_HOST}:{HTTP_PORT}",
            "now": datetime.now(timezone.utc),
        },
        "charge_points": to_jsonable(STATE)
    }
    return web.json_response(to_jsonable(body))

def _get_cp(cpid: str) -> Optional[CentralSystemCP]:
    cp = CPS.get(cpid)
    if not cp:
        raise web.HTTPNotFound(text=f"CP {cpid} not connected.")
    return cp

@routes.post("/trigger_boot")
async def http_trigger_boot(request: web.Request):
    data = await _read_body(request)
    cpid = data.get("cpid") or data.get("cp_id")
    log.info("HTTP %s %s from %s body=%s", request.method, request.path, request.remote, dict(data))
    cp = _get_cp(cpid)
    log.info("[%s] TriggerMessage(BootNotification)", cpid)
    resp = await cp.safe_call(call.TriggerMessagePayload(
        requested_message=MessageTrigger.boot_notification
    ))
    return web.json_response({"status": getattr(resp, "status", "NoResponse")})

@routes.post("/reset")
async def http_reset(request: web.Request):
    data = await _read_body(request)
    cpid = data.get("cpid")
    reset_type_raw = data.get("type", data.get("reset_type", "Hard"))
    log.info("HTTP %s %s from %s body=%s", request.method, request.path, request.remote, dict(data))
    reset_type = enum_ci(ResetType, reset_type_raw, default=ResetType.hard)
    cp = _get_cp(cpid)
    log.info("[%s] Reset(%s)", cpid, reset_type.name)
    resp = await cp.safe_call(call.ResetPayload(type=reset_type))
    return web.json_response({"status": getattr(resp, "status", "NoResponse")})

@routes.post("/availability")
async def http_availability(request: web.Request):
    data = await _read_body(request)
    cpid = data.get("cpid")
    connector_id = int(data.get("connector_id", 1))
    # hem "availability" hem "type" destekle
    avail_raw = data.get("availability", data.get("type", "Operative"))
    avail_type = enum_ci(AvailabilityType, avail_raw, default=AvailabilityType.operative)

    cp = _get_cp(cpid)
    log.info("[%s] ChangeAvailability(%s) on connector %s", cpid, avail_type.name, connector_id)

    resp = await cp.safe_call(call.ChangeAvailabilityPayload(
        connector_id=connector_id,
        type=avail_type
    ))
    return web.json_response({"status": getattr(resp, "status", "NoResponse")})

@routes.post("/start")
async def http_remote_start(request: web.Request):
    data = await _read_body(request)
    cpid = data.get("cpid")
    connector_id = int(data.get("connector_id", 1))
    id_tag = data.get("id_tag", "TEST1234")
    cp = _get_cp(cpid)
    log.info("[%s] RemoteStartTransaction on connector %s (idTag=%s)", cpid, connector_id, id_tag)
    resp = await cp.safe_call(call.RemoteStartTransactionPayload(
        id_tag=id_tag, connector_id=connector_id
    ))
    return web.json_response({"status": getattr(resp, "status", "NoResponse")})

@routes.post("/stop")
async def http_remote_stop(request: web.Request):
    data = await _read_body(request)
    cpid = data.get("cpid")
    tx_id = int(data.get("transaction_id", 1))
    log.info("HTTP %s %s from %s body=%s", request.method, request.path, request.remote, dict(data))
    cp = _get_cp(cpid)
    log.info("[%s] RemoteStopTransaction(tx_id=%s)", cpid, tx_id)
    resp = await cp.safe_call(call.RemoteStopTransactionPayload(transaction_id=tx_id))
    return web.json_response({"status": getattr(resp, "status", "NoResponse")})

@routes.post("/unlock")
async def http_unlock(request: web.Request):
    data = await _read_body(request)
    cpid = data.get("cpid")
    connector_id = int(data.get("connector_id", 1))
    cp = _get_cp(cpid)
    log.info("[%s] UnlockConnector(connectorId=%s)", cpid, connector_id)
    resp = await cp.safe_call(call.UnlockConnectorPayload(connector_id=connector_id))
    resp = await cp.safe_call(call.UnlockConnectorPayload(connector_id=connector_id))
    return web.json_response({"status": getattr(resp, "status", "NoResponse")})

@routes.get("/logs/stream")
async def http_logs_stream(request: web.Request):
    resp = web.StreamResponse(
        status=200,
        reason="OK",
        headers={
            "Content-Type": "text/event-stream",
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",
        },
    )
    await resp.prepare(request)

    # Yeni abone kuyruğu
    q: asyncio.Queue[str] = asyncio.Queue()
    LOG_QUEUES.add(q)

    # Önce buffer’daki son satırları gönder
    for line in list(LOG_BUFFER):
        await resp.write(f"data: {line}\n\n".encode())

    # Keepalive ping (yorum satırı) – bazı proxy’ler için faydalı
    async def keepalive():
        try:
            while True:
                await asyncio.sleep(15)
                await resp.write(b": ping\n\n")
        except asyncio.CancelledError:
            pass

    ka_task = asyncio.create_task(keepalive())

    try:
        while True:
            line = await q.get()
            await resp.write(f"data: {line}\n\n".encode())
    except asyncio.CancelledError:
        pass
    finally:
        ka_task.cancel()
        LOG_QUEUES.discard(q)
        with contextlib.suppress(Exception):
            await resp.write_eof()

    return resp

@routes.get("/logs")
async def http_logs(request: web.Request):
    return web.json_response({"lines": list(LOG_BUFFER)})

# -------------------------
# Başlatma
# -------------------------
async def start_servers():
    ws_server = await websockets.serve(
        ws_handler, OCPP_HOST, OCPP_PORT,
        ping_interval=WS_PING_INTERVAL,
        ping_timeout=WS_PING_TIMEOUT,
        max_size=2**20
    )
    log.info("Starting OCPP 1.6J CS on ws://%s:%s%s/<CPID>  +  HTTP on http://%s:%s",
             OCPP_HOST, OCPP_PORT, OCPP_PATH_PREFIX, HTTP_HOST, HTTP_PORT)

    app = web.Application()
    app.add_routes(routes)
    runner = web.AppRunner(app)
    await runner.setup()
    http_site = web.TCPSite(runner, HTTP_HOST, HTTP_PORT)
    await http_site.start()

    return ws_server, runner

async def main():
    ws_server, http_runner = await start_servers()

    loop = asyncio.get_running_loop()
    stop_event = asyncio.Event()

    def _graceful_shutdown():
        # Birden fazla Ctrl+C / sinyale karşı idempotent
        if not stop_event.is_set():
            stop_event.set()

    for sig in (signal.SIGINT, signal.SIGTERM):
        try:
            loop.add_signal_handler(sig, _graceful_shutdown)
        except NotImplementedError:
            # (Windows vs.) desteklenmiyorsa KeyboardInterrupt'a bırak
            pass

    try:
        await stop_event.wait()
    finally:
        await http_runner.cleanup()
        ws_server.close()
        await ws_server.wait_closed()

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass