Hallo zusammen, ich habe es zwar geschafft mit dem entsprechenden Video von @simon42 das Python-Script zum Ändern von Entitäten einzubinden, habe aber jetzt ein Problem mit einem Script zum Auslesen von Ecoflow-Komponenten über die API-Methode.
Wenn ich das Script ausführe, erhalte ich die Fehlermeldung:
Die Aktion python_script.ecoflow konnte nicht ausgeführt werden. Error executing script: Not allowed to import sys
Ich habe auch mal versucht, in die configuration.yaml den folgenden Eintrag zu ergänzen:
python_script:
allow_all_imports: true
hass_is_global: true
(normalerweise steht da nur python_script:)
Das Script, um das es geht, ist folgendes:
# Example Code to obtain data from API for specific sensors on an EcoFlow device
# Mark Hicks - 01/29/2024
# written by mail@sven-erbe.de - 14/02/2024
# short description
# script collect all reqistered devices from ecoflow api and creates a entity_id sensor.ecoflow_myapi_{device_name}
# prerequest:
# Pyscript: Python scripting integration - https://github.com/custom-components/pyscript
# ecoflow AccessKey and SecretKey - needs to request on https://developer-eu.ecoflow.com/
# please change accesskey and secret in the script
import sys
import json
import requests
import hashlib
import hmac
import random
import time
import binascii
from urllib.parse import urlencode
def hmac_sha256(data, key):
hashed = hmac.new(key.encode('utf-8'), data.encode('utf-8'), hashlib.sha256).digest()
sign = binascii.hexlify(hashed).decode('utf-8')
return sign
def get_map(json_obj, prefix=""):
def flatten(obj, pre=""):
result = {}
if isinstance(obj, dict):
for k, v in obj.items():
result.update(flatten(v, f"{pre}.{k}" if pre else k))
elif isinstance(obj, list):
for i, item in enumerate(obj):
result.update(flatten(item, f"{pre}[{i}]"))
else: result[pre] = obj
return result
return flatten(json_obj, prefix)
def get_qstr(params): return '&'.join([f"{key}={params[key]}" for key in sorted(params.keys())])
def get_api(url, key, secret, params=None):
nonce = str(random.randint(100000, 999999))
timestamp = str(int(time.time() * 1000))
headers = {'accessKey':key,'nonce':nonce,'timestamp':timestamp}
sign_str = (get_qstr(get_map(params)) + '&' if params else '') + get_qstr(headers)
print(sign_str)
headers['sign'] = hmac_sha256(sign_str, secret)
response = task.executor(requests.get, url, headers=headers, json=params)
if response.status_code == 200: return response.json()
else: print(f"get_api: {response.text}")
@service
def get_ef_device_list_all(return_response=True):
url = 'https://api.ecoflow.com/iot-open/sign/device/list'
url_details = 'https://api.ecoflow.com/iot-open/sign/device/quota/all'
quotas = ["inv.cfgAcEnabled","inv.cfgAcXboost","bms_bmsStatus.hwVersion"]
#params = {"cmdSet":32,"id":66,"quotas":quotas}
# Replace with valid access/secret keys and device SN
accesskey = 'geheim'
secret = 'noch viel geheimer'
payload = get_api(url,accesskey,secret)
response_data = payload
devices = response_data['data']
if 'data' in payload:
devices = payload['data']
for device in devices:
device_sn = device['sn']
device_name = device['deviceName']
if device_name:
device_name_clean = device_name.replace(" ", "").replace("ü", "ue").replace("ö", "oe").replace("ä", "ae")
else:
device_name_clean = ""
device_name = device_name_clean
if device_name: # Überprüfen, ob der Gerätenamen nicht leer ist
sensor_entity_id = f'sensor.ecoflow_myapi_{device_name}'
else:
sensor_entity_id = f'sensor.ecoflow_myapi_{device_sn}'
# get details
url_string = url_details + '?sn=' + device_sn
payload_details = get_api(url_string,accesskey,secret,{"sn":device_sn})
attributes = {'device_name': device_name, 'online': device['online'], 'serialnumber': device_sn}
if 'data' in payload_details:
details = payload_details['data']
for key, value in details.items():
if key.startswith('2_1.') or key.startswith('20_1.'):
attributes[key] = value
hass.states.async_set(sensor_entity_id, 'State', attributes=attributes)