Merge pull request #32 from BrianPugh/package-manager

break up cli into multiple modules

belay/cli/__init__.py

@@ -1 +1 @@
-from .main import app, exec, identify, info, run, sync
+from .main import app

belay/cli/common.py

@@ -0,0 +1,4 @@
+help_port = "Port (like /dev/ttyUSB0) or WebSocket (like ws://192.168.1.100) of device."
+help_password = (  # nosec
+    "Password for communication methods (like WebREPL) that require authentication."
+)

belay/cli/exec.py

@@ -0,0 +1,16 @@
+from pathlib import Path
+
+from typer import Argument, Option
+
+from belay import Device
+from belay.cli.common import help_password, help_port
+
+
+def exec(
+    port: str = Argument(..., help=help_port),
+    statement: str = Argument(..., help="Statement to execute on-device."),
+    password: str = Option("", help=help_password),
+):
+    """Execute python statement on-device."""
+    device = Device(port, password=password)
+    device(statement)

belay/cli/identify.py

@@ -0,0 +1,61 @@
+from pathlib import Path
+from time import sleep
+
+from typer import Argument, Option
+
+from belay import Device
+from belay.cli.common import help_password, help_port
+from belay.cli.info import info
+
+
+def identify(
+    port: str = Argument(..., help=help_port),
+    pin: int = Argument(..., help="GPIO pin to flash LED on."),
+    password: str = Option("", help=help_password),
+    neopixel: bool = Option(False, help="Indicator is a neopixel."),
+):
+    """Display device firmware information and blink an LED."""
+    device = info(port=port, password=password)
+
+    if device.implementation.name == "circuitpython":
+        device(f"led = digitalio.DigitalInOut(board.GP{pin})")
+        device("led.direction = digitalio.Direction.OUTPUT")
+
+        if neopixel:
+            device("from neopixel_write import neopixel_write")
+
+            @device.task
+            def set_led(state):
+                val = (255, 255, 255) if state else (0, 0, 0)
+                val = bytearray(val)
+                neopixel_write(led, val)  # noqa: F821
+
+        else:
+
+            @device.task
+            def set_led(state):
+                led.value = state  # noqa: F821
+
+    else:
+        device(f"led = Pin({pin}, Pin.OUT)")
+
+        if neopixel:
+            device("import neopixel")
+            device("pixel = neopixel.NeoPixel(led, 1)")
+
+            @device.task
+            def set_led(state):
+                pixel[0] = (255, 255, 255) if state else (0, 0, 0)  # noqa: F821
+                pixel.write()  # noqa: F821
+
+        else:
+
+            @device.task
+            def set_led(state):
+                led.value(state)  # noqa: F821
+
+    while True:
+        set_led(True)
+        sleep(0.5)
+        set_led(False)
+        sleep(0.5)

belay/cli/info.py

@@ -0,0 +1,17 @@
+from typer import Argument, Option
+
+from belay import Device
+from belay.cli.common import help_password, help_port
+
+
+def info(
+    port: str = Argument(..., help=help_port),
+    password: str = Option("", help=help_password),
+):
+    """Display device firmware information."""
+    device = Device(port, password=password)
+    version_str = "v" + ".".join(str(x) for x in device.implementation.version)
+    print(
+        f"{device.implementation.name} {version_str} - {device.implementation.platform}"
+    )
+    return device

belay/cli/main.py

@@ -1,164 +1,16 @@
-from functools import partial
-from pathlib import Path
-from time import sleep
-from typing import List, Optional, Union
-
 import typer
-from rich.console import Console
-from rich.progress import Progress
-from typer import Argument, Option
 
 from belay import Device
-
-Arg = partial(Argument, ..., show_default=False)
-Opt = partial(Option)
+from belay.cli.common import help_password, help_port
+from belay.cli.exec import exec
+from belay.cli.identify import identify
+from belay.cli.info import info
+from belay.cli.run import run
+from belay.cli.sync import sync
 
 app = typer.Typer()
-state = {}
-console: Console
-
-_help_port = (
-    "Port (like /dev/ttyUSB0) or WebSocket (like ws://192.168.1.100) of device."
-)
-_help_password = (  # nosec
-    "Password for communication methods (like WebREPL) that require authentication."
-)
-
-
-@app.callback()
-def callback(silent: bool = False):
-    """Tool to interact with MicroPython hardware."""
-    global console, Progress
-    state["silent"] = silent
-    console_kwargs = {}
-    if state["silent"]:
-        console_kwargs["quiet"] = True
-    console = Console(**console_kwargs)
-    Progress = partial(Progress, console=console)
-
-
-@app.command()
-def sync(
-    port: str = Arg(help=_help_port),
-    folder: Path = Arg(help="Path of local file or folder to sync."),
-    dst: str = Opt("/", help="Destination directory to unpack folder contents to."),
-    password: str = Opt("", help=_help_password),
-    keep: Optional[List[str]] = Opt(None, help="Files to keep."),
-    ignore: Optional[List[str]] = Opt(None, help="Files to ignore."),
-    mpy_cross_binary: Optional[Path] = Opt(
-        None, help="Compile py files with this executable."
-    ),
-):
-    """Synchronize a folder to device."""
-    # Typer issues: https://github.com/tiangolo/typer/issues/410
-    keep = keep if keep else None
-    ignore = ignore if ignore else None
-
-    with Progress() as progress:
-        task_id = progress.add_task("")
-        progress_update = partial(progress.update, task_id)
-        progress_update(description=f"Connecting to {port}")
-        device = Device(port, password=password)
-        progress_update(description=f"Connected to {port}.")
-
-        device.sync(
-            folder,
-            dst=dst,
-            keep=keep,
-            ignore=ignore,
-            mpy_cross_binary=mpy_cross_binary,
-            progress_update=progress_update,
-        )
-
-        progress_update(description="Sync complete.")
-
-
-@app.command()
-def run(
-    port: str = Arg(help=_help_port),
-    file: Path = Arg(help="File to run on-device."),
-    password: str = Opt("", help=_help_password),
-):
-    """Run file on-device."""
-    device = Device(port, password=password)
-    content = file.read_text()
-    device(content)
-
-
-@app.command()
-def exec(
-    port: str = Arg(help=_help_port),
-    statement: str = Arg(help="Statement to execute on-device."),
-    password: str = Opt("", help=_help_password),
-):
-    """Execute python statement on-device."""
-    device = Device(port, password=password)
-    device(statement)
-
-
-@app.command()
-def info(
-    port: str = Arg(help=_help_port),
-    password: str = Opt("", help=_help_password),
-):
-    """Display device firmware information."""
-    device = Device(port, password=password)
-    version_str = "v" + ".".join(str(x) for x in device.implementation.version)
-    print(
-        f"{device.implementation.name} {version_str} - {device.implementation.platform}"
-    )
-    return device
-
-
-@app.command()
-def identify(
-    port: str = Arg(help=_help_port),
-    pin: int = Arg(help="GPIO pin to flash LED on."),
-    password: str = Opt("", help=_help_password),
-    neopixel: bool = Option(False, help="Indicator is a neopixel."),
-):
-    """Display device firmware information and blink an LED."""
-    device = info(port=port, password=password)
-
-    if device.implementation.name == "circuitpython":
-        device(f"led = digitalio.DigitalInOut(board.GP{pin})")
-        device("led.direction = digitalio.Direction.OUTPUT")
-
-        if neopixel:
-            device("from neopixel_write import neopixel_write")
-
-            @device.task
-            def set_led(state):
-                val = (255, 255, 255) if state else (0, 0, 0)
-                val = bytearray(val)
-                neopixel_write(led, val)  # noqa: F821
-
-        else:
-
-            @device.task
-            def set_led(state):
-                led.value = state  # noqa: F821
-
-    else:
-        device(f"led = Pin({pin}, Pin.OUT)")
-
-        if neopixel:
-            device("import neopixel")
-            device("pixel = neopixel.NeoPixel(led, 1)")
-
-            @device.task
-            def set_led(state):
-                pixel[0] = (255, 255, 255) if state else (0, 0, 0)  # noqa: F821
-                pixel.write()  # noqa: F821
-
-        else:
-
-            @device.task
-            def set_led(state):
-                led.value(state)  # noqa: F821
-
-    while True:
-        set_led(True)
-        sleep(0.5)
-        set_led(False)
-        sleep(0.5)
+app.command()(sync)
+app.command()(run)
+app.command()(exec)
+app.command()(info)
+app.command()(identify)

belay/cli/run.py

@@ -0,0 +1,17 @@
+from pathlib import Path
+
+from typer import Argument, Option
+
+from belay import Device
+from belay.cli.common import help_password, help_port
+
+
+def run(
+    port: str = Argument(..., help=help_port),
+    file: Path = Argument(..., help="File to run on-device."),
+    password: str = Option("", help=help_password),
+):
+    """Run file on-device."""
+    device = Device(port, password=password)
+    content = file.read_text()
+    device(content)

belay/cli/sync.py

@@ -0,0 +1,44 @@
+from functools import partial
+from pathlib import Path
+from typing import List, Optional
+
+from rich.progress import Progress
+from typer import Argument, Option
+
+from belay import Device
+from belay.cli.common import help_password, help_port
+
+
+def sync(
+    port: str = Argument(..., help=help_port),
+    folder: Path = Argument(..., help="Path of local file or folder to sync."),
+    dst: str = Option("/", help="Destination directory to unpack folder contents to."),
+    password: str = Option("", help=help_password),
+    keep: Optional[List[str]] = Option(None, help="Files to keep."),
+    ignore: Optional[List[str]] = Option(None, help="Files to ignore."),
+    mpy_cross_binary: Optional[Path] = Option(
+        None, help="Compile py files with this executable."
+    ),
+):
+    """Synchronize a folder to device."""
+    # Typer issues: https://github.com/tiangolo/typer/issues/410
+    keep = keep if keep else None
+    ignore = ignore if ignore else None
+
+    with Progress() as progress:
+        task_id = progress.add_task("")
+        progress_update = partial(progress.update, task_id)
+        progress_update(description=f"Connecting to {port}")
+        device = Device(port, password=password)
+        progress_update(description=f"Connected to {port}.")
+
+        device.sync(
+            folder,
+            dst=dst,
+            keep=keep,
+            ignore=ignore,
+            mpy_cross_binary=mpy_cross_binary,
+            progress_update=progress_update,
+        )
+
+        progress_update(description="Sync complete.")

tests/cli/test_exec.py

@@ -1,5 +1,5 @@
 def test_exec_basic(mocker, mock_device, cli_runner):
-    mock_device.patch("belay.cli.main.Device")
+    mock_device.patch("belay.cli.exec.Device")
     result = cli_runner("exec", "print('hello world')")
     assert result.exit_code == 0
     mock_device.inst.assert_called_once_with("print('hello world')")

tests/cli/test_info.py

@@ -1,5 +1,5 @@
 def test_info_basic(mocker, mock_device, cli_runner, tmp_path):
-    mock_device.patch("belay.cli.main.Device")
+    mock_device.patch("belay.cli.info.Device")
     mock_device.inst.implementation.name = "testingpython"
     mock_device.inst.implementation.version = (4, 7, 9)
     mock_device.inst.implementation.platform = "pytest"

tests/cli/test_run.py

@@ -1,5 +1,5 @@
 def test_run_basic(mocker, mock_device, cli_runner, tmp_path):
-    mock_device.patch("belay.cli.main.Device")
+    mock_device.patch("belay.cli.run.Device")
     py_file = tmp_path / "foo.py"
     py_file.write_text("print('hello')\nprint('world')")
     result = cli_runner("run", str(py_file))

tests/cli/test_sync.py

@@ -2,7 +2,7 @@
 
 
 def test_sync_basic(mocker, mock_device, cli_runner):
-    mock_device.patch("belay.cli.main.Device")
+    mock_device.patch("belay.cli.sync.Device")
     result = cli_runner("sync", "foo")
     assert result.exit_code == 0
     mock_device.inst.sync.assert_called_once_with(