V1 CLI Architecture

V1 CLI Architecture#

This directory contains the modularized CLI implementation for Haniwers v1.

Structure#

cli/
├── __init__.py      # Package exports
├── main.py          # Entry point with global options
├── config.py        # Configuration management commands
├── port.py          # Port search command
├── daq.py           # Data acquisition command
├── threshold.py     # Threshold setting command
├── scan.py          # Threshold scanning command
├── fitter.py        # Threshold scanning command (planned)
├── mock.py          # Mocker management command
└── README.md        # This file

Adding New Commands#

Follow these established patterns to add new commands:

Pattern 1: Single Command Module#

For commands with a single action (like daq, scan):

1. Create the command module (cli/new_command.py):

"""New command implementation."""

import typer
from typer import Context
from pathlib import Path
from haniwers.v1.config.loader import ConfigLoader

def new_command(ctx: Context):
    """Brief description of what this command does.

    Longer description with details about the command's purpose
    and behavior.
    """
    config_path = ctx.obj.get("config_path")
    typer.echo(f"[v1] NEW_COMMAND start with config: {config_path}")

    # Load configuration
    try:
        loader = ConfigLoader(config_path)
        cfg = loader.config
    except Exception as e:
        msg = f"[ERROR] Failed to load config: {e}"
        typer.echo(msg, err=True)
        raise typer.Exit(code=1)

    # Command logic here
    typer.echo("[v1] NEW_COMMAND complete")

2. Register in main.py:

# Add import
from haniwers.v1.cli import config, daq, scan, new_command

# Add registration (after existing commands)
app.command()(new_command.new_command)

3. Create tests (tests/v1/unit/cli/new_command/test_new_command.py):

"""Unit tests for new_command.

What is new_command?
    Brief description of the command's purpose.

What are we testing?
    - Success cases: normal operation
    - Failure cases: error handling
    - Edge cases: boundary conditions

Coverage Target: 80%+
"""

import pytest
from typer.testing import CliRunner
from unittest.mock import patch, MagicMock
from haniwers.v1.cli.main import app

runner = CliRunner()


class TestNewCommand:
    """Test new_command success cases."""

    @patch("haniwers.v1.cli.new_command.ConfigLoader")
    def test_new_command_success(self, mock_loader, tmp_path, mock_config):
        """Test new_command executes successfully."""
        # Arrange
        mock_loader.return_value.config = mock_config
        config_path = tmp_path / "test.toml"

        # Act
        result = runner.invoke(app, ["new-command", "--config", str(config_path)])

        # Assert
        assert result.exit_code == 0
        assert "[v1] NEW_COMMAND start" in result.output
        mock_loader.assert_called_once_with(config_path)


class TestNewCommandFailures:
    """Test new_command failure cases."""

    def test_new_command_missing_config(self, tmp_path):
        """Test handling of missing configuration file."""
        # Arrange
        missing_file = tmp_path / "does_not_exist.toml"

        # Act
        result = runner.invoke(app, ["new-command", "--config", str(missing_file)])

        # Assert
        assert result.exit_code == 1
        assert "[ERROR]" in result.output


@pytest.fixture
def mock_config():
    """Create mock configuration for testing."""
    config = MagicMock()
    # Configure mock attributes as needed
    return config

Pattern 2: Multi-Command Module#

For commands with subcommands (like config show, config init):

1. Create the command module (cli/new_group.py):

"""New command group with subcommands."""

import typer

# Create Typer app for this command group
app = typer.Typer()


@app.command()
def subcommand1(ctx: typer.Context):
    """First subcommand description."""
    config_path = ctx.obj.get("config_path")
    typer.echo(f"[v1] Subcommand1 with config: {config_path}")


@app.command()
def subcommand2(ctx: typer.Context):
    """Second subcommand description."""
    config_path = ctx.obj.get("config_path")
    typer.echo(f"[v1] Subcommand2 with config: {config_path}")

2. Register in main.py:

# Add import
from haniwers.v1.cli import config, daq, scan, new_group

# Add registration
app.add_typer(new_group.app, name="new-group")

3. Create tests (one file per subcommand):

  • tests/v1/unit/cli/new_group/test_subcommand1.py

  • tests/v1/unit/cli/new_group/test_subcommand2.py

Testing Guidelines#

Test Organization#

  • One test file per function/command

  • Structure: Success Cases → Failure Cases → Edge Cases

  • Coverage target: 80%+ for all modules

Test Documentation#

Each test file must include:

"""Unit tests for command_name.

What is command_name?
    Brief description of what the command does.

What are we testing?
    - List of test categories
    - Key scenarios covered
    - Important edge cases

Coverage Target: 80%+
"""

Mocking Strategy#

  • Mock external dependencies: ConfigLoader, run_session, etc.

  • Mock time-dependent functions: datetime.now()

  • Use unittest.mock.patch for isolation

  • Create fixtures for common mock configurations

Running Tests#

# Run all CLI tests
poetry run pytest tests/v1/unit/cli/

# Run specific command tests
poetry run pytest tests/v1/unit/cli/new_command/

# Run with coverage
poetry run pytest tests/v1/unit/cli/ --cov=haniwers.v1.cli --cov-report=term

Code Quality Standards#

Formatting#

# Format code
poetry run ruff format src/haniwers/v1/cli/

# Check formatting
poetry run ruff format --check src/haniwers/v1/cli/

Linting#

# Run linter
poetry run ruff check src/haniwers/v1/cli/

# Fix auto-fixable issues
poetry run ruff check --fix src/haniwers/v1/cli/

Pre-commit Hooks#

# Run all pre-commit hooks
poetry run pre-commit run --all-files

Architecture Patterns#

Global Options (main.py)#

  • --env: Environment file path (default: .env.haniwers)

  • --config: Configuration file path (default: hnw.toml)

  • Stored in ctx.obj dictionary for command access

Configuration Loading#

All commands should:

  1. Get config path from ctx.obj

  2. Use ConfigLoader to load and validate configuration

  3. Handle loading errors with helpful messages

  4. Exit with code 1 on errors

Error Handling#

try:
    loader = ConfigLoader(config_path)
    cfg = loader.config
except Exception as e:
    msg = f"[ERROR] Failed to load config: {e}"
    typer.echo(msg, err=True)
    raise typer.Exit(code=1)

Output Conventions#

  • Success messages: typer.echo("[v1] MESSAGE")

  • Error messages: typer.echo("[ERROR] MESSAGE", err=True)

  • Progress updates: typer.echo("[v1] STEP description")

Command Options with ConfigLoader Precedence#

When a command has options that can come from both CLI arguments and configuration files, follow this simple rule:

CLI Options > ConfigLoader values > Hardcoded defaults

The easiest way: Load config from files, then override with CLI values in one place (the command function). This keeps precedence logic simple and clear.

The Simple Pattern#

For commands with config loading (like scan serial):

Step 1: Set CLI option defaults to None

def serial(
    channel: int = typer.Option(None, "--channel", min=1, max=3),
    center: int = typer.Option(None, "--center", min=1, max=1023),
    nsteps: int = typer.Option(None, "--nsteps", min=1),
    duration: int = typer.Option(None, "--duration", min=1),
    port: str = typer.Option(None, "--port"),
    config_path: Path = typer.Option(None, "--config"),
) -> None:

Why? None means “user didn’t specify this”. Non-None means “user specified this value”.

Step 2: Load config and override with CLI values

try:
    # Load from config file (or use defaults if no file)
    # ConfigLoader accepts "config_path=None"
    loader = ConfigLoader(config_path)
    cfg = loader.config

    # Override config values if user specified them
    # IMPORTANT: Use 'is not None' not truthiness (to handle 0 as valid)
    if nsteps is not None:
        cfg.scan.nsteps = nsteps
    if duration is not None:
        cfg.scan.duration = duration
    if port is not None:
        cfg.device.port = port

    # Now cfg has the final merged values
    # Run your scan logic with cfg

except FileNotFoundError as e:
    typer.echo(f"[ERROR] Config file not found: {e}", err=True)
    raise typer.Exit(code=1)
except Exception as e:
    typer.echo(f"[ERROR] {e}", err=True)
    raise typer.Exit(code=1)

Step 3: Use the normalized config

After overrides, cfg has all final values. Pass it to your scan functions without worrying about None values.

Key Rules#

  1. Always use is not None - Never use if nsteps: because 0 is valid

    if nsteps is not None:  # ✓ Correct
if nsteps:              # ✗ Wrong - fails for 0

  2. Override only specified channels - In multi-channel mode, only update channels the user specified

    # If user specified --centers "1:280;3:320", only update ch1 and ch3
# Leave ch2 at its config value

  3. One place for all precedence logic - All “CLI vs config” decisions in the command function, nowhere else

Testing Precedence Rules#

For comprehensive testing of CLI option precedence:

class TestSerialOptionPrecedence:
    """Test CLI > ConfigLoader > Hardcoded Default precedence."""

    @patch("haniwers.v1.cli.scan.ConfigLoader")
    def test_cli_option_overrides_config(self, mock_loader):
        """CLI option takes precedence over config file."""
        # Arrange
        mock_cfg = MagicMock()
        mock_cfg.scan.nsteps = 20  # Config has 20
        mock_loader.return_value.config = mock_cfg

        # Act: run with --nsteps 30
        result = runner.invoke(app, [
            "scan", "serial",
            "--channel", "1", "--center", "280",
            "--nsteps", "30"
        ])

        # Assert: should use 30 (CLI value)
        assert mock_cfg.scan.nsteps == 30

    @patch("haniwers.v1.cli.scan.ConfigLoader")
    def test_config_value_used_when_no_cli(self, mock_loader):
        """ConfigLoader value used when CLI option not specified."""
        # Arrange
        mock_cfg = MagicMock()
        mock_cfg.scan.nsteps = 20  # Config has 20
        mock_loader.return_value.config = mock_cfg

        # Act: run without --nsteps
        result = runner.invoke(app, [
            "scan", "serial",
            "--channel", "1", "--center", "280"
        ])

        # Assert: should still be 20 (config value, unchanged)
        assert mock_cfg.scan.nsteps == 20

    def test_hardcoded_default_when_no_cli_or_config(self):
        """Hardcoded default used when both CLI and config missing."""
        # If config file doesn't exist, ConfigLoader uses internal defaults
        # Verify that command still runs with hardcoded defaults as fallback
        pass

Configuration Modes#

When a command supports multiple modes, detect at the CLI layer:

  • Mode 1: Single-channel CLI: --channel 1 --center 280 --nsteps 10

  • Mode 2: Multi-channel CLI: --centers "1:280;2:300;3:320" --nsteps 10

  • Mode 3: Config-only: --config config.toml (no CLI options)

  • Mode 4: Config + Port override: --config config.toml --port /dev/ttyUSB0

All modes use the same precedence: CLI values override config values.

Parsing Multi-Value Options#

For options that accept multiple delimited values (like --centers "1:300;2:400;3:500"), create a reusable helper function to parse and validate:

Pattern: Delimiter-separated channel:value pairs

def parse_channel_options(option_str: str) -> dict[str, int]:
    """Parse 'channel:value' pairs separated by semicolons.

    Generic parser for any delimited channel option. Can be used for multiple
    CLI options with the same "channel:value;channel:value" format.

    Input format: "1:300;2:400;3:500"
    Output: {"ch1": 300, "ch2": 400, "ch3": 500}

    Raises:
        ValueError: If format is invalid (wrong delimiter or bad channel/value)
    """
    if not option_str:
        return {}

    pairs: dict[str, int] = {}
    for item in option_str.split(";"):
        try:
            channel_str, value_str = item.split(":")
            channel = int(channel_str)
            value = int(value_str)

            # Validate channel range (1-3 for OSECHI detector)
            if not (1 <= channel <= 3):
                raise ValueError(f"Channel {channel} out of range 1-3")

            pairs[f"ch{channel}"] = value
        except ValueError as e:
            raise ValueError(f"Invalid option format: {e}") from e

    return pairs

Usage in command (for centers):

# In serial() command function:
if centers is not None:
    try:
        channel_centers = parse_channel_options(centers)
        # Apply to config
        for ch_key, center_value in channel_centers.items():
            cfg.sensors[ch_key].center = center_value
    except ValueError as e:
        typer.echo(f"[ERROR] {e}", err=True)
        raise typer.Exit(code=1)

Key points:

  1. Generic parsing function - Reusable for any “channel:value” delimited option, not just centers

  2. Separate parsing from application - Parse first, validate, then apply to config

  3. Clear error messages - Tell user exactly what went wrong with their input

  4. Normalize output - Return a consistent data structure (dict with “ch1”, “ch2” keys)

  5. Validate ranges - Check channel bounds (1-3) during parsing, validate specific values in caller

Real-World Example: scan serial command#

See src/haniwers/v1/cli/scan.py for the simplified implementation:

  • ConfigLoader conditional loading (loads config if needed, handles errors)

  • Direct config override: CLI values override cfg attributes directly

  • Selective sensor updates (only specified channels)

  • All precedence logic in one place (the serial() function)

File Changes Required#

When adding a new command, you need to modify:

  1. Create: src/haniwers/v1/cli/new_command.py (command implementation)

  2. Modify: src/haniwers/v1/cli/main.py (2 lines: import + registration)

  3. Create: tests/v1/unit/cli/new_command/ (test directory)

  4. Create: tests/v1/unit/cli/new_command/test_*.py (test files)

That’s it! Just 2 lines changed in main.py for each new command.

Examples#

Minimal Single Command#

# cli/hello.py
def hello(ctx: typer.Context):
    """Say hello."""
    typer.echo("[v1] Hello, World!")

# main.py additions
from haniwers.v1.cli import hello
app.command()(hello.hello)

Command with Configuration#

# cli/process.py
def process(ctx: typer.Context):
    """Process data with configuration."""
    config_path = ctx.obj.get("config_path")
    loader = ConfigLoader(config_path)
    cfg = loader.config

    # Use configuration
    typer.echo(f"[v1] Processing with workspace: {cfg.daq.workspace}")

# main.py additions
from haniwers.v1.cli import process
app.command()(process.process)

Next Steps#

After implementing your command:

  1. Run tests: poetry run pytest tests/v1/unit/cli/your_command/

  2. Check coverage: Ensure 80%+ coverage

  3. Format code: poetry run ruff format

  4. Run linting: poetry run ruff check

  5. Run pre-commit: poetry run pre-commit run --all-files

  6. Update documentation if needed

  7. Create commit with conventional commit format

目次