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"""