Add HTML source support (#38)

* Update the documentation

* Add support for HTML sources

* Update example and add exception

* Change js to full name javascript

* Update the documentation

README.md

@@ -51,7 +51,12 @@ from omnikinverter import OmnikInverter
 
 async def main():
     """Show example on getting Omnik Inverter data."""
-    async with OmnikInverter(host="example_host", use_json=False) as client:
+    async with OmnikInverter(
+        host="example_host",
+        source_type="javascript",
+        username="omnik",
+        password="inverter",
+    ) as client:
         inverter = await client.inverter()
         device = await client.device()
         print(inverter)
@@ -63,6 +68,8 @@ if __name__ == "__main__":
     loop.run_until_complete(main())
 ```
 
+For the **source type** you can choose between: `javascript` (default), `json` and `html`.
+
 ## Data
 
 You can read the following data with this package:
@@ -80,7 +87,7 @@ You can read the following data with this package:
 
 ### Device
 
-- Signal Quality (only JS)
+- Signal Quality (only with JS)
 - Firmware Version
 - IP Address
 
@@ -94,20 +101,28 @@ You need at least:
 - Python 3.8+
 - [Poetry][poetry-install]
 
-To install all packages, including all development requirements:
+Install all packages, including all development requirements:
 
 ```bash
 poetry install
-pre-commit install
 ```
 
-Poetry creates by default an virtual environment where it installs all necessary pip packages, to enter or exit the venv run the following commands:
+Poetry creates by default an virtual environment where it installs all
+necessary pip packages, to enter or exit the venv run the following commands:
 
 ```bash
 poetry shell
 exit
 ```
 
+Setup the pre-commit check, you must run this inside the virtual environment:
+
+```bash
+pre-commit install
+```
+
+*Now you're all set to get started!*
+
 As this repository uses the [pre-commit][pre-commit] framework, all changes
 are linted and tested with each commit. You can run all checks and tests
 manually, using the following command:

omnikinverter/exceptions.py

@@ -5,6 +5,10 @@ class OmnikInverterError(Exception):
     """Generic Omnik Inverter exception."""
 
 
+class OmnikInverterAuthError(OmnikInverterError):
+    """Omnik Inverter Authentication exception."""
+
+
 class OmnikInverterConnectionError(OmnikInverterError):
     """Omnik Inverter connection exception."""
 

omnikinverter/models.py

@@ -71,6 +71,44 @@ def validation(data_list):
             solar_energy_total=float(get_value("i_eall")),
         )
 
+    @staticmethod
+    def from_html(data: dict[str, Any]) -> Inverter:
+        """Return Inverter object from the Omnik Inverter response.
+
+        Args:
+            data: The HTML (webscraping) data from the Omnik Inverter.
+
+        Returns:
+            An Inverter object.
+        """
+
+        def get_value(search_key):
+            match = re.search(f'(?<={search_key}=").*?(?=";)', data.replace(" ", ""))
+            try:
+                value = match.group(0)
+                if value != "":
+                    if search_key in ["webdata_now_p", "webdata_rate_p"]:
+                        return int(value)
+                    if search_key in ["webdata_today_e", "webdata_total_e"]:
+                        return float(value)
+                    return value
+                return None
+            except AttributeError as exception:
+                raise OmnikInverterWrongSourceError(
+                    "Your inverter has no data source from a html file."
+                ) from exception
+
+        return Inverter(
+            serial_number=get_value("webdata_sn"),
+            model=get_value("webdata_pv_type"),
+            firmware=get_value("webdata_msvn"),
+            firmware_slave=get_value("webdata_ssvn"),
+            solar_rated_power=get_value("webdata_rate_p"),
+            solar_current_power=get_value("webdata_now_p"),
+            solar_energy_today=get_value("webdata_today_e"),
+            solar_energy_total=get_value("webdata_total_e"),
+        )
+
     @staticmethod
     def from_js(data: dict[str, Any]) -> Inverter:
         """Return Inverter object from the Omnik Inverter response.
@@ -145,6 +183,33 @@ def from_json(data: dict[str, Any]) -> Device:
             ip_address=data["ip"],
         )
 
+    @staticmethod
+    def from_html(data: dict[str, Any]) -> Device:
+        """Return Device object from the Omnik Inverter response.
+
+        Args:
+            data: The HTML (webscraping) data from the Omnik Inverter.
+
+        Returns:
+            An Device object.
+        """
+
+        for correction in [" ", "%"]:
+            data = data.replace(correction, "")
+
+        def get_value(search_key):
+            match = re.search(f'(?<={search_key}=").*?(?=";)', data)
+            value = match.group(0)
+            if value != "":
+                return value
+            return None
+
+        return Device(
+            signal_quality=get_value("cover_sta_rssi"),
+            firmware=get_value("cover_ver"),
+            ip_address=get_value("cover_sta_ip"),
+        )
+
     @staticmethod
     def from_js(data: dict[str, Any]) -> Device:
         """Return Device object from the Omnik Inverter response.
@@ -155,22 +220,20 @@ def from_js(data: dict[str, Any]) -> Device:
         Returns:
             An Device object.
         """
-
-        def get_value(value_type):
-            if value_type == "ip" and data.find("wanIp") != -1:
-                match = re.search(r'(?<=wanIp=").*?(?=";)', data.replace(" ", ""))
-                value = match.group(0)
-            if value_type == "signal" and data.find("m2mRssi") != -1:
-                match = re.search(r'(?<=m2mRssi=").*?(?=";)', data.replace(" ", ""))
-                value = match.group(0).replace("%", "")
-                return int(value)
-            if value_type == "version" and data.find("version") != -1:
-                match = re.search(r'(?<=version=").*?(?=";)', data.replace(" ", ""))
-                value = match.group(0)
-            return value
+        for correction in [" ", "%"]:
+            data = data.replace(correction, "")
+
+        def get_value(search_key):
+            match = re.search(f'(?<={search_key}=").*?(?=";)', data)
+            value = match.group(0)
+            if value != "":
+                if search_key == "m2mRssi":
+                    return int(value)
+                return value
+            return None
 
         return Device(
-            signal_quality=get_value("signal"),
+            signal_quality=get_value("m2mRssi"),
             firmware=get_value("version"),
-            ip_address=get_value("ip"),
+            ip_address=get_value("wanIp"),
         )

omnikinverter/omnikinverter.py

@@ -6,6 +6,7 @@
 from dataclasses import dataclass
 from typing import Any
 
+import aiohttp
 import async_timeout
 from aiohttp.client import ClientError, ClientResponseError, ClientSession
 from yarl import URL
@@ -18,26 +19,34 @@
 class OmnikInverter:
     """Main class for handling connections with the Omnik Inverter."""
 
+    # pylint: disable=too-many-arguments
     def __init__(
         self,
         host: str,
-        use_json: bool = False,
+        username: str | None = None,
+        password: str | None = None,
+        source_type: str = "javascript",
         request_timeout: int = 10,
         session: ClientSession | None = None,
     ) -> None:
         """Initialize connection with the Omnik Inverter.
 
         Args:
             host: Hostname or IP address of the Omnik Inverter.
-            use_json: Boolean to confirm you use a JSON source.
+            username: Username for HTTP auth, if enabled.
+            password: Password for HTTP auth, if enabled.
+            source_type: Whisch source your inverter uses
+                [javascript (default), html, json].
             request_timeout: An integer with the request timeout in seconds.
             session: Optional, shared, aiohttp client session.
         """
         self._session = session
         self._close_session = False
 
         self.host = host
-        self.use_json = use_json
+        self.username = username
+        self.password = password
+        self.source_type = source_type
         self.request_timeout = request_timeout
 
     async def request(
@@ -71,11 +80,16 @@ async def request(
             self._session = ClientSession()
             self._close_session = True
 
+        auth = None
+        if self.username and self.password:
+            auth = aiohttp.BasicAuth(self.username, self.password)
+
         try:
             with async_timeout.timeout(self.request_timeout):
                 response = await self._session.request(
                     "GET",
                     url,
+                    auth=auth,
                     params=params,
                     headers=headers,
                 )
@@ -89,7 +103,7 @@ async def request(
                 "Error occurred while communicating with Omnik Inverter device"
             ) from exception
 
-        types = ["application/json", "application/x-javascript"]
+        types = ["application/json", "application/x-javascript", "text/html"]
         content_type = response.headers.get("Content-Type", "")
         if not any(item in content_type for item in types):
             text = await response.text()
@@ -106,9 +120,12 @@ async def inverter(self) -> Inverter:
         Returns:
             A Inverter data object from the Omnik Inverter.
         """
-        if self.use_json:
+        if self.source_type == "json":
             data = await self.request("status.json", params={"CMD": "inv_query"})
             return Inverter.from_json(data)
+        if self.source_type == "html":
+            data = await self.request("status.html")
+            return Inverter.from_html(data)
         data = await self.request("js/status.js")
         return Inverter.from_js(data)
 
@@ -118,9 +135,12 @@ async def device(self) -> Device:
         Returns:
             A Device data object from the Omnik Inverter.
         """
-        if self.use_json:
+        if self.source_type == "json":
             data = await self.request("status.json", params={"CMD": "inv_query"})
             return Device.from_json(data)
+        if self.source_type == "html":
+            data = await self.request("status.html")
+            return Device.from_html(data)
         data = await self.request("js/status.js")
         return Device.from_js(data)
 

test_output.py

@@ -8,7 +8,10 @@
 
 async def main():
     """Test."""
-    async with OmnikInverter(host="example.com", use_json=False) as client:
+    async with OmnikInverter(
+        host="examples.com",
+        source_type="js",
+    ) as client:
         inverter: OmnikInverter = await client.inverter()
         device: OmnikInverter = await client.device()
         print(inverter)

tests/fixtures/status.html

@@ -0,0 +1 @@
+var webdata_sn = "12345678910";var webdata_msvn = "V5.07Build245";var webdata_ssvn = "";var webdata_pv_type = "Omnik2500tl ";var webdata_rate_p = "2500";var webdata_now_p = "219";var webdata_today_e = "0.23";var webdata_total_e = "6454.5";var webdata_alarm = "";var webdata_utime = "1";var cover_mid = "902000501";var cover_ver = "ME_08_0102_2.03";var cover_wmode = "";var cover_ap_ssid = "";var cover_ap_ip = "";var cover_ap_mac = "F0FE6B79BA5B";var cover_sta_ssid = "";var cover_sta_rssi = "";var cover_sta_ip = "192.168.0.106";var cover_sta_mac = "F0FE6B79BA5A";var status_a = "1";var status_b = "0";var status_c = "0";

tests/fixtures/status_devicearray.js

@@ -1 +1 @@
-var version= "H4.01.51MW.2.01W1.0.64(2018-01-251-D)";var m2mRssi= "39%";var wanIp= "192.168.0.10";var myDeviceArray=new Array();myDeviceArray[0]="12345678910,V4.08Build215,V4.12Build246,Omnik1500tl ,,850,232,52002,,1,";
+var version= "H4.01.51MW.2.01W1.0.64(2018-01-251-D)";var m2mRssi= "39%";var wanIp= "";var myDeviceArray=new Array();myDeviceArray[0]="12345678910,V4.08Build215,V4.12Build246,Omnik1500tl ,,850,232,52002,,1,";

tests/fixtures/wrong_status.html

@@ -0,0 +1 @@
+Lol it's empty!