Merge pull request #6 from cryptk/pydantic

feat: output data parsed into pydantic models

.github/workflows/ci.yml

@@ -40,8 +40,8 @@ jobs:
           - "3.11"
         os:
           - ubuntu-latest
-          - windows-latest
-          - macOS-latest
+          # - windows-latest
+          # - macOS-latest
     runs-on: ${{ matrix.os }}
     steps:
       - uses: actions/checkout@v3
@@ -56,28 +56,28 @@ jobs:
       # - name: Test with Pytest
       #   run: poetry run pytest
       #   shell: bash
-  # release:
-  #   runs-on: ubuntu-latest
-  #   environment: release
-  #   if: github.ref == 'refs/heads/main'
-  #   needs:
-  #     - test
+  release:
+    runs-on: ubuntu-latest
+    environment: release
+    if: github.ref == 'refs/heads/main'
+    needs:
+      - test
 
-  #   steps:
-  #     - uses: actions/checkout@v3
-  #       with:
-  #         fetch-depth: 0
-  #         persist-credentials: false
+    steps:
+      - uses: actions/checkout@v3
+        with:
+          fetch-depth: 0
+          persist-credentials: false
 
-  #     # Run semantic release:
-  #     # - Update CHANGELOG.md
-  #     # - Update version in code
-  #     # - Create git tag
-  #     # - Create GitHub release
-  #     # - Publish to PyPI
-  #     - name: Python Semantic Release
-  #       uses: relekang/python-semantic-release@v7.33.2
-  #       with:
-  #         github_token: ${{ secrets.GH_TOKEN }}
-  #         repository_username: __token__
-  #         repository_password: ${{ secrets.PYPI_TOKEN }}
+      # Run semantic release:
+      # - Update CHANGELOG.md
+      # - Update version in code
+      # - Create git tag
+      # - Create GitHub release
+      # - Publish to PyPI
+      - name: Python Semantic Release
+        uses: relekang/python-semantic-release@v7.33.2
+        with:
+          github_token: ${{ secrets.GH_TOKEN }}
+          repository_username: __token__
+          repository_password: ${{ secrets.REPOSITORY_PASSWORD }}

.pre-commit-config.yaml

@@ -39,13 +39,11 @@ repos:
             -   id: ruff
                 args:
                     - --fix
-    # -   repo: local
-    #     hooks:
-    #         -   id: pylint
-    #             name: pylint
-    #             entry: poetry run -- pylint
-    #             language: system
-    #             types: [python]
+    -   repo: https://github.com/pylint-dev/pylint
+        rev: v2.17.0
+        hooks:
+            -   id: pylint
+                additional_dependencies: [ "pydantic", "xmltodict" ]
     -   repo: https://github.com/pre-commit/mirrors-mypy
         rev: v1.2.0
         hooks:

pyomnilogic_local/api.py

@@ -5,6 +5,10 @@
 from typing import Literal, overload
 import xml.etree.ElementTree as ET
 
+from .models.filter_diagnostics import FilterDiagnostics
+from .models.mspconfig import MSPConfig
+from .models.telemetry import Telemetry
+from .models.util import to_pydantic
 from .protocol import OmniLogicProtocol
 from .types import ColorLogicBrightness, ColorLogicShow, ColorLogicSpeed, MessageType
 
@@ -52,6 +56,7 @@ async def async_get_alarm_list(self) -> str:
 
         return await self.async_send_message(MessageType.GET_ALARM_LIST, req_body, True)
 
+    @to_pydantic(pydantic_type=MSPConfig)
     async def async_get_config(self) -> str:
         body_element = ET.Element("Request", {"xmlns": "http://nextgen.hayward.com/api"})
 
@@ -62,6 +67,7 @@ async def async_get_config(self) -> str:
 
         return await self.async_send_message(MessageType.REQUEST_CONFIGURATION, req_body, True)
 
+    @to_pydantic(pydantic_type=FilterDiagnostics)
     async def async_get_filter_diagnostics(self, pool_id: int, equipment_id: int) -> str:
         """async_get_filter_diagnostics handles sending a GetUIFilterDiagnosticInfo XML API call to the Hayward Omni pool controller
 
@@ -90,6 +96,7 @@ async def async_get_filter_diagnostics(self, pool_id: int, equipment_id: int) ->
     async def async_get_log_config(self) -> str:
         return await self.async_send_message(MessageType.REQUEST_LOG_CONFIG, None, True)
 
+    @to_pydantic(pydantic_type=Telemetry)
     async def async_get_telemetry(self) -> str:
         body_element = ET.Element("Request", {"xmlns": "http://nextgen.hayward.com/api"})
 

pyomnilogic_local/cli.py

@@ -6,34 +6,58 @@
 
 from pyomnilogic_local.api import OmniLogicAPI
 
-# from pyomnilogic_local.models.telemetry import Telemetry
-
+from .models.filter_diagnostics import FilterDiagnostics
+from .models.mspconfig import MSPConfig
+from .models.telemetry import Telemetry
 
 POOL_ID = 7
 PUMP_EQUIPMENT_ID = 8
 LIGHT_EQUIPMENT_ID = 10
 HEATER_EQUIPMENT_ID = 18
+RELAY_SYSTEM_ID = 13
 
 
 async def async_main() -> None:
+    diags: FilterDiagnostics  # noqa: F842
+    mspconfig: MSPConfig  # noqa: F842
+    telem: Telemetry  # noqa: F842
+
     omni = OmniLogicAPI(os.environ.get("OMNILOGIC_HOST", "127.0.0.1"), 10444, 5.0)
 
     # Some basic calls to run some testing against the library
-    # Fetch the MSPConfig data
-    config = await omni.async_get_config()
-    print(config)
-    # parsed_config = MSPConfig.load_xml(config)
 
-    # Fetch the current telemetry data
+    # Fetch the MSPConfig data parsed into a model
+    mspconfig = await omni.async_get_config()
+    # Fetch the MSPConfig data as the raw XML
+    # mspconfig = await omni.async_get_config(raw=True)
+    print(mspconfig)
+
+    # Fetch the current telemetry data parsed into a model
     telem = await omni.async_get_telemetry()
+    # Fetch the current telemetry data as the raw XML
+    # telem = await omni.async_get_telemetry(raw=True)
     print(telem)
-    # parsed_telem = Telemetry.load_xml(xml=telem)
 
     # Fetch a list of current alarms
     # print(await omni.async_get_alarm_list())
 
     # Fetch diagnostic data for a filter pump
-    # print(await omni.async_get_filter_diagnostics(POOL_ID, PUMP_EQUIPMENT_ID))
+    diags = await omni.async_get_filter_diagnostics(POOL_ID, PUMP_EQUIPMENT_ID)
+    print(diags)
+    # Decode the filter display revision
+    # b1=chr(diags.get_param_by_name("DisplayFWRevisionB1"))
+    # b2=chr(diags.get_param_by_name("DisplayFWRevisionB2"))
+    # b3=chr(diags.get_param_by_name("DisplayFWRevisionB3"))
+    # b4=chr(diags.get_param_by_name("DisplayFWRevisionB4"))
+    # # b5 and b6 are whitespace and a null terminator
+    # # b5=chr(diags.get_param_by_name("DisplayFWRevisionB5"))
+    # # b6=chr(diags.get_param_by_name("DisplayFWRevisionB6"))
+    # print(f"{b1}{b2}.{b3}.{b4}")
+    # # Decode the filter power consumption (don't do this, it's returned already decoded in the telemetry)
+    # p1=diags.get_param_by_name("PowerMSB")
+    # p2=diags.get_param_by_name("PowerLSB")
+    # # The f-string below converts the bytes to hex and displays them.  Just get this value from the telemetry, it's easier
+    # print(f"{p1:x}{p2:x}")
 
     # Fetch logging configuration
     # print(await omni.async_get_log_config())
@@ -58,6 +82,10 @@ async def async_main() -> None:
     # await omni.async_set_equipment(POOL_ID, LIGHT_EQUIPMENT_ID, True)
     # await omni.async_set_equipment(POOL_ID, LIGHT_EQUIPMENT_ID, False)
 
+    # Turn a relay on/off
+    # await omni.async_set_equipment(POOL_ID, RELAY_SYSTEM_ID, True)
+    # await omni.async_set_equipment(POOL_ID, RELAY_SYSTEM_ID, False)
+
 
 def main() -> None:
     logging.basicConfig(format="%(asctime)s %(levelname)s: %(message)s", level=logging.DEBUG)

pyomnilogic_local/models/filter_diagnostics.py

@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+from xmltodict import parse as xml_parse
+
+
+class FilterDiagnosticsParameter(BaseModel):
+    name: str = Field(alias="@name")
+    dataType: str = Field(alias="@dataType")
+    value: int = Field(alias="#text")
+
+
+class FilterDiagnosticsParameters(BaseModel):
+    parameter: list[FilterDiagnosticsParameter] = Field(alias="Parameter")
+
+
+class FilterDiagnostics(BaseModel):
+    name: str = Field(alias="Name")
+    # parameters: FilterDiagnosticsParameters = Field(alias="Parameters")
+    parameters: list[FilterDiagnosticsParameter] = Field(alias="Parameters")
+
+    class Config:
+        orm_mode = True
+
+    def get_param_by_name(self, name: str) -> int:
+        return [param.value for param in self.parameters if param.name == name][0]
+
+    @staticmethod
+    def load_xml(xml: str) -> FilterDiagnostics:
+        data = xml_parse(
+            xml,
+            # Some things will be lists or not depending on if a pool has more than one of that piece of equipment.  Here we are coercing
+            # everything that *could* be a list into a list to make the parsing more consistent.
+            force_list=("Parameter"),
+        )
+        # The XML nests the Parameter entries under a Parameters entry, this is annoying to work with.  Here we are adjusting the data to
+        # remove that extra level in the data
+        data["Response"]["Parameters"] = data["Response"]["Parameters"]["Parameter"]
+        return FilterDiagnostics.parse_obj(data["Response"])