<filename>src/helheimr_heating/helheimr_heating/heating/district_heating.py
#!/usr/bin/python
# coding=utf-8
"""
Our district/tele-heating system offers a slightly outdated web-interface (CMI:
Control & Monitoring Interface which allows to query the state of our solar
cells and the teleheating inlet, as well as adjusting the inlet temperature to
our central heating.
"""


import logging
from attributedict.collections import AttributeDict
from enum import Enum
from html.parser import HTMLParser
from ..utils import broadcast, common, config, networking, translation

_ = translation.translator_fx()


_logger = logging.getLogger('DistrictHeating')


class DistrictHeatingQueryParser(HTMLParser):
    """The district heating system gateway provides a quite crappy website.
    This parser takes care of the *really* dirty work to translate a status
    query into into something we can use programmatically.
    """
    def __init__(self):
        super(DistrictHeatingQueryParser, self).__init__()
        # If you extend the parser, do NOT forget to
        # register the corresponding _div_mapping!
        self._status = {
            'btn_eco_status': None,         # Button Eco
            'btn_medium_status': None,      # Button Mittel
            'btn_medium_time': None,
            'btn_high_status': None,        # Button Hoch
            'btn_high_time': None,
            'btn_very_high_status': None,   # Button Sehr Hoch
            'btn_very_high_time': None,
            'btn_transition_status': None,  # Button Übergang

            'src_solar_temperature': None,  # Herkunft Solar °C
            'src_solar_power': None,        # Herkunft Solar kW
            'src_tele_temperature': None,   # Herkunft Fernwärme °C
            'src_tele_power': None,         # Herkunft Fernwärme kW

            'consumption_state': None,        # Verbrauch 'ein/ausgeschaltet'
            'consumption_heating_temperature': None,  # Verbrauch Heizung °C
            'consumption_heating_power': None,        # Verbrauch Heizung kW
            'consumption_boiler_temperature': None,   # Verbrauch Boiler °C
            'consumption_boiler_power': None          # Verbrauch Boiler kW
        }

        # During parsing, this holds the key into self._status - will be
        # set upon handling starting tags
        self._store_data_to = None

        self._div_mapping = {
            # Button states (on/off)
            'pos38': 'btn_eco_status',
            'pos34': 'btn_medium_status',
            'pos36': 'btn_high_status',
            'pos37': 'btn_very_high_status',
            'pos39': 'btn_transition_status',

            # Remaining time per button (relevant if they were pressed)
            'pos35': 'btn_medium_time',
            'pos29': 'btn_high_time',
            'pos31': 'btn_very_high_time',

            # Source (solar or district heating)
            'pos7': 'src_solar_temperature',
            'pos8': 'src_solar_power',
            'pos5': 'src_tele_temperature',
            'pos6': 'src_tele_power',

            # Consumption (heating or boiler)
            'pos42': 'consumption_state',
            'pos40': 'consumption_heating_temperature',
            'pos41': 'consumption_heating_power',
            'pos13': 'consumption_boiler_temperature',
            'pos14': 'consumption_boiler_power'
        }
        self._interesting_divs = self._div_mapping.keys()

    @property
    def status_dict(self):
        return AttributeDict(self._status)

    def handle_starttag(self, tag, attrs):
        if tag == 'div':
            div_id = None
            for attr in attrs:
                name, val = attr
                if name == 'id':
                    div_id = val
                    break
            if div_id in self._interesting_divs:
                self._store_data_to = self._div_mapping[div_id]
            else:
                self._store_data_to = None

    def handle_endtag(self, tag):
        self._store_data_to = None

    def handle_data(self, data):
        if self._store_data_to is None:
            return
        trimmed = data.strip()
        if self._store_data_to == 'consumption_state':
            self._status[self._store_data_to] = trimmed.lower() == 'ongeschaltet'  # sic!
        elif self._store_data_to.endswith('_status'):
            # Need to parse an on/off string
            self._status[self._store_data_to] = True if trimmed.lower() == 'on' else False
        elif self._store_data_to.endswith('_time'):
            # Need to parse an "remaining time" string which should contain 'Xm Ys'
            times = common.extract_integers(trimmed)
            if len(times) == 2:  # pragma: no cover (can only be tested if we turn on the inlet "boost")
                self._status[self._store_data_to] = times[0] * 60 + times[1]
            elif len(times) == 1:
                self._status[self._store_data_to] = times[0]
            else:  # pragma: no cover
                _logger.error(f"Invalid 'remaining time' string: '{trimmed}'.")
                broadcast.error(_("Invalid character sequence '%(seq)s' while parsing the district heating CMI (expected 'remaining time' string).",
                                  seq=trimmed), source=broadcast.SRC_DISTRICT_HEATING)
        elif self._store_data_to.endswith('_temperature') or \
                self._store_data_to.endswith('_power'):
            # Should be a float
            nums = common.extract_floats(trimmed.replace(',', '.'))  # Even the decimal point for floats is inconsistently used on this "wonderful" CMI website...
            if len(nums) == 1:
                self._status[self._store_data_to] = nums[0]
            else:  # pragma: no cover
                _logger.error(f"Invalid data string for a single float: '{trimmed}'.")
                broadcast.error(_("Invalid input '%(seq)s' for parsing a float from the district heating CMI.",
                                  seq=trimmed), source=broadcast.SRC_DISTRICT_HEATING)
        else:
            # Implement other handlers if needed
            pass


class DistrictHeatingRequest(Enum):
    """
    Type of the district heating's CMI buttons.
    * "Eco" and "transition" don't have an assigned inlet temperature. Thus, we
      assign them rather arbitrary values (as long as they differ sufficiently
      from the other - temperature-assigned - buttons' values).
    * Medium, high and "very high" buttons have an inlet temperature assigned,
      which we will use to verify/validate the heating request parameters.
    """
    ECO = 1
    TRANSITION = 2
    MEDIUM = 55
    HIGH = 60
    VERY_HIGH = 65

    @classmethod
    def from_int(cls, value):
        if value == 1:
            return cls.ECO
        elif value == 2:
            return cls.TRANSITION
        elif value == 55:
            return cls.MEDIUM
        elif value == 60:
            return cls.HIGH
        elif value == 65:
            return cls.VERY_HIGH
        else:
            raise ValueError(_("Cannot convert integer (%(value)d) to DistrictHeatingRequest.", value=value))


class _DistrictHeating:
    def __init__(self):
        # Load headers to replay the cURL requests
        cfg = config.load_configuration(config.HELHEIMR_CONFIG_FILE_CONTROL)
        dhcfg = cfg.district_heating
        self._headers = {k: dhcfg.http_headers[k] for k in dhcfg.http_headers}

        # Prepare the button mapping
        self._buttons = dict()
        for request_type in DistrictHeatingRequest:
            self._buttons[request_type] = dhcfg[f'button_{request_type.name.lower()}']

        # Parameters which need to be set properly to switch on district heating
        self._param_change = (dhcfg.param_name_change, dhcfg.param_value_change)
        self._param_name_button = dhcfg.param_name_button

        # URLs of the district heating CMI
        self._url_change = dhcfg.url_change
        self._url_query = dhcfg.url_query

        _logger.info('Initialized district heating wrapper.')

    def get_inlet_control_options(self, as_int=True):
        """
        Returns the subset of "CMI buttons" which we support to "click"
        programmatically. As there is no actual API, but only a simplistic
        web interface, we have to replay an HTTP request which simulates
        a button click in order to adjust the inlet temperature.
        """
        return [
            AttributeDict({'label': '55\u200a°', 'value': DistrictHeatingRequest.MEDIUM}),
            AttributeDict({'label': '60\u200a°', 'value': DistrictHeatingRequest.HIGH}),
            AttributeDict({'label': '65\u200a°', 'value': DistrictHeatingRequest.VERY_HIGH})
        ]

    def start_heating(self, request_type, language=None):  # pragma: no cover
        """request_type is a DistrictHeatingRequest, specifying which physical button press should be simulated."""
        _ = translation.translator_fx(language)
        try:
            if isinstance(request_type, int):
                request_type = DistrictHeatingRequest.from_int(request_type)
            elif not isinstance(request_type, DistrictHeatingRequest):
                raise ValueError(_('Input to start_heating() must be integer or a DistrictHeatingRequest'))
            btn_id = self._buttons[request_type]
        except ValueError as err:
            return common.make_response(False, message=_('Invalid district heating request: %(err)s', err=err))

        params = (
            (self._param_name_button, btn_id),
            self._param_change
        )
        response = networking.safe_http_get(self._url_change, headers=self._headers, params=params, verify=False)
        if response is None:
            return common.make_response(False, message=_('Network error while contacting the district heating CMI.'))
        else:
            if response.status_code == 200:
                return common.make_response(True, message=_('Successfully adjusted the inlet temperature.'))
            else:
                return common.make_response(False, message=_('HTTP error %(status_code)d while contacting the district heating CMI.',
                                                             status_code=response.status_code))

    def query_heating(self, language=None):
        _ = translation.translator_fx(language)
        response = networking.safe_http_get(self._url_query, headers=self._headers, verify=False)
        if response is None:  # pragma: no cover
            return common.make_response(False, message=_('Network error while querying the district heating CMI status.'))

        if response:
            query = DistrictHeatingQueryParser()
            query.feed(response.text)
            return common.make_response(True, district_heating_status=query.status_dict)
        else:  # pragma: no cover
            return common.make_response(False, message=_('HTTP error %(status_code)d while querying the district heating CMI status.',
                                                         status_code=response.status_code))


# Singleton instance
district_heater = _DistrictHeating()

<filename>src/helheimr_heating/helheimr_heating/heating/district_heating.py
#!/usr/bin/python
# coding=utf-8
"""
Our district/tele-heating system offers a slightly outdated web-interface (CMI:
Control & Monitoring Interface which allows to query the state of our solar
cells and the teleheating inlet, as well as adjusting the inlet temperature to
our central heating.
"""


import logging
from attributedict.collections import AttributeDict
from enum import Enum
from html.parser import HTMLParser
from ..utils import broadcast, common, config, networking, translation

_ = translation.translator_fx()


_logger = logging.getLogger('DistrictHeating')


class DistrictHeatingQueryParser(HTMLParser):
    """The district heating system gateway provides a quite crappy website.
    This parser takes care of the *really* dirty work to translate a status
    query into into something we can use programmatically.
    """
    def __init__(self):
        super(DistrictHeatingQueryParser, self).__init__()
        # If you extend the parser, do NOT forget to
        # register the corresponding _div_mapping!
        self._status = {
            'btn_eco_status': None,         # Button Eco
            'btn_medium_status': None,      # Button Mittel
            'btn_medium_time': None,
            'btn_high_status': None,        # Button Hoch
            'btn_high_time': None,
            'btn_very_high_status': None,   # Button Sehr Hoch
            'btn_very_high_time': None,
            'btn_transition_status': None,  # Button Übergang

            'src_solar_temperature': None,  # Herkunft Solar °C
            'src_solar_power': None,        # Herkunft Solar kW
            'src_tele_temperature': None,   # Herkunft Fernwärme °C
            'src_tele_power': None,         # Herkunft Fernwärme kW

            'consumption_state': None,        # Verbrauch 'ein/ausgeschaltet'
            'consumption_heating_temperature': None,  # Verbrauch Heizung °C
            'consumption_heating_power': None,        # Verbrauch Heizung kW
            'consumption_boiler_temperature': None,   # Verbrauch Boiler °C
            'consumption_boiler_power': None          # Verbrauch Boiler kW
        }

        # During parsing, this holds the key into self._status - will be
        # set upon handling starting tags
        self._store_data_to = None

        self._div_mapping = {
            # Button states (on/off)
            'pos38': 'btn_eco_status',
            'pos34': 'btn_medium_status',
            'pos36': 'btn_high_status',
            'pos37': 'btn_very_high_status',
            'pos39': 'btn_transition_status',

            # Remaining time per button (relevant if they were pressed)
            'pos35': 'btn_medium_time',
            'pos29': 'btn_high_time',
            'pos31': 'btn_very_high_time',

            # Source (solar or district heating)
            'pos7': 'src_solar_temperature',
            'pos8': 'src_solar_power',
            'pos5': 'src_tele_temperature',
            'pos6': 'src_tele_power',

            # Consumption (heating or boiler)
            'pos42': 'consumption_state',
            'pos40': 'consumption_heating_temperature',
            'pos41': 'consumption_heating_power',
            'pos13': 'consumption_boiler_temperature',
            'pos14': 'consumption_boiler_power'
        }
        self._interesting_divs = self._div_mapping.keys()

    @property
    def status_dict(self):
        return AttributeDict(self._status)

    def handle_starttag(self, tag, attrs):
        if tag == 'div':
            div_id = None
            for attr in attrs:
                name, val = attr
                if name == 'id':
                    div_id = val
                    break
            if div_id in self._interesting_divs:
                self._store_data_to = self._div_mapping[div_id]
            else:
                self._store_data_to = None

    def handle_endtag(self, tag):
        self._store_data_to = None

    def handle_data(self, data):
        if self._store_data_to is None:
            return
        trimmed = data.strip()
        if self._store_data_to == 'consumption_state':
            self._status[self._store_data_to] = trimmed.lower() == 'ongeschaltet'  # sic!
        elif self._store_data_to.endswith('_status'):
            # Need to parse an on/off string
            self._status[self._store_data_to] = True if trimmed.lower() == 'on' else False
        elif self._store_data_to.endswith('_time'):
            # Need to parse an "remaining time" string which should contain 'Xm Ys'
            times = common.extract_integers(trimmed)
            if len(times) == 2:  # pragma: no cover (can only be tested if we turn on the inlet "boost")
                self._status[self._store_data_to] = times[0] * 60 + times[1]
            elif len(times) == 1:
                self._status[self._store_data_to] = times[0]
            else:  # pragma: no cover
                _logger.error(f"Invalid 'remaining time' string: '{trimmed}'.")
                broadcast.error(_("Invalid character sequence '%(seq)s' while parsing the district heating CMI (expected 'remaining time' string).",
                                  seq=trimmed), source=broadcast.SRC_DISTRICT_HEATING)
        elif self._store_data_to.endswith('_temperature') or \
                self._store_data_to.endswith('_power'):
            # Should be a float
            nums = common.extract_floats(trimmed.replace(',', '.'))  # Even the decimal point for floats is inconsistently used on this "wonderful" CMI website...
            if len(nums) == 1:
                self._status[self._store_data_to] = nums[0]
            else:  # pragma: no cover
                _logger.error(f"Invalid data string for a single float: '{trimmed}'.")
                broadcast.error(_("Invalid input '%(seq)s' for parsing a float from the district heating CMI.",
                                  seq=trimmed), source=broadcast.SRC_DISTRICT_HEATING)
        else:
            # Implement other handlers if needed
            pass


class DistrictHeatingRequest(Enum):
    """
    Type of the district heating's CMI buttons.
    * "Eco" and "transition" don't have an assigned inlet temperature. Thus, we
      assign them rather arbitrary values (as long as they differ sufficiently
      from the other - temperature-assigned - buttons' values).
    * Medium, high and "very high" buttons have an inlet temperature assigned,
      which we will use to verify/validate the heating request parameters.
    """
    ECO = 1
    TRANSITION = 2
    MEDIUM = 55
    HIGH = 60
    VERY_HIGH = 65

    @classmethod
    def from_int(cls, value):
        if value == 1:
            return cls.ECO
        elif value == 2:
            return cls.TRANSITION
        elif value == 55:
            return cls.MEDIUM
        elif value == 60:
            return cls.HIGH
        elif value == 65:
            return cls.VERY_HIGH
        else:
            raise ValueError(_("Cannot convert integer (%(value)d) to DistrictHeatingRequest.", value=value))


class _DistrictHeating:
    def __init__(self):
        # Load headers to replay the cURL requests
        cfg = config.load_configuration(config.HELHEIMR_CONFIG_FILE_CONTROL)
        dhcfg = cfg.district_heating
        self._headers = {k: dhcfg.http_headers[k] for k in dhcfg.http_headers}

        # Prepare the button mapping
        self._buttons = dict()
        for request_type in DistrictHeatingRequest:
            self._buttons[request_type] = dhcfg[f'button_{request_type.name.lower()}']

        # Parameters which need to be set properly to switch on district heating
        self._param_change = (dhcfg.param_name_change, dhcfg.param_value_change)
        self._param_name_button = dhcfg.param_name_button

        # URLs of the district heating CMI
        self._url_change = dhcfg.url_change
        self._url_query = dhcfg.url_query

        _logger.info('Initialized district heating wrapper.')

    def get_inlet_control_options(self, as_int=True):
        """
        Returns the subset of "CMI buttons" which we support to "click"
        programmatically. As there is no actual API, but only a simplistic
        web interface, we have to replay an HTTP request which simulates
        a button click in order to adjust the inlet temperature.
        """
        return [
            AttributeDict({'label': '55\u200a°', 'value': DistrictHeatingRequest.MEDIUM}),
            AttributeDict({'label': '60\u200a°', 'value': DistrictHeatingRequest.HIGH}),
            AttributeDict({'label': '65\u200a°', 'value': DistrictHeatingRequest.VERY_HIGH})
        ]

    def start_heating(self, request_type, language=None):  # pragma: no cover
        """request_type is a DistrictHeatingRequest, specifying which physical button press should be simulated."""
        _ = translation.translator_fx(language)
        try:
            if isinstance(request_type, int):
                request_type = DistrictHeatingRequest.from_int(request_type)
            elif not isinstance(request_type, DistrictHeatingRequest):
                raise ValueError(_('Input to start_heating() must be integer or a DistrictHeatingRequest'))
            btn_id = self._buttons[request_type]
        except ValueError as err:
            return common.make_response(False, message=_('Invalid district heating request: %(err)s', err=err))

        params = (
            (self._param_name_button, btn_id),
            self._param_change
        )
        response = networking.safe_http_get(self._url_change, headers=self._headers, params=params, verify=False)
        if response is None:
            return common.make_response(False, message=_('Network error while contacting the district heating CMI.'))
        else:
            if response.status_code == 200:
                return common.make_response(True, message=_('Successfully adjusted the inlet temperature.'))
            else:
                return common.make_response(False, message=_('HTTP error %(status_code)d while contacting the district heating CMI.',
                                                             status_code=response.status_code))

    def query_heating(self, language=None):
        _ = translation.translator_fx(language)
        response = networking.safe_http_get(self._url_query, headers=self._headers, verify=False)
        if response is None:  # pragma: no cover
            return common.make_response(False, message=_('Network error while querying the district heating CMI status.'))

        if response:
            query = DistrictHeatingQueryParser()
            query.feed(response.text)
            return common.make_response(True, district_heating_status=query.status_dict)
        else:  # pragma: no cover
            return common.make_response(False, message=_('HTTP error %(status_code)d while querying the district heating CMI status.',
                                                         status_code=response.status_code))


# Singleton instance
district_heater = _DistrictHeating()

#!/usr/bin/python # coding=utf-8 Our district/tele-heating system offers a slightly outdated web-interface (CMI: Control & Monitoring Interface which allows to query the state of our solar cells and the teleheating inlet, as well as adjusting the inlet temperature to our central heating. The district heating system gateway provides a quite crappy website.     This parser takes care of the *really* dirty work to translate a status     query into into something we can use programmatically. # If you extend the parser, do NOT forget to # register the corresponding _div_mapping! # Button Eco # Button Mittel # Button Hoch # Button Sehr Hoch # Button Übergang # Herkunft Solar °C # Herkunft Solar kW # Herkunft Fernwärme °C # Herkunft Fernwärme kW # Verbrauch 'ein/ausgeschaltet' # Verbrauch Heizung °C # Verbrauch Heizung kW # Verbrauch Boiler °C # Verbrauch Boiler kW # During parsing, this holds the key into self._status - will be # set upon handling starting tags # Button states (on/off) # Remaining time per button (relevant if they were pressed) # Source (solar or district heating) # Consumption (heating or boiler) # sic! # Need to parse an on/off string # Need to parse an "remaining time" string which should contain 'Xm Ys' # pragma: no cover (can only be tested if we turn on the inlet "boost") # pragma: no cover # Should be a float # Even the decimal point for floats is inconsistently used on this "wonderful" CMI website... # pragma: no cover # Implement other handlers if needed Type of the district heating's CMI buttons.     * "Eco" and "transition" don't have an assigned inlet temperature. Thus, we       assign them rather arbitrary values (as long as they differ sufficiently       from the other - temperature-assigned - buttons' values).     * Medium, high and "very high" buttons have an inlet temperature assigned,       which we will use to verify/validate the heating request parameters. # Load headers to replay the cURL requests # Prepare the button mapping # Parameters which need to be set properly to switch on district heating # URLs of the district heating CMI Returns the subset of "CMI buttons" which we support to "click"         programmatically. As there is no actual API, but only a simplistic         web interface, we have to replay an HTTP request which simulates         a button click in order to adjust the inlet temperature. # pragma: no cover request_type is a DistrictHeatingRequest, specifying which physical button press should be simulated. # pragma: no cover # pragma: no cover # Singleton instance