From 956009f0e78a597bdca76b7621a7780b0119c3b2 Mon Sep 17 00:00:00 2001 From: Tony Fruzza Date: Tue, 19 May 2026 05:45:42 +0000 Subject: [PATCH] feat(cli): add set command group for equipment control MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a new 'set' command group to the CLI that exposes equipment control operations that were previously only available through the Python API: - set heater-temp — Set heater target temp (°F) - set solar-temp — Set solar heater target (°F) - set speed — Set pump/filter speed (0-100%) - set on — Turn equipment on - set off — Turn equipment off This allows scripting and automation of pool control without writing Python code, complementing the existing read-only 'get' and 'debug' command groups. Resolves #145 --- pyomnilogic_local/cli/cli.py | 2 + pyomnilogic_local/cli/set/__init__.py | 0 pyomnilogic_local/cli/set/commands.py | 31 ++++++++++++ pyomnilogic_local/cli/set/equipment.py | 61 ++++++++++++++++++++++++ pyomnilogic_local/cli/set/heater_temp.py | 57 ++++++++++++++++++++++ pyomnilogic_local/cli/set/speed.py | 46 ++++++++++++++++++ 6 files changed, 197 insertions(+) create mode 100644 pyomnilogic_local/cli/set/__init__.py create mode 100644 pyomnilogic_local/cli/set/commands.py create mode 100644 pyomnilogic_local/cli/set/equipment.py create mode 100644 pyomnilogic_local/cli/set/heater_temp.py create mode 100644 pyomnilogic_local/cli/set/speed.py diff --git a/pyomnilogic_local/cli/cli.py b/pyomnilogic_local/cli/cli.py index ab05751..60f4954 100644 --- a/pyomnilogic_local/cli/cli.py +++ b/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) diff --git a/pyomnilogic_local/cli/set/__init__.py b/pyomnilogic_local/cli/set/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/pyomnilogic_local/cli/set/commands.py b/pyomnilogic_local/cli/set/commands.py new file mode 100644 index 0000000..0cc51e4 --- /dev/null +++ b/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) diff --git a/pyomnilogic_local/cli/set/equipment.py b/pyomnilogic_local/cli/set/equipment.py new file mode 100644 index 0000000..896fe5b --- /dev/null +++ b/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})") diff --git a/pyomnilogic_local/cli/set/heater_temp.py b/pyomnilogic_local/cli/set/heater_temp.py new file mode 100644 index 0000000..ab75a50 --- /dev/null +++ b/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") diff --git a/pyomnilogic_local/cli/set/speed.py b/pyomnilogic_local/cli/set/speed.py new file mode 100644 index 0000000..7d83272 --- /dev/null +++ b/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}%")