SoC Fisker OCEAN

[danmacfly]

Hallo zusammen,

ist es bereits irgendwie möglich, oder angedacht, dass der SoC des Fisker OCEAN's implementiert wird? Ich weiß, bin da sicherlich ein Exote zurzeit, mit der Kombi des Wagens und der openWB, aber ich habe mich echt sehr an das vollintegrierte Verwöhnprogramm von Euch gewöhnt. Auch die tibber-Integration ist genial umgesetzt.

Falls da was in der pipeline ist, wäre es toll. Falls ich irgendetwas dazu beisteuern kann, bitte einfach melden.

Vielen Dank und beste Grüße

[Michael_F]

Ein Bekannter von mir interessiert sich für die openWB. Allerdings wäre für ihn die SoC Abfrage sehr wichtig.
Habe mal etwas recherchiert zu den Schnittstellen/API. Habe ein Projekt realsiert für Home Assistant gefunden, 54 Sensorwerte können ausgelesen werden:

https://github.com/MichaelOE/home-assistant-MyFisker

TOKEN_URL = "https://auth.fiskerdps.com/auth/login"
WSS_URL_EU = "wss://gw.cec-euprd.fiskerinc.com/mobile"

Habe leider die notwendigen Programmierkenntnisse nicht. Vielleicht kennt ja jemand jemanden der sich damit spielen könnte.

[derNeueDet]

Alternative: Homeassistant betreiben und den SoC von dort per MQTT in de openWB leifern. Mache ich mit Mercedes EQ so.

VG
Det

[Michael_F]

Der Bekannte ist computertechnisch eher ein Laie. Er hätte gerne eine integrierte "no-brainer" Lösung.

[knichiknaxl]

Guten Abend
Ich besitze einen Fisker und habe mich mal etwas beschäftigt mit der Entwicklungsumgebung von openWB 2. Die läuft nun soweit auf einer VM/Debian 11. Habe mich auch eingelesen in die SoC Module der 2er Version (Templates und primär Polestar, BMW). Muss dazu sagen dass ich in python und Github keine Erfahrungen habe. Auch mit der Tokenauthentifizierung kenne ich mich nicht gut aus.

Leider bekomme ich den Fehler "ModuleNotFoundError" beim ausführen von "sudo phyton3 soc.py" aber auch von api.py
Habe die Pfade mehrmals kontrolliert, die scheinen zu passen (wenn ich nicht komplett auf dem Holzweg bin).

Unten das was ich versucht habe anzupassen und per FTP auf die Entwicklungsumgebung draufgespielt habe. Eine leere __init__.py habe ich auch im Verzeichnis /fisker/ erstellt.

Hier die Doku zu der API-Schnittstelle: https://www.fiskerapi.io/fisker-api

config.py

Code: Alles auswählen

from typing import Optional
class FiskerOceanConfiguration:
    def __init__(self, user_id: Optional[str] = None, password: Optional[str] = None, vin: Optional[str] = None):
        self.user_id = user_id
        self.password = password
        self.vin = vin

class FiskerOcean:
    def __init__(self,
                 name: str = "FiskerOcean",
                 type: str = "fisker",
                 configuration: FiskerOceanConfiguration = None) -> None:
        self.name = name
        self.type = type
        self.configuration = configuration or FiskerOceanConfiguration()

soc.py

Code: Alles auswählen

from typing import List

import logging

from helpermodules.cli import run_using_positional_cli_args
from modules.common import store
from modules.common.abstract_device import DeviceDescriptor
from modules.common.abstract_vehicle import VehicleUpdateData
from modules.common.component_state import CarState
from modules.common.configurable_vehicle import ConfigurableVehicle
from modules.vehicles.fisker import api
from modules.vehicles.fisker.config import FiskerOcean, FiskerOceanConfiguration

log = logging.getLogger(__name__)


def create_vehicle(vehicle_config: FiskerOcean, vehicle: int):
    def updater(vehicle_update_data: VehicleUpdateData) -> CarState:
        return api.fetch_soc(
            vehicle_config.configuration.user_id,
            vehicle_config.configuration.password,
            vehicle_config.configuration.vin,
            vehicle)
    return ConfigurableVehicle(vehicle_config=vehicle_config, component_updater=updater, vehicle=vehicle)

def FiskerOcean_update(user_id: str, password: str, vin: str, charge_point: int):
    log.debug("FiskerOcean: user_id="+user_id+"vin="+vin+"charge_point="+str(charge_point))
    vehicle_config = FiskerOcean(configuration=FiskerOceanConfiguration(user_id, password, vin))
    store.get_car_value_store(charge_point).store.set(api.fetch_soc(
         vehicle_config.configuration.user_id,
         vehicle_config.configuration.password,
         vehicle_config.configuration.vin,
         charge_point))

def main(argv: List[str]):
    run_using_positional_cli_args(FiskerOcean_update, argv)

device_descriptor = DeviceDescriptor(configuration_factory=FiskerOcean)

api.py

Code: Alles auswählen

import logging
from modules.common import req
import time
from modules.vehicles.fisker.auth import FiskerAuth
from modules.common.component_state import CarState


log = logging.getLogger(__name__)


class FiskerApi:

    def __init__(self, username: str, password: str, vin: str) -> None:
        self.auth = FiskerAuth(username, password, vin)
        self.vin = vin
        self.client_session = req.get_http_session()

    def query_params(self, params: dict, url='https://auth.fiskerdps.com/auth/login') -> dict or None:
        access_token = self.auth.get_auth_token()
        if access_token is None:
            raise Exception("query_params error:could not get auth token")

        headers = {
            "Content-Type": "application/json",
            "authorization": f"Bearer {self.auth.access_token}"
        }

        log.info("query_params:%s", params['query'])
        try:
            result = self.client_session.get(url=url, params=params, headers=headers)
        except Exception as e:
            if result.status_code == 401:
                self.auth.delete_token()
            if self.auth.access_token is not None:
                # if we got an access code but the query failed, VIN could be wrong, so let`s check it
                self.check_vin()
            raise e

        result_data = result.json()
        if result_data.get('errors'):
            error_message = result_data['errors'][0]['message']
            raise Exception("query_params error: %s", error_message)

        return result_data

    def get_battery_data(self) -> dict or None:
        params = {
            "query": "query GetBatteryData($vin: String!) { getBatteryData(vin: $vin) { "
            + "batteryChargeLevelPercentage  estimatedDistanceToEmptyKm } }",
            "operationName": "GetBatteryData",
            "variables": "{\"vin\":\"" + self.vin + "\"}"
        }

        result = self.query_params(params)

        return result['data']['getBatteryData']

    def check_vin(self) -> None:
        # get Vehicle Data
        params = {
            "query": "query GetConsumerCarsV2 { getConsumerCarsV2 { vin internalVehicleIdentifier __typename }}",
            "operationName": "GetConsumerCarsV2",
            "variables": "{}"
        }
        result = self.query_params(params, url='https://api.fiskerdps.com/api/v2/users/me')
        if result is not None and result['data'] is not None:
            vins = []
            # get list of cars and store the ones not matching our vin
            cars = result['data']['getConsumerCarsV2']
            if len(cars) == 0:
                raise Exception("Es konnten keine Fahrzeuge im Account gefunden werden. Bitte in den Einstellungen " +
                                "prüfen, ob der Besitzer-Account des Fisker Ocean eingetragen ist.")

            for i in range(0, len(cars)):
                if cars[i]['vin'] == self.vin:
                    pass
                else:
                    vins.append(cars[i]['vin'])
            if len(vins) > 0:
                raise Exception("You probably specified a wrong VIN. We only found:%s", ",".join(vins))


def fetch_soc(user_id: str, password: str, vin: str, vehicle: int) -> CarState:
    api = FiskerApi(user_id, password, vin)
    bat_data = api.get_battery_data()
    soc = bat_data['batteryChargeLevelPercentage']
    est_range = bat_data['estimatedDistanceToEmptyKm']

    return CarState(soc, est_range, time.strftime("%m/%d/%Y, %H:%M:%S"))

auth.py (nur tlw. angepasst)

Code: Alles auswählen

import logging
import json
import requests
import os
import re
from datetime import datetime, timedelta
from modules.common.store import RAMDISK_PATH

log = logging.getLogger(__name__)


class FiskerAuth:
    """ base class for Fisker authentication"""

    def __init__(self, username: str, password: str, vin: str) -> None:
        self.username = username
        self.password = password
        self.vin = vin
        self.access_token = None
        self.refresh_token = None
        self.token_expiry = None
        self.client_session = requests.session()
        self.resume_path = None
        self.code = None
        self.token = None
        self.token_file = str(RAMDISK_PATH)+'/fiskerocean_token_'+vin+'.json'
        self.token_store: dict = {}

    def delete_token(self) -> None:
        self.access_token = None
        self.refresh_token = None
        self.token_expiry = None
        # remove from ramdisk
        if os.path.exists(self.token_file):
            try:
                os.remove(self.token_file)
            except IOError:
                log.error("delete_token:error deleting token store %s", self.token_file)

    def _load_token_from_ramdisk(self) -> None:
        if os.path.exists(self.token_file):
            log.info("_load_token_from_ramdisk:loading from file %s", self.token_file)
            self.token_store = {}
            with open(self.token_file, "r") as tf:
                try:
                    self.token_store = json.load(tf)
                    self.access_token = self.token_store['access_token']
                    self.refresh_token = self.token_store['refresh_token']
                    self.token_expiry = datetime.strptime(self.token_store['token_expiry'], "%d.%m.%Y %H:%M:%S")
                except json.JSONDecodeError as e:
                    log.error("_load_token_from_ramdisk:error loading token store %s:%s",
                              self.token_file, e)

            if 'access_token' not in self.token_store:
                self.access_token = None
            if 'refresh_token' not in self.token_store:
                self.refresh_token = None
            if 'token_expiry' not in self.token_store:
                self.token_expiry = None

    def _save_token_to_ramdisk(self) -> None:
        try:
            tf = open(self.token_file, mode='w', encoding='utf-8')
        except IOError as e:
            log.error("_save_token_to_ramdisk:error saving token store %s:%s", self.token_file, e)
            return
        try:
            json.dump(self.token_store, tf, ensure_ascii=False, indent=4)
        except json.JSONDecodeError as e:
            log.error("_save_token_to_ramdisk:error saving token store %s:%s", self.token_file, e)

    # auth step 3: get token
    def get_auth_token(self) -> str or None:
        # first try to load token from ramdisk
        self._load_token_from_ramdisk()

        if self.token_expiry is not None and self.token_expiry > datetime.now():
            log.info("get_auth_token:using token from file %s", self.token_file)
            return self.access_token
        else:
            log.info("get_auth_token:token from file %s expired. New authentication required", self.token_file)

        code = self._get_auth_code()
        if code is None:
            return None

        # get token
        params = {
            "query": "query getAuthToken($code: String) { getAuthToken(code: $code) { id_token access_token \
            refresh_token expires_in }}",
            "operationName": "getAuthToken",
            "variables": json.dumps({"code": code})
        }

        headers = {
            "Content-Type": "application/json"
        }
        log.info("get_auth_token:attempting to get new token")
        try:
            result = self.client_session.get("https://auth.fiskerdps.com/auth/login",
                                             params=params, headers=headers)
        except requests.RequestException as e:
            log.error("get_auth_token:http error:%s", e)
            return None
        if result.status_code != 200:
            log.error("get_auth_token:error:get response:%d", result.status_code)
            return None

        result_data = result.json()
        log.info(result_data)

        if result_data['data']['getAuthToken'] is not None:
            self.access_token = result_data['data']['getAuthToken']['access_token']
            self.refresh_token = result_data['data']['getAuthToken']['refresh_token']
            self.token_expiry = datetime.now(
            ) + timedelta(seconds=result_data['data']['getAuthToken']['expires_in'])
            # save tokens to ramdisk
            self.token_store['access_token'] = self.access_token
            self.token_store['refresh_token'] = self.refresh_token
            self.token_store['token_expiry'] = self.token_expiry.strftime("%d.%m.%Y %H:%M:%S")
            self._save_token_to_ramdisk()
        else:
            log.error("get_auth_token:error getting token:no valid data in http response")
            return None

        log.info("get_auth_token:got token:%s", self.access_token)
        return self.access_token

    # auth step 2: get code
    def _get_auth_code(self) -> str or None:
        self.resume_path = self._get_auth_resumePath()
        if self.resume_path is None:
            return None

        params = {
            'client_id': 'polmystar'
        }
        data = {
            'pf.username': self.username,
            'pf.pass': self.password
        }

        log.info("_get_auth_code:attempting to get new code")
        try:
            result = self.client_session.post(
                f"https://polestarid.eu.polestar.com/as/{self.resume_path}/resume/as/authorization.ping",
                params=params,
                data=data
            )
        except requests.RequestException as e:
            log.error("_get_auth_code:http error:%s", e)
            return None
        if result.status_code != 200:
            log.error("_get_auth_code:error getting auth code: post response:%d", result.status_code)
            return None
        if re.search(r"ERR", result.request.path_url, flags=re.IGNORECASE) is not None:
            log.error("_get_auth_code:error:check username/password")
            return None
        # get code
        m = re.search(r"code=(.+)", result.request.path_url)
        if m is not None:
            code = m.group(1)
            log.info("_get_auth_code:got code %s", code)
        else:
            code = None
            log.info("_get_auth_code:error getting auth code")

        return code

    # auth step 1: get resumePath
    def _get_auth_resumePath(self) -> str or None:
        # Get Resume Path
        params = {
            "response_type": "code",
            "client_id": "polmystar",
            "redirect_uri": "https://www.polestar.com/sign-in-callback"
        }
        log.info("_get_auth_resumePath:attempting to get resumePath")
        try:
            result = self.client_session.get("https://polestarid.eu.polestar.com/as/authorization.oauth2",
                                             params=params)
        except requests.RequestException as e:
            log.error("_get_auth_resumePath:http error:%s", e)
            return None
        if result.status_code != 200:
            log.error("_get_auth_resumePath:get response:%d", result.status_code)
            return None
        m = re.search(r"resumePath=([^&]+)", result.url)
        if m is not None:
            resume_path = m.group(1)
            log.info("_get_auth_resumePath:got resumePath %s", resume_path)
        else:
            resume_path = None
            log.info("_get_auth_resumePath:error getting resumePath")

        return resume_path

Irgendwie stehe ich an und komm nicht weiter. Kann mir jemand hier weiterhelfen?

[LenaK]

Es ist nicht vorgesehen, die Module isoliert vom restlichen Programm auszuführen. Du musst die main.py ausführen und dann im UI das Fisker-Modul konfigurieren.

[knichiknaxl]

Danke für den Hinweis. Ich komm nicht ganz klar.
Bekomme hier auch einen ModuleNotFoundError.

Code: Alles auswählen

pi@debian11:/var/www/html/openWB/packages$ python3 main.py
Traceback (most recent call last):
  File "/var/www/html/openWB/packages/main.py", line 7, in <module>
    import schedule
ModuleNotFoundError: No module named 'schedule'

Was muss ich ausserdem machen um im UI das Fisker-Modul zu konfigurieren?

[LenaK]

Läuft der system-Service? sudo systemctl status openwb2
Dann diesen entweder neu starten sudo systemctl restart openwb2
oder stoppen sudo systemctl stop openwb2 und mit ./runs/atreboot.sh und main.py händisch die main starten.

Bei dir sind die Python-Packages nicht installiert. Das sollte beim Starten passieren. Bitte nach dem Start ins main.log schauen.

[knichiknaxl]

Läuft der system-Service? sudo systemctl status openwb2
Dann diesen entweder neu starten sudo systemctl restart openwb2
oder stoppen sudo systemctl stop openwb2 und mit ./runs/atreboot.sh und main.py händisch die main starten.

Ja läuft:

Code: Alles auswählen

pi@debian11:/var/www/html/openWB/packages$ sudo systemctl status openwb2
[sudo] Passwort für pi:
● openwb2.service - "Regelung openWB 2.0"
     Loaded: loaded (/var/www/html/openWB/data/config/openwb2.service; enabled;>
     Active: active (running) since Thu 2024-06-27 20:31:14 CEST; 16h ago
    Process: 451 ExecStartPre=/var/www/html/openWB/runs/atreboot.sh (code=exite>
   Main PID: 1179 (python3)
      Tasks: 9 (limit: 4644)
     Memory: 233.9M
        CPU: 4min 3.978s
     CGroup: /system.slice/openwb2.service
             └─1179 python3 /var/www/html/openWB/packages/main.py

Restart und stop funktioniert, ./runs/atreboot.sh (?) und main.py funktionieren nicht:

Code: Alles auswählen

root@debian11:/var/www/html/openWB# ./runs/atreboot.sh
./runs/atreboot.sh: Zeile 423: /var/www/html/openWB/ramdisk/main.log: Keine Berechtigung
root@debian11:/var/www/html/openWB#

Code: Alles auswählen

root@debian11:/var/www/html/openWB/packages# python3 main.py
Traceback (most recent call last):
  File "/var/www/html/openWB/packages/main.py", line 7, in <module>
    import schedule
ModuleNotFoundError: No module named 'schedule'
root@debian11:/var/www/html/openWB/packages#

Bei dir sind die Python-Packages nicht installiert. Das sollte beim Starten passieren. Bitte nach dem Start ins main.log schauen.

Die Python-Packages dürften installiert sein:

Code: Alles auswählen

install required python packages with 'pip3'...
Requirement already satisfied: typing-extensions==4.4.0 in /home/openwb/.local/lib/python3.9/site-packages (from -r /var/www/html/openWB/requirements.txt (line 1)) (4.4.0)
Requirement already satisfied: jq==1.1.3 in /home/openwb/.local/lib/python3.9/site-packages (from -r /var/www/html/openWB/requirements.txt (line 2)) (1.1.3)
Requirement already satisfied: paho_mqtt==1.6.1 in /home/openwb/.local/lib/python3.9/site-packages (from -r /var/www/html/openWB/requirements.txt (line 3)) (1.6.1)
Requirement already satisfied: pymodbus==2.5.2 in /home/openwb/.local/lib/python3.9/site-packages (from -r /var/www/html/openWB/requirements.txt (line 4)) (2.5.2)
Requirement already satisfied: pytest==6.2.5 in /home/openwb/.local/lib/python3.9/site-packages (from -r /var/www/html/openWB/requirements.txt (line 5)) (6.2.5)
Requirement already satisfied: requests_mock==1.9.3 in /home/openwb/.local/lib/python3.9/site-packages (from -r /var/www/html/openWB/requirements.txt (line 6)) (1.9.3)
Requirement already satisfied: lxml==4.9.1 in /home/openwb/.local/lib/python3.9/site-packages (from -r /var/www/html/openWB/requirements.txt (line 7)) (4.9.1)
[.......]
Requirement already satisfied: exceptiongroup>=1.0.2 in /home/openwb/.local/lib/python3.9/site-packages (from anyio->httpx->bimmer_connected==0.15.1->-r /var/www/html/openWB/requirements.txt (line 25)) (1.2.1)
done

[derNeueDet]

Nicht als root ausführen, vermute dein User heißt auch openwb. Mit dem User musst du es auch probieren.

EDIT: Scheint, dass du die Software unter dem User pi installiert hast, dann halt als pi ausführen. Allerdings müsste man dann auch mal die Service Descriptions anschauen, ob die den richtigen User eingestellt haben


VG
Det