"""PySAJ interacts as a library to communicate with SAJ inverters""" import aiohttp import asyncio import concurrent import csv from io import StringIO from datetime import date import logging import xml.etree.ElementTree as ET _LOGGER = logging.getLogger(__name__) MAPPER_STATES = { "0": "Not connected", "1": "Waiting", "2": "Normal", "3": "Error", "4": "Upgrading", } URL_PATH_ETHERNET = "real_time_data.xml" URL_PATH_ETHERNET_INFO = "equipment_data.xml" URL_PATH_WIFI = "status/status.php" URL_PATH_WIFI_INFO = "info.php" class Sensor(object): """Sensor definition""" def __init__(self, key, csv_1_key, csv_2_key, factor, name, unit='', per_day_basis=False, per_total_basis=False): self.key = key self.csv_1_key = csv_1_key self.csv_2_key = csv_2_key self.factor = factor self.name = name self.unit = unit self.value = None self.per_day_basis = per_day_basis self.per_total_basis = per_total_basis self.date = date.today() self.enabled = False class Sensors(object): """SAJ sensors""" def __init__(self, wifi=False): self.__s = [] self.add( ( Sensor("p-ac", 11, 23, "", "current_power", "W"), Sensor("e-today", 3, 3, "/100", "today_yield", "kWh", True), Sensor("e-total", 1, 1, "/100", "total_yield", "kWh", False, True), Sensor("t-today", 4, 4, "/10", "today_time", "h", True), Sensor("t-total", 2, 2, "/10", "total_time", "h", False, True), Sensor("CO2", 21, 33, "/10", "total_co2_reduced", "kg", False, True), Sensor("temp", 20, 32, "/10", "temperature", "°C"), Sensor("state", 22, 34, "", "state"), Sensor("maxPower", -1, -1, "", "today_max_current", "W", True) ) ) def __len__(self): """Length.""" return len(self.__s) def __contains__(self, key): """Get a sensor using either the name or key.""" try: if self[key]: return True except KeyError: return False def __getitem__(self, key): """Get a sensor using either the name or key.""" for sen in self.__s: if sen.name == key or sen.key == key: return sen raise KeyError(key) def __iter__(self): """Iterator.""" return self.__s.__iter__() def add(self, sensor): """Add a sensor, warning if it exists.""" if isinstance(sensor, (list, tuple)): for sss in sensor: self.add(sss) return if not isinstance(sensor, Sensor): raise TypeError("pysaj.Sensor expected") if sensor.name in self: old = self[sensor.name] self.__s.remove(old) _LOGGER.warning("Replacing sensor %s with %s", old, sensor) if sensor.key in self: _LOGGER.warning("Duplicate SAJ sensor key %s", sensor.key) self.__s.append(sensor) class SAJ(object): """Provides access to SAJ inverter data""" def __init__(self, host, wifi=False, username='admin', password='admin'): self.host = host self.wifi = wifi self.username = username self.password = password self.serialnumber = "XXXXXXXXXXXXXXXXX" self.url = "http://{0}/".format(self.host) if self.wifi: if (len(self.username) > 0 and len(self.password) > 0): self.url = "http://{0}:{1}@{2}/".format(self.username, self.password, self.host) self.url_info = self.url + URL_PATH_WIFI_INFO self.url += URL_PATH_WIFI else: self.url_info = self.url + URL_PATH_ETHERNET_INFO self.url += URL_PATH_ETHERNET async def read(self, sensors): """Returns necessary sensors from SAJ inverter""" try: timeout = aiohttp.ClientTimeout(total=5) async with aiohttp.ClientSession(timeout=timeout, raise_for_status=True) as session: current_url = self.url_info async with session.get(current_url) as response: data = await response.text() if self.wifi: csv_data = StringIO(data) reader = csv.reader(csv_data) for row in reader: self.serialnumber = row.pop(0) else: xml = ET.fromstring(data) find = xml.find("SN") if find is not None: self.serialnumber = find.text _LOGGER.debug("Inverter SN: %s", self.serialnumber) current_url = self.url async with session.get(current_url) as response: data = await response.text() at_least_one_enabled = False if self.wifi: csv_data = StringIO(data) reader = csv.reader(csv_data) ncol = len(next(reader)) csv_data.seek(0) values = [] for row in reader: for (i, v) in enumerate(row): values.append(v) for sen in sensors: if ncol < 24: if sen.csv_1_key != -1: try: v = values[sen.csv_1_key] except IndexError: v = None else: v = None else: if sen.csv_2_key != -1: try: v = values[sen.csv_2_key] except IndexError: v = None else: v = None if v is not None: if sen.name == "state": sen.value = MAPPER_STATES[v] else: sen.value = eval( "{0}{1}".format(v, sen.factor) ) sen.date = date.today() sen.enabled = True at_least_one_enabled = True else: xml = ET.fromstring(data) for sen in sensors: find = xml.find(sen.key) if find is not None: sen.value = find.text sen.date = date.today() sen.enabled = True at_least_one_enabled = True if not at_least_one_enabled: if self.wifi: raise csv.Error else: raise ET.ParseError if sen.enabled: _LOGGER.debug("Got new value for sensor %s: %s", sen.name, sen.value) return True except (aiohttp.client_exceptions.ClientConnectorError, concurrent.futures._base.TimeoutError): # Connection to inverter not possible. # This can be "normal" - so warning instead of error - as SAJ # inverters are powered by DC and thus have no power after the sun # has set. _LOGGER.warning("Connection to SAJ inverter is not possible. " + "The inverter may be offline due to darkness. " + "Otherwise check host/ip address.") return False except aiohttp.client_exceptions.ClientResponseError as err: # 401 Unauthorized: wrong username/password if err.status == 401: raise UnauthorizedException(err) else: raise UnexpectedResponseException(err) except csv.Error: # CSV is not valid raise UnexpectedResponseException( str.format("No valid CSV received from {0} at {1}", self.host, current_url) ) except ET.ParseError: # XML is not valid or even no XML at all raise UnexpectedResponseException( str.format("No valid XML received from {0} at {1}", self.host, current_url) ) class UnauthorizedException(Exception): """Exception for Unauthorized 401 status code""" def __init__(self, message): Exception.__init__(self, message) class UnexpectedResponseException(Exception): """Exception for unexpected status code""" def __init__(self, message): Exception.__init__(self, message)