CLI Enhancements

- new debug command to parse PCAP data
- remove functions that no longer appear to be supported by modern
  firmware
- general cleanup and improved error handling

Author: cryptk

poetry.lock

@@ -61,21 +61,6 @@ async def async_send_message(self, message_type: MessageType, message: str | Non
 
         return resp
 
-    async def async_get_alarm_list(self) -> str:
-        """Retrieve a list of alarms from the Omni.
-
-        Returns:
-            str: An XML body indicating any alarms that are present
-        """
-        body_element = ET.Element("Request", {"xmlns": "http://nextgen.hayward.com/api"})
-
-        name_element = ET.SubElement(body_element, "Name")
-        name_element.text = "GetAllAlarmList"
-
-        req_body = ET.tostring(body_element, xml_declaration=True, encoding="unicode")
-
-        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:
         """Retrieve the MSPConfig from the Omni, optionally parse it into a pydantic model.
@@ -125,14 +110,6 @@ async def async_get_filter_diagnostics(
 
         return await self.async_send_message(MessageType.GET_FILTER_DIAGNOSTIC_INFO, req_body, True)
 
-    async def async_get_log_config(self) -> str:
-        """Retrieve the logging configuration from the Omni.
-
-        Returns:
-            str: An XML body describing the logging configuration
-        """
-        return await self.async_send_message(MessageType.REQUEST_LOG_CONFIG, None, True)
-
     @to_pydantic(pydantic_type=Telemetry)
     async def async_get_telemetry(self) -> str:
         """Retrieve the current telemetry data from the Omni, optionally parse it into a pydantic model.

pyomnilogic_local/api.py

@@ -1,24 +1,43 @@
 import asyncio
+from typing import Any
 
 import click
 
 from pyomnilogic_local.api import OmniLogicAPI
 from pyomnilogic_local.cli.debug import commands as debug
 from pyomnilogic_local.cli.get import commands as get
+from pyomnilogic_local.cli.utils import async_get_mspconfig, async_get_telemetry
 
 
 async def get_omni(host: str) -> OmniLogicAPI:
     return OmniLogicAPI(host, 10444, 5.0)
 
 
+async def fetch_startup_data(omni: OmniLogicAPI) -> tuple[Any, Any]:
+    """Fetch MSPConfig and Telemetry from the controller."""
+    try:
+        mspconfig = await async_get_mspconfig(omni)
+        telemetry = await async_get_telemetry(omni)
+    except Exception as exc:
+        raise RuntimeError(f"[ERROR] Failed to fetch config or telemetry from controller: {exc}") from exc
+    return mspconfig, telemetry
+
+
 @click.group()
 @click.pass_context
 @click.option("--host", default="127.0.0.1", help="Hostname or IP address of omnilogic system")
 def entrypoint(ctx: click.Context, host: str) -> None:
+    """Main CLI entrypoint for OmniLogic local control."""
     ctx.ensure_object(dict)
-    omni = asyncio.run(get_omni(host))
-
+    try:
+        omni = asyncio.run(get_omni(host))
+        mspconfig, telemetry = asyncio.run(fetch_startup_data(omni))
+    except Exception as exc:  # pylint: disable=broad-except
+        click.secho(str(exc), fg="red", err=True)
+        ctx.exit(1)
     ctx.obj["OMNI"] = omni
+    ctx.obj["MSPCONFIG"] = mspconfig
+    ctx.obj["TELEMETRY"] = telemetry
 
 
 entrypoint.add_command(debug.debug)

pyomnilogic_local/cli/cli.py

@@ -1,13 +1,22 @@
 # Need to figure out how to resolve the 'Untyped decorator makes function "..." untyped' errors in mypy when using click decorators
 # mypy: disable-error-code="misc"
 import asyncio
+import xml.etree.ElementTree as ET
+import zlib
+from collections import defaultdict
+from pathlib import Path
 from typing import Literal, overload
 
 import click
+from scapy.layers.inet import UDP
+from scapy.utils import rdpcap
 
 from pyomnilogic_local.api import OmniLogicAPI
 from pyomnilogic_local.cli.utils import async_get_mspconfig, async_get_telemetry
 from pyomnilogic_local.models.filter_diagnostics import FilterDiagnostics
+from pyomnilogic_local.models.leadmessage import LeadMessage
+from pyomnilogic_local.omnitypes import MessageType
+from pyomnilogic_local.protocol import OmniLogicMessage
 
 
 @click.group()
@@ -34,18 +43,6 @@ def get_telemetry(ctx: click.Context) -> None:
     click.echo(telemetry)
 
 
-@debug.command()
-@click.pass_context
-def get_alarm_list(ctx: click.Context) -> None:
-    alarm_list = asyncio.run(async_get_alarm_list(ctx.obj["OMNI"]))
-    click.echo(alarm_list)
-
-
-async def async_get_alarm_list(omni: OmniLogicAPI) -> str:
-    alarm_list = await omni.async_get_alarm_list()
-    return alarm_list
-
-
 @debug.command()
 @click.option("--pool-id", help="System ID of the Body Of Water the filter is associated with")
 @click.option("--filter-id", help="System ID of the filter to request diagnostics for")
@@ -84,12 +81,107 @@ async def async_get_filter_diagnostics(omni: OmniLogicAPI, pool_id: int, filter_
 
 
 @debug.command()
+@click.argument("pcap_file", type=click.Path(exists=True, path_type=Path))
 @click.pass_context
-def get_log_config(ctx: click.Context) -> None:
-    log_config = asyncio.run(async_get_log_config(ctx.obj["OMNI"]))
-    click.echo(log_config)
-
-
-async def async_get_log_config(omni: OmniLogicAPI) -> str:
-    log_config = await omni.async_get_log_config()
-    return log_config
+def parse_pcap(ctx: click.Context, pcap_file: Path) -> None:
+    """Parse a PCAP file and reconstruct Omnilogic protocol communication."""
+    # Read the PCAP file
+    try:
+        packets = rdpcap(str(pcap_file))
+    except Exception as e:
+        click.echo(f"Error reading PCAP file: {e}", err=True)
+        raise click.Abort()
+
+    # Track multi-message sequences (LeadMessage + BlockMessages)
+    # Key: (src_ip, dst_ip, msg_id), Value: list of messages
+    message_sequences: dict[tuple[str, str, int], list[OmniLogicMessage]] = defaultdict(list)
+
+    # Process packets in order
+    for packet in packets:
+        if not packet.haslayer(UDP):
+            click.echo("Not a UDP packet, skipping...", err=True)
+            continue
+
+        udp = packet[UDP]
+        src_ip = packet.payload.src
+        dst_ip = packet.payload.dst
+
+        # Parse the Omnilogic message
+        try:
+            omni_msg = OmniLogicMessage.from_bytes(bytes(udp.payload))
+            click.echo(f"Parsed Omnilogic message: {omni_msg}")
+        except Exception:  # pylint: disable=broad-except
+            # Not an Omnilogic message, skip it
+            click.echo("Not an Omnilogic message, skipping...", err=True)
+            continue
+
+        # Print the basic packet info
+        click.echo(f"{src_ip} sent {omni_msg.type.name} to {dst_ip}")
+
+        # Track LeadMessage/BlockMessage sequences
+        if omni_msg.type == MessageType.MSP_LEADMESSAGE:
+            # Start a new sequence
+            seq_key = (src_ip, dst_ip, omni_msg.id)
+            message_sequences[seq_key] = [omni_msg]
+        elif omni_msg.type == MessageType.MSP_BLOCKMESSAGE:
+            # Find the matching LeadMessage sequence
+            # We need to find the sequence with the same src/dst and highest ID less than or equal to this message
+            matching_seq: tuple[str, str, int] = ("", "", 0)
+            for seq_key in message_sequences:
+                if seq_key[0] == src_ip and seq_key[1] == dst_ip:
+                    # Check if this is the right sequence (the LeadMessage should have been received before this block)
+                    if not matching_seq or seq_key[2] > matching_seq[2]:
+                        matching_seq = seq_key
+
+            if matching_seq:
+                message_sequences[matching_seq].append(omni_msg)
+
+                # Check if we have all the blocks
+                lead_msg = message_sequences[matching_seq][0]
+                lead_data = LeadMessage.from_orm(ET.fromstring(lead_msg.payload[:-1]))
+
+                # We have LeadMessage + all BlockMessages
+                if len(message_sequences[matching_seq]) == lead_data.msg_block_count + 1:
+                    # Reassemble and decode
+                    try:
+                        decoded_msg = _reassemble_and_decode(message_sequences[matching_seq])
+                        click.echo(f"\nMessage from {src_ip} decoded:")
+                        click.echo(decoded_msg)
+                        click.echo()  # Extra newline for readability
+                    except Exception as e:  # pylint: disable=broad-except
+                        click.echo(f"Error decoding message: {e}", err=True)
+
+                    # Clean up this sequence
+                    del message_sequences[matching_seq]
+
+
+def _reassemble_and_decode(messages: list[OmniLogicMessage]) -> str:
+    """
+    Reassemble a LeadMessage + BlockMessages sequence and decode the payload.
+
+    Args:
+        messages: List containing LeadMessage followed by BlockMessages
+
+    Returns:
+        Decoded message content as string
+    """
+    lead_msg = messages[0]
+    block_msgs = messages[1:]
+
+    # Reassemble the blocks
+    # Sort by message ID to ensure correct order
+    sorted_blocks = sorted(block_msgs, key=lambda m: m.id)
+
+    # Concatenate the block payloads (skip the 8-byte header on each block)
+    reassembled = b""
+    for block_msg in sorted_blocks:
+        reassembled += block_msg.payload[8:]
+
+    # Decompress if necessary
+    if lead_msg.compressed:
+        reassembled = zlib.decompress(reassembled)
+
+    # Decode to string
+    decoded = reassembled.decode("utf-8").strip("\x00")
+
+    return decoded

pyomnilogic_local/cli/debug/commands.py

@@ -1,14 +1,18 @@
+from functools import cache
+
 from pyomnilogic_local.api import OmniLogicAPI
 from pyomnilogic_local.models.mspconfig import MSPConfig
 from pyomnilogic_local.models.telemetry import Telemetry
 
 
+@cache
 async def async_get_mspconfig(omni: OmniLogicAPI, raw: bool = False) -> MSPConfig:
     mspconfig: MSPConfig
     mspconfig = await omni.async_get_config(raw=raw)
     return mspconfig
 
 
+@cache
 async def async_get_telemetry(omni: OmniLogicAPI, raw: bool = False) -> Telemetry:
     telemetry: Telemetry
     telemetry = await omni.async_get_telemetry(raw=raw)