Merge branch 'main' of https://github.com/dala318/python_ph803w

Author: dala318

.gitignore

@@ -1 +1,4 @@
-.vscode
+.vscode
+HIDDEN
+OLD
+*__pycache__

__init__.py

@@ -0,0 +1,47 @@
+"""The Detailed Hello World Push integration."""
+from __future__ import annotations
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.typing import ConfigType
+
+from . import hub
+from .const import DOMAIN
+
+VERSION = "0.0.1"
+
+# List of platforms to support. There should be a matching .py file for each,
+# eg <cover.py> and <sensor.py>
+PLATFORMS: list[str] = ["sensor"]
+
+
+# async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
+#     """Set up a skeleton component."""
+#     hass.data.setdefault(DOMAIN, {})[entry.entry_id] = hub.Hub(hass, None)
+#     hass.config_entries.async_setup_platforms(entry, PLATFORMS)
+
+#     return True
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up Hello World from a config entry."""
+    # Store an instance of the "connecting" class that does the work of speaking
+    # with your actual devices.
+    hass.data.setdefault(DOMAIN, {})[entry.entry_id] = hub.Hub(hass, entry.data["host"])
+
+    # This creates each HA object for each platform your device requires.
+    # It's done by calling the `async_setup_entry` function in each platform module.
+    hass.config_entries.async_setup_platforms(entry, PLATFORMS)
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    # This is called when an entry/configured device is to be removed. The class
+    # needs to unload itself, and remove callbacks. See the classes for further
+    # details
+    unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
+    if unload_ok:
+        hass.data[DOMAIN].pop(entry.entry_id)
+
+    return unload_ok

config_flow.py

@@ -0,0 +1,119 @@
+"""Config flow for Hello World integration."""
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+import voluptuous as vol
+
+from homeassistant import config_entries, exceptions
+from homeassistant.core import HomeAssistant
+
+from .const import DOMAIN  # pylint:disable=unused-import
+from .hub import Hub
+
+_LOGGER = logging.getLogger(__name__)
+
+# This is the schema that used to display the UI to the user. This simple
+# schema has a single required host field, but it could include a number of fields
+# such as username, password etc. See other components in the HA core code for
+# further examples.
+# Note the input displayed to the user will be translated. See the
+# translations/<lang>.json file and strings.json. See here for further information:
+# https://developers.home-assistant.io/docs/config_entries_config_flow_handler/#translations
+# At the time of writing I found the translations created by the scaffold didn't
+# quite work as documented and always gave me the "Lokalise key references" string
+# (in square brackets), rather than the actual translated value. I did not attempt to
+# figure this out or look further into it.
+DATA_SCHEMA = vol.Schema({("host"): str})
+# DATA_SCHEMA = vol.Schema({})
+
+
+async def validate_input(hass: HomeAssistant, data: dict) -> dict[str, Any]:
+    """Validate the user input allows us to connect.
+
+    Data has the keys from DATA_SCHEMA with values provided by the user.
+    """
+    # Validate the data can be used to set up a connection.
+
+    # This is a simple example to show an error in the UI for a short hostname
+    # The exceptions are defined at the end of this file, and are used in the
+    # `async_step_user` method below.
+    if len(data["host"]) < 3:
+        raise InvalidHost
+
+    hub = Hub(hass, data["host"])
+    # The dummy hub provides a `test_connection` method to ensure it's working
+    # as expected
+    result = await hub.test_connection()
+    if not result:
+        # If there is an error, raise an exception to notify HA that there was a
+        # problem. The UI will also show there was a problem
+        raise CannotConnect
+
+    # If your PyPI package is not built with async, pass your methods
+    # to the executor:
+    # await hass.async_add_executor_job(
+    #     your_validate_func, data["username"], data["password"]
+    # )
+
+    # If you cannot connect:
+    # throw CannotConnect
+    # If the authentication is wrong:
+    # InvalidAuth
+
+    # Return info that you want to store in the config entry.
+    # "Title" is what is displayed to the user for this hub device
+    # It is stored internally in HA as part of the device config.
+    # See `async_step_user` below for how this is used
+    return {"title": data["host"]}
+
+
+class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for Hello World."""
+
+    VERSION = 1
+    # Pick one of the available connection classes in homeassistant/config_entries.py
+    # This tells HA if it should be asking for updates, or it'll be notified of updates
+    # automatically. This example uses PUSH, as the dummy hub will notify HA of
+    # changes.
+    CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_PUSH
+
+    async def async_step_user(self, user_input=None):
+        """Handle the initial step."""
+        # This goes through the steps to take the user through the setup process.
+        # Using this it is possible to update the UI and prompt for additional
+        # information. This example provides a single form (built from `DATA_SCHEMA`),
+        # and when that has some validated input, it calls `async_create_entry` to
+        # actually create the HA config entry. Note the "title" value is returned by
+        # `validate_input` above.
+        errors = {}
+        if user_input is not None:
+            try:
+                info = await validate_input(self.hass, user_input)
+
+                return self.async_create_entry(title=info["title"], data=user_input)
+            except CannotConnect:
+                errors["base"] = "cannot_connect"
+            except InvalidHost:
+                # The error string is set here, and should be translated.
+                # This example does not currently cover translations, see the
+                # comments on `DATA_SCHEMA` for further details.
+                # Set the error on the `host` field, not the entire form.
+                errors["host"] = "cannot_connect"
+            except Exception:  # pylint: disable=broad-except
+                _LOGGER.exception("Unexpected exception")
+                errors["base"] = "unknown"
+
+        # If there is no user input or there were errors, show the form again, including any errors that were found with the input.
+        return self.async_show_form(
+            step_id="user", data_schema=DATA_SCHEMA, errors=errors
+        )
+
+
+class CannotConnect(exceptions.HomeAssistantError):
+    """Error to indicate we cannot connect."""
+
+
+class InvalidHost(exceptions.HomeAssistantError):
+    """Error to indicate there is an invalid hostname."""

const.py

@@ -0,0 +1 @@
+DOMAIN = "ph803w"

hub.py

@@ -0,0 +1,175 @@
+"""A demonstration 'hub' that connects several devices."""
+from __future__ import annotations
+
+# In a real implementation, this would be in an external library that's on PyPI.
+# The PyPI package needs to be included in the `requirements` section of manifest.json
+# See https://developers.home-assistant.io/docs/creating_integration_manifest
+# for more information.
+# This dummy hub always returns 3 rollers.
+import asyncio
+import socket
+import random
+
+from .lib import discovery, device
+
+from homeassistant.core import HomeAssistant
+
+PH803W_DEFAULT_TCP_PORT = 12416
+
+
+class Hub:
+    """Dummy hub for Hello World example."""
+
+    manufacturer = "Demonstration Corp"
+
+    def __init__(self, hass: HomeAssistant, host: str) -> None:
+        """Init dummy hub."""
+        # discover = discovery.Discovery()
+        # res = discover.run()
+        dev = device.Device(host)
+        res = dev.run()
+        # dev.close()
+
+        self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self.socket.connect((host, PH803W_DEFAULT_TCP_PORT))
+
+        data = bytes.fromhex("0000000303000006")
+        self.socket.sendall(data)
+        response = self.socket.recv(1024)
+        passcode_lenth = response[9]
+        passcode_raw = response[10 : 10 + passcode_lenth]
+        passcode = passcode_raw.decode("utf-8")
+
+        data = (
+            bytes.fromhex("000000030f00000800")
+            + passcode_lenth.to_bytes(1, "little")
+            + passcode_raw
+        )
+        self.socket.sendall(data)
+        response = self.socket.recv(1024)
+        if response[8] != 0:
+            print("Error connecting")
+
+        # Connection established, from now on some cyclig bahavior
+        data = bytes.fromhex("000000030400009002")
+        self.socket.sendall(data)
+        empty_counter = 0
+        data = bytes.fromhex("0000000303000015")
+        response = self.socket.recv(1024)
+        # if len(response) == 0:
+        #     empty_counter += 1
+        #     continue
+        empty_counter = 0
+        # print(response)
+        if len(response) == 18:
+            flag1 = response[8]
+            if flag1 & 0b0000_0100:
+                print("In water")
+            flag2 = response[9]
+            if flag2 & 0b0000_0010:
+                print("ORP on")
+            if flag2 & 0b0000_0001:
+                print("PH on")
+            # state_raw = response[8 : 9]
+            ph_raw = response[10:12]
+            self.ph = int.from_bytes(ph_raw, "big") * 0.01
+            redox_raw = response[12:14]
+            self.redox = int.from_bytes(redox_raw, "big") - 2000
+            unknown1_raw = response[14:16]
+            unknown1 = int.from_bytes(unknown1_raw, "big")
+            unknown2_raw = response[15:18]
+            unknown2 = int.from_bytes(unknown2_raw, "big")
+
+        self._host = host
+        self._hass = hass
+        self._name = host
+        self._id = host.lower()
+        self.probe = Probe(f"{self._id}_1", f"{self._name} 1", self)
+        self.online = True
+
+    @property
+    def hub_id(self) -> str:
+        """ID for dummy hub."""
+        return self._id
+
+    async def test_connection(self) -> bool:
+        """Test connectivity to the Dummy hub is OK."""
+        await asyncio.sleep(1)
+        return True
+
+
+class Probe:
+    """Dummy roller (device for HA) for Hello World example."""
+
+    def __init__(self, id: str, name: str, hub: Hub) -> None:
+        """Init dummy roller."""
+        self._id = id
+        self.hub = hub
+        self.name = name
+        self._ph_value = hub.ph
+        self._orp_value = hub.redox
+        self._callbacks = set()
+        self._loop = asyncio.get_event_loop()
+
+        # Some static information about this device
+        self.firmware_version = f"0.0.{random.randint(1, 9)}"
+        self.model = "Test Device"
+
+    @property
+    def id(self) -> str:
+        """Return ID for roller."""
+        return self._id
+
+    # async def set_position(self, position: int) -> None:
+    #     """
+    #     Set dummy cover to the given position.
+
+    #     State is announced a random number of seconds later.
+    #     """
+    #     self._target_position = position
+
+    #     # Update the moving status, and broadcast the update
+    #     self.moving = position - 50
+    #     await self.publish_updates()
+
+    #     self._loop.create_task(self.delayed_update())
+
+    async def delayed_update(self) -> None:
+        """Publish updates, with a random delay to emulate interaction with device."""
+        await asyncio.sleep(random.randint(1, 10))
+        self.moving = 0
+        await self.publish_updates()
+
+    def register_callback(self, callback: Callable[[], None]) -> None:
+        """Register callback, called when Roller changes state."""
+        self._callbacks.add(callback)
+
+    def remove_callback(self, callback: Callable[[], None]) -> None:
+        """Remove previously registered callback."""
+        self._callbacks.discard(callback)
+
+    # In a real implementation, this library would call it's call backs when it was
+    # notified of any state changeds for the relevant device.
+    async def publish_updates(self) -> None:
+        """Schedule call all registered callbacks."""
+        pass
+        # self._current_position = self._target_position
+        # for callback in self._callbacks:
+        #     callback()
+
+    @property
+    def online(self) -> float:
+        """Roller is online."""
+        # The dummy roller is offline about 10% of the time. Returns True if online,
+        # False if offline.
+        return random.random() > 0.1
+
+    @property
+    def ph(self) -> float:
+        """Battery level as a percentage."""
+        return self._ph_value
+
+    @property
+    def orp(self) -> float:
+        """Return a random voltage roughly that of a 12v battery."""
+        return self._orp_value