Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- import logging
- import requests
- import voluptuous as vol
- from homeassistant.components.sensor import (PLATFORM_SCHEMA, ENTITY_ID_FORMAT)
- from homeassistant.const import CONF_MONITORED_CONDITIONS, CONF_NAME, CONF_ADDRESS, CONF_DEVICE_ID, ATTR_ATTRIBUTION
- import homeassistant.helpers.config_validation as cv
- from homeassistant.helpers.entity import Entity
- from homeassistant.helpers.entity import async_generate_entity_id
- _LOGGER = logging.getLogger(__name__)
- DEFAULT_NAME = 'Fronius'
- SENSOR_TYPES = {
- 'fac': {"jsonPath": "Body.Data.FAC.Value", "name": "FAC", "unit": "Hz", "icon": "mdi:weather-lightning"},
- 'iac': {"jsonPath": "Body.Data.IAC.Value", "name": "IAC", "unit": "A", "icon": "mdi:weather-lightning"},
- 'status_code': {"jsonPath": "Body.Data.DeviceStatus.StatusCode", "name": "Status Code", "unit": " ", "icon": "mdi:weather-lightning"},
- 'day_energy': {"jsonPath": "Body.Data.DAY_ENERGY.Value", "name": "Day Energy", "unit": "Wh", "icon": "mdi:weather-lightning"},
- }
- PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
- vol.Required(CONF_NAME, default=DEFAULT_NAME): cv.string,
- vol.Required(CONF_ADDRESS): cv.string,
- vol.Required(CONF_DEVICE_ID): cv.string,
- vol.Required(CONF_MONITORED_CONDITIONS, default=[]):
- vol.All(cv.ensure_list, [vol.In(SENSOR_TYPES)])
- })
- def setup_platform(hass, config, add_entities, discovery_info=None):
- address = config.get(CONF_ADDRESS)
- device_id = config.get(CONF_DEVICE_ID)
- name = config.get(CONF_NAME)
- dev = []
- for monitored_condition in config[CONF_MONITORED_CONDITIONS]:
- uid = '{}_{}_{}'.format(DEFAULT_NAME, device_id, monitored_condition)
- entity_id = async_generate_entity_id(ENTITY_ID_FORMAT, uid, hass=hass)
- dev.append(FroniusSensor(entity_id, name, address, device_id, monitored_condition))
- add_entities(dev, True)
- class FroniusSensor(Entity):
- def __init__(self, entity_id, name, address, device_id, sensor_type):
- self.entity_id = entity_id
- self._platform_name = name
- self._address = address
- self._device_id = device_id
- self.sensor_type = sensor_type
- self._data = None
- self._state = None
- self._jsonPath = SENSOR_TYPES[sensor_type]["jsonPath"]
- self._sensor_name = SENSOR_TYPES[sensor_type]["name"]
- self._unit_of_measurement = SENSOR_TYPES[sensor_type]["unit"]
- self._icon = SENSOR_TYPES[sensor_type]["icon"]
- @property
- def name(self):
- return '{} - {}'.format(self._platform_name, self._sensor_name)
- @property
- def icon(self):
- return self._icon
- @property
- def state(self):
- if self._data is not None:
- self._state = FroniusSensor.extractor(self._data, self._jsonPath)
- return self._state
- @property
- def unit_of_measurement(self):
- return self._unit_of_measurement
- def update(self):
- address = '{}/solar_api/v1/GetInverterRealtimeData.cgi?Scope=Device&DeviceId={}&DataCollection=CommonInverterData'.format(
- self._address, self._device_id)
- request = requests.get(address)
- if request.status_code == 200 and request.content.__len__() > 0:
- self._data = request.json()
- @staticmethod
- def extractor(json, path):
- def extractor_arr(json_obj, path_array):
- if len(path_array) > 1:
- return extractor_arr(json_obj[path_array[0]], path_array[1:])
- return json_obj[path_array[0]]
- return extractor_arr(json, path.split("."))
Add Comment
Please, Sign In to add comment