Add support for new Ginlong/Solis Wifi stick (S3-WIFI-ST)

omnikinverter/models.py

@@ -200,6 +200,40 @@ def from_tcp(data: dict[str, Any]) -> Inverter:
             ),
         )
 
+    @staticmethod
+    def from_cgi(data: str) -> Inverter:
+        """Return Inverter object from the Omnik Inverter response.
+
+        Args:
+            data: The CGI (webscraping) data from the Omnik Inverter.
+
+        Returns:
+            An Inverter object.
+        """
+
+        split_data = data.split(";")
+        if len(split_data) < 7:
+            raise OmnikInverterWrongSourceError(
+                    "Your inverter has no data source from cgi."
+                )      
+
+        def try_parse_float(item):
+            try:
+                return float(item)
+            except:
+                return None
+
+        return Inverter(
+            serial_number=split_data[0],
+            model=split_data[2],
+            firmware=split_data[1],
+            firmware_slave=None,
+            solar_rated_power=None,
+            solar_current_power=try_parse_float(split_data[4]),
+            solar_energy_today=try_parse_float(split_data[5]),
+            solar_energy_total=try_parse_float(split_data[6]),
+        )
+
 
 @dataclass
 class Device:
@@ -285,3 +319,32 @@ def get_value(search_key: str) -> Any:
             firmware=get_value("version"),
             ip_address=get_value("wanIp"),
         )
+
+    @staticmethod
+    def from_cgi(data: str) -> Device:
+        """Return Device object from the Omnik Inverter response.
+
+        Args:
+            data: The CGI (webscraping) data from the Omnik Inverter.
+
+        Returns:
+            A Device object.
+        """
+
+        split_data = data.split(";")
+        if len(split_data) < 9:
+            raise OmnikInverterWrongSourceError(
+                    "Your inverter has no device data from cgi."
+                )
+
+        def try_parse_int(item):
+            try:
+                return int(item)
+            except:
+                return None
+
+        return Device(
+            signal_quality=try_parse_int(split_data[8]),
+            firmware=split_data[1],
+            ip_address=split_data[9],
+        )

omnikinverter/omnikinverter.py

@@ -75,7 +75,7 @@ async def request(
 
         # Use big try to make sure manual session is always cleaned up
         try:
-            if self.source_type == "html" and (
+            if (self.source_type == "html" or self.source_type == "cgi") and (
                 self.username is None or self.password is None
             ):
                 raise OmnikInverterAuthError(
@@ -179,6 +179,9 @@ async def inverter(self) -> Inverter:
         if self.source_type == "html":
             data = await self.request("status.html")
             return Inverter.from_html(data)
+        if self.source_type == "cgi":
+            data = await self.request("inverter.cgi", params={"t": "123"})
+            return Inverter.from_cgi(data)
         if self.source_type == "javascript":
             data = await self.request("js/status.js")
             return Inverter.from_js(data)
@@ -203,6 +206,9 @@ async def device(self) -> Device:
         if self.source_type == "html":
             data = await self.request("status.html")
             return Device.from_html(data)
+        if self.source_type == "cgi":
+            data = await self.request("moniter.cgi", params={"t": "123"})
+            return Device.from_cgi(data)
         if self.source_type == "javascript":
             data = await self.request("js/status.js")
             return Device.from_js(data)

tests/fixtures/inverter.cgi

@@ -0,0 +1 @@
+12345678910;3280031;205;38.4;780;23.799999;d;NO; 

tests/fixtures/moniter.cgi

@@ -0,0 +1 @@
+5A1227251CB01447;00010130;Disable;null;null;null;Enable;SSID NAME;60;192.168.0.106;BE:BA:DE:FF:1C:6B;Connected;Unconnected; 

tests/test_models.py

@@ -130,6 +130,67 @@ async def test_device_html(aresponses: ResponsesMockServer) -> None:
         assert device.firmware == "ME_08_0102_2.03"
         assert device.ip_address == "192.168.0.106"
 
+@pytest.mark.asyncio
+async def test_inverter_cgi(aresponses: ResponsesMockServer) -> None:
+    """Test request from an Inverter - CGI source."""
+    aresponses.add(
+        "example.com",
+        "/inverter.cgi",
+        "GET",
+        aresponses.Response(
+            status=200,
+            headers={"Content-Type": "text/html"},
+            text=load_fixtures("inverter.cgi"),
+        ),
+    )
+
+    async with aiohttp.ClientSession() as session:
+        client = OmnikInverter(
+            host="example.com",
+            source_type="cgi",
+            username="klaas",
+            password="supercool",
+            session=session,
+        )
+        inverter: Inverter = await client.inverter()
+        assert inverter
+        assert inverter.serial_number == "12345678910"
+        assert inverter.firmware == "3280031"
+        assert inverter.firmware_slave is None
+        assert inverter.model == "205"
+        assert inverter.solar_rated_power is None
+        assert inverter.solar_current_power == 780
+        assert inverter.solar_energy_today == 23.799999
+        assert inverter.solar_energy_total is None
+
+
+@pytest.mark.asyncio
+async def test_device_cgi(aresponses: ResponsesMockServer) -> None:
+    """Test request from a Device - CGI source."""
+    aresponses.add(
+        "example.com",
+        "/moniter.cgi",
+        "GET",
+        aresponses.Response(
+            status=200,
+            headers={"Content-Type": "text/html"},
+            text=load_fixtures("moniter.cgi"),
+        ),
+    )
+
+    async with aiohttp.ClientSession() as session:
+        client = OmnikInverter(
+            host="example.com",
+            source_type="cgi",
+            username="klaas",
+            password="supercool",
+            session=session,
+        )
+        device: Device = await client.device()
+        assert device
+        assert device.signal_quality == 60
+        assert device.firmware == "00010130"
+        assert device.ip_address == "192.168.0.106"
 
 @pytest.mark.asyncio
 async def test_inverter_without_session(aresponses: ResponsesMockServer) -> None: