SoC Fisker OCEAN

Alles rund um SoC (Ladezustand des Fahrzeuges). Probleme, Fragen, Fehlfunktionen gehören hier hin
knichiknaxl
Beiträge: 10
Registriert: Do Jun 27, 2024 7:48 pm

Re: SoC Fisker OCEAN

Beitrag von knichiknaxl »

Bin am Wochenende über die Arbeit von ddraeyer (Zip-File mit VM/openWB Installation) gestoßen und habe die VM neu aufgesetzt. Tolles Image und funktioniert bisher einwandfrei. https://github.com/openWB/core/pull/168 ... 2187347232 Danke dafür!
Scheinbar war bei meiner Installation irgendetwas nicht ganz sauber.

Bin auch einen ganzen Schritt weiter. Das Fahrzeug ist im UI auswählbar und user_id, password und die vin können in der Konfiguration eingegeben und gespeichert werden.

Nun stehe ich bei der Authentifizierung an. Bekomme folgenden Fehler:

Code: Alles auswählen

  File "/var/www/html/openWB/packages/modules/vehicles/fisker/api.py", line 21, in query_params
    raise Exception("query_params error:could not get auth token")
Exception: query_params error:could not get auth token
Mit dem MQTT Explorer komme ich ebenfalls drauf und sehe folgende Fehlermeldung: "<class 'Exception'> ('query_params error:could not get auth token',)"

Kann mir hier jemand weiterhelfen? Die Auth-URL https://auth.fiskerdps.com/auth/login sollte passen. EDIT: Es gibt noch eine zweite URL: https://www.fiskerapi.io/fisker-api/authentication
Die Doku zu der API-Schnittstelle: https://www.fiskerapi.io/fisker-api
Dateianhänge
MQTT-vehicle-get_fault_str.png
MQTT-vehicle-get_fault_str.png (35.05 KiB) 1289 mal betrachtet
rleidner
Beiträge: 1001
Registriert: Mo Nov 02, 2020 9:50 am
Has thanked: 5 times
Been thanked: 38 times

Re: SoC Fisker OCEAN

Beitrag von rleidner »

Ich fürchte, bei dieser Problematik wird Dir hier wenig konkret geholfen werden können.
Ich habe schon einige SOC-Module gebaut und etwas Erfahrung damit.

Die referenzierte "API-Beschreibung" ist inoffiziell und unvollständig, d.h. für Deine Problemstellung nicht wirklich hilfreich.
Es scheint auch keine einfache oauth/rest Geschichte zu sein. Irgendwo werden auch "websockets" erwähnt, das ist eine Komplexität für sich.
Ich habe eine websockets-API im smarteq -Modul implementiert, ist z.Zt. noch PR und ich bin sehr unsicher, ob das vom openwb-Team akzeptiert werden wird.

Gibt es denn eine funktionierende Lösung für Fisker, die die Fahrzeugdaten in irgendein Home Automation System holt?
Falls ja, würde ich empfehlen, dort "abzuschreiben".

Wenn nicht wird Dir nur der mühsame Weg bleiben, die Lücken der API-Beschreibung selbst zu füllen.
Da wirst Du evtl. nicht darum herumkommen, erst mal den Traffic zwischen Fisker App und Server zu verstehen.
Wie das gehen kann ist z.B. für BMW hier beschreiben:
https://github.com/bimmerconnected/bimm ... _mybmw.rst

Noch ein Tipp:
Wenn ich mit einem SOC-Abruf für ein ein neues Fahrzeug beginne und kein funktionierendes Muster habe, baue ich erst mal ein einfaches python script, das den gesamten Abruf erfolgreich durchführt.
Erst wenn das sauber funktioniert, baue ich das in die openwb-Umgebung ein.

EDIT: Ich sehe gerade, dass es eine Lösung für HA gibt, also sogar in python.
Das ist doch ein guter Startpunkt.
https://github.com/MichaelOE/home-assistant-MyFisker
openWB-2 Standard+ | openWB EVU Kit v2 MID| 9,9kWp mit Kostal Plenticore 8.5 plus | VW ID.3, Kia EV6, Smart EQ forfour
knichiknaxl
Beiträge: 10
Registriert: Do Jun 27, 2024 7:48 pm

Re: SoC Fisker OCEAN

Beitrag von knichiknaxl »

Danke rleidner für deinen Input.

Genau, es gibt eine gute Lösung in HA. Auf diese hat Michael_F auch am 26.3. verwiesen. Habe diese nun angepasst und bin wieder ein Stück weiter mit der api.py

Nun gekomme ich folgende Fehlermeldung im main-log [EDIT]:

Code: Alles auswählen

EvData(set=Set(soc_error_counter=16), charge_template=0, ev_template=0, name='Standard-Fahrzeug', tag_id=[], get=Get(soc=0, soc_timestamp=1719915274.228868, force_soc_update=False, range=0, fault_state=2, fault_str='<class \'AttributeError\'> ("\'FiskerApi\' object has no attribute \'data_battery\'",)'))
Hier der relevante Code-Schnipsel dazu:

Code: Alles auswählen

def get_battery_data(self) -> dict or None:
        # get Vehicle Data (example from Polestar Config), NOCH ANPASSEN
        params = {
            "query": "query GetBatteryData($vin: String!) { getBatteryData(vin: $vin) { "
            + "battery_percent  battery_max_miles } }",
            "operationName": "GetBatteryData",
            "variables": "{\"vin\":\"" + self.vin + "\"}"
        }

        result = self.query_params(params)

        return result['data']['getBatteryData']
    
def fetch_soc(username: str, password: str, region: str, vehicle: int) -> CarState:
    # example from Polestar Config, NOCH ANPASSEN
    api = FiskerApi(username, password, region)
    bat_data = api.data_battery()
    soc = bat_data['battery_percent']
    est_range = bat_data['battery_max_miles"']

    return CarState(soc, est_range, time.strftime("%m/%d/%Y, %H:%M:%S"))
EDIT: im soc-log steht folgendes:

Code: Alles auswählen

2024-07-02 12:16:34,652 - {modules.common.configurable_vehicle:56} - {DEBUG:fetch soc_ev0} - Vehicle Instance <class 'modules.vehicles.fisker.config.FiskerOcean'>
2024-07-02 12:16:34,652 - {modules.common.configurable_vehicle:57} - {DEBUG:fetch soc_ev0} - Calculated SoC-State CalculatedSocState(imported_start=0, manual_soc=None, soc_start=0)
2024-07-02 12:16:34,653 - {modules.common.configurable_vehicle:58} - {DEBUG:fetch soc_ev0} - Vehicle Update Data VehicleUpdateData(plug_state=False, charge_state=False, imported=None, battery_capacity=82000, efficiency=90, soc_from_cp=None, timestamp_soc_from_cp=None)
2024-07-02 12:16:34,653 - {modules.common.configurable_vehicle:59} - {DEBUG:fetch soc_ev0} - General Config GeneralVehicleConfig(use_soc_from_cp=False, request_interval_charging=300, request_interval_not_charging=120, request_only_plugged=False)
2024-07-02 12:16:34,653 - {modules.common.component_context:25} - {DEBUG:fetch soc_ev0} - Update Komponente ['FiskerOcean']
2024-07-02 12:16:34,653 - {modules.vehicles.fisker.api:49} - {DEBUG:fetch soc_ev0} - FiskerApi init
2024-07-02 12:16:34,654 - {modules.common.fault_state:49} - {ERROR:fetch soc_ev0} - FiskerOcean: FaultState FaultStateLevel.ERROR, FaultStr <class 'AttributeError'> ("'FiskerApi' object has no attribute 'data_battery'",), Traceback: 
Traceback (most recent call last):
  File "/var/www/html/openWB/packages/modules/common/configurable_vehicle.py", line 66, in update
    car_state = self._get_carstate_by_source(vehicle_update_data, source)
  File "/var/www/html/openWB/packages/modules/common/configurable_vehicle.py", line 110, in _get_carstate_by_source
    return self.__component_updater(vehicle_update_data)
  File "/var/www/html/openWB/packages/modules/vehicles/fisker/soc.py", line 19, in updater
    return api.fetch_soc(
  File "/var/www/html/openWB/packages/modules/vehicles/fisker/api.py", line 327, in fetch_soc
    bat_data = api.data_battery()
AttributeError: 'FiskerApi' object has no attribute 'data_battery'
Ich kann allerdings nicht sagen, ob die Authentifizierung funktioniert. Wie kann ich das testen oder sehen (im main.log sehe ich nichts dazu)?

Hier das modifizierte api.py Skript (funktioniert noch nicht):

Code: Alles auswählen

# references:
# https://github.com/MichaelOE/home-assistant-MyFisker/blob/main/custom_components/my_fisker/api.py

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

# DOMAIN = "my_fisker"

API_TIMEOUT = 10
DEFAULT_SCAN_INTERVAL = 60

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

# TRIM_EXTREME_ULTRA_BATT_CAPACITY = 113
# TRIM_SPORT_BATT_CAPACITY = 80

CAR_SETTINGS = "car_settings"
DIGITAL_TWIN = "digital_twin"
PROFILES = "profiles"

log = logging.getLogger(__name__)

import aiohttp


_LOGGER = logging.getLogger(__name__)

headers = {"User-Agent": "MOBILE 1.0.0.0"}

HasAUTH = False
HasVIN = False

    
class FiskerApi:
    """Handle connection towards Fisker API servers."""

    # Global variable to store the WebSocket connection
    global_websocket = None

    vin = ""

    def __init__(self, username: str, password: str, region: str):
        _LOGGER.debug("FiskerApi init")
        self._username = username
        self._password = password
        self._region = region

        self._token = ""
        self._timeout = aiohttp.ClientTimeout(total=API_TIMEOUT)
        self.data = {}

    async def GetAuthTokenAsync(self):
        """Get the Authentification token from Fisker, is used towards the WebSocket connection."""

        params = {"username": self._username, "password": self._password}
        async with aiohttp.ClientSession() as session, session.post(
            TOKEN_URL, data=params
        ) as response:
            data = await response.json()

            # Check if a key exists
            if "accessToken" in data:
                retVal = data["accessToken"]
            else:
                retVal = data["message"]

            self._token = retVal
            return self._token

    async def tokenReturn(self):
        return self._token

    def GetCarSettings(self):
        try:
            data = json.loads(self.data[CAR_SETTINGS])
            _LOGGER.debug(data)
            return data
        except NameError:
            _LOGGER.warning("Self.data['CAR_SETTINGS'] is not available")
            return None
            
    async def GetDigitalTwin(self):
        self.data[DIGITAL_TWIN] = self.flatten_json(
            self.ParseDigitalTwinResponse(
                await self.__GetWebsocketResponse(DIGITAL_TWIN)
            )
        )
        return self.data[DIGITAL_TWIN]

    async def GetProfiles(self):
        self.data[PROFILES] = self.ParseProfilesResponse(
            await self.__GetWebsocketResponse(PROFILES)
        )
        return self.data[PROFILES]

    def ParseDigitalTwinResponse(self, jsonMsg):
        # _LOGGER.debug('Start ParseDigitalTwinResponse()')
        # Parse the JSON response into a Python dictionary
        data = json.loads(jsonMsg)
        _LOGGER.debug(data)

        if data["handler"] != DIGITAL_TWIN:
            _LOGGER.debug("ParseDigitalTwinResponse: Wrong answer from websocket")
            _LOGGER.debug(data)
            return "Wrong answer from websocket"

        # Now you can access the items in the JSON response as you would with a Python dictionary
        DIGITAL_TWIN = data["data"]

        # Use the jsonpath expression to find the value in the data
        _LOGGER.debug(DIGITAL_TWIN)  # Outputs: value1
        return DIGITAL_TWIN

    def GenerateVerifyRequest(self):
        # _LOGGER.debug('Start GenerateVerifyRequest()')
        data = {}
        messageData = {}

        # token = self.GetAuthToken(username, password)
        token = self._token

        data["token"] = token
        messageData["data"] = data
        messageData["handler"] = "verify"
        # print (messageData)
        return messageData

    def ParseVerifyResponse(self, jsonMsg):
        # Parse the JSON response into a Python dictionary
        data = json.loads(jsonMsg)

        if data["handler"] != "verify":
            return "Wrong answer from websocket"

        # Now you can access the items in the JSON response as you would with a Python dictionary
        item1 = data["data"]["authenticated"]

        if item1 != "true":
            return "Not authenticated"

        result = item1
        _LOGGER.debug(result)  # Outputs: value1
        return True
            
    def GenerateProfilesRequest(self):
        # _LOGGER.debug('Start GenerateProfilesRequest()')
        messageData = {}

        messageData["handler"] = PROFILES
        # print (messageData)
        return messageData

    def DigitalTwinRequest(self, vin):
        # _LOGGER.debug('Start DigitalTwinRequest()')
        data = {}
        messageData = {}
        data["vin"] = self.vin
        messageData["data"] = data
        messageData["handler"] = DIGITAL_TWIN
        return messageData

    def ParseProfilesResponse(self, jsonMsg):
        # _LOGGER.debug('Start ParseProfilesResponse()')
        # Parse the JSON response into a Python dictionary
        data = json.loads(jsonMsg)
        # print (data)
        if data["handler"] != PROFILES:
            _LOGGER.debug("ParseProfilesResponse: Wrong answer from websocket")
            _LOGGER.debug(data)
            return "Wrong answer from websocket"

        # Now you can access the items in the JSON response as you would with a Python dictionary
        item1 = data["data"][0]["vin"]

        # Use the jsonpath expression to find the value in the data
        result = item1
        return result

    async def SendCommandRequest(self, command):
        # _LOGGER.debug('Start SendCommandRequest()')
        data = {}
        messageData = {}
        data["vin"] = self.vin
        data["command"] = command
        messageData["data"] = data
        messageData["handler"] = "remote_command"
        return await self.__SendWebsocketRequest(messageData)

    def __GetRegionURL(self):
        if self._region == "EU":
            return WSS_URL_EU
        else:
            raise WSS_URL_US

    async def __GetWebsocketResponse(self, responseToReturn: str):
        HasAUTH = False
        HasVIN = HasAUTH

        wssUrl = self.__GetRegionURL()

        async with aiohttp.ClientSession() as session:
            async with session.ws_connect(wssUrl, headers=headers) as ws:
                await ws.send_str(json.dumps(self.GenerateVerifyRequest()))
                while True:
                    response = await ws.receive_str()
                    handler = json.loads(response)["handler"]

                    if handler == CAR_SETTINGS:
                        self.data[CAR_SETTINGS] = response

                    if handler == responseToReturn:
                        try:
                            await ws.close()
                        except Exception as e:
                            _LOGGER.debug(
                                f"Error occurred while closing WebSocket: {e}"
                            )
                        return response

                    if HasAUTH is not True:
                        if handler == "verify":
                            HasAUTH = (
                                json.loads(response)["data"]["authenticated"] is True
                            )
                            # Send a message
                            # _LOGGER.debug(f"Sending 'GenerateProfilesRequest'")
                            await ws.send_str(
                                json.dumps(self.GenerateProfilesRequest())
                            )

                    if HasAUTH is True and HasVIN is not True:
                        if handler == PROFILES:
                            self.vin = self.ParseProfilesResponse(response)
                            # print (f"vin = {vin}")
                            if self.vin != "":
                                self.HasVIN = True
                                # Send a message
                                _LOGGER.debug(
                                    f"Auth & VIN ok - Sending 'DigitalTwinRequest' to vin={self.vin}"
                                )
                                await ws.send_str(
                                    json.dumps(self.DigitalTwinRequest(self.vin))
                                )

                    if HasAUTH is True and HasVIN is True:
                        # _LOGGER.debug(f"Received message: {message}")
                        if handler == responseToReturn:
                            _LOGGER.error(response)
                            try:
                                await ws.close()
                            except Exception as e:
                                _LOGGER.error(
                                    f"Error occurred while closing WebSocket: {e}"
                                )

                            return response

    async def __SendWebsocketRequest(self, commandToSend: str):
        HasAUTH = False
        wssUrl = self.__GetRegionURL()

        async with aiohttp.ClientSession() as session:
            async with session.ws_connect(wssUrl, headers=headers) as ws:
                await ws.send_str(json.dumps(self.GenerateVerifyRequest()))
                while True:
                    response = await ws.receive_str()
                    handler = json.loads(response)["handler"]

                    if handler in (DIGITAL_TWIN, CAR_SETTINGS):
                        try:
                            await ws.close()
                        except Exception as e:
                            _LOGGER.debug(
                                f"Error occurred while closing WebSocket: {e}"
                            )
                        return response

                    if HasAUTH is not True:
                        if handler == "verify":
                            HasAUTH = (
                                json.loads(response)["data"]["authenticated"] is True
                            )
                            # Send a message
                            # _LOGGER.debug(f"Sending 'GenerateProfilesRequest'")
                            await ws.send_str(json.dumps(commandToSend))

    def flatten_json(self, jsonIn):
        out = {}

        def flatten(x, name=""):
            if type(x) is dict:
                for a in x:
                    flatten(x[a], name + a + "_")
            elif type(x) is list:
                i = 0
                for a in x:
                    flatten(a, name + str(i) + "_")
                    i += 1
            else:
                out[name[:-1]] = x

        flatten(jsonIn)
        return out
    
    def get_battery_data(self) -> dict or None:
        # get Vehicle Data (example from Polestar Config), NOCH ANPASSEN
        params = {
            "query": "query GetBatteryData($vin: String!) { getBatteryData(vin: $vin) { "
            + "battery_percent  battery_max_miles } }",
            "operationName": "GetBatteryData",
            "variables": "{\"vin\":\"" + self.vin + "\"}"
        }

        result = self.query_params(params)

        return result['data']['getBatteryData']
    
def fetch_soc(username: str, password: str, region: str, vehicle: int) -> CarState:
    # example from Polestar Config, NOCH ANPASSEN
    api = FiskerApi(username, password, region)
    bat_data = api.data_battery()
    soc = bat_data['battery_percent']
    est_range = bat_data['battery_max_miles"']

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


class FiskerApiError(Exception):
    """Base exception for all MyFisker API errors"""


class AuthenticationError(FiskerApiError):
    """Authenatication failed"""


class RequestError(FiskerApiError):
    """Failed to get the results from the API"""

    def __init__(self, message, error_code):
        super().__init__(message)
        self.error_code = error_code


class RequestConnectionError(FiskerApiError):
    """Failed to make the request to the API"""


class RequestTimeoutError(FiskerApiError):
    """Failed to get the results from the API"""


class RequestRetryError(FiskerApiError):
    """Retries too many times"""


class RequestDataError(FiskerApiError):
    """Data is not valid"""
Zuletzt geändert von knichiknaxl am Do Jul 04, 2024 1:21 pm, insgesamt 1-mal geändert.
rleidner
Beiträge: 1001
Registriert: Mo Nov 02, 2020 9:50 am
Has thanked: 5 times
Been thanked: 38 times

Re: SoC Fisker OCEAN

Beitrag von rleidner »

Ich kann meinen Tipp nur wiederholen: :
Wenn ich mit einem SOC-Abruf für ein ein neues Fahrzeug beginne und kein funktionierendes Muster habe, baue ich erst mal ein einfaches python script, das den gesamten Abruf erfolgreich durchführt.
Erst wenn das sauber funktioniert, baue ich das in die openwb-Umgebung ein.
...
und in dem python script JEDEN einzelnen Schritt protokollieren bis es funktioniert!
openWB-2 Standard+ | openWB EVU Kit v2 MID| 9,9kWp mit Kostal Plenticore 8.5 plus | VW ID.3, Kia EV6, Smart EQ forfour
Antworten