Add Sunlit Solar integration with configuration flow, API handling, and sensor support
This commit is contained in:
commit
f7edc0c41d
12
.gitignore
vendored
Normal file
12
.gitignore
vendored
Normal file
@ -0,0 +1,12 @@
|
||||
.vscode
|
||||
.devcontainer
|
||||
/config/.cloud
|
||||
/config/.storage
|
||||
/config/deps
|
||||
/config/tts
|
||||
/config/blueprints
|
||||
/config/*.yaml
|
||||
/config/.HA_VERSION
|
||||
/config/*.db*
|
||||
/config/*.log*
|
||||
__pycache__
|
49
config/custom_components/sunlit_solar/__init__.py
Normal file
49
config/custom_components/sunlit_solar/__init__.py
Normal file
@ -0,0 +1,49 @@
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
||||
from datetime import timedelta
|
||||
import aiohttp
|
||||
import logging
|
||||
from .const import KEY_SCAN_INTERVAL
|
||||
from .api import async_get_device_list
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
from .const import DOMAIN
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry):
|
||||
session = aiohttp.ClientSession()
|
||||
|
||||
async def fetch_data():
|
||||
try:
|
||||
return await async_get_device_list(entry, session)
|
||||
except Exception as e:
|
||||
logging.error(f"Error fetching data: {e}")
|
||||
raise UpdateFailed(f"API error: {e}")
|
||||
|
||||
update_interval = timedelta(seconds=entry.data.get(KEY_SCAN_INTERVAL, 2))
|
||||
coordinator = DataUpdateCoordinator(
|
||||
hass,
|
||||
_LOGGER,
|
||||
name="SunlitSolar Sensoren",
|
||||
update_method=fetch_data,
|
||||
update_interval=update_interval,
|
||||
)
|
||||
|
||||
await coordinator.async_config_entry_first_refresh()
|
||||
|
||||
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = {
|
||||
"coordinator": coordinator,
|
||||
"session": session
|
||||
}
|
||||
await hass.config_entries.async_forward_entry_setup(entry, "sensor")
|
||||
await hass.config_entries.async_forward_entry_setup(entry, "binary_sensor")
|
||||
# hass.async_create_task(
|
||||
# hass.config_entries.async_forward_entry_setup(entry, "sensor")
|
||||
# )
|
||||
|
||||
return True
|
||||
|
||||
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
|
||||
await hass.data[DOMAIN][entry.entry_id]["session"].close()
|
||||
await hass.config_entries.async_forward_entry_unload(entry, "sensor")
|
||||
return await hass.config_entries.async_forward_entry_unload(entry, "binary_sensor")
|
97
config/custom_components/sunlit_solar/api.py
Normal file
97
config/custom_components/sunlit_solar/api.py
Normal file
@ -0,0 +1,97 @@
|
||||
import logging
|
||||
import aiohttp
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
API_URL = "https://api.sunlitsolar.de/rest"
|
||||
isMock = False # Set to True for testing with mock data
|
||||
|
||||
class SunlitResponseCode:
|
||||
"""Enum for Sunlit Solar API response codes."""
|
||||
SUCCESS = 0
|
||||
# ERROR = 1
|
||||
INVALID_CREDENTIALS = 100003
|
||||
# USER_NOT_FOUND = 3
|
||||
# SERVER_ERROR = 4
|
||||
|
||||
def get_message_from_response(response):
|
||||
"""Extracts the message from the Sunlit Solar API response."""
|
||||
if 'message' in response:
|
||||
for key, value in response["message"].items():
|
||||
return value
|
||||
return response['message']
|
||||
else:
|
||||
return "Unbekannter Fehler"
|
||||
|
||||
async def async_login(username, password):
|
||||
"""Asynchronous login to the Sunlit Solar API."""
|
||||
try:
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.post(API_URL+"/user/login", json={'account': username, 'password': password, 'isMock': False}) as response:
|
||||
response.raise_for_status()
|
||||
return await response.json()
|
||||
except aiohttp.ClientError as e:
|
||||
_LOGGER.error("Fehler beim Abrufen der Daten: %s", e)
|
||||
raise
|
||||
|
||||
async def async_get_device_list(entry: ConfigEntry, session: aiohttp.ClientSession):
|
||||
try:
|
||||
token = entry.data["access_token"]
|
||||
familys = []
|
||||
# get family-id -> get devices from family -> get device statistics from device id
|
||||
async with session.post(API_URL+"/family/list", headers={"Authorization": f"Bearer {token}"}, json={'isMock': isMock}) as response:
|
||||
response.raise_for_status()
|
||||
resp_familys = await response.json()
|
||||
if resp_familys['code'] != SunlitResponseCode.SUCCESS:
|
||||
_LOGGER.error("Fehler beim Abrufen der Familien: %s", get_message_from_response(resp_familys))
|
||||
else:
|
||||
familys = resp_familys['content']
|
||||
if familys:
|
||||
# Get devices for each family
|
||||
for f_key,family in enumerate(familys):
|
||||
print(family['id'])
|
||||
async with session.post(API_URL+"/v1.2/device/list", headers={"Authorization": f"Bearer {token}"}, json={
|
||||
"familyId": family['id'],
|
||||
"deviceType": "ALL",
|
||||
"isMock": isMock,
|
||||
"size": 20,
|
||||
"page": 0
|
||||
}) as response:
|
||||
response.raise_for_status()
|
||||
resp_devices = await response.json()
|
||||
if resp_devices['code'] != SunlitResponseCode.SUCCESS:
|
||||
_LOGGER.error("Fehler beim Abrufen der Geräte: %s", get_message_from_response(resp_devices))
|
||||
else:
|
||||
# todo: implement pagination
|
||||
familys[f_key]['devices'] = resp_devices['content']['content']
|
||||
if 'spaces' not in familys[f_key]:
|
||||
familys[f_key]['spaces'] = {}
|
||||
if familys[f_key]['devices'] and len(familys[f_key]['devices']) > 0:
|
||||
# Get statistics for each device
|
||||
for d_key,device in enumerate(family['devices']):
|
||||
async with session.post(API_URL+"/v1.1/statistics/static/device", headers={"Authorization": f"Bearer {token}"}, json={
|
||||
"deviceId": device['deviceId'],
|
||||
"isMock": isMock,
|
||||
}) as response:
|
||||
response.raise_for_status()
|
||||
resp_statistics = await response.json()
|
||||
if resp_statistics['code'] != SunlitResponseCode.SUCCESS:
|
||||
_LOGGER.error("Fehler beim Abrufen der Statistiken: %s", get_message_from_response(resp_statistics))
|
||||
else:
|
||||
familys[f_key]['devices'][d_key]['statistics'] = resp_statistics['content']
|
||||
|
||||
if 'spaceId' in device and device['spaceId'] not in familys[f_key]['spaces']:
|
||||
async with session.post(API_URL+"/v1.5/space/index", headers={"Authorization": f"Bearer {token}"}, json={
|
||||
"spaceId": device['spaceId'],
|
||||
"isMock": isMock,
|
||||
}) as response:
|
||||
response.raise_for_status()
|
||||
resp_space = await response.json()
|
||||
if resp_space['code'] != SunlitResponseCode.SUCCESS:
|
||||
_LOGGER.error("Fehler beim Abrufen des Raums: %s", get_message_from_response(resp_space))
|
||||
else:
|
||||
familys[f_key]['spaces'][device['spaceId']] = resp_space['content']
|
||||
|
||||
return familys
|
||||
except aiohttp.ClientError as e:
|
||||
_LOGGER.error("Fehler beim Abrufen der Daten: %s", e)
|
||||
raise
|
116
config/custom_components/sunlit_solar/binary_sensor.py
Normal file
116
config/custom_components/sunlit_solar/binary_sensor.py
Normal file
@ -0,0 +1,116 @@
|
||||
from homeassistant.components.binary_sensor import (
|
||||
BinarySensorEntity,
|
||||
BinarySensorDeviceClass,
|
||||
)
|
||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
||||
from homeassistant.helpers.entity import DeviceInfo
|
||||
|
||||
from .helper import get_merged_device_data
|
||||
from .const import DOMAIN
|
||||
|
||||
|
||||
async def async_setup_entry(hass, entry, async_add_entities):
|
||||
coordinator = hass.data[DOMAIN][entry.entry_id]["coordinator"]
|
||||
data = get_merged_device_data(coordinator)
|
||||
sensors = []
|
||||
|
||||
for family_key in data:
|
||||
family = data[family_key]
|
||||
if "devices" not in family:
|
||||
continue
|
||||
for device_key in family["devices"]:
|
||||
device = family["devices"][device_key]
|
||||
device_info = DeviceInfo(
|
||||
identifiers={(DOMAIN, device["deviceId"], device.get("deviceSn",""))},
|
||||
name=device.get("stationName", "Unbekanntes Gerät"),
|
||||
serial_number=device.get("deviceSn", "Unbekannt"),
|
||||
suggested_area=family.get("name", "Unbekannt"),
|
||||
manufacturer="Sunlit Solar",
|
||||
model=device.get("deviceType", "Unbekanntes Modell"),
|
||||
)
|
||||
if device["deviceType"] == "ENERGY_STORAGE_BATTERY":
|
||||
device_info = DeviceInfo(
|
||||
identifiers={(DOMAIN, device["deviceId"], device.get("deviceSn",""))},
|
||||
name="Speicher",
|
||||
serial_number=device.get("deviceSn", "Unbekannt"),
|
||||
suggested_area=family.get("name", "Unbekannt"),
|
||||
manufacturer="Sunlit Solar",
|
||||
model=device.get("deviceType", "Unbekanntes Modell"),
|
||||
)
|
||||
sensor_keys = [
|
||||
{'id': 'fault', 'name': 'Fehler', 'device_class': None, 'state_class': None, 'native_unit_of_measurement': None},
|
||||
{'id': 'off', 'name': 'Abgeschaltet', 'device_class': None, 'state_class': None, 'native_unit_of_measurement': None},
|
||||
{'id': 'bypass', 'name': 'Bypass aktiv', 'device_class': None, 'state_class': None, 'native_unit_of_measurement': None},
|
||||
{'id': 'isChargingStatus', 'name': 'Lädt', 'device_class': None, 'state_class': None, 'native_unit_of_measurement': None},
|
||||
]
|
||||
|
||||
for sensor in sensor_keys:
|
||||
sensors.append(Sensor(
|
||||
coordinator,
|
||||
device_info,
|
||||
entry.entry_id,
|
||||
f"{sensor['id']}",
|
||||
family_key,
|
||||
device_key,
|
||||
sensor['name'],
|
||||
sensor['device_class'],
|
||||
sensor['state_class'],
|
||||
sensor['native_unit_of_measurement']
|
||||
))
|
||||
|
||||
# heaterStatusList
|
||||
for i,key in enumerate(device['heaterStatusList']):
|
||||
sensors.append(Sensor(
|
||||
coordinator,
|
||||
device_info,
|
||||
entry.entry_id,
|
||||
key,
|
||||
family_key,
|
||||
device_key,
|
||||
f'Heizung Speicher {i+1}',
|
||||
None, #sensor['device_class'],
|
||||
None, #sensor['state_class'],
|
||||
None, #sensor['native_unit_of_measurement']
|
||||
"heaterStatusList"
|
||||
))
|
||||
|
||||
|
||||
async_add_entities(sensors)
|
||||
|
||||
class Sensor(CoordinatorEntity, BinarySensorEntity):
|
||||
def __init__(self, coordinator, deviceinfo, entry_id, id, family_key, device_key, name, device_class, state_class, native_unit_of_measurement, subkey=None):
|
||||
super().__init__(coordinator)
|
||||
self._entry_id = entry_id
|
||||
self._deviceinfo = deviceinfo
|
||||
self._family_key = family_key
|
||||
self._device_key = device_key
|
||||
self._subkey = subkey
|
||||
self._id = id
|
||||
|
||||
self._attr_name = name
|
||||
self._attr_unique_id = f"{entry_id}_{family_key}_{device_key}_{id}"
|
||||
if subkey:
|
||||
self._attr_unique_id += f"{entry_id}_{family_key}_{device_key}_{subkey}_{id}"
|
||||
self._attr_native_unit_of_measurement = native_unit_of_measurement
|
||||
self._attr_device_class = device_class
|
||||
self._attr_state_class = state_class
|
||||
|
||||
@property
|
||||
def is_on(self):
|
||||
data = get_merged_device_data(self.coordinator)
|
||||
if self._family_key not in data:
|
||||
return None
|
||||
family = data[self._family_key]
|
||||
if "devices" not in family or self._device_key not in family["devices"]:
|
||||
return None
|
||||
device = family["devices"][self._device_key]
|
||||
if self._subkey:
|
||||
if self._subkey in device and isinstance(device[self._subkey], dict):
|
||||
return device[self._subkey].get(self._id, 'Unbekannt')
|
||||
else:
|
||||
return device.get(self._subkey, [])[self._id] if isinstance(device.get(self._subkey, []), list) else device.get(self._subkey, {}).get(self._id, 'Unbekannt')
|
||||
return device.get(self._id, 'Unbekannt')
|
||||
|
||||
@property
|
||||
def device_info(self) -> DeviceInfo:
|
||||
return self._deviceinfo
|
61
config/custom_components/sunlit_solar/config_flow.py
Normal file
61
config/custom_components/sunlit_solar/config_flow.py
Normal file
@ -0,0 +1,61 @@
|
||||
from homeassistant import config_entries
|
||||
import voluptuous as vol
|
||||
from .api import async_login, SunlitResponseCode
|
||||
|
||||
from .const import DOMAIN
|
||||
from .const import KEY_USERNAME, KEY_PASSWORD, KEY_SCAN_INTERVAL
|
||||
from homeassistant.helpers.selector import (
|
||||
TextSelector,
|
||||
TextSelectorConfig,
|
||||
TextSelectorType,
|
||||
)
|
||||
|
||||
class MyIntegrationConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
|
||||
async def async_step_user(self, user_input=None):
|
||||
if user_input is not None:
|
||||
print("User input received:", user_input)
|
||||
err_msg = "api_unreachable"
|
||||
# Here you would typically validate the user input, e.g., check credentials
|
||||
if user_input.get(KEY_USERNAME) and user_input.get(KEY_PASSWORD) and "@" in user_input[KEY_USERNAME]:
|
||||
login_resp = await async_login(user_input[KEY_USERNAME], user_input[KEY_PASSWORD])
|
||||
if login_resp is not None and login_resp["code"] == SunlitResponseCode.SUCCESS:
|
||||
# If validation is successful, create the entry
|
||||
user_input[KEY_SCAN_INTERVAL] = int(user_input.get(KEY_SCAN_INTERVAL, 60))
|
||||
return self.async_create_entry(title=user_input[KEY_USERNAME], data={**login_resp["content"], **{KEY_SCAN_INTERVAL: user_input[KEY_SCAN_INTERVAL]}})
|
||||
err_msg = "invalid_credentials"
|
||||
if login_resp is not None and login_resp["code"] != SunlitResponseCode.SUCCESS:
|
||||
for key, value in login_resp["message"].items():
|
||||
err_msg = value
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="user",
|
||||
data_schema=vol.Schema({
|
||||
vol.Required(KEY_USERNAME): TextSelector(
|
||||
TextSelectorConfig(type=TextSelectorType.EMAIL, autocomplete="username")
|
||||
),
|
||||
vol.Required(KEY_PASSWORD): TextSelector(
|
||||
TextSelectorConfig(
|
||||
type=TextSelectorType.PASSWORD, autocomplete="current-password"
|
||||
)
|
||||
),
|
||||
vol.Optional(KEY_SCAN_INTERVAL, default=60): int,
|
||||
}),
|
||||
errors={"base": err_msg}
|
||||
)
|
||||
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="user",
|
||||
data_schema=vol.Schema({
|
||||
vol.Required(KEY_USERNAME): TextSelector(
|
||||
TextSelectorConfig(type=TextSelectorType.EMAIL, autocomplete="username")
|
||||
),
|
||||
vol.Required(KEY_PASSWORD): TextSelector(
|
||||
TextSelectorConfig(
|
||||
type=TextSelectorType.PASSWORD, autocomplete="current-password"
|
||||
)
|
||||
),
|
||||
vol.Optional(KEY_SCAN_INTERVAL, default=60): int,
|
||||
}),
|
||||
)
|
4
config/custom_components/sunlit_solar/const.py
Normal file
4
config/custom_components/sunlit_solar/const.py
Normal file
@ -0,0 +1,4 @@
|
||||
DOMAIN = "sunlit_solar"
|
||||
KEY_USERNAME = "username"
|
||||
KEY_PASSWORD = "password"
|
||||
KEY_SCAN_INTERVAL = "scan_interval"
|
17
config/custom_components/sunlit_solar/helper.py
Normal file
17
config/custom_components/sunlit_solar/helper.py
Normal file
@ -0,0 +1,17 @@
|
||||
|
||||
def get_merged_device_data(coordinator):
|
||||
"""Fetch and merge data from the coordinator for the given entry."""
|
||||
ret = {}
|
||||
for family in coordinator.data:
|
||||
ret[family['id']] = family.copy() # Use copy to avoid modifying the original family data
|
||||
_devices = family.get("devices", []).copy() # Use copy to avoid modifying the original list
|
||||
ret[family['id']]['devices'] = {}
|
||||
if "devices" not in family:
|
||||
continue
|
||||
for device in _devices:
|
||||
device = {**device, **device.get("statistics", {})} # Merge statistics into device
|
||||
if device["deviceType"] == "ENERGY_STORAGE_BATTERY":
|
||||
space = family["spaces"].get(device["spaceId"], {})
|
||||
device = {**device, **space.get("battery",{})} # Merge statistics into device
|
||||
ret[family['id']]['devices'][device['deviceId']] = device
|
||||
return ret
|
12
config/custom_components/sunlit_solar/manifest.json
Normal file
12
config/custom_components/sunlit_solar/manifest.json
Normal file
@ -0,0 +1,12 @@
|
||||
{
|
||||
"domain": "sunlit_solar",
|
||||
"name": "Sunlit Solar",
|
||||
"version": "1.0",
|
||||
"documentation": "https://github.com/dezeyer23",
|
||||
"dependencies": [],
|
||||
"codeowners": ["@dezeyer23"],
|
||||
"requirements": ["aiohttp"],
|
||||
"config_flow": true,
|
||||
"iot_class": "cloud_polling",
|
||||
"domains": ["binary_sensor", "sensor"]
|
||||
}
|
316
config/custom_components/sunlit_solar/sensor.py
Normal file
316
config/custom_components/sunlit_solar/sensor.py
Normal file
@ -0,0 +1,316 @@
|
||||
from homeassistant.components.sensor import SensorEntity
|
||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
||||
from homeassistant.helpers.entity import DeviceInfo
|
||||
from .const import DOMAIN
|
||||
from .helper import get_merged_device_data
|
||||
|
||||
|
||||
async def async_setup_entry(hass, entry, async_add_entities):
|
||||
coordinator = hass.data[DOMAIN][entry.entry_id]["coordinator"]
|
||||
data = get_merged_device_data(coordinator)
|
||||
sensors = []
|
||||
|
||||
for family_key in data:
|
||||
family = data[family_key]
|
||||
if "devices" not in family:
|
||||
continue
|
||||
for device_key in family["devices"]:
|
||||
device = family["devices"][device_key]
|
||||
device_info = DeviceInfo(
|
||||
identifiers={(DOMAIN, device["deviceId"], device.get("deviceSn",""))},
|
||||
name=device.get("stationName", "Unbekanntes Gerät"),
|
||||
serial_number=device.get("deviceSn", "Unbekannt"),
|
||||
suggested_area=family.get("name", "Unbekannt"),
|
||||
manufacturer="Sunlit Solar",
|
||||
model=device.get("deviceType", "Unbekanntes Modell"),
|
||||
)
|
||||
if device["deviceType"] == "ENERGY_STORAGE_BATTERY":
|
||||
device_info = DeviceInfo(
|
||||
identifiers={(DOMAIN, device["deviceId"], device.get("deviceSn",""))},
|
||||
name="Speicher",
|
||||
serial_number=device.get("deviceSn", "Unbekannt"),
|
||||
suggested_area=family.get("name", "Unbekannt"),
|
||||
manufacturer="Sunlit Solar",
|
||||
model=device.get("deviceType", "Unbekanntes Modell"),
|
||||
)
|
||||
sensor_keys = [
|
||||
{'id': 'status', 'name': 'Status', 'device_class': None, 'state_class': None, 'native_unit_of_measurement': None},
|
||||
{'id': 'batteryLevel', 'name': 'Ladung Speicher komplett', 'device_class': "battery", 'state_class': None, 'native_unit_of_measurement': "%"},
|
||||
{'id': 'batterySoc', 'name': 'Ladung Speicher 1 (Kopf)', 'device_class': "battery", 'state_class': None, 'native_unit_of_measurement': "%"},
|
||||
{'id': 'deviceCount', 'name': 'Speicheranzahl', 'device_class': None, 'state_class': None, 'native_unit_of_measurement': None},
|
||||
{'id': 'chargeRemaining', 'name': 'verbleibende Ladung', 'device_class': "battery", 'state_class': None, 'native_unit_of_measurement': "%"},
|
||||
{'id': 'dischargeRemaining', 'name': 'verbleibende Entladung', 'device_class': "battery", 'state_class': None, 'native_unit_of_measurement': "%"},
|
||||
{'id': 'inputPower', 'name': 'Eingangsleistung', 'device_class': "power", 'state_class': None, 'native_unit_of_measurement': "W"},
|
||||
{'id': 'outputPower', 'name': 'Ausgangsleistung', 'device_class': "power", 'state_class': None, 'native_unit_of_measurement': "W"},
|
||||
{'id': 'firmwareVersion', 'name': 'Firmware-Version', 'device_class': None, 'state_class': None, 'native_unit_of_measurement': None},
|
||||
# {'id': 'status', 'name': 'Status', 'value': device.get('status', 'Unbekannt'), 'device_class': None, 'state_class': None, 'native_unit_of_measurement': None},
|
||||
# {'id': 'status', 'name': 'Status', 'value': device.get('status', 'Unbekannt'), 'device_class': None, 'state_class': None, 'native_unit_of_measurement': None},
|
||||
# {'id': 'status', 'name': 'Status', 'value': device.get('status', 'Unbekannt'), 'device_class': None, 'state_class': None, 'native_unit_of_measurement': None},
|
||||
]
|
||||
if device.get('battery1Soc',None) != None:
|
||||
sensor_keys.append(
|
||||
{'id': 'battery1Soc', 'name': 'Ladung Speicher 2', 'device_class': "battery", 'state_class': None, 'native_unit_of_measurement': "%"},
|
||||
)
|
||||
if device.get('battery2Soc',None) != None:
|
||||
sensor_keys.append(
|
||||
{'id': 'battery2Soc', 'name': 'Ladung Speicher 3', 'device_class': "battery", 'state_class': None, 'native_unit_of_measurement': "%"},
|
||||
)
|
||||
if device.get('battery3Soc',None) != None:
|
||||
sensor_keys.append(
|
||||
{'id': 'battery3Soc', 'name': 'Ladung Speicher 4', 'device_class': "battery", 'state_class': None, 'native_unit_of_measurement': "%"},
|
||||
)
|
||||
|
||||
for sensor in sensor_keys:
|
||||
sensors.append(Sensor(
|
||||
coordinator,
|
||||
device_info,
|
||||
entry.entry_id,
|
||||
f"{sensor['id']}",
|
||||
family_key,
|
||||
device_key,
|
||||
sensor['name'],
|
||||
sensor['device_class'],
|
||||
sensor['state_class'],
|
||||
sensor['native_unit_of_measurement']
|
||||
))
|
||||
if isinstance(device['batteryMppt1Data'], dict):
|
||||
sensors.append(Sensor(
|
||||
coordinator,
|
||||
device_info,
|
||||
entry.entry_id,
|
||||
f"batteryMpptInVol",
|
||||
family_key,
|
||||
device_key,
|
||||
f'Speicher 1 MPPT 1 Eingangsspannung',
|
||||
"voltage", #sensor['device_class'],
|
||||
None, #sensor['state_class'],
|
||||
"V", #sensor['native_unit_of_measurement']
|
||||
"batteryMppt1Data"
|
||||
))
|
||||
sensors.append(Sensor(
|
||||
coordinator,
|
||||
device_info,
|
||||
entry.entry_id,
|
||||
f"batteryMpptInCur",
|
||||
family_key,
|
||||
device_key,
|
||||
f'Speicher 1 MPPT 1 Eingangsstrom ',
|
||||
"current", #sensor['device_class'],
|
||||
None, #sensor['state_class'],
|
||||
"A", #sensor['native_unit_of_measurement']
|
||||
"batteryMppt1Data"
|
||||
))
|
||||
sensors.append(Sensor(
|
||||
coordinator,
|
||||
device_info,
|
||||
entry.entry_id,
|
||||
f"batteryMpptInPower",
|
||||
family_key,
|
||||
device_key,
|
||||
f'Speicher 1 MPPT 1 Eingangsleistung',
|
||||
"power", #sensor['device_class'],
|
||||
None, #sensor['state_class'],
|
||||
"W", #sensor['native_unit_of_measurement']
|
||||
"batteryMppt1Data"
|
||||
))
|
||||
if isinstance(device['batteryMppt2Data'], dict):
|
||||
sensors.append(Sensor(
|
||||
coordinator,
|
||||
device_info,
|
||||
entry.entry_id,
|
||||
"batteryMpptInVol",
|
||||
family_key,
|
||||
device_key,
|
||||
f'Speicher 1 MPPT 2 Eingangsspannung',
|
||||
"voltage", #sensor['device_class'],
|
||||
None, #sensor['state_class'],
|
||||
"V", #sensor['native_unit_of_measurement']
|
||||
"batteryMppt2Data"
|
||||
))
|
||||
sensors.append(Sensor(
|
||||
coordinator,
|
||||
device_info,
|
||||
entry.entry_id,
|
||||
f"batteryMpptInCur",
|
||||
family_key,
|
||||
device_key,
|
||||
f'Speicher 1 MPPT 2 Eingangsstrom ',
|
||||
"current", #sensor['device_class'],
|
||||
None, #sensor['state_class'],
|
||||
"A", #sensor['native_unit_of_measurement']
|
||||
"batteryMppt2Data"
|
||||
))
|
||||
sensors.append(Sensor(
|
||||
coordinator,
|
||||
device_info,
|
||||
entry.entry_id,
|
||||
f"batteryMpptInPower",
|
||||
family_key,
|
||||
device_key,
|
||||
f'Speicher 1 MPPT 2 Eingangsleistung',
|
||||
"power", #sensor['device_class'],
|
||||
None, #sensor['state_class'],
|
||||
"W", #sensor['native_unit_of_measurement']
|
||||
"batteryMppt2Data"
|
||||
))
|
||||
if isinstance(device['battery1MpptData'], dict):
|
||||
sensors.append(Sensor(
|
||||
coordinator,
|
||||
device_info,
|
||||
entry.entry_id,
|
||||
f"batteryMpptInVol",
|
||||
family_key,
|
||||
device_key,
|
||||
f'Speicher 2 MPPT 1 Eingangsspannung',
|
||||
"voltage", #sensor['device_class'],
|
||||
None, #sensor['state_class'],
|
||||
"V", #sensor['native_unit_of_measurement']
|
||||
"battery1MpptData"
|
||||
))
|
||||
sensors.append(Sensor(
|
||||
coordinator,
|
||||
device_info,
|
||||
entry.entry_id,
|
||||
f"batteryMpptInCur",
|
||||
family_key,
|
||||
device_key,
|
||||
f'Speicher 2 MPPT 1 Eingangsstrom ',
|
||||
"current", #sensor['device_class'],
|
||||
None, #sensor['state_class'],
|
||||
"A", #sensor['native_unit_of_measurement']
|
||||
"battery1MpptData"
|
||||
))
|
||||
sensors.append(Sensor(
|
||||
coordinator,
|
||||
device_info,
|
||||
entry.entry_id,
|
||||
f"batteryMpptInPower",
|
||||
family_key,
|
||||
device_key,
|
||||
f'Speicher 2 MPPT 1 Eingangsleistung',
|
||||
"power", #sensor['device_class'],
|
||||
None, #sensor['state_class'],
|
||||
"W", #sensor['native_unit_of_measurement']
|
||||
"battery1MpptData"
|
||||
))
|
||||
if isinstance(device['battery2MpptData'], dict):
|
||||
sensors.append(Sensor(
|
||||
coordinator,
|
||||
device_info,
|
||||
entry.entry_id,
|
||||
f"batteryMpptInVol",
|
||||
family_key,
|
||||
device_key,
|
||||
f'Speicher 3 MPPT 1 Eingangsspannung',
|
||||
"voltage", #sensor['device_class'],
|
||||
None, #sensor['state_class'],
|
||||
"V", #sensor['native_unit_of_measurement']
|
||||
"battery2MpptData"
|
||||
))
|
||||
sensors.append(Sensor(
|
||||
coordinator,
|
||||
device_info,
|
||||
entry.entry_id,
|
||||
"batteryMpptInCur",
|
||||
family_key,
|
||||
device_key,
|
||||
f'Speicher 3 MPPT 1 Eingangsstrom ',
|
||||
"current", #sensor['device_class'],
|
||||
None, #sensor['state_class'],
|
||||
"A", #sensor['native_unit_of_measurement']
|
||||
"battery2MpptData"
|
||||
))
|
||||
sensors.append(Sensor(
|
||||
coordinator,
|
||||
device_info,
|
||||
entry.entry_id,
|
||||
f"batteryMpptInPower",
|
||||
family_key,
|
||||
device_key,
|
||||
f'Speicher 3 MPPT 1 Eingangsleistung',
|
||||
"power", #sensor['device_class'],
|
||||
None, #sensor['state_class'],
|
||||
"W", #sensor['native_unit_of_measurement']
|
||||
"battery2MpptData"
|
||||
))
|
||||
if isinstance(device['battery3MpptData'], dict):
|
||||
sensors.append(Sensor(
|
||||
coordinator,
|
||||
device_info,
|
||||
entry.entry_id,
|
||||
"batteryMpptInVol",
|
||||
family_key,
|
||||
device_key,
|
||||
f'Speicher 4 MPPT 1 Eingangsspannung',
|
||||
"voltage", #sensor['device_class'],
|
||||
None, #sensor['state_class'],
|
||||
"V", #sensor['native_unit_of_measurement']
|
||||
"battery3MpptData"
|
||||
))
|
||||
sensors.append(Sensor(
|
||||
coordinator,
|
||||
device_info,
|
||||
entry.entry_id,
|
||||
f"batteryMpptInCur",
|
||||
family_key,
|
||||
device_key,
|
||||
f'Speicher 4 MPPT 1 Eingangsstrom ',
|
||||
"current", #sensor['device_class'],
|
||||
None, #sensor['state_class'],
|
||||
"A", #sensor['native_unit_of_measurement']
|
||||
"battery3MpptData"
|
||||
))
|
||||
sensors.append(Sensor(
|
||||
coordinator,
|
||||
device_info,
|
||||
entry.entry_id,
|
||||
"batteryMpptInPower",
|
||||
family_key,
|
||||
device_key,
|
||||
f'Speicher 4 MPPT 1 Eingangsleistung',
|
||||
"power", #sensor['device_class'],
|
||||
None, #sensor['state_class'],
|
||||
"W", #sensor['native_unit_of_measurement']
|
||||
"battery3MpptData"
|
||||
))
|
||||
|
||||
|
||||
async_add_entities(sensors)
|
||||
|
||||
class Sensor(CoordinatorEntity, SensorEntity):
|
||||
def __init__(self, coordinator, deviceinfo, entry_id, id, family_key, device_key, name, device_class, state_class, native_unit_of_measurement, subkey=None):
|
||||
super().__init__(coordinator)
|
||||
self._entry_id = entry_id
|
||||
self._deviceinfo = deviceinfo
|
||||
self._family_key = family_key
|
||||
self._device_key = device_key
|
||||
self._subkey = subkey
|
||||
self._id = id
|
||||
|
||||
self._attr_name = name
|
||||
self._attr_unique_id = f"{entry_id}_{family_key}_{device_key}_{id}"
|
||||
if subkey:
|
||||
self._attr_unique_id += f"{entry_id}_{family_key}_{device_key}_{subkey}_{id}"
|
||||
self._attr_native_unit_of_measurement = native_unit_of_measurement
|
||||
self._attr_device_class = device_class
|
||||
self._attr_state_class = state_class
|
||||
|
||||
@property
|
||||
def native_value(self):
|
||||
data = get_merged_device_data(self.coordinator)
|
||||
if self._family_key not in data:
|
||||
return None
|
||||
family = data[self._family_key]
|
||||
if "devices" not in family or self._device_key not in family["devices"]:
|
||||
return None
|
||||
device = family["devices"][self._device_key]
|
||||
if self._subkey:
|
||||
if self._subkey in device and isinstance(device[self._subkey], dict):
|
||||
return device[self._subkey].get(self._id, 'Unbekannt')
|
||||
else:
|
||||
return device.get(self._subkey, [])[self._id] if isinstance(device.get(self._subkey, []), list) else device.get(self._subkey, {}).get(self._id, 'Unbekannt')
|
||||
return device.get(self._id, 'Unbekannt')
|
||||
|
||||
@property
|
||||
def device_info(self) -> DeviceInfo:
|
||||
return self._deviceinfo
|
30
config/custom_components/sunlit_solar/strings.json
Normal file
30
config/custom_components/sunlit_solar/strings.json
Normal file
@ -0,0 +1,30 @@
|
||||
{
|
||||
"config": {
|
||||
"step": {
|
||||
"user": {
|
||||
"title": "Add Group",
|
||||
"description": "Some description",
|
||||
"data": {
|
||||
"username": "Benutzername",
|
||||
"password": "Kennwort",
|
||||
"scan_interval": "Abruf-Interval"
|
||||
},
|
||||
"data_description": {
|
||||
"scan_interval": "Abruf alle x Sekunden"
|
||||
},
|
||||
"sections": {
|
||||
"additional_options": {
|
||||
"name": "Additional options",
|
||||
"description": "A description of the section",
|
||||
"data": {
|
||||
"advanced_group_option": "Advanced group option"
|
||||
},
|
||||
"data_description": {
|
||||
"advanced_group_option": "A very complicated option which does abc"
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
Loading…
x
Reference in New Issue
Block a user