feat(cli): add set command group for equipment control

pyomnilogic_local/cli/cli.py

@@ -7,6 +7,7 @@
 from pyomnilogic_local import OmniLogic
 from pyomnilogic_local.cli.debug import commands as debug
 from pyomnilogic_local.cli.get import commands as get
+from pyomnilogic_local.cli.set import commands as set_cmds
 
 
 @click.group()
@@ -55,3 +56,4 @@ def entrypoint(ctx: click.Context, host: str, port: int, timeout: int) -> None:
 
 entrypoint.add_command(debug.debug)
 entrypoint.add_command(get.get)
+entrypoint.add_command(set_cmds.set)

pyomnilogic_local/cli/set/commands.py

@@ -0,0 +1,31 @@
+# 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"
+
+from __future__ import annotations
+
+import click
+
+from pyomnilogic_local.cli.set.equipment import equipment_off, equipment_on
+from pyomnilogic_local.cli.set.heater_temp import heater_temp, solar_temp
+from pyomnilogic_local.cli.set.speed import speed
+
+
+@click.group()
+@click.pass_context
+def set(ctx: click.Context) -> None:
+    """Control pool equipment (turn on/off, set temperature, set speed).
+
+    These commands send control signals to pool equipment. They require
+    equipment system IDs which can be found using the 'get' commands.
+
+    Use with caution — these commands directly control physical equipment.
+    """
+    ctx.ensure_object(dict)
+
+
+# Register subcommands
+set.add_command(equipment_on)
+set.add_command(equipment_off)
+set.add_command(heater_temp)
+set.add_command(solar_temp)
+set.add_command(speed)

pyomnilogic_local/cli/set/equipment.py

@@ -0,0 +1,61 @@
+# mypy: disable-error-code="misc"
+
+from __future__ import annotations
+
+import asyncio
+from typing import TYPE_CHECKING
+
+import click
+
+if TYPE_CHECKING:
+    from pyomnilogic_local import OmniLogic
+
+
+@click.command("on")
+@click.argument("system_id", type=int)
+@click.pass_context
+def equipment_on(ctx: click.Context, system_id: int) -> None:
+    """Turn equipment on.
+
+    SYSTEM_ID is the equipment's system ID. Works with heaters, pumps, filters,
+    lights, and relays. Use the appropriate 'get' command to find system IDs.
+
+    Example:
+        omnilogic set on 5
+    """
+    omnilogic: OmniLogic = ctx.obj["OMNILOGIC"]
+
+    equipment = omnilogic.get_equipment_by_id(system_id)
+    if equipment is None:
+        raise click.ClickException(f"No equipment found with system_id {system_id}.")
+
+    if not hasattr(equipment, "turn_on"):
+        raise click.ClickException(f"Equipment '{equipment.name}' (system_id={system_id}) does not support turn_on.")
+
+    asyncio.run(equipment.turn_on())
+    click.echo(f"Turned on '{equipment.name}' (system_id={system_id})")
+
+
+@click.command("off")
+@click.argument("system_id", type=int)
+@click.pass_context
+def equipment_off(ctx: click.Context, system_id: int) -> None:
+    """Turn equipment off.
+
+    SYSTEM_ID is the equipment's system ID. Works with heaters, pumps, filters,
+    lights, and relays. Use the appropriate 'get' command to find system IDs.
+
+    Example:
+        omnilogic set off 5
+    """
+    omnilogic: OmniLogic = ctx.obj["OMNILOGIC"]
+
+    equipment = omnilogic.get_equipment_by_id(system_id)
+    if equipment is None:
+        raise click.ClickException(f"No equipment found with system_id {system_id}.")
+
+    if not hasattr(equipment, "turn_off"):
+        raise click.ClickException(f"Equipment '{equipment.name}' (system_id={system_id}) does not support turn_off.")
+
+    asyncio.run(equipment.turn_off())
+    click.echo(f"Turned off '{equipment.name}' (system_id={system_id})")

pyomnilogic_local/cli/set/heater_temp.py

@@ -0,0 +1,57 @@
+# mypy: disable-error-code="misc"
+
+from __future__ import annotations
+
+import asyncio
+from typing import TYPE_CHECKING
+
+import click
+
+if TYPE_CHECKING:
+    from pyomnilogic_local import OmniLogic
+
+
+@click.command("heater-temp")
+@click.argument("system_id", type=int)
+@click.argument("temperature", type=int)
+@click.pass_context
+def heater_temp(ctx: click.Context, system_id: int, temperature: int) -> None:
+    """Set heater target temperature (Fahrenheit).
+
+    SYSTEM_ID is the virtual heater's system ID (use 'get heaters' to find it).
+    TEMPERATURE is the target temperature in Fahrenheit.
+
+    Example:
+        omnilogic set heater-temp 4 82
+    """
+    omnilogic: OmniLogic = ctx.obj["OMNILOGIC"]
+
+    heater = omnilogic.all_heaters.get_by_id(system_id)
+    if heater is None:
+        raise click.ClickException(f"No heater found with system_id {system_id}. Use 'omnilogic get heaters' to list available heaters.")
+
+    asyncio.run(heater.set_temperature(temperature))
+    click.echo(f"Set heater '{heater.name}' (system_id={system_id}) to {temperature}°F")
+
+
+@click.command("solar-temp")
+@click.argument("system_id", type=int)
+@click.argument("temperature", type=int)
+@click.pass_context
+def solar_temp(ctx: click.Context, system_id: int, temperature: int) -> None:
+    """Set solar heater target temperature (Fahrenheit).
+
+    SYSTEM_ID is the virtual heater's system ID (use 'get heaters' to find it).
+    TEMPERATURE is the target solar temperature in Fahrenheit.
+
+    Example:
+        omnilogic set solar-temp 4 90
+    """
+    omnilogic: OmniLogic = ctx.obj["OMNILOGIC"]
+
+    heater = omnilogic.all_heaters.get_by_id(system_id)
+    if heater is None:
+        raise click.ClickException(f"No heater found with system_id {system_id}. Use 'omnilogic get heaters' to list available heaters.")
+
+    asyncio.run(heater.set_solar_temperature(temperature))
+    click.echo(f"Set solar temperature for '{heater.name}' (system_id={system_id}) to {temperature}°F")

pyomnilogic_local/cli/set/speed.py

@@ -0,0 +1,46 @@
+# mypy: disable-error-code="misc"
+
+from __future__ import annotations
+
+import asyncio
+from typing import TYPE_CHECKING
+
+import click
+
+from pyomnilogic_local.filter import Filter
+from pyomnilogic_local.pump import Pump
+
+if TYPE_CHECKING:
+    from pyomnilogic_local import OmniLogic
+
+
+@click.command("speed")
+@click.argument("system_id", type=int)
+@click.argument("percent", type=int)
+@click.pass_context
+def speed(ctx: click.Context, system_id: int, percent: int) -> None:
+    """Set pump or filter speed (0-100 percent).
+
+    SYSTEM_ID is the pump or filter's system ID (use 'get pumps' or 'get filters' to find it).
+    PERCENT is the speed percentage (0 will turn the pump off).
+
+    Example:
+        omnilogic set speed 3 75
+    """
+    omnilogic: OmniLogic = ctx.obj["OMNILOGIC"]
+
+    equipment = omnilogic.all_pumps.get_by_id(system_id)
+    if equipment is None:
+        equipment = omnilogic.all_filters.get_by_id(system_id)
+
+    if equipment is None:
+        raise click.ClickException(
+            f"No pump or filter found with system_id {system_id}. "
+            "Use 'omnilogic get pumps' or 'omnilogic get filters' to list available equipment."
+        )
+
+    if not isinstance(equipment, (Pump, Filter)):
+        raise click.ClickException(f"Equipment with system_id {system_id} is not a pump or filter.")
+
+    asyncio.run(equipment.set_speed(percent))
+    click.echo(f"Set '{equipment.name}' (system_id={system_id}) to {percent}%")