Easee Local OCPP Configuration Tool

easee_configure_local_ocpp.py

#!/bin/python3
"""
Easee Local OCPP Configuration Tool
This script allows you to enable or disable the local OCPP feature on your Easee charger.
It also allows you to configure the URL and charge point ID for the local OCPP server.
Usage:
    python easee_configure_local_ocpp.py <user_name> <password>

Details: https://github.com/easee/connect/discussions/2
"""

# MIT License
#
# Copyright (c) 2025 Andreas Merkle ([email protected])
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

import sys
import os
import json
import time
from typing import Optional
import requests

EASEE_API_BASE_URL = "https://api.easee.com/api"
EASEE_LOCAL_OCPP_BASE_URL = "https://api.easee.com/local-ocpp"
INDENT_LEVEL = 4

# pylint: disable=too-few-public-methods
class AccessToken:
    """
    Represents an access token.
    """

    TOKEN_FILE_NAME = "access_token.json"

    def __init__(self, user_name: str, password: str) -> None:
        """
        Initialize the access token instance.

        Args:
            user_name (str): The username for the Easee account.
            password (str): The password for the Easee account.
        """
        self.user_name: str = user_name
        self.password: str = password
        self.access_token: Optional[str] = None

    def get_access_token(self) -> Optional[str]:
        """
        Get the access token. If the access token is not available or expired,
        it will be requested from the Easee API.

        Raises:
            requests.exceptions.HTTPError: If the request to the Easee API fails.

        Returns:
            Optional[str]: The access token string.
        """
        if self.access_token is None:
            self.access_token = self._load_access_token_from_file()

        if (self.access_token is None) or (self._is_access_token_valid() is False):
            self.access_token = self._request_access_token()

            if self.access_token is not None:
                # Save the access token to a file for later use.
                self._save_access_token(self.access_token)

        return self.access_token

    def _is_access_token_valid(self) -> bool:
        """
        Check if the access token is valid.

        Returns:
            bool: True if the access token is valid, False otherwise.
        """
        is_valid = False

        if self.access_token is not None:
            # Check if the access token file exists and is not older than 1 hour.
            file_age = time.time() - os.path.getmtime(AccessToken.TOKEN_FILE_NAME)
            if file_age < 3600:
                is_valid = True

        return is_valid

    def _load_access_token_from_file(self) -> Optional[str]:
        """
        Get the access token from the file.

        Returns:
            Optional[str]: The access token if it exists, None otherwise.
        """
        access_token = None

        # Check if the token file exists.
        if os.path.exists(AccessToken.TOKEN_FILE_NAME):
            with open(AccessToken.TOKEN_FILE_NAME, "r", encoding="utf-8") as fd:
                data = json.load(fd)
                access_token = data.get("accessToken")

        return access_token

    def _save_access_token(self, access_token: str) -> None:
        """
        Save the access token to a file.

        Args:
            access_token (str): The access token to save.
        """
        with open(AccessToken.TOKEN_FILE_NAME, "w", encoding="utf-8") as fd:
            json.dump({"accessToken": access_token}, fd)

    def _request_access_token(self) -> str:
        """
        Request a new access token from the Easee API.

        Raises:
            requests.exceptions.HTTPError: If the request to the Easee API fails.

        Returns:
            str: The access token.
        """
        login_url = f"{EASEE_API_BASE_URL}/accounts/login"
        headers = {
            "accept": "application/json",
            "content-type": "application/*+json"
        }
        data = {
            "userName": self.user_name,
            "password": self.password
        }

        print("Requesting access token...")
        response = requests.post(login_url, headers=headers, json=data, timeout=10)
        response.raise_for_status()
        access_token = response.json().get("accessToken")

        return access_token

# pylint: disable=too-many-instance-attributes
class Charger:
    """
    Represents a charger with its details.
    """
    # pylint: disable=too-many-arguments, too-many-positional-arguments
    def __init__(self,
                 access_token: AccessToken,
                 charger_id: str,
                 name: str,
                 backplate_id: str,
                 master_backplate_id: str) -> None:
        """
        Initialize the charger instance.

        Args:
            access_token (AccessToken): The access token instance to authenticate the request.
            charger_id (str): The unique identifier of the charger.
            name (str): The name of the charger.
            backplate_id (str): The ID of the backplate associated with the charger.
            master_backplate_id (str): The ID of the master backplate if applicable.
        """
        self._access_token: AccessToken = access_token
        self.charger_id: str = charger_id
        self.name: str = name
        self.backplate_id: str = backplate_id
        self.master_backplate_id: str = master_backplate_id

        self.url: Optional[str] = None
        self.charge_point_id: Optional[str] = None
        self.connectivity_mode: Optional[str] = None

        self.operator: str = ""

        self.is_basic_charge_plan_enabled: bool = False
        self.is_weekly_charge_plan_enabled: bool = False

    def _update_local_ocpp_configuration(self) -> None:
        """
        Update the charger details by requesting the configuration details from the Easee API.
        """
        try:
            configuration_details_json = self._request_connection_details()

            self.url = configuration_details_json.get("websocketConnectionArgs", {}).get("url")
            self.charge_point_id = self.url.split("/")[-1] if self.url else None
            self.url = self.url[:-(len(self.charge_point_id) + 1)]

            # pylint: disable=line-too-long
            self.connectivity_mode = configuration_details_json.get("connectivityMode", "OcppOff")

        except requests.exceptions.HTTPError:
            self.url = None
            self.charge_point_id = None
            self.connectivity_mode = None

    def update(self) -> None:
        """
        Update the charger details by requesting the configuration details from the Easee API.
        """
        self._update_local_ocpp_configuration()

        try:
            operator_json = self._request_operator()
            self.operator = operator_json.get("name", "")

        except requests.exceptions.HTTPError:
            self.operator = ""

        try:
            basic_charge_plan_json = self._request_basic_charge_plan()
            self.is_basic_charge_plan_enabled = basic_charge_plan_json.get("isEnabled", False)

        except requests.exceptions.HTTPError:
            self.is_basic_charge_plan_enabled = False

        try:
            weekly_charge_plan_json = self._request_weekly_charge_plan()
            self.is_weekly_charge_plan_enabled = weekly_charge_plan_json.get("isEnabled", False)

        except requests.exceptions.HTTPError:
            self.is_weekly_charge_plan_enabled = False

    def _request_connection_details(self) -> dict:
        """
        Request the connection details of the charger.

        Raises:
            requests.exceptions.HTTPError: If the request to the Easee API fails.

        Returns:
            dict: The JSON response containing the connection details of the charger.
        """
        url = f"{EASEE_LOCAL_OCPP_BASE_URL}/v1/connection-details/{self.charger_id}"
        headers = {
            "accept": "application/json",
            "authorization": f"Bearer {self._access_token.get_access_token()}"
        }

        print(f"Requesting connection details for charger {self.name}...")
        response = requests.get(url, headers=headers, timeout=10)
        response.raise_for_status()

        return response.json()

    def _request_operator(self) -> dict:
        """
        Request the operator details of the charger.

        Raises:
            requests.exceptions.HTTPError: If the request to the Easee API fails.

        Returns:
            dict: The JSON response containing the operator details of the charger.
        """
        url = f"{EASEE_API_BASE_URL}/chargers/{self.charger_id}/partners"
        headers = {
            "Accept": "application/json",
            "Authorization": f"Bearer {self._access_token.get_access_token()}"
        }

        print(f"Requesting operator for charger {self.name}...")
        response = requests.get(url, headers=headers, timeout=10)
        response.raise_for_status()

        return response.json()

    def _request_basic_charge_plan(self) -> dict:
        """
        Request the basic charge plan details of the charger.

        Raises:
            requests.exceptions.HTTPError: If the request to the Easee API fails.

        Returns:
            dict: The JSON response containing the basic charge plan details of the charger.
        """
        url = f"{EASEE_API_BASE_URL}/chargers/{self.charger_id}/basic_charge_plan"
        headers = {
            "Accept": "application/json",
            "Authorization": f"Bearer {self._access_token.get_access_token()}"
        }

        print(f"Requesting basic charge plan for charger {self.name}...")
        response = requests.get(url, headers=headers, timeout=10)
        response.raise_for_status()

        return response.json()

    def _request_weekly_charge_plan(self) -> dict:
        """
        Request the weekly charge plan details of the charger.

        Raises:
            requests.exceptions.HTTPError: If the request to the Easee API fails.

        Returns:
            dict: The JSON response containing the weekly charge plan details of the charger.
        """
        url = f"{EASEE_API_BASE_URL}/chargers/{self.charger_id}/weekly_charge_plan"
        headers = {
            "Accept": "application/json",
            "Authorization": f"Bearer {self._access_token.get_access_token()}"
        }

        print(f"Requesting weekly charge plan for charger {self.name}...")
        response = requests.get(url, headers=headers, timeout=10)
        response.raise_for_status()

        return response.json()

    def _update_connection_details(self,
                                      connectivity_mode: str,
                                      local_ocpp_url: str,
                                      charge_point_id: Optional[str]) -> dict:
        """
        Update the connection details of the charger.

        Args:
            connectivity_mode (str): The connectivity mode to set for the charger.
            local_ocpp_url (str): The local OCPP URL to send the update request to.
            charge_point_id (str): The charge point ID of the charger.

        Raises:
            requests.exceptions.HTTPError: If the request to the Easee API fails.

        Returns:
            dict: The JSON response containing the updated connection details of the charger.
        """
        url = f"{EASEE_LOCAL_OCPP_BASE_URL}/v1/connection-details/{self.charger_id}"
        headers = {
            "Accept": "application/json",
            "Authorization": f"Bearer {self._access_token.get_access_token()}"
        }
        data = {
            "connectivityMode": connectivity_mode,
            "websocketConnectionArgs": {
                "url": local_ocpp_url
            },
            "chargePointId": charge_point_id
        }

        print(f"Updating connection details for charger {self.name}...")
        response = requests.post(url, headers=headers, json=data, timeout=10)
        response.raise_for_status()

        return response.json()

    def _apply_configuration_local_ocpp(self, version: str) -> dict:
        """
        Apply the local OCPP configuration to the charger.

        Args:
            version (str): The OCPP version to apply.
        
        Raises:
            requests.exceptions.HTTPError: If the request to the Easee API fails.

        Returns:
            dict: The JSON response containing the applied configuration details of the charger.
        """
        url = f"{EASEE_LOCAL_OCPP_BASE_URL}/v1/connections/chargers/{self.charger_id}"
        headers = {
            "accept": "application/json",
            "authorization": f"Bearer {self._access_token.get_access_token()}"
        }
        data = {
            "version": version
        }

        print(f"Applying local OCPP configuration for charger {self.name}...")
        response = requests.post(url, headers=headers, json=data, timeout=10)
        response.raise_for_status()

        return response.json()

    def is_local_ocpp_enabled(self) -> bool:
        """
        Check if local OCPP is enabled for the charger.

        Returns:
            bool: True if local OCPP is enabled, False otherwise.
        """
        is_enabled = False

        if self.connectivity_mode is not None:
            if self.connectivity_mode == "DualProtocol":
                is_enabled = True

        return is_enabled

    def is_master_charger(self) -> bool:
        """
        Check if the charger is a master charger.

        Returns:
            bool: True if the charger is a master charger, False otherwise.
        """
        is_master = False

        if self.backplate_id is not None and self.master_backplate_id is not None:
            if self.backplate_id == self.master_backplate_id:
                is_master = True

        return is_master

    def enable_local_ocpp(self) -> None:
        """
        Enable local OCPP for the charger by requesting the connection details from the Easee API.

        Raises:
            requests.exceptions.HTTPError: If the request to the Easee API fails.
        """
        charge_point_id = self.charge_point_id
        if charge_point_id is not None:
            if charge_point_id == "":
                charge_point_id = None

        connection_details_json = self._update_connection_details(
            "DualProtocol",
            self.url,
            charge_point_id
        )

        version = connection_details_json.get("version")
        self._apply_configuration_local_ocpp(version)
        self._update_local_ocpp_configuration()

    def disable_local_ocpp(self) -> None:
        """
        Disable local OCPP for the charger by requesting the connection details from the Easee API.

        Raises:
            requests.exceptions.HTTPError: If the request to the Easee API fails.
        """
        # If the local OCPP URL is not provided, use the default URL.
        # Otherwise the request will fail.
        local_ocpp_url = self.url
        if local_ocpp_url is None:
            local_ocpp_url = "ws://127.0.0.1:9000"

        connection_details_json = self._update_connection_details(
            "OcppOff",
            local_ocpp_url,
            self.charge_point_id
        )

        version = connection_details_json.get("version")
        self._apply_configuration_local_ocpp(version)
        self._update_local_ocpp_configuration()

def _request_chargers(access_token: AccessToken) -> dict:
    """
    Request the list of chargers from the Easee API.

    Args:
        access_token (AccessToken): The access token instance to authenticate the request.

    Raises:
        requests.exceptions.HTTPError: If the request to the Easee API fails.

    Returns:
        dict: The JSON response containing the list of chargers.
    """
    url = f"{EASEE_API_BASE_URL}/accounts/chargers"
    headers = {
        "Accept": "application/json",
        "Authorization": f"Bearer {access_token.get_access_token()}"
    }

    print("Requesting chargers...")
    response = requests.get(url, headers=headers, timeout=10)
    response.raise_for_status()

    return response.json()

def _show_charger_info(charger: Charger, indent: int) -> None:
    """
    Display the information of the selected charger.

    Args:
        charger (Charger): The charger instance to display information for.
        indent (int): The indentation level for the output.
    """
    label_width = 32
    print(f"{' ' * indent}{'Charger Name'.ljust(label_width)}: {charger.name}")
    print(f"{' ' * indent}{'Charger ID'.ljust(label_width)}: {charger.charger_id}")
    print(f"{' ' * indent}{'Local OCPP'.ljust(label_width)}: "
          f"{'Enabled' if charger.is_local_ocpp_enabled() else 'Disabled'}")
    print(f"{' ' * indent}{'Is Master Charger'.ljust(label_width)}: "
          f"{'Yes' if charger.is_master_charger() else 'No'}")
    print(f"{' ' * indent}{'Local OCPP URL'.ljust(label_width)}: "
          f"{charger.url if charger.url else 'Not set'}")
    print(f"{' ' * indent}{'Local OCPP Charge Point ID'.ljust(label_width)}: "
          f"{charger.charge_point_id if charger.charge_point_id else 'Not set'}")
    print(f"{' ' * indent}{'Operator'.ljust(label_width)}: {charger.operator}")
    print(f"{' ' * indent}{'Basic Charge Plan Enabled'.ljust(label_width)}: "
          f"{charger.is_basic_charge_plan_enabled}")
    print(f"{' ' * indent}{'Weekly Charge Plan Enabled'.ljust(label_width)}: "
          f"{charger.is_weekly_charge_plan_enabled}")

def _user_select_charger(chargers: list) -> Optional[Charger]:
    """
    Prompt the user to select a charger from the list of available chargers.

    Args:
        chargers (list): A list of Charger instances to choose from.

    Returns:
        Optional[Charger]: The selected Charger instance or None if the user chooses to exit.
    """
    print("Select a charger:")

    for idx, charger in enumerate(chargers):
        print(f"{idx}: {charger.name}")
        _show_charger_info(charger, INDENT_LEVEL)

    print("e: Exit")

    operator_name = "Easee"
    charger = None
    while True:
        try:
            user_input = input("Enter the number of the charger: ").strip()

            # Exit requested?
            if 'e' == user_input.lower():
                print("Exiting...")
                break

            choice = int(user_input)

            if 0 <= choice < len(chargers):
                charger = chargers[choice]
            else:
                print("Invalid choice. Please try again.")

            if charger is not None:
                if charger.operator != operator_name:
                    print(f"The selected charger is not operated by {operator_name}."
                          " Please select another charger or exit.")
                else:
                    break

        except ValueError:
            print("Invalid input. Please enter a number.")

    return charger

def _user_input_url(charger: Charger) -> None:
    """
    Prompt the user to enter a URL for the charger.
    If the charger already has a URL, the user can choose to keep it or enter a new one.

    Args:
        charger (Charger): The charger instance to configure.
    """
    url = charger.url
    if url is not None:
        print(f"Current URL: {url}")
        url = input("Enter the new URL or press Enter to keep current: ").strip()

        if url != "":
            charger.url = url
    else:
        url = input("Enter the URL (e.g., ws://localhost:9000): ").strip()
        charger.url = url

def _user_select_charge_point_id(charger: Charger) -> None:
    """
    Prompt the user to select or enter a charge point ID for the charger.

    Args:
        charger (Charger): The charger instance to configure.
    """
    charge_point_id = charger.charge_point_id
    if charge_point_id is not None:
        print(f"Current Charge Point ID: {charge_point_id}")
        charge_point_id = input("Enter the new Charge Point ID"
                                " or press Enter to keep current"
                                " or - to use the charger id: ").strip()

        # Does the user want to use the charger id as charge point id?
        if charge_point_id == "-":
            charger.charge_point_id = ""

        # If the user entered a new charge point id, update it.
        elif charge_point_id != "":
            charger.charge_point_id = charge_point_id
    else:
        charge_point_id = input("Enter the Charge Point ID (e.g., my-charge-point): ").strip()
        charger.charge_point_id = charge_point_id

def _user_select_action(charger: Charger) -> None:
    """
    Prompt the user to select an action for the charger.

    Args:
        charger (Charger): The charger instance to configure.
    """
    print("Select an action:")
    if charger.is_local_ocpp_enabled():
        print("1: Change local OCPP URL/Charge Point ID")
        print("2: Disable Local OCPP")
    else:
        print("1: Enable Local OCPP")

    print("b: Back")

    while True:
        user_input = input("Enter the number of the action: ").strip()

        if 'b' == user_input.lower():
            print("Back...")
            break

        if user_input == "1":
            _user_input_url(charger)
            _user_select_charge_point_id(charger)
            print("Enabling local OCPP...")
            charger.enable_local_ocpp()
            break

        if (user_input == "2") and (charger.is_local_ocpp_enabled() is True):
            print("Disabling local OCPP...")
            charger.disable_local_ocpp()
            break

        print("Invalid choice. Please try again.")

def _run(user_name: str, password: str) -> int:
    """
    Run the Easee Local OCPP Configuration Tool.

    Args:
        user_name (str): The username for the Easee account.
        password (str): The password for the Easee account.

    Returns:
        int: The exit status code (0 for success, non-zero for failure).
    """
    status = 0
    access_token = AccessToken(user_name, password)
    chargers = []
    chargers_json = _request_chargers(access_token)

    for site_json in chargers_json:
        for circuit_json in site_json.get("circuits", []):
            for charger_json in circuit_json.get("chargers", []):
                backplate = charger_json.get("backPlate", {})

                charger = Charger(access_token,
                                  charger_json.get("id"),
                                  charger_json.get("name"),
                                  backplate.get("id"),
                                  backplate.get("masterBackPlateId"))

                charger.update()

                chargers.append(charger)

    if not chargers:
        print("No chargers found. Please check your credentials or network connection.")
        status = 1
    else:
        while True:
            charger = _user_select_charger(chargers)

            if charger is not None:
                _user_select_action(charger)
            else:
                break

    return status

def _main():
    """
    Main function to run the Easee Local OCPP Configuration Tool.
    """
    status = 0
    if len(sys.argv) < 3:
        print("Usage: easee_configure_local_ocpp.py <user_name> <password>")
        status = 1
    else:
        user_name = sys.argv[1]
        password = sys.argv[2]
        status = _run(user_name, password)

    return status

if __name__ == "__main__":
    sys.exit(_main())