diff --git a/belay/cli/__init__.py b/belay/cli/__init__.py index 52cc189..e4c7aa5 100644 --- a/belay/cli/__init__.py +++ b/belay/cli/__init__.py @@ -1 +1 @@ -from .main import app, exec, identify, info, run, sync +from .main import app diff --git a/belay/cli/common.py b/belay/cli/common.py new file mode 100644 index 0000000..db42ce0 --- /dev/null +++ b/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." +) diff --git a/belay/cli/exec.py b/belay/cli/exec.py new file mode 100644 index 0000000..031a53c --- /dev/null +++ b/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) diff --git a/belay/cli/identify.py b/belay/cli/identify.py new file mode 100644 index 0000000..c04953d --- /dev/null +++ b/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) diff --git a/belay/cli/info.py b/belay/cli/info.py new file mode 100644 index 0000000..919cd82 --- /dev/null +++ b/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 diff --git a/belay/cli/main.py b/belay/cli/main.py index c4820f1..e56295d 100644 --- a/belay/cli/main.py +++ b/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) diff --git a/belay/cli/run.py b/belay/cli/run.py new file mode 100644 index 0000000..2c8983b --- /dev/null +++ b/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) diff --git a/belay/cli/sync.py b/belay/cli/sync.py new file mode 100644 index 0000000..88dedac --- /dev/null +++ b/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.") diff --git a/tests/cli/test_exec.py b/tests/cli/test_exec.py index 15c74c8..0d90b7d 100644 --- a/tests/cli/test_exec.py +++ b/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')") diff --git a/tests/cli/test_info.py b/tests/cli/test_info.py index 49c69ee..d65dd0d 100644 --- a/tests/cli/test_info.py +++ b/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" diff --git a/tests/cli/test_run.py b/tests/cli/test_run.py index 713022c..465cc49 100644 --- a/tests/cli/test_run.py +++ b/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)) diff --git a/tests/cli/test_sync.py b/tests/cli/test_sync.py index 25b7d3c..20ed1af 100644 --- a/tests/cli/test_sync.py +++ b/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(