DEVELOPMENT ENVIRONMENT

~liljamo/ha-ouman-eh800

084a34e5eb98102d3beeff54d9c1368155cec938 — Jonni Liljamo 19 days ago a21df45
format
M custom_components/ouman_eh800/__init__.py => custom_components/ouman_eh800/__init__.py +4 -0
@@ 1,6 1,7 @@
"""
ouman_eh800
"""

import logging

from homeassistant.core import HomeAssistant


@@ 12,7 13,10 @@ _LOGGER = logging.getLogger(__name__)

PLATFORMS = ["sensor"]


async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
    """TODO"""

    _LOGGER.debug("Setting up Ouman EH-800")

    hass.data.setdefault(DOMAIN, {})

M custom_components/ouman_eh800/config_flow.py => custom_components/ouman_eh800/config_flow.py +45 -23
@@ 1,44 1,66 @@
"""Ouman EH-800 config flow"""

import logging
import voluptuous as vol

from typing import Any, Dict, Optional
from typing import Any

import voluptuous as vol

from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from .const import DOMAIN, DEFAULT_PORT, CONF_HOST, CONF_PORT, CONF_USERNAME, CONF_PASSWORD

from .const import (
    DOMAIN,
    DEFAULT_PORT,
    CONF_HOST,
    CONF_PORT,
    CONF_USERNAME,
    CONF_PASSWORD,
)

_LOGGER = logging.getLogger(__name__)

USER_SCHEMA = vol.Schema(
        {
            vol.Required(CONF_HOST): str,
            vol.Required(CONF_PORT, default=DEFAULT_PORT): int,
            vol.Required(CONF_USERNAME): str,
            vol.Required(CONF_PASSWORD): str,
        }
    {
        vol.Required(CONF_HOST): str,
        vol.Required(CONF_PORT, default=DEFAULT_PORT): int,
        vol.Required(CONF_USERNAME): str,
        vol.Required(CONF_PASSWORD): str,
    }
)

class OumanEH800ConfigFlow(ConfigFlow, domain=DOMAIN):

class OumanEH800ConfigFlow(
    ConfigFlow, domain=DOMAIN
):  # pylint: disable=too-few-public-methods
    """Ouman EH-800 config flow"""

    VERSION = 1

    async def _create_entry(self, host: str, port: int, username: str, password: str) -> ConfigFlowResult:
    async def _create_entry(
        self, host: str, port: int, username: str, password: str
    ) -> ConfigFlowResult:
        """Register new entry."""
        return self.async_create_entry(
                title=f"Ouman {host}",
                data={
                    CONF_HOST: host,
                    CONF_PORT: port,
                    CONF_USERNAME: username,
                    CONF_PASSWORD: password
                }
            title=f"Ouman {host}",
            data={
                CONF_HOST: host,
                CONF_PORT: port,
                CONF_USERNAME: username,
                CONF_PASSWORD: password,
            },
        )

    async def async_step_user(self, user_input: dict[str, Any] | None = None) -> ConfigFlowResult:
    async def async_step_user(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """TODO"""
        if user_input is None:
            return self.async_show_form(
                step_id="user", data_schema=USER_SCHEMA
            )
            return self.async_show_form(step_id="user", data_schema=USER_SCHEMA)

        _LOGGER.debug(user_input)
        return await self._create_entry(user_input[CONF_HOST], user_input[CONF_PORT], user_input[CONF_USERNAME], user_input[CONF_PASSWORD])
        return await self._create_entry(
            user_input[CONF_HOST],
            user_input[CONF_PORT],
            user_input[CONF_USERNAME],
            user_input[CONF_PASSWORD],
        )

M custom_components/ouman_eh800/const.py => custom_components/ouman_eh800/const.py +5 -3
@@ 1,3 1,5 @@
""" Ouman EH-800 consts """

DOMAIN = "ouman_eh800"

DEFAULT_PORT = 80


@@ 10,6 12,6 @@ CONF_PASSWORD = "password"
TEMPERATURE_SENSOR_TYPE_OUTSIDE = "outside"
TEMPERATURE_SENSOR_TYPE_L1_ROOM = "l1_room"
TEMPERATURE_SENSOR_TYPE_L1_SUPPLY = "l1_supply"
#TEMPERATURE_SENSOR_TYPE_ = ""
#TEMPERATURE_SENSOR_TYPE_ = ""
#TEMPERATURE_SENSOR_TYPE_ = ""
# TEMPERATURE_SENSOR_TYPE_ = ""
# TEMPERATURE_SENSOR_TYPE_ = ""
# TEMPERATURE_SENSOR_TYPE_ = ""

M custom_components/ouman_eh800/eh800.py => custom_components/ouman_eh800/eh800.py +26 -6
@@ 1,44 1,64 @@
"""TODO"""

import logging
import requests

_LOGGER = logging.getLogger(__name__)


class EH800:
    """TODO"""

    def __init__(self, host: str, port: int, username: str, password: str) -> None:
        """TODO"""
        self._uri = f"http://{host}:{port}"
        self._login = f"uid={username};pwd={password};"

        self._outside_temp = 0.0
        self._l1_room_temp = 0.0
        self._l1_supply_temp = 0.0

    def _refresh_login(self) -> bool:
        """TODO"""
        r = requests.get(f"{self._uri}/login?{self._login}")

        if r.text[:-1] == "login?result=ok;":
            _LOGGER.debug("Login ok")
            return True
        else:
            _LOGGER.debug("Login error")
            return False

        _LOGGER.debug("Login error")
        return False

    def _request_value(self, register) -> str:
        """TODO"""
        if not self._refresh_login():
            return
            return ""

        r = requests.get(f"{self._uri}/request?{register}")
        eq_index = r.text.find("=")
        sc_index = r.text.find(";")
        return r.text[eq_index+1:sc_index]
        return r.text[eq_index + 1 : sc_index]

    def get_outside_temp(self) -> float:
        """TODO"""
        return self._outside_temp

    def update_outside_temp(self):
        """TODO"""
        self._outside_temp = self._request_value("S_227_85")

    def get_l1_room_temp(self) -> float:
        """TODO"""
        return self._l1_room_temp

    def update_l1_room_temp(self):
        """TODO"""
        self._l1_room_temp = self._request_value("S_261_85")

    def get_l1_supply_temp(self) -> float:
        """TODO"""
        return self._l1_supply_temp

    def update_l1_supply_temp(self):
        """TODO"""
        self._l1_supply_temp = self._request_value("S_259_85")


M custom_components/ouman_eh800/sensor.py => custom_components/ouman_eh800/sensor.py +43 -9
@@ 1,28 1,51 @@
"""
TODO
"""

import logging

from datetime import timedelta

from homeassistant.components.sensor import SensorDeviceClass, SensorEntity, SensorStateClass
from homeassistant.components.sensor import (
    SensorDeviceClass,
    SensorEntity,
    SensorStateClass,
)
from homeassistant.core import HomeAssistant
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import UnitOfTemperature
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.util import Throttle

from. const import *
from .const import (
    DOMAIN,
    CONF_HOST,
    CONF_PORT,
    CONF_USERNAME,
    CONF_PASSWORD,
    TEMPERATURE_SENSOR_TYPE_L1_SUPPLY,
    TEMPERATURE_SENSOR_TYPE_L1_ROOM,
    TEMPERATURE_SENSOR_TYPE_OUTSIDE,
)
from .eh800 import EH800

_LOGGER = logging.getLogger(__name__)


async def async_setup_entry(
        hass: HomeAssistant,
        entry: ConfigEntry,
        async_add_entities: AddEntitiesCallback,
    ) -> None:
    hass: HomeAssistant,
    entry: ConfigEntry,
    async_add_entities: AddEntitiesCallback,
) -> None:
    """TODO"""
    config = hass.data[DOMAIN][entry.entry_id]

    eh800 = EH800(config[CONF_HOST], config[CONF_PORT], config[CONF_USERNAME], config[CONF_PASSWORD])
    eh800 = EH800(
        config[CONF_HOST],
        config[CONF_PORT],
        config[CONF_USERNAME],
        config[CONF_PASSWORD],
    )

    entities = []
    entities.append(TemperatureSensor(hass, eh800, TEMPERATURE_SENSOR_TYPE_OUTSIDE))


@@ 33,7 56,10 @@ async def async_setup_entry(

    return True


class TemperatureSensor(SensorEntity):
    """Temperature sensor"""

    def __init__(self, hass: HomeAssistant, api: EH800, sensor_type: str):
        self._hass = hass
        self._api = api


@@ 42,13 68,15 @@ class TemperatureSensor(SensorEntity):
        self._unique_id = f"{DOMAIN}_temperature_{sensor_type}".lower()

        self._state = 0.0
    

    @property
    def device_class(self):
        """TODO"""
        return SensorDeviceClass.TEMPERATURE

    @property
    def device_info(self):
        """TODO"""
        return {
            "identifiers": {(DOMAIN, self.unique_id)},
            "name": self.name,


@@ 56,26 84,32 @@ class TemperatureSensor(SensorEntity):

    @property
    def name(self):
        """TODO"""
        return self.unique_id

    @property
    def native_unit_of_measurement(self):
        """TODO"""
        return UnitOfTemperature.CELSIUS

    @property
    def state(self):
        """TODO"""
        return self._state

    @property
    def state_class(self):
        """TODO"""
        return SensorStateClass.MEASUREMENT

    @property
    def unique_id(self):
        """TODO"""
        return self._unique_id

    @Throttle(timedelta(minutes=1))
    async def async_update(self):
        """TODO"""
        if self._sensor_type == TEMPERATURE_SENSOR_TYPE_OUTSIDE:
            await self._hass.async_add_executor_job(self._api.update_outside_temp)
            self._state = self._api.get_outside_temp()

M flake.nix => flake.nix +6 -4
@@ 34,10 34,12 @@
            # Nix formatting
            alejandra.enable = true;

            # Toml formatting
            taplo.enable = true;

            # TODO: Python linting and formatting
            # Python linting and formatting
            pylint = {
              enable = true;
              args = ["--disable=import-error"];
            };
            black.enable = true;

            # Spell checking
            typos = {