diff --git a/pyblnet/__init__.py b/pyblnet/__init__.py index b3840c4..2bfd39f 100644 --- a/pyblnet/__init__.py +++ b/pyblnet/__init__.py @@ -12,9 +12,9 @@ VERSION = (0, 9, 1) -__version__ = '.'.join([str(i) for i in VERSION]) -__author__ = 'nielstron' -__author_email__ = 'n.muendler@web.de' -__copyright__ = 'Copyright (C) 2019 nielstron' +__version__ = ".".join([str(i) for i in VERSION]) +__author__ = "nielstron" +__author_email__ = "n.muendler@web.de" +__copyright__ = "Copyright (C) 2019 nielstron" __license__ = "MIT" __url__ = "https://github.com/nielstron/pyblnet" diff --git a/pyblnet/blnet.py b/pyblnet/blnet.py index 6e30da9..b894435 100644 --- a/pyblnet/blnet.py +++ b/pyblnet/blnet.py @@ -19,7 +19,7 @@ class BLNET(object): """ General high-level BLNET interface. - + Abstracts from actual type of connection to the BLNET and provides as much functionality as possible. Attributes: @@ -33,33 +33,36 @@ class BLNET(object): use_ta boolean about whether to make use of the (buggy) PC-BLNET interface """ - def __init__(self, - address, - web_port=80, - password=None, - ta_port=40000, - timeout=5, - max_retries=5, - use_web=True, - use_ta=False): + def __init__( + self, + address, + web_port=80, + password=None, + ta_port=40000, + timeout=5, + max_retries=5, + use_web=True, + use_ta=False, + ): """ If a connection (Web or TA/Direct) should not be used, set the corresponding use_* to False Params: @param ta_port: Port for direct TCP Connection """ - assert (isinstance(address, str)) - assert (web_port is None or isinstance(web_port, int)) - assert (ta_port is None or isinstance(ta_port, int)) - assert (timeout is None or isinstance(timeout, int)) + assert isinstance(address, str) + assert web_port is None or isinstance(web_port, int) + assert ta_port is None or isinstance(ta_port, int) + assert timeout is None or isinstance(timeout, int) self.address = address self.timeout = timeout self.max_retries = max_retries self.blnet_web = None self.blnet_direct = None if use_web: - self.blnet_web = BLNETWeb("{}:{}".format(address, web_port), - password, timeout) + self.blnet_web = BLNETWeb( + "{}:{}".format(address, web_port), password, timeout + ) if use_ta: # The address might not have a resulting hostname # especially not if not prefixed with http:// @@ -73,33 +76,31 @@ def fetch(self, node=None): (defaults to active node on the device) """ data = { - 'analog': {}, - 'digital': {}, - 'speed': {}, - 'energy': {}, - 'power': {}, + "analog": {}, + "digital": {}, + "speed": {}, + "energy": {}, + "power": {}, } if self.blnet_web: with self.blnet_web as blnet_session: if node is not None: blnet_session.set_node(node) - data['analog'] = self._convert_web( - blnet_session.read_analog_values()) - data['digital'] = self._convert_web( - blnet_session.read_digital_values()) + data["analog"] = self._convert_web(blnet_session.read_analog_values()) + data["digital"] = self._convert_web(blnet_session.read_digital_values()) if self.blnet_direct: direct = self.blnet_direct.get_latest(self.max_retries)[0] # Override values for analog and digital as values are # expected to be more precise here - for domain in ['analog', 'digital']: + for domain in ["analog", "digital"]: for id, value in direct[domain].items(): if data[domain].get(id) is not None: - data[domain][id]['value'] = value - for domain in ['speed', 'energy', 'power']: + data[domain][id]["value"] = value + for domain in ["speed", "energy", "power"]: for id, value in direct[domain].items(): if value is None: continue - data[domain][id] = {'value': value} + data[domain][id] = {"value": value} return data def turn_on(self, digital_id, can_node=None): @@ -107,59 +108,61 @@ def turn_on(self, digital_id, can_node=None): Turn switch with given id on given node on Return: no error during set operation """ - return self._turn(digital_id, 'EIN', can_node) + return self._turn(digital_id, "EIN", can_node) def turn_off(self, digital_id, can_node=None): """ Turn switch with given id on given node off Return: no error during set operation """ - return self._turn(digital_id, 'AUS', can_node) + return self._turn(digital_id, "AUS", can_node) def turn_auto(self, digital_id, can_node=None): """ Turn switch with given id on given node to "AUTO"/ give control over to UVR Return: no error during set operation """ - return self._turn(digital_id, 'AUTO', can_node) + return self._turn(digital_id, "AUTO", can_node) def _turn(self, digital_id, value, can_node=None): if self.blnet_web: with self.blnet_web as blnet_session: if not blnet_session.logged_in(): - raise ConnectionError('Could not log in') + raise ConnectionError("Could not log in") if can_node is not None: if not blnet_session.set_node(can_node): - raise ConnectionError('Could not set node') + raise ConnectionError( + "Could not set can node to {}".format(can_node) + ) if not blnet_session.set_digital_value(digital_id, value): - raise ConnectionError('Failed to set value') + raise ConnectionError( + "Failed to set digital switch {} to value {}".format( + digital_id, value + ) + ) return True else: - raise EnvironmentError('Can\'t set values with blnet web disabled') - - def get_value(self, - name=None, - id=None, - type='digital', - ret='value', - cached=None): + raise EnvironmentError("Can't set values with blnet web disabled") + + def get_value(self, name=None, id=None, type="digital", ret="value", cached=None): """ - higher level interface to get a value or mode by name or id - ret: can be 'value', 'mode' - returns the value if ret='value' or the mode if ret='mode' as first return value - and the dictionary containing name, value, id and mode as a second return value - cached: in order to prevent polling data from BLNet with every call, - the data can be fetched once and stored and passed to this function + higher level interface to get a value or mode by name or id + ret: can be 'value', 'mode' + returns the value if ret='value' or the mode if ret='mode' as first return value + and the dictionary containing name, value, id and mode as a second return value + cached: in order to prevent polling data from BLNet with every call, + the data can be fetched once and stored and passed to this function """ val = None dic = None - if name is None and id is None: return val + if name is None and id is None: + return val if cached is None: cached = self.fetch() for key, v in cached[type].items(): - if str(v['name']) == str(name) or str(v['id']) == str(id): + if str(v["name"]) == str(name) or str(v["id"]) == str(id): val = v[ret] dic = v @@ -167,27 +170,33 @@ def get_value(self, def get_digital_value(self, name=None, id=None, cached=None): return self.get_value( - type='digital', ret='value', name=name, id=id, cached=cached) + type="digital", ret="value", name=name, id=id, cached=cached + ) def get_digital_mode(self, name=None, id=None, cached=None): return self.get_value( - type='digital', ret='mode', name=name, id=id, cached=cached) + type="digital", ret="mode", name=name, id=id, cached=cached + ) def get_analog_value(self, name=None, id=None, cached=None): return self.get_value( - type='analog', ret='value', name=name, id=id, cached=cached) + type="analog", ret="value", name=name, id=id, cached=cached + ) def get_energy_value(self, name=None, id=None, cached=None): return self.get_value( - type='energy', ret='value', name=name, id=id, cached=cached) + type="energy", ret="value", name=name, id=id, cached=cached + ) def get_speed_value(self, name=None, id=None, cached=None): return self.get_value( - type='speed', ret='value', name=name, id=id, cached=cached) + type="speed", ret="value", name=name, id=id, cached=cached + ) def get_power_value(self, name=None, id=None, cached=None): return self.get_value( - type='power', ret='value', name=name, id=id, cached=cached) + type="power", ret="value", name=name, id=id, cached=cached + ) @staticmethod def _convert_web(values): @@ -197,7 +206,7 @@ def _convert_web(values): data = {} try: for sensor in values: - data[int(sensor['id'])] = sensor + data[int(sensor["id"])] = sensor except TypeError: pass return data diff --git a/pyblnet/blnet_conn.py b/pyblnet/blnet_conn.py index 2d7ad75..b98972c 100644 --- a/pyblnet/blnet_conn.py +++ b/pyblnet/blnet_conn.py @@ -36,7 +36,7 @@ class BLNETDirect(object): """ A class for establishing a direct connection to the BLNET (rather than scraping the web interface) - + It uses this protocol and is still very buggy www.haus-terra.at/heizung/download/Schnittstelle/Schnittstelle_PC_Bootloader.pdf """ @@ -48,9 +48,9 @@ def __init__(self, address, port=40000, reset=False): @param port: integer, Port of the bl-net device for connection @param reset: boolean, delete data on BL-Net after data receive """ - assert (isinstance(address, str)) - assert (isinstance(port, int)) - assert (isinstance(reset, bool)) + assert isinstance(address, str) + assert isinstance(port, int) + assert isinstance(reset, bool) self.address = address self.port = port self.reset = reset @@ -70,56 +70,60 @@ def get_count(self): if self._checksum(data): if self._mode == CAN_MODE: - frame_count = struct.unpack('<{}B'.format(len(data)), - data)[5] - (type, version, timestamp, frame_count, _, start_address, - end_address, checksum) = struct.unpack( - ' start_address: # calculate count - self._count = ((end_address - start_address) / - self._address_inc) + 1 + self._count = ( + (end_address - start_address) / self._address_inc + ) + 1 else: self._count = ( - (self._address_end + start_address - end_address) / - self._address_inc) + self._address_end + start_address - end_address + ) / self._address_inc self._address = end_address if self._count: self._count = int(self._count) return self._count else: - raise ConnectionError('Could not retreive count') + raise ConnectionError("Could not retreive count") def _get_data(self, max_count=None): data = [] @@ -139,14 +143,14 @@ def _check_mode(self): """ Check if Bootloader Mode is supported @throws ConnectionError Mode not supported - """ + """ self._connect() self._mode = self._query(GET_MODE, 1) self._disconnect() if self._mode in [CAN_MODE, DL2_MODE, DL_MODE]: return True - raise ConnectionError('BL-Net mode is not supported') + raise ConnectionError("BL-Net mode is not supported") def _connect(self): """ @@ -154,8 +158,9 @@ def _connect(self): @throws ConnectionError Connection failed """ if self._socket is None: - available = getaddrinfo(self.address, self.port, 0, SOCK_STREAM, - IPPROTO_TCP) + available = getaddrinfo( + self.address, self.port, 0, SOCK_STREAM, IPPROTO_TCP + ) for (family, socktype, proto, _, sockaddr) in available: try: self._socket = socket(family, socktype, proto) @@ -164,7 +169,7 @@ def _connect(self): except: self._socket = None if self._socket is None: - raise ConnectionError('Could not connect to BLNET') + raise ConnectionError("Could not connect to BLNET") def _disconnect(self): """ @@ -192,8 +197,7 @@ def _query(self, command, length): return data self._disconnect() - raise ConnectionError( - 'Error while querying command {}'.format(command)) + raise ConnectionError("Error while querying command {}".format(command)) def _start_read(self): """ @@ -208,7 +212,7 @@ def _checksum(self, data): @param byte string $data Binary string to check @return boolean """ - binary = struct.unpack('<{}B'.format(len(data)), data) + binary = struct.unpack("<{}B".format(len(data)), data) checksum = binary[-1] sum = 0 @@ -229,14 +233,21 @@ def _fetch_data(self): # build address for bootloader addresses = [ - self._address & 0xFF, (self._address & 0x7F00) >> 7, - (self._address & 0xFF8000) >> 15 + self._address & 0xFF, + (self._address & 0x7F00) >> 7, + (self._address & 0xFF8000) >> 15, ] # build command - command = struct.pack('<6B', READ_DATA, addresses[0], addresses[1], - addresses[2], 1, - (READ_DATA + 1 + sum(addresses)) % 256) + command = struct.pack( + "<6B", + READ_DATA, + addresses[0], + addresses[1], + addresses[2], + 1, + (READ_DATA + 1 + sum(addresses)) % 256, + ) data = self._query(command, self._fetch_size) @@ -247,7 +258,7 @@ def _fetch_data(self): self._address = self._address_end self._count -= 1 return self._split_datasets(data) - raise ConnectionError('Could not retreive data') + raise ConnectionError("Could not retreive data") def _split_datasets(self, data): """ @@ -258,21 +269,22 @@ def _split_datasets(self, data): frames = {} if self._mode == CAN_MODE: for frame in range(0, self._can_frames): - frames[frame] = BLNETParser(data[3 + DATASET_SIZE * frame:3 + - DATASET_SIZE * (frame + 1)]) + frames[frame] = BLNETParser( + data[3 + DATASET_SIZE * frame : 3 + DATASET_SIZE * (frame + 1)] + ) elif self._mode == DL_MODE: frames[0] = BLNETParser(data[:DATASET_SIZE]) elif self._mode == DL2_MODE: frames[0] = BLNETParser(data[:DATASET_SIZE]) frames[1] = BLNETParser( - data[3 + DATASET_SIZE:3 + DATASET_SIZE + DATASET_SIZE]) + data[3 + DATASET_SIZE : 3 + DATASET_SIZE + DATASET_SIZE] + ) start = 0 if self._mode == CAN_MODE: start = 3 - if data[start:start + DATASET_SIZE - - 6] == b'0x00' * (DATASET_SIZE - 6): + if data[start : start + DATASET_SIZE - 6] == b"0x00" * (DATASET_SIZE - 6): return False else: return {k: v.to_dict() for k, v in frames.items()} @@ -284,11 +296,11 @@ def _end_read(self, success=True): self._connect() # Send end read command if self._query(END_READ, 1) != END_READ: - raise ConnectionError('End read command failed') + raise ConnectionError("End read command failed") # reset data if configured if success and self.reset: if self._query(RESET_DATA, 1) != RESET_DATA: - raise ConnectionError('Reset memory failed') + raise ConnectionError("Reset memory failed") self._count = None self._address = None self._disconnect() @@ -303,7 +315,7 @@ def get_latest(self, max_retries=MAX_RETRYS): self._connect() self.get_count() frames = {} - info = {'sleep': {}, 'got': {}} + info = {"sleep": {}, "got": {}} # for all frames for frame in range(0, self._can_frames): @@ -324,34 +336,34 @@ def get_latest(self, max_retries=MAX_RETRYS): sleep(binary[1]) self._connect() else: - info['got'][frame] = n + info["got"][frame] = n frames.update(self._split_latest(data, frame)) break # TODO this looks suspicious if n == max_retries - 1: - frames[frame] = 'timeout' - info['sleep'][frame] = sleeps + frames[frame] = "timeout" + info["sleep"][frame] = sleeps self._end_read(True) if len(frames) > 0: - frames['date'] = datetime.now() - frames['info'] = info + frames["date"] = datetime.now() + frames["info"] = info return frames raise ConnectionError("Could not get latest data") def _split_latest(self, data, frame): """ - Split binary string by fetch latest + Split binary string by fetch latest @param data: bytestring @param frame: int @return Array of frame -> value mappings """ frames = {} if self._mode == CAN_MODE: - frames[frame] = BLNETParser(data[1:LATEST_SIZE + 1]) + frames[frame] = BLNETParser(data[1 : LATEST_SIZE + 1]) elif self._mode == DL_MODE: - frames[0] = BLNETParser(data[1:LATEST_SIZE + 1]) + frames[0] = BLNETParser(data[1 : LATEST_SIZE + 1]) elif self._mode == DL2_MODE: - frames[0] = BLNETParser(data[1:LATEST_SIZE + 1]) - frames[1] = BLNETParser(data[LATEST_SIZE + 1:2 * LATEST_SIZE + 1]) + frames[0] = BLNETParser(data[1 : LATEST_SIZE + 1]) + frames[1] = BLNETParser(data[LATEST_SIZE + 1 : 2 * LATEST_SIZE + 1]) return {k: v.to_dict() for k, v in frames.items()} diff --git a/pyblnet/blnet_parser.py b/pyblnet/blnet_parser.py index 95e79ef..d52a51c 100644 --- a/pyblnet/blnet_parser.py +++ b/pyblnet/blnet_parser.py @@ -46,25 +46,33 @@ def __init__(self, data): # (fetched from bootloader storage) if len(data) == 61: (_, seconds, minutes, hours, days, months, years) = struct.unpack( - '<55sBBBBBB', data) - self.date = datetime(2000 + years, months, days, hours, minutes, - seconds) + "<55sBBBBBB", data + ) + self.date = datetime(2000 + years, months, days, hours, minutes, seconds) # Only parse preceding data data = data[:55] power = [0, 0] kWh = [0, 0] MWh = [0, 0] - (_, digital, speed, active, power[0], kWh[0], MWh[0], power[1], kWh[1], - MWh[1]) = struct.unpack('<32sH4sBLHHLHH', data) - - analog = struct.unpack( - '<{}{}'.format('H' * 16, 'x' * (len(data) - 32)), data) + ( + _, + digital, + speed, + active, + power[0], + kWh[0], + MWh[0], + power[1], + kWh[1], + MWh[1], + ) = struct.unpack("<32sH4sBLHHLHH", data) + + analog = struct.unpack("<{}{}".format("H" * 16, "x" * (len(data) - 32)), data) self.analog = {} for channel in range(0, 16): - self.analog[channel + 1] = round( - self._convert_analog(analog[channel]), 3) + self.analog[channel + 1] = round(self._convert_analog(analog[channel]), 3) self.digital = {} for channel in range(0, 16): @@ -72,19 +80,19 @@ def __init__(self, data): self.speed = {} for channel in range(0, 4): - self.speed[channel + 1] = round( - self._convert_speed(speed[channel]), 3) + self.speed[channel + 1] = round(self._convert_speed(speed[channel]), 3) self.energy = {} for channel in range(0, 2): self.energy[channel + 1] = round( - self._convert_energy(MWh[channel], kWh[channel], active, - channel), 3) + self._convert_energy(MWh[channel], kWh[channel], active, channel), 3 + ) self.power = {} for channel in range(0, 2): self.power[channel + 1] = round( - self._convert_power(power[channel], active, channel), 3) + self._convert_power(power[channel], active, channel), 3 + ) def to_dict(self): """ @@ -150,16 +158,13 @@ def _convert_power(self, value, active, position): @return its power """ if active & position: - return self._calculate_value(value, 1 / 2560, INT32_MASK, - INT32_SIGN) + return self._calculate_value(value, 1 / 2560, INT32_MASK, INT32_SIGN) else: return None - def _calculate_value(self, - value, - multiplier=1, - positive_mask=POSITIVE_VALUE_MASK, - signbit=SIGN_BIT): + def _calculate_value( + self, value, multiplier=1, positive_mask=POSITIVE_VALUE_MASK, signbit=SIGN_BIT + ): result = value & positive_mask if value & signbit: result = -((result ^ positive_mask) + 1) diff --git a/pyblnet/blnet_web.py b/pyblnet/blnet_web.py index 8e50de1..93231a6 100644 --- a/pyblnet/blnet_web.py +++ b/pyblnet/blnet_web.py @@ -32,23 +32,24 @@ def test_blnet(ip, timeout=5, id=0): # Parse DOM object from HTMLCode dom = htmldom.HtmlDom().createDom(r.text) # either a 'Zugriff verweigert' message is shown - if 'BL-Net'.lower() in dom.find('title').text().lower(): + if "BL-Net".lower() in dom.find("title").text().lower(): return True # or (more often) the BL-NET Menü - if 'BL-NET' in dom.find('div#head').text(): + if "BL-NET" in dom.find("div#head").text(): return True return False class BLNETWeb(object): """ - Interface for connecting with, collecting data from and controlling the BLNet + Interface for connecting with, collecting data from and controlling the BLNet via it's HTTP-interface Attributes: ip the ip/domain of the BL-Net to connect to password the password to log into the web interface provided timeout timeout for http requests """ + ip = "" _def_password = "0128" # default password is 0128 password = "" @@ -58,14 +59,13 @@ def __init__(self, ip, password=_def_password, timeout=5): """ Constructor """ - assert (isinstance(ip, str)) - assert (password is None or isinstance(password, str)) - assert (timeout is None or isinstance(timeout, int)) + assert isinstance(ip, str) + assert password is None or isinstance(password, str) + assert timeout is None or isinstance(timeout, int) if not ip.startswith("http://") and not ip.startswith("https://"): ip = "http://" + ip if not test_blnet(ip): - raise ValueError( - 'No BLNET found under given address: {}'.format(ip)) + raise ValueError("No BLNET found under given address: {}".format(ip)) self.ip = ip self.password = password self._timeout = timeout @@ -93,41 +93,39 @@ def logged_in(self): # so that it can be quickly loaded self.ip + "/par.htm?blp=A1200101&1238653", headers=self.cookie_header(), - timeout=self._timeout) + timeout=self._timeout, + ) except requests.exceptions.RequestException: return False - return r.headers.get('Set-Cookie') is not None + return r.headers.get("Set-Cookie") is not None def cookie_header(self): """ Creates the header to pass the session TAID as cookie """ - headers = {'Cookie': self.current_taid} + headers = {"Cookie": self.current_taid} return headers def log_in(self): """ Logs into the BLNET interface, renews the TAID - + Return: Login successful """ if self.logged_in(): return True - payload = { - 'blu': 1, # log in as experte - 'blp': self.password, - 'bll': "Login" - } - headers = {'Content-Type': 'application/x-www-form-urlencoded'} + payload = {"blu": 1, "blp": self.password, "bll": "Login"} # log in as experte + headers = {"Content-Type": "application/x-www-form-urlencoded"} try: r = requests.post( - self.ip + '/main.html', + self.ip + "/main.html", data=payload, headers=headers, - timeout=self._timeout) + timeout=self._timeout, + ) except requests.exceptions.RequestException: return False - self.current_taid = r.headers.get('Set-Cookie') + self.current_taid = r.headers.get("Set-Cookie") # try two times to log in i = 0 while i < 2: @@ -139,7 +137,7 @@ def log_in(self): def log_out(self): """ Logs out of the BLNET interface - + Return: successful log out """ if self.password is None: @@ -148,7 +146,8 @@ def log_out(self): requests.get( self.ip + "/main.html?blL=1", headers=self.cookie_header(), - timeout=self._timeout) + timeout=self._timeout, + ) except requests.exceptions.RequestException: return False return not self.logged_in() @@ -157,7 +156,7 @@ def set_node(self, node): """ Selects the node at which the UVR of interest lies future requests will be sent at this particular UVR - + Return: Still logged in (indicating successful node change) """ # send the request to change the node @@ -165,11 +164,12 @@ def set_node(self, node): r = requests.get( self.ip + "/can.htm?blaA=" + str(node), headers=self.cookie_header(), - timeout=self._timeout) + timeout=self._timeout, + ) except requests.exceptions.RequestException: return False # return whether we we're still logged in => setting went well - return self.password is None or r.headers.get('Set-Cookie') is not None + return self.password is None or r.headers.get("Set-Cookie") is not None def read_analog_values(self): """ @@ -180,13 +180,14 @@ def read_analog_values(self): r = requests.get( self.ip + "/580500.htm", headers=self.cookie_header(), - timeout=self._timeout) + timeout=self._timeout, + ) except requests.exceptions.RequestException: return None # Parse DOM object from HTMLCode dom = htmldom.HtmlDom().createDom(r.text) # Check if we didn't fail in access - #if 'BL-Net Zugang verweigert' in dom.find('title').text(): + # if 'BL-Net Zugang verweigert' in dom.find('title').text(): # return None # get the element containing the interesting information dom = dom.find("div.c")[1] @@ -202,7 +203,8 @@ def read_analog_values(self): # search for data by regular expression match_iter = re.finditer( r"(?P\d+): (?P.+)\n( ){3,6}(?P(- )?\d+,\d+) (?P.+?)   PAR?", - data_raw) + data_raw, + ) # parse a dict of the match and save them all in a list for match in match_iter: match_dict = match.groupdict() @@ -222,20 +224,21 @@ def read_analog_values(self): def read_digital_values(self): """ Reads all digital values (switches) from the web interface - and returns list of quadruples of id, name, mode (AUTO/HAND), value + and returns list of quadruples of id, name, mode (AUTO/HAND), value (EIN/AUS) """ try: r = requests.get( self.ip + "/580600.htm", headers=self.cookie_header(), - timeout=self._timeout) + timeout=self._timeout, + ) except requests.exceptions.RequestException: return None # Parse DOM object from HTMLCode dom = htmldom.HtmlDom().createDom(r.text) # Check if we didn't fail in access - if 'BL-Net Zugang verweigert' in dom.find('title').text(): + if "BL-Net Zugang verweigert" in dom.find("title").text(): return None # get the element containing the interesting information dom = dom.find("div.c")[1] @@ -249,7 +252,8 @@ def read_digital_values(self): # search for data by regular expression match_iter = re.finditer( r"(?P\d+): (?P.+)\n    (?P(AUTO|HAND))/(?P(AUS|EIN))", - data_raw) + data_raw, + ) # parse a dict of the match and save them all in a list for match in match_iter: match_dict = match.groupdict() @@ -263,7 +267,7 @@ def read_digital_values(self): def set_digital_value(self, digital_id, value): """ - Sets a digital value with given id to given value + Sets a digital value with given id to given value Accepts 'EIN' and everything evaluating to True as well as 'AUS' and everything evaluating to False and 'AUTO' as values @@ -277,53 +281,59 @@ def set_digital_value(self, digital_id, value): # throw error for wrong id's if digital_id < 1: raise ValueError( - 'Device id can\'t be smaller than 1, was {}'.format( - digital_id)) + "Device id can't be smaller than 1, was {}".format(digital_id) + ) if digital_id > 15: raise ValueError( - 'Device id can\'t be larger than 15, was {}'.format( - digital_id)) + "Device id can't be larger than 15, was {}".format(digital_id) + ) # transform input value to 'EIN' or 'AUS' if isinstance(value, str): - if value.lower() == 'AUTO'.lower() or value == '3': - value = '3' # 3 means auto - elif value.lower() == 'EIN'.lower() or value == '2' or value.lower( - ) == 'on'.lower(): - value = '2' # 2 means turn on - elif value.lower() == 'AUS'.lower() or value == '1' or value.lower( - ) == 'off'.lower(): - value = '1' # 1 means turn off + if value.lower() == "AUTO".lower() or value == "3": + value = "3" # 3 means auto + elif ( + value.lower() == "EIN".lower() + or value == "2" + or value.lower() == "on".lower() + ): + value = "2" # 2 means turn on + elif ( + value.lower() == "AUS".lower() + or value == "1" + or value.lower() == "off".lower() + ): + value = "1" # 1 means turn off else: raise ValueError("Illegal input string {}".format(value)) elif isinstance(value, int) and not isinstance(value, bool): if value in (1, 2, 3): value = str(value) elif value == 0: - value = '1' + value = "1" else: raise ValueError("Illegal input integer {}".format(value)) else: # value can be interpreted as a true value if value: - value = '2' # 2 means turn on + value = "2" # 2 means turn on else: - value = '1' # 1 means turn off - assert (value in ['1', '2', '3']) + value = "1" # 1 means turn off + assert value in ["1", "2", "3"] # convert id to hexvalue so that 10 etc become A... - hex_repr = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 'A', 'B', 'C', 'D', 'E', 'F'] + hex_repr = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, "A", "B", "C", "D", "E", "F"] if digital_id > 9: digital_id = hex_repr[digital_id] # submit data to website try: r = requests.get( - "{}/580600.htm?blw91A1200{}={}".format(self.ip, digital_id, - value), + "{}/580600.htm?blw91A1200{}={}".format(self.ip, digital_id, value), headers=self.cookie_header(), - timeout=self._timeout) + timeout=self._timeout, + ) except requests.exceptions.RequestException: return False # return whether we we're still logged in => setting went well - return self.password is None or r.headers.get('Set-Cookie') is not None + return self.password is None or r.headers.get("Set-Cookie") is not None diff --git a/pyblnet/tests/test_structure/blnet_mock_server.py b/pyblnet/tests/test_structure/blnet_mock_server.py index b3f0407..4323781 100644 --- a/pyblnet/tests/test_structure/blnet_mock_server.py +++ b/pyblnet/tests/test_structure/blnet_mock_server.py @@ -1,6 +1,7 @@ import os import urllib.parse from http.server import SimpleHTTPRequestHandler, HTTPServer + try: from http import HTTPStatus except ImportError: @@ -10,8 +11,8 @@ import posixpath from pathlib import Path -SERVER_DIR = Path(__file__).parent or Path('.') -PASSWORD = '0123' +SERVER_DIR = Path(__file__).parent or Path(".") +PASSWORD = "0123" COOKIE_RANGE = (0xAAAA, 0xFFFF) COOKIE_NUM = COOKIE_RANGE[1] - COOKIE_RANGE[0] @@ -78,21 +79,26 @@ def do_GET(self): self.send_error(403, "Access denied because server is blocked") return path = self.translate_path(self.path) - if (self.server.is_logged_in(self.headers.get("cookie")) - and "?blL=1" in self.requestline): + if ( + self.server.is_logged_in(self.headers.get("cookie")) + and "?blL=1" in self.requestline + ): self.server.log_out() # Only access that is allowed without login is main.html - if (not Path(path) == SERVER_DIR - and not Path(path) == SERVER_DIR.joinpath('main.htm') - and not Path(path) == SERVER_DIR.joinpath('main.html')): + if ( + not Path(path) == SERVER_DIR + and not Path(path) == SERVER_DIR.joinpath("main.htm") + and not Path(path) == SERVER_DIR.joinpath("main.html") + ): if not self.server.is_logged_in(self.headers.get("cookie")): self.send_error(403, "Not logged in, access denied") return # Parse node sets node_reg = re.compile( - r'[?&]blw91A1200(?P[0-9a-fA-F])=(?P[1-3])') + r"[?&]blw91A1200(?P[0-9a-fA-F])=(?P[1-3])" + ) for match in node_reg.finditer(self.path): - self.server.set_node(match.group('node'), match.group('value')) + self.server.set_node(match.group("node"), match.group("value")) # print(path) super().do_GET() @@ -103,9 +109,9 @@ def do_POST(self): pass # Result of self.rfile.read() if correct POST request: # b'blu=1&blp=0123&bll=Login' - perfect = b'blu=1&blp=0123&bll=Login' + perfect = b"blu=1&blp=0123&bll=Login" request_raw = self.rfile.read(len(perfect)) - request_string = request_raw.decode(encoding='utf-8') + request_string = request_raw.decode(encoding="utf-8") request_data = request_string.split("&") blu = False @@ -115,15 +121,18 @@ def do_POST(self): if query.startswith("blu"): if query.split("=")[1] != "1": self.send_error( - 403, - "Wrong user set: expected blu=1, got {}".format(query)) + 403, "Wrong user set: expected blu=1, got {}".format(query) + ) return blu = True elif query.startswith("blp"): if query.split("=")[1] != self.server.password: self.send_error( - 403, "Wrong password: expected blp={}, got {}".format( - self.server.password, query)) + 403, + "Wrong password: expected blp={}, got {}".format( + self.server.password, query + ), + ) return blp = True elif query.startswith("bll"): @@ -131,16 +140,16 @@ def do_POST(self): self.send_error( 403, "Wrong bll spec, expected bll=Login, got {}".format( - request_string)) + request_string + ), + ) return bll = True if not (blu and blp and bll): - self.send_error(403, - "Missing query param in {}".format(request_string)) + self.send_error(403, "Missing query param in {}".format(request_string)) return # Check for content type - if not self.headers.get( - 'Content-Type') == 'application/x-www-form-urlencoded': + if not self.headers.get("Content-Type") == "application/x-www-form-urlencoded": self.send_error(403, "Wrong content-type") return # All checks passed? set Set-Cookie header and do_GET @@ -149,7 +158,7 @@ def do_POST(self): if not cookie: self.send_error(403, "Previous user did not log out") return - self.headers.add_header('Cookie', cookie) + self.headers.add_header("Cookie", cookie) self.do_GET() def translate_path(self, path): @@ -162,16 +171,16 @@ def translate_path(self, path): only slightly changed method of the standard library """ # abandon query parameters - path = path.split('?', 1)[0] - path = path.split('#', 1)[0] + path = path.split("?", 1)[0] + path = path.split("#", 1)[0] # Don't forget explicit trailing slash when normalizing. Issue17324 - trailing_slash = path.rstrip().endswith('/') + trailing_slash = path.rstrip().endswith("/") try: - path = urllib.parse.unquote(path, errors='surrogatepass') + path = urllib.parse.unquote(path, errors="surrogatepass") except UnicodeDecodeError: path = urllib.parse.unquote(path) path = posixpath.normpath(path) - words = path.split('/') + words = path.split("/") words = filter(None, words) path = str(SERVER_DIR.absolute()) for word in words: @@ -180,7 +189,7 @@ def translate_path(self, path): continue path = os.path.join(path, word) if trailing_slash: - path += '/' + path += "/" return path def send_head(self): @@ -199,11 +208,10 @@ def send_head(self): f = None if os.path.isdir(path): parts = urllib.parse.urlsplit(self.path) - if not parts.path.endswith('/'): + if not parts.path.endswith("/"): # redirect browser - doing basically what apache does self.send_response(HTTPStatus.MOVED_PERMANENTLY) - new_parts = (parts[0], parts[1], parts[2] + '/', parts[3], - parts[4]) + new_parts = (parts[0], parts[1], parts[2] + "/", parts[3], parts[4]) new_url = urllib.parse.urlunsplit(new_parts) self.send_header("Location", new_url) self.end_headers() @@ -218,7 +226,7 @@ def send_head(self): return self.list_directory(path) ctype = self.guess_type(path) try: - f = open(path, 'rb') + f = open(path, "rb") except OSError: self.send_error(HTTPStatus.NOT_FOUND, "File not found") return None @@ -227,11 +235,10 @@ def send_head(self): self.send_header("Content-type", ctype) fs = os.fstat(f.fileno()) self.send_header("Content-Length", str(fs[6])) - self.send_header("Last-Modified", - self.date_time_string(fs.st_mtime)) + self.send_header("Last-Modified", self.date_time_string(fs.st_mtime)) # Addition: send cookie - if self.server.is_logged_in(self.headers.get('Cookie')): - self.send_header('Set-Cookie', self.server.logged_in) + if self.server.is_logged_in(self.headers.get("Cookie")): + self.send_header("Set-Cookie", self.server.logged_in) self.end_headers() return f except: @@ -248,23 +255,25 @@ def send_error(self, code, message=None, explain=None): """ self.log_error("code %d, message %s", code, message) self.send_response(code, message) - self.send_header('Connection', 'close') + self.send_header("Connection", "close") # Message body is omitted for cases described in: # - RFC7230: 3.3. 1xx, 204(No Content), 304(Not Modified) # - RFC7231: 6.3.6. 205(Reset Content) body = None - if (code >= 200 and - code not in (HTTPStatus.NO_CONTENT, HTTPStatus.RESET_CONTENT, - HTTPStatus.NOT_MODIFIED)): + if code >= 200 and code not in ( + HTTPStatus.NO_CONTENT, + HTTPStatus.RESET_CONTENT, + HTTPStatus.NOT_MODIFIED, + ): # HTML encode to prevent Cross Site Scripting attacks # (see bug #1100201) # Specialized error method for BLNET - with SERVER_DIR.joinpath(".error.html").open('rb') as file: + with SERVER_DIR.joinpath(".error.html").open("rb") as file: body = file.read() self.send_header("Content-Type", self.error_content_type) - self.send_header('Content-Length', int(len(body))) + self.send_header("Content-Length", int(len(body))) self.end_headers() - if self.command != 'HEAD' and body: + if self.command != "HEAD" and body: self.wfile.write(body) diff --git a/pyblnet/tests/test_web.py b/pyblnet/tests/test_web.py index d619e34..fdca2e7 100644 --- a/pyblnet/tests/test_web.py +++ b/pyblnet/tests/test_web.py @@ -4,9 +4,7 @@ # general requirements import unittest from .test_structure.server_control import Server -from .test_structure.blnet_mock_server import ( - BLNETServer, BLNETRequestHandler, PASSWORD -) +from .test_structure.blnet_mock_server import BLNETServer, BLNETRequestHandler, PASSWORD # For the server in this case import time @@ -15,7 +13,7 @@ from pyblnet import BLNET, test_blnet, BLNETWeb from .web_raw.web_state import STATE, STATE_ANALOG, STATE_DIGITAL -ADDRESS = 'localhost' +ADDRESS = "localhost" class OfflineTest(unittest.TestCase): @@ -39,7 +37,7 @@ class BLNETWebTest(unittest.TestCase): server = None server_control = None port = 0 - url = 'http://localhost:80' + url = "http://localhost:80" def setUp(self): # Create an arbitrary subclass of TCP Server as the server to be started @@ -80,7 +78,9 @@ def test_blnet_web_log_in(self): with BLNETWeb(self.url, password=PASSWORD, timeout=10) as blnet: self.assertTrue(blnet.logged_in()) # test without http - blnet = BLNETWeb("{}:{}".format(ADDRESS, self.port), password=PASSWORD, timeout=10) + blnet = BLNETWeb( + "{}:{}".format(ADDRESS, self.port), password=PASSWORD, timeout=10 + ) with blnet as blnet: self.assertTrue(blnet.logged_in()) @@ -88,11 +88,10 @@ def test_blnet_fetch(self): """ Test fetching data in higher level class """ self.assertEqual( BLNET( - ADDRESS, - password=PASSWORD, - timeout=10, - use_ta=False, - web_port=self.port).fetch(), STATE) + ADDRESS, password=PASSWORD, timeout=10, use_ta=False, web_port=self.port + ).fetch(), + STATE, + ) def test_blnet_fetch_fine_grained(self): """ Test fetching data in higher level class """ @@ -102,56 +101,56 @@ def scratch_tuple(type, id, ret): return fetched[type][id][ret], fetched[type][id], fetched blnet = BLNET( - ADDRESS, - password=PASSWORD, - timeout=10, - use_ta=False, - web_port=self.port) + ADDRESS, password=PASSWORD, timeout=10, use_ta=False, web_port=self.port + ) self.assertEqual( blnet.get_analog_value(id=5, cached=fetched), - scratch_tuple('analog', 5, 'value')) + scratch_tuple("analog", 5, "value"), + ) self.assertEqual( - blnet.get_analog_value(name='TSP.oben', cached=fetched), - scratch_tuple('analog', 2, 'value')) + blnet.get_analog_value(name="TSP.oben", cached=fetched), + scratch_tuple("analog", 2, "value"), + ) self.assertEqual( blnet.get_digital_value(id=1, cached=fetched), - scratch_tuple('digital', 1, 'value')) + scratch_tuple("digital", 1, "value"), + ) self.assertEqual( blnet.get_digital_mode(id=1, cached=fetched), - scratch_tuple('digital', 1, 'mode')) + scratch_tuple("digital", 1, "mode"), + ) # test auto-fetching self.assertEqual( - blnet.get_digital_mode(id=1), scratch_tuple('digital', 1, 'mode')) + blnet.get_digital_mode(id=1), scratch_tuple("digital", 1, "mode") + ) def test_blnet_turn(self): """ Test higher level turn methods """ blnet = BLNET( - ADDRESS, - password=PASSWORD, - timeout=10, - use_ta=False, - web_port=self.port) + ADDRESS, password=PASSWORD, timeout=10, use_ta=False, web_port=self.port + ) blnet.turn_on(10) - self.assertEqual(self.server.get_node('A'), '2') + self.assertEqual(self.server.get_node("A"), "2") blnet.turn_on(9) - self.assertEqual(self.server.get_node('9'), '2') + self.assertEqual(self.server.get_node("9"), "2") blnet.turn_auto(8) - self.assertEqual(self.server.get_node('8'), '3') + self.assertEqual(self.server.get_node("8"), "3") blnet.turn_off(1) - self.assertEqual(self.server.get_node('1'), '1') + self.assertEqual(self.server.get_node("1"), "1") def test_blnet_fetch_error(self): """ Test fetching data in higher level class with missing password (or otherwise denied access to the data) """ self.assertEqual( - BLNET(ADDRESS, password=None, timeout=10, use_ta=False, web_port=self.port).fetch(), - {'analog': {}, 'digital': {}, 'energy': {}, 'power': {}, 'speed': {}} + BLNET( + ADDRESS, password=None, timeout=10, use_ta=False, web_port=self.port + ).fetch(), + {"analog": {}, "digital": {}, "energy": {}, "power": {}, "speed": {}}, ) def test_blnet_web_analog(self): """ Test reading analog values """ with BLNETWeb(self.url, password=PASSWORD, timeout=10) as blnet: - self.assertEqual( - blnet.read_analog_values(), STATE_ANALOG) + self.assertEqual(blnet.read_analog_values(), STATE_ANALOG) def test_blnet_web_digital(self): """ Test reading digital values""" @@ -161,29 +160,29 @@ def test_blnet_web_digital(self): def test_blnet_web_set_digital(self): """ Test setting digital values """ with BLNETWeb(self.url, password=PASSWORD, timeout=10) as blnet: - blnet.set_digital_value(10, '2') - self.assertEqual(self.server.get_node('A'), '2') - blnet.set_digital_value(9, 'EIN') - self.assertEqual(self.server.get_node('9'), '2') - blnet.set_digital_value(8, 'auto') - self.assertEqual(self.server.get_node('8'), '3') - blnet.set_digital_value(1, 'on') - self.assertEqual(self.server.get_node('1'), '2') - blnet.set_digital_value(1, 'AUS') - self.assertEqual(self.server.get_node('1'), '1') + blnet.set_digital_value(10, "2") + self.assertEqual(self.server.get_node("A"), "2") + blnet.set_digital_value(9, "EIN") + self.assertEqual(self.server.get_node("9"), "2") + blnet.set_digital_value(8, "auto") + self.assertEqual(self.server.get_node("8"), "3") + blnet.set_digital_value(1, "on") + self.assertEqual(self.server.get_node("1"), "2") + blnet.set_digital_value(1, "AUS") + self.assertEqual(self.server.get_node("1"), "1") blnet.set_digital_value(5, 3) - self.assertEqual(self.server.get_node('5'), '3') + self.assertEqual(self.server.get_node("5"), "3") blnet.set_digital_value(4, True) - self.assertEqual(self.server.get_node('4'), '2') + self.assertEqual(self.server.get_node("4"), "2") blnet.set_digital_value(6, False) - self.assertEqual(self.server.get_node('6'), '1') + self.assertEqual(self.server.get_node("6"), "1") try: - blnet.set_digital_value(0, 'EIN') + blnet.set_digital_value(0, "EIN") self.fail("Didn't catch wrong id 0") except ValueError: pass try: - blnet.set_digital_value(16, 'EIN') + blnet.set_digital_value(16, "EIN") self.fail("Didn't catch wrong id 16") except ValueError: pass diff --git a/pyblnet/tests/web_raw/web_state.py b/pyblnet/tests/web_raw/web_state.py index b272f49..fea4126 100644 --- a/pyblnet/tests/web_raw/web_state.py +++ b/pyblnet/tests/web_raw/web_state.py @@ -1,240 +1,151 @@ -STATE = {'analog': {1: {'id': '1', - 'name': 'TKollektor', - 'unit_of_measurement': '°C', - 'value': '5.3'}, - 2: {'id': '2', - 'name': 'TSP.oben', - 'unit_of_measurement': '°C', - 'value': '46.3'}, - 3: {'id': '3', - 'name': 'TSP.unten', - 'unit_of_measurement': '°C', - 'value': '53.1'}, - 4: {'id': '4', - 'name': 'THeizkr.VL', - 'unit_of_measurement': '°C', - 'value': '44.8'}, - 5: {'id': '5', - 'name': 'Temp.Aussen', - 'unit_of_measurement': '°C', - 'value': '-72.3'}, - 6: {'id': '6', - 'name': 'Temp.Raum', - 'unit_of_measurement': '°C', - 'value': '24.6'}, - 7: {'id': '7', - 'name': 'T Kaminofen', - 'unit_of_measurement': '°C', - 'value': '20.9'}, - 9: {'id': '9', - 'name': 'TZirku.RL', - 'unit_of_measurement': '°C', - 'value': '36.6'}}, - 'digital': {1: {'id': '1', - 'mode': 'AUTO', - 'name': 'Pumpe-Solar', - 'value': 'AUS'}, - 2: {'id': '2', - 'mode': 'AUTO', - 'name': 'Pumpe-Hzkr', - 'value': 'EIN'}, - 5: {'id': '5', - 'mode': 'AUTO', - 'name': 'Anf.Kessel', - 'value': 'AUS'}, - 6: {'id': '6', - 'mode': 'HAND', - 'name': 'Vent.Solar', - 'value': 'AUS'}, - 7: {'id': '7', - 'mode': 'AUTO', - 'name': 'P Kaminofen', - 'value': 'EIN'}, - 10: {'id': '10', - 'mode': 'HAND', - 'name': 'WW-Pumpe1', - 'value': 'AUS'}}, - 'energy': {}, - 'power': {}, - 'speed': {}} - -STATE_ALTERNATIVE = { - 'analog': { +STATE = { + "analog": { + 1: { + "id": "1", + "name": "TKollektor", + "unit_of_measurement": "°C", + "value": "5.3", + }, 2: { - 'id': '2', - 'name': 'TSP.oben', - 'value': '47.1', - 'unit_of_measurement': '°C' + "id": "2", + "name": "TSP.oben", + "unit_of_measurement": "°C", + "value": "46.3", }, 3: { - 'id': '3', - 'name': 'TSP.unten', - 'value': '57.8', - 'unit_of_measurement': '°C' + "id": "3", + "name": "TSP.unten", + "unit_of_measurement": "°C", + "value": "53.1", }, 4: { - 'id': '4', - 'name': 'THeizkr.VL', - 'value': '45.8', - 'unit_of_measurement': '°C' + "id": "4", + "name": "THeizkr.VL", + "unit_of_measurement": "°C", + "value": "44.8", }, 5: { - 'id': '5', - 'name': 'Temp.Aussen', - 'value': '1.1', - 'unit_of_measurement': '°C' + "id": "5", + "name": "Temp.Aussen", + "unit_of_measurement": "°C", + "value": "-72.3", }, 6: { - 'id': '6', - 'name': 'Temp.Raum', - 'value': '24.1', - 'unit_of_measurement': '°C' + "id": "6", + "name": "Temp.Raum", + "unit_of_measurement": "°C", + "value": "24.6", }, 7: { - 'id': '7', - 'name': 'T Kaminofen', - 'value': '55.8', - 'unit_of_measurement': '°C' + "id": "7", + "name": "T Kaminofen", + "unit_of_measurement": "°C", + "value": "20.9", }, 9: { - 'id': '9', - 'name': 'TZirku.RL', - 'value': '22.4', - 'unit_of_measurement': '°C' - } - }, - 'digital': { - 1: { - 'id': '1', - 'name': 'Pumpe-Solar', - 'mode': 'AUTO', - 'value': 'AUS' + "id": "9", + "name": "TZirku.RL", + "unit_of_measurement": "°C", + "value": "36.6", }, + }, + "digital": { + 1: {"id": "1", "mode": "AUTO", "name": "Pumpe-Solar", "value": "AUS"}, + 2: {"id": "2", "mode": "AUTO", "name": "Pumpe-Hzkr", "value": "EIN"}, + 5: {"id": "5", "mode": "AUTO", "name": "Anf.Kessel", "value": "AUS"}, + 6: {"id": "6", "mode": "HAND", "name": "Vent.Solar", "value": "AUS"}, + 7: {"id": "7", "mode": "AUTO", "name": "P Kaminofen", "value": "EIN"}, + 10: {"id": "10", "mode": "HAND", "name": "WW-Pumpe1", "value": "AUS"}, + }, + "energy": {}, + "power": {}, + "speed": {}, +} + +STATE_ALTERNATIVE = { + "analog": { 2: { - 'id': '2', - 'name': 'Pumpe-Hzkr', - 'mode': 'AUTO', - 'value': 'EIN' + "id": "2", + "name": "TSP.oben", + "value": "47.1", + "unit_of_measurement": "°C", + }, + 3: { + "id": "3", + "name": "TSP.unten", + "value": "57.8", + "unit_of_measurement": "°C", + }, + 4: { + "id": "4", + "name": "THeizkr.VL", + "value": "45.8", + "unit_of_measurement": "°C", }, 5: { - 'id': '5', - 'name': 'Anf.Kessel', - 'mode': 'AUTO', - 'value': 'AUS' + "id": "5", + "name": "Temp.Aussen", + "value": "1.1", + "unit_of_measurement": "°C", }, 6: { - 'id': '6', - 'name': 'Vent.Solar', - 'mode': 'HAND', - 'value': 'AUS' + "id": "6", + "name": "Temp.Raum", + "value": "24.1", + "unit_of_measurement": "°C", }, 7: { - 'id': '7', - 'name': 'P Kaminofen', - 'mode': 'AUTO', - 'value': 'EIN' + "id": "7", + "name": "T Kaminofen", + "value": "55.8", + "unit_of_measurement": "°C", + }, + 9: { + "id": "9", + "name": "TZirku.RL", + "value": "22.4", + "unit_of_measurement": "°C", }, - 10: { - 'id': '10', - 'name': 'WW-Pumpe1', - 'mode': 'HAND', - 'value': 'AUS' - } }, - 'speed': {}, - 'energy': {}, - 'power': {} + "digital": { + 1: {"id": "1", "name": "Pumpe-Solar", "mode": "AUTO", "value": "AUS"}, + 2: {"id": "2", "name": "Pumpe-Hzkr", "mode": "AUTO", "value": "EIN"}, + 5: {"id": "5", "name": "Anf.Kessel", "mode": "AUTO", "value": "AUS"}, + 6: {"id": "6", "name": "Vent.Solar", "mode": "HAND", "value": "AUS"}, + 7: {"id": "7", "name": "P Kaminofen", "mode": "AUTO", "value": "EIN"}, + 10: {"id": "10", "name": "WW-Pumpe1", "mode": "HAND", "value": "AUS"}, + }, + "speed": {}, + "energy": {}, + "power": {}, } -STATE_ANALOG = [{'id': '1', 'name': 'TKollektor', 'unit_of_measurement': '°C', 'value': '5.3'}, - {'id': '2', 'name': 'TSP.oben', 'unit_of_measurement': '°C', 'value': '46.3'}, - {'id': '3', 'name': 'TSP.unten', 'unit_of_measurement': '°C', 'value': '53.1'}, - {'id': '4', - 'name': 'THeizkr.VL', - 'unit_of_measurement': '°C', - 'value': '44.8'}, - {'id': '5', - 'name': 'Temp.Aussen', - 'unit_of_measurement': '°C', - 'value': '-72.3'}, - {'id': '6', 'name': 'Temp.Raum', 'unit_of_measurement': '°C', 'value': '24.6'}, - {'id': '7', - 'name': 'T Kaminofen', - 'unit_of_measurement': '°C', - 'value': '20.9'}, - {'id': '9', 'name': 'TZirku.RL', 'unit_of_measurement': '°C', 'value': '36.6'}] +STATE_ANALOG = [ + {"id": "1", "name": "TKollektor", "unit_of_measurement": "°C", "value": "5.3"}, + {"id": "2", "name": "TSP.oben", "unit_of_measurement": "°C", "value": "46.3"}, + {"id": "3", "name": "TSP.unten", "unit_of_measurement": "°C", "value": "53.1"}, + {"id": "4", "name": "THeizkr.VL", "unit_of_measurement": "°C", "value": "44.8"}, + {"id": "5", "name": "Temp.Aussen", "unit_of_measurement": "°C", "value": "-72.3"}, + {"id": "6", "name": "Temp.Raum", "unit_of_measurement": "°C", "value": "24.6"}, + {"id": "7", "name": "T Kaminofen", "unit_of_measurement": "°C", "value": "20.9"}, + {"id": "9", "name": "TZirku.RL", "unit_of_measurement": "°C", "value": "36.6"}, +] -STATE_ANALOG_ALTERNATIVE = [{ - 'id': '2', - 'name': 'TSP.oben', - 'unit_of_measurement': '°C', - 'value': '47.1' -}, -{ - 'id': '3', - 'name': 'TSP.unten', - 'unit_of_measurement': '°C', - 'value': '57.8' -}, -{ - 'id': '4', - 'name': 'THeizkr.VL', - 'unit_of_measurement': '°C', - 'value': '45.8' -}, -{ - 'id': '5', - 'name': 'Temp.Aussen', - 'unit_of_measurement': '°C', - 'value': '1.1' -}, -{ - 'id': '6', - 'name': 'Temp.Raum', - 'unit_of_measurement': '°C', - 'value': '24.1' -}, -{ - 'id': '7', - 'name': 'T Kaminofen', - 'unit_of_measurement': '°C', - 'value': '55.8' -}, -{ - 'id': '9', - 'name': 'TZirku.RL', - 'unit_of_measurement': '°C', - 'value': '22.4' -}] +STATE_ANALOG_ALTERNATIVE = [ + {"id": "2", "name": "TSP.oben", "unit_of_measurement": "°C", "value": "47.1"}, + {"id": "3", "name": "TSP.unten", "unit_of_measurement": "°C", "value": "57.8"}, + {"id": "4", "name": "THeizkr.VL", "unit_of_measurement": "°C", "value": "45.8"}, + {"id": "5", "name": "Temp.Aussen", "unit_of_measurement": "°C", "value": "1.1"}, + {"id": "6", "name": "Temp.Raum", "unit_of_measurement": "°C", "value": "24.1"}, + {"id": "7", "name": "T Kaminofen", "unit_of_measurement": "°C", "value": "55.8"}, + {"id": "9", "name": "TZirku.RL", "unit_of_measurement": "°C", "value": "22.4"}, +] -STATE_DIGITAL = [{ -'id': '1', -'mode': 'AUTO', - 'name': 'Pumpe-Solar', - 'value': 'AUS' -}, { - 'id': '2', - 'mode': 'AUTO', - 'name': 'Pumpe-Hzkr', - 'value': 'EIN' -}, { - 'id': '5', - 'mode': 'AUTO', - 'name': 'Anf.Kessel', - 'value': 'AUS' -}, { - 'id': '6', - 'mode': 'HAND', - 'name': 'Vent.Solar', - 'value': 'AUS' -}, { - 'id': '7', - 'mode': 'AUTO', - 'name': 'P Kaminofen', - 'value': 'EIN' -}, { - 'id': '10', - 'mode': 'HAND', - 'name': 'WW-Pumpe1', - 'value': 'AUS' -}] +STATE_DIGITAL = [ + {"id": "1", "mode": "AUTO", "name": "Pumpe-Solar", "value": "AUS"}, + {"id": "2", "mode": "AUTO", "name": "Pumpe-Hzkr", "value": "EIN"}, + {"id": "5", "mode": "AUTO", "name": "Anf.Kessel", "value": "AUS"}, + {"id": "6", "mode": "HAND", "name": "Vent.Solar", "value": "AUS"}, + {"id": "7", "mode": "AUTO", "name": "P Kaminofen", "value": "EIN"}, + {"id": "10", "mode": "HAND", "name": "WW-Pumpe1", "value": "AUS"}, +] diff --git a/setup.py b/setup.py index 03386ee..beb923c 100644 --- a/setup.py +++ b/setup.py @@ -12,41 +12,39 @@ here = path.abspath(path.dirname(__file__)) # Get the long description from the README file -with open(path.join(here, 'README.md'), encoding='utf-8') as f: +with open(path.join(here, "README.md"), encoding="utf-8") as f: long_description = f.read() setup( - name='PyBLNET', + name="PyBLNET", version=pyblnet.__version__, - description='Automate wireless communication to UVR1611 via BL-NET', + description="Automate wireless communication to UVR1611 via BL-NET", author=pyblnet.__author__, author_email=pyblnet.__author_email__, - url='https://github.com/nielstron/pyblnet/', - py_modules=['pyblnet'], + url="https://github.com/nielstron/pyblnet/", + py_modules=["pyblnet"], packages=find_packages(), - package_data={'': ['*.html', '*.htm']}, - install_requires=['htmldom', 'requests'], + package_data={"": ["*.html", "*.htm"]}, + install_requires=["htmldom", "requests"], long_description=long_description, - long_description_content_type='text/markdown', + long_description_content_type="text/markdown", license=pyblnet.__license__, classifiers=[ - 'Development Status :: 4 - Beta', + "Development Status :: 4 - Beta", # Indicate who your project is intended for - 'Intended Audience :: Developers', - 'Topic :: Software Development :: Object Brokering', - + "Intended Audience :: Developers", + "Topic :: Software Development :: Object Brokering", # Pick your license as you wish (should match "license" above) - 'License :: OSI Approved :: MIT License', - + "License :: OSI Approved :: MIT License", # Specify the Python versions you support here. In particular, ensure # that you indicate whether you support Python 2, Python 3 or both. - 'Programming Language :: Python :: 3.4', - 'Programming Language :: Python :: 3.5', - 'Programming Language :: Python :: 3.6', - 'Programming Language :: Python :: 3.7', - 'Programming Language :: Python :: 3.8', + "Programming Language :: Python :: 3.4", + "Programming Language :: Python :: 3.5", + "Programming Language :: Python :: 3.6", + "Programming Language :: Python :: 3.7", + "Programming Language :: Python :: 3.8", ], - keywords='python uvr1611 blnet technische alternative home automation iot', - python_requires='>=3', - test_suite='pyblnet.tests', + keywords="python uvr1611 blnet technische alternative home automation iot", + python_requires=">=3", + test_suite="pyblnet.tests", )