cariflex/scripts/citrineos_fm_integration.py

#!/usr/bin/env python3
"""
Cariflex - CitrineOS ↔ FlexMeasures Integration Service
Links OCPP charging sessions to FM sensors for energy management.
"""

import json, logging, os, re, sys
from datetime import datetime, timezone, timedelta
import requests

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("cariflex-citrineos-fm")

# Config
FM_HOST = os.getenv("FM_HOST", "https://cariflex.digitribe.fr")
FM_EMAIL = os.getenv("FM_EMAIL", "admin@digitribe.fr")
FM_PWD = os.getenv("FM_PWD", "Digitribe972")
CITRINEOS_HOST = os.getenv("CITRINEOS_HOST", "http://cariflex-citrineos-server:8080")

# OCPP Charge Point ID -> FM Sensor ID mapping
# Each charge point (EVSE) maps to a sensor in FlexMeasures
CHARGEPOINT_SENSOR_MAP = {
    "CP001": 61,  # EV_01
    "CP002": 62,  # EV_02
    "CP003": 63,  # EV_03
    "CP004": 64,  # EV_04
    "CP005": 65,  # EV_05
    "CP006": 66,  # EV_06
    "CP007": 67,  # EV_07
    "CP008": 68,  # EV_08
    "CP009": 69,  # EV_09
    "CP010": 70,  # EV_10
}

fm_session = None


def fm_login():
    global fm_session
    try:
        s = requests.Session(); s.verify = False
        r = s.get(f"{FM_HOST}/login", timeout=15)
        m = re.search(r"csrf_token[^>]*value=[\"\\']([^\"\\']+)", r.text)
        if m:
            r = s.post(f"{FM_HOST}/login", data={"email":FM_EMAIL,"password":FM_PWD,"csrf_token":m.group(1),"remember":"y"}, allow_redirects=True, timeout=15)
            if "dashboard" in r.url or r.status_code == 200:
                fm_session = s
                return True
    except Exception as e:
        logger.error(f"FM login: {e}")
    return False


def post_sensor_data(sensor_id, value, unit, start, duration="PT1H"):
    if not fm_session:
        return False
    try:
        r = fm_session.post(f"{FM_HOST}/api/v3_0/sensors/{sensor_id}/data",
            json={"values":[value],"start":start,"duration":duration,"unit":unit}, timeout=30)
        return r.status_code in [200,201,202]
    except Exception as e:
        logger.error(f"Post: {e}")
        return False


def get_citrineos_sessions():
    """Get active charging sessions from CitrineOS."""
    try:
        r = requests.get(f"{CITRINEOS_HOST}/api/v1/sessions", timeout=10)
        if r.status_code == 200:
            return r.json()
    except Exception as e:
        logger.error(f"CitrineOS sessions: {e}")
    return []


def get_citrineos_meter_values(session_id):
    """Get meter values for a charging session."""
    try:
        r = requests.get(f"{CITRINEOS_HOST}/api/v1/sessions/{session_id}/meter-values", timeout=10)
        if r.status_code == 200:
            return r.json()
    except Exception as e:
        logger.error(f"CitrineOS meter values: {e}")
    return []


def process_charging_session(session):
    """Process a charging session and update FM sensors."""
    charge_point_id = session.get("chargePointId")
    session_id = session.get("id")
    
    if charge_point_id not in CHARGEPOINT_SENSOR_MAP:
        logger.warning(f"Unknown charge point: {charge_point_id}")
        return
    
    fm_sensor_id = CHARGEPOINT_SENSOR_MAP[charge_point_id]
    
    # Get meter values
    meter_values = get_citrineos_meter_values(session_id)
    
    for mv in meter_values:
        value = mv.get("value", 0)
        unit = mv.get("unit", "W")
        timestamp = mv.get("timestamp", datetime.now(timezone.utc).isoformat())
        
        # Convert W to kW
        if unit == "W":
            value = value / 1000
            unit = "kW"
        
        # Post to FM
        ok = post_sensor_data(fm_sensor_id, value, unit, timestamp)
        logger.info(f"CitrineOS {charge_point_id} -> FM sensor {fm_sensor_id}: {value} {unit}: {'OK' if ok else 'FAIL'}")


def send_charging_profile(charge_point_id, power_limit_kw):
    """Send a charging profile to a charge point via CitrineOS."""
    try:
        payload = {
            "chargePointId": charge_point_id,
            "chargingProfile": {
                "chargingProfileId": 1,
                "stackLevel": 0,
                "chargingProfilePurpose": "TxDefaultProfile",
                "chargingProfileKind": "Absolute",
                "chargingSchedule": {
                    "chargingRateUnit": "W",
                    "chargingSchedulePeriod": [{
                        "startPeriod": 0,
                        "limit": int(power_limit_kw * 1000)  # Convert kW to W
                    }]
                }
            }
        }
        r = requests.post(f"{CITRINEOS_HOST}/api/v1/charge-points/{charge_pointId}/charging-profiles",
            json=payload, timeout=10)
        logger.info(f"Charging profile sent to {charge_point_id}: {power_limit_kw} kW -> {r.status_code}")
        return r.status_code == 200
    except Exception as e:
        logger.error(f"Send charging profile: {e}")
        return False


def process_fm_schedules():
    """Read FM schedules and send to CitrineOS charge points."""
    for cp_id, sensor_id in CHARGEPOINT_SENSOR_MAP.items():
        try:
            # Get latest schedule from FM
            r = fm_session.get(f"{FM_HOST}/api/v3_0/sensors/{sensor_id}/data?limit=1", timeout=10)
            if r.status_code == 200:
                data = r.json()
                if data and len(data) > 0:
                    latest = data[-1]
                    power = float(latest.get("event_value", 0))
                    if power > 0:
                        send_charging_profile(cp_id, power)
        except Exception as e:
            logger.error(f"FM schedule for {cp_id}: {e}")


async def main():
    logger.info("Cariflex CitrineOS-FM Integration starting")
    logger.info(f"FM: {FM_HOST}")
    logger.info(f"CitrineOS: {CITRINEOS_HOST}")
    logger.info(f"Charge points: {len(CHARGEPOINT_SENSOR_MAP)}")
    
    fm_login()
    
    while True:
        try:
            # 1. Get charging sessions from CitrineOS
            sessions = get_citrineos_sessions()
            for session in sessions:
                if session.get("status") == "Active":
                    process_charging_session(session)
            
            # 2. Send FM schedules to CitrineOS
            process_fm_schedules()
            
        except Exception as e:
            logger.error(f"Integration error: {e}")
            fm_login()
        
        await asyncio.sleep(30)


if __name__ == "__main__":
    asyncio.run(main())