WIP: CLI

pyproject.toml

@@ -17,11 +17,16 @@ Homepage = "https://github.com/clssn/numato-gpio"
 Repository = "https://github.com/clssn/numato-gpio.git"
 
 [tool.poetry.scripts]
-numato-discover = "numato_gpio.__main__:main"
+numato = "numato_gpio.cli.cmd_numato:app"
+numato-discover = "numato_gpio.cli.cmd_discover:app"
 
 [tool.poetry.dependencies]
 python = "^3.8.1"
 pyserial = "^3.1"
+click = "^8.1.7"
+rich = "^13.7.1"
+typer = "^0.12.3"
+anyio = "^4.3.0"
 
 [tool.poetry.group.dev.dependencies]
 black = "^23.7.0"

src/numato_gpio/__main__.py

@@ -1,17 +1,19 @@
 """Main module printing detected numato devices on the command-line."""
-from numato_gpio import discover, cleanup, devices
 
+import asyncio
+from pathlib import Path
+from typing import List, Optional
 
-def main():
-    """Print out information about all discovered devices."""
-    try:
-        discover()
-        print(f"Discovered devices: {'(None)' if not devices else ''}")
-        for device in devices.values():
-            print(device)
-    finally:
-        cleanup()
+import rich
+import typer
+from numato_gpio.cli import cmd_numato
+from typing_extensions import Annotated
 
+import numato_gpio
+from numato_gpio.cli import cmd_numato
+
+main = cmd_numato.app
 
 if __name__ == "__main__":
     main()
+

src/numato_gpio/cli/cmd_discover.py

@@ -0,0 +1,55 @@
+import asyncio
+from pathlib import Path
+from typing import List, Optional
+import rich
+import typer
+from typing_extensions import Annotated
+import numato_gpio
+from numato_gpio.cli import utils
+
+
+async def list_devices() -> List[str]:
+    return [str(p) for p in Path("/dev").iterdir()]
+
+
+app = typer.Typer()
+
+
+@utils.asyncio_run
+async def discover(
+    verbose: Annotated[bool, typer.Option("--verbose", help="Verbose output")] = False,
+    devices: Annotated[
+        Optional[List[str]],
+        typer.Option(
+            "--device",
+            "-d",
+            metavar="DEVICE",
+            help="Device file path, option can be used multiple times",
+            show_default=True,
+            autocompletion=utils.async_completion(list_devices),
+        ),
+    ] = numato_gpio.DEFAULT_DEVICES,
+):
+    """Print information about all discovered Numato GPIO devices."""
+    try:
+        if verbose:
+            rich.print("[bold]Scanning devices:[/bold]\n")
+            rich.print("\n".join(dev for dev in devices))
+            rich.print()
+
+        await asyncio.get_event_loop().run_in_executor(None, numato_gpio.discover)
+
+        rich.print(
+            f"[bold]Discovered devices[/bold]: {'(None)' if not numato_gpio.devices else ''}"
+        )
+        rich.print()
+        for device in numato_gpio.devices.values():
+            rich.print(device)
+    finally:
+        await asyncio.get_event_loop().run_in_executor(None, numato_gpio.cleanup)
+
+
+app.command()(discover)
+
+if __name__ == "__main__":
+    app()

src/numato_gpio/cli/cmd_gpio.py

@@ -0,0 +1,101 @@
+import typer
+from typing_extensions import Annotated
+
+import numato_gpio
+from numato_gpio.cli import utils
+import click
+
+app = typer.Typer(
+    help="Operate device IO and ADC ports.",
+    no_args_is_help=True,
+)
+
+PORT_LOGIC_LEVEL_HIGH = 1
+PORT_LOGIC_LEVEL_LOW = 0
+
+
+@app.command(name="set")
+def set_(
+    device: Annotated[numato_gpio.NumatoUsbGpio, utils.numato_device_option],
+    port: Annotated[
+        int,
+        typer.Argument(
+            callback=utils.validate_port_range,
+            show_default=False,
+        ),
+    ],
+):
+    """Turn a port on (high)."""
+    try:
+        device.write(port, PORT_LOGIC_LEVEL_HIGH)
+    except numato_gpio.NumatoGpioError as err:
+        raise click.ClickException("Failed to write to the device.") from err
+
+
+@app.command()
+def clear(
+    device: Annotated[numato_gpio.NumatoUsbGpio, utils.numato_device_option],
+    port: Annotated[
+        int,
+        typer.Argument(
+            callback=utils.validate_port_range,
+            show_default=False,
+        ),
+    ],
+):
+    """Turn a port off (low)."""
+    try:
+        device.write(port, PORT_LOGIC_LEVEL_LOW)
+    except numato_gpio.NumatoGpioError as err:
+        raise click.ClickException(f"Failed to write to the device.") from err
+
+
+@app.command()
+def read(
+    device: Annotated[numato_gpio.NumatoUsbGpio, utils.numato_device_option],
+    port: int,
+):
+    """Read port level."""
+    try:
+        print(device.read(port))
+    except numato_gpio.NumatoGpioError as err:
+        raise click.ClickException("Failed to write to the device.") from err
+
+
+@app.command()
+def iomask():
+    """Configure the IO mask for all ports."""
+
+
+@app.command()
+def iodir():
+    """Configure the IO direction for all ports."""
+
+
+@app.command()
+def readall():
+    """Read all ports at once."""
+
+
+@app.command()
+def writeall():
+    """Write all ports at once."""
+
+
+app_notify = typer.Typer(help="Manage notifications.")
+app.add_typer(app_notify, name="notify")
+
+
+@app_notify.command()
+def on():
+    """Turn notifications on."""
+
+
+@app_notify.command()
+def off():
+    """Turn notifications off."""
+
+
+@app_notify.command()
+def get():
+    """Determine whether notifications are turned on or off."""

src/numato_gpio/cli/cmd_id.py

@@ -0,0 +1,38 @@
+import asyncio
+from pathlib import Path
+from typing import List, Optional
+
+import rich
+import typer
+from typing_extensions import Annotated
+
+import numato_gpio
+from numato_gpio.cli import utils
+
+app = typer.Typer(
+    help="Manage device IDs.",
+    no_args_is_help=True,
+)
+
+
+@app.command()
+@utils.asyncio_run
+async def get(
+        device: Annotated[numato_gpio.NumatoUsbGpio, utils.numato_device_option],
+):
+    """Print the device ID."""
+    rich.print(device.id)
+
+
+@app.command(name="set")
+@utils.asyncio_run
+async def set_(
+        device: Annotated[numato_gpio.NumatoUsbGpio, utils.numato_device_option],
+        id_: Annotated[int, typer.Argument(hidden=True, show_default=False, metavar="ID")],
+):
+    """Write an ID to the device."""
+    old_id = device.id
+    if old_id == id_:
+        rich.print(f"Warning: Re-writing same ID {id_} to device. ID will be unchanged.")
+    device.id=id_
+    rich.print(f"Device {old_id} now has ID {id_}")

src/numato_gpio/cli/cmd_numato.py

@@ -0,0 +1,14 @@
+import typer
+from numato_gpio.cli import cmd_gpio, cmd_id, cmd_ver, cmd_discover
+
+
+app = typer.Typer(no_args_is_help=True)
+
+app.command()(cmd_discover.discover)
+app.command()(cmd_ver.ver)
+
+app.add_typer(cmd_id.app, name="id")
+app.add_typer(cmd_gpio.app, name="gpio")
+
+if __name__ == "__main__":
+    app()