cli

CLI module for gwframe.

StageGroup

Bases: TyperGroup

Command group that runs a chain of stage subcommands in one invocation.

Two things keep this from being a stock group. A variadic positional (the input paths) cannot coexist with subcommands, since it would swallow the stage tokens, so :meth:parse_args splits the argument list at the first token naming a stage: everything before it is the group's own arguments and options (in any order), everything from it on is the stage chain. And Click's chain mode is gone from Typer 0.27, which ships its own copy of Click instead of depending on the package, so :meth:invoke parses and runs the stages itself, then hands the resulting :class:Stage objects to :func:_run_transform.

Only Typer's public surface is used; the click package may not be installed. Contexts are left unannotated because their class differs between the two Typer generations.

call_operation

call_operation(operation_func, input_files: list[Path], output_dir: Path | None, in_place: bool, **operation_kwargs) -> list[str]

Call an operation function, handling single-file output case.

Parameters:

Name	Type	Description	Default
operation_func	callable	Operation function from operations module	required
input_files	list[Path]	Input files to process	required
output_dir	Path or None	Output directory or file path	required
in_place	bool	Whether to modify in place	required
**operation_kwargs		Additional keyword arguments for operation_func	{}

Returns:

Name	Type	Description
output_files	list[str]	List of output file paths

Source code in gwframe/cli.py

def call_operation(
    operation_func,
    input_files: list[Path],
    output_dir: Path | None,
    in_place: bool,
    **operation_kwargs,
) -> list[str]:
    """
    Call an operation function, handling single-file output case.

    Parameters
    ----------
    operation_func : callable
        Operation function from operations module
    input_files : list[Path]
        Input files to process
    output_dir : Path or None
        Output directory or file path
    in_place : bool
        Whether to modify in place
    **operation_kwargs
        Additional keyword arguments for operation_func

    Returns
    -------
    output_files : list[str]
        List of output file paths
    """
    # Handle single-file output case
    if not in_place and output_dir is not None and output_dir.suffix == ".gwf":
        if len(input_files) != 1:
            console.print(
                "[red]Error: Single file output requires single file input[/red]"
            )
            raise typer.Exit(1)

        # Use temporary directory for operation to avoid filename collisions
        with tempfile.TemporaryDirectory() as temp_dir_name:
            temp_dir = Path(temp_dir_name)

            output_files = operation_func(
                [str(f) for f in input_files],
                str(temp_dir),
                **operation_kwargs,
            )

            # Move to final location
            temp_output = Path(output_files[0])
            final_output = output_dir
            final_output.parent.mkdir(parents=True, exist_ok=True)
            shutil.move(str(temp_output), str(final_output))

            return [str(final_output)]

    # Standard operation (directory output or in-place)
    return operation_func(
        [str(f) for f in input_files],
        str(output_dir) if output_dir else None,
        in_place=in_place,
        **operation_kwargs,
    )

combine

combine(input_sources: list[Path] = Argument(..., help='N source files or N source directories to combine (N >= 2)', exists=True), output_dir: Path = Option(..., '--output-dir', '-o', help='Output directory for combined files'), keep: list[str] | None = Option(None, '--keep', '-k', help='Only include these channels (can be specified multiple times)'), drop: list[str] | None = Option(None, '--drop', '-d', help='Exclude these channels (can be specified multiple times)'))

Combine channels from N sources covering the same time ranges.

Takes N files (covering the same time) or N directories (with matching frame sets) and merges their channels. All sources must be the same type.

Examples: gwframe combine file1.gwf file2.gwf -o output/ gwframe combine dir1/ dir2/ -o output/ --keep L1:STRAIN --keep L1:LSC gwframe combine dir1/ dir2/ -o output/ --drop L1:UNWANTED

Source code in gwframe/cli.py

@app.command()
def combine(
    input_sources: list[Path] = typer.Argument(
        ...,
        help="N source files or N source directories to combine (N >= 2)",
        exists=True,
    ),
    output_dir: Path = typer.Option(
        ...,
        "--output-dir",
        "-o",
        help="Output directory for combined files",
    ),
    keep: list[str] | None = typer.Option(
        None,
        "--keep",
        "-k",
        help="Only include these channels (can be specified multiple times)",
    ),
    drop: list[str] | None = typer.Option(
        None,
        "--drop",
        "-d",
        help="Exclude these channels (can be specified multiple times)",
    ),
):
    """
    Combine channels from N sources covering the same time ranges.

    Takes N files (covering the same time) or N directories (with matching
    frame sets) and merges their channels. All sources must be the same type.

    Examples:
        gwframe combine file1.gwf file2.gwf -o output/
        gwframe combine dir1/ dir2/ -o output/ --keep L1:STRAIN --keep L1:LSC
        gwframe combine dir1/ dir2/ -o output/ --drop L1:UNWANTED
    """
    if len(input_sources) < 2:
        console.print("[red]Error: combine requires at least 2 sources[/red]")
        raise typer.Exit(1)

    # Check that keep and drop are mutually exclusive
    if keep is not None and drop is not None:
        console.print("[red]Error: --keep and --drop are mutually exclusive[/red]")
        raise typer.Exit(1)

    # Check that all sources are the same type
    are_files = [p.is_file() for p in input_sources]
    are_dirs = [p.is_dir() for p in input_sources]

    if not (all(are_files) or all(are_dirs)):
        console.print(
            "[red]Error: All sources must be same type (files or directories)[/red]"
        )
        raise typer.Exit(1)

    source_type = "files" if all(are_files) else "directories"

    # Build status message
    status_msg = f"Combining channels from {len(input_sources)} {source_type}"
    if keep:
        status_msg += f" (keeping {len(keep)} channel(s))"
    elif drop:
        status_msg += f" (dropping {len(drop)} channel(s))"
    console.print(f"[cyan]{status_msg}...[/cyan]")

    try:
        with frame_progress("Combining") as progress:
            output_files = operations.combine_channels(
                [str(s) for s in input_sources],
                str(output_dir),
                keep_channels=keep,
                drop_channels=drop,
                progress=progress,
            )
        console.print(
            f"[green]Wrote {len(output_files)} combined file(s) to {output_dir}[/green]"
        )
    except (ValueError, OSError) as e:
        console.print(f"[red]Error: {e}[/red]")
        raise typer.Exit(1) from e

drop

drop(input_paths: list[Path] = Argument(..., help='Input GWF file(s) or directory/directories to process', exists=True), output_dir: Path | None = Option(None, '--output-dir', '-o', help='Output directory or file for processed files'), channels: list[str] = Option(..., '--channel', '-c', help='Channel(s) to drop (can be specified multiple times)'), in_place: bool = Option(False, '--in-place', '-i', help='Modify files in place instead of creating new ones'), recursive: bool = Option(False, '--recursive', '-r', help='Recurse into subdirectories when processing directories'))

Remove specified channels from frame files.

Accepts files or directories.

Examples: gwframe drop input.gwf -o output.gwf -c L1:UNWANTED_CHANNEL gwframe drop input.gwf --in-place -c L1:UNWANTED_CHANNEL gwframe drop data/ -o output/ -c L1:CHAN1 -c L1:CHAN2

Source code in gwframe/cli.py

@app.command()
def drop(
    input_paths: list[Path] = typer.Argument(
        ...,
        help="Input GWF file(s) or directory/directories to process",
        exists=True,
    ),
    output_dir: Path | None = typer.Option(
        None,
        "--output-dir",
        "-o",
        help="Output directory or file for processed files",
    ),
    channels: list[str] = typer.Option(
        ...,
        "--channel",
        "-c",
        help="Channel(s) to drop (can be specified multiple times)",
    ),
    in_place: bool = typer.Option(
        False,
        "--in-place",
        "-i",
        help="Modify files in place instead of creating new ones",
    ),
    recursive: bool = typer.Option(
        False,
        "--recursive",
        "-r",
        help="Recurse into subdirectories when processing directories",
    ),
):
    """
    Remove specified channels from frame files.

    Accepts files or directories.

    Examples:
        gwframe drop input.gwf -o output.gwf -c L1:UNWANTED_CHANNEL
        gwframe drop input.gwf --in-place -c L1:UNWANTED_CHANNEL
        gwframe drop data/ -o output/ -c L1:CHAN1 -c L1:CHAN2
    """
    # Expand paths to files
    input_files = expand_paths(input_paths, recursive=recursive)

    if not input_files:
        console.print("[red]Error: No files found matching criteria[/red]")
        raise typer.Exit(1)

    # Validate output options
    validate_output_options(output_dir, in_place)

    console.print(
        f"[cyan]Dropping {len(channels)} channel(s) in "
        f"{len(input_files)} file(s)...[/cyan]"
    )

    try:
        with frame_progress("Dropping") as progress:
            output_files = call_operation(
                operations.drop_channels,
                input_files,
                output_dir,
                in_place,
                channels_to_drop=channels,
                progress=progress,
            )
        if in_place:
            console.print(
                f"[green]Modified {len(output_files)} file(s) in place[/green]"
            )
        else:
            console.print(
                f"[green]Wrote {len(output_files)} file(s) to {output_dir}[/green]"
            )
    except (ValueError, OSError) as e:
        console.print(f"[red]Error: {e}[/red]")
        raise typer.Exit(1) from e

expand_paths

expand_paths(paths: list[Path], recursive: bool = False) -> list[Path]

Expand paths to list of files, handling both files and directories.

Directories are searched for *.gwf files.

Parameters:

Name	Type	Description	Default
paths	list[Path]	List of file or directory paths	required
recursive	bool	If True, recurse into subdirectories (default: False)	False

Returns:

Name	Type	Description
files	list[Path]	Expanded list of file paths

Source code in gwframe/cli.py

def expand_paths(
    paths: list[Path],
    recursive: bool = False,
) -> list[Path]:
    """
    Expand paths to list of files, handling both files and directories.

    Directories are searched for *.gwf files.

    Parameters
    ----------
    paths : list[Path]
        List of file or directory paths
    recursive : bool, optional
        If True, recurse into subdirectories (default: False)

    Returns
    -------
    files : list[Path]
        Expanded list of file paths
    """
    files = []
    for path in paths:
        if path.is_file():
            files.append(path)
        elif path.is_dir():
            if recursive:
                files.extend(sorted(path.rglob("*.gwf")))
            else:
                files.extend(sorted(path.glob("*.gwf")))
        else:
            console.print(f"[yellow]Warning: Skipping invalid path: {path}[/yellow]")
    return files

frame_progress

frame_progress(label: str)

Show a per-frame progress bar, yielding the callback the operations take.

The bar is disabled when stdout is not a terminal, so scripted and piped runs get the plain summary lines only. The total is supplied by the first callback, so the bar starts indeterminate.

Source code in gwframe/cli.py

@contextmanager
def frame_progress(label: str):
    """
    Show a per-frame progress bar, yielding the callback the operations take.

    The bar is disabled when stdout is not a terminal, so scripted and
    piped runs get the plain summary lines only. The total is supplied by
    the first callback, so the bar starts indeterminate.
    """
    bar = Progress(
        SpinnerColumn(),
        TextColumn("[progress.description]{task.description}"),
        BarColumn(),
        MofNCompleteColumn(),
        TextColumn("frames"),
        TimeElapsedColumn(),
        TimeRemainingColumn(),
        console=console,
        transient=True,
        disable=not console.is_terminal,
    )
    with bar:
        task = bar.add_task(label, total=None)

        def update(done: int, total: int, path: str) -> None:
            bar.update(task, completed=done, total=total)

        yield update

impute

impute(input_paths: list[Path] = Argument(..., help='Input GWF file(s) or directory/directories to process', exists=True), output_dir: Path | None = Option(None, '--output-dir', '-o', help='Output directory or file for processed files'), replace_value: float = Option(float('nan'), '--replace-value', '-r', help='Value to replace (default: NaN)'), fill_value: float = Option(0.0, '--fill-value', '-f', help='Value to use for replacement (will be cast to appropriate dtype)'), channels: list[str] | None = Option(None, '--channel', '-c', help='Channel(s) to impute (can be specified multiple times)'), in_place: bool = Option(False, '--in-place', '-i', help='Modify files in place instead of creating new ones'), recursive: bool = Option(False, '--recursive', help='Recurse into subdirectories when processing directories'))

Replace specific values in frame file channels with a fill value.

Accepts files or directories.

Examples: gwframe impute input.gwf -o output.gwf gwframe impute input.gwf --in-place --fill-value 0.0 --channel L1:STRAIN gwframe impute data/ -o output/ --replace-value -999.0 --fill-value 0.0

Source code in gwframe/cli.py

@app.command()
def impute(
    input_paths: list[Path] = typer.Argument(
        ...,
        help="Input GWF file(s) or directory/directories to process",
        exists=True,
    ),
    output_dir: Path | None = typer.Option(
        None,
        "--output-dir",
        "-o",
        help="Output directory or file for processed files",
    ),
    replace_value: float = typer.Option(
        float("nan"),
        "--replace-value",
        "-r",
        help="Value to replace (default: NaN)",
    ),
    fill_value: float = typer.Option(
        0.0,
        "--fill-value",
        "-f",
        help="Value to use for replacement (will be cast to appropriate dtype)",
    ),
    channels: list[str] | None = typer.Option(
        None,
        "--channel",
        "-c",
        help="Channel(s) to impute (can be specified multiple times)",
    ),
    in_place: bool = typer.Option(
        False,
        "--in-place",
        "-i",
        help="Modify files in place instead of creating new ones",
    ),
    recursive: bool = typer.Option(
        False,
        "--recursive",
        help="Recurse into subdirectories when processing directories",
    ),
):
    """
    Replace specific values in frame file channels with a fill value.

    Accepts files or directories.

    Examples:
        gwframe impute input.gwf -o output.gwf
        gwframe impute input.gwf --in-place --fill-value 0.0 --channel L1:STRAIN
        gwframe impute data/ -o output/ --replace-value -999.0 --fill-value 0.0
    """
    # Expand paths to files
    input_files = expand_paths(input_paths, recursive=recursive)

    if not input_files:
        console.print("[red]Error: No files found matching criteria[/red]")
        raise typer.Exit(1)

    # Validate output options
    validate_output_options(output_dir, in_place)

    # Build status message
    replace_str = (
        "NaN" if replace_value != replace_value else str(replace_value)
    )  # NaN != NaN
    status_msg = f"Replacing {replace_str} with {fill_value}"
    if channels:
        status_msg += f" in {len(channels)} channel(s)"
    console.print(f"[cyan]{status_msg}...[/cyan]")

    try:
        with frame_progress("Imputing") as progress:
            output_files = call_operation(
                operations.impute_missing_data,
                input_files,
                output_dir,
                in_place,
                replace_value=replace_value,
                fill_value=fill_value,
                channels=channels,
                progress=progress,
            )
        if in_place:
            console.print(
                f"[green]Modified {len(output_files)} file(s) in place[/green]"
            )
        else:
            console.print(
                f"[green]Wrote {len(output_files)} file(s) to {output_dir}[/green]"
            )
    except (ValueError, OSError) as e:
        console.print(f"[red]Error: {e}[/red]")
        raise typer.Exit(1) from e

inspect

inspect(input_path: Path = Argument(..., help='GWF file to inspect', exists=True), verbose: int = Option(0, '--verbose', '-v', count=True, help='Increase verbosity (-v channels, -vv frames, -vvv per-channel detail, -vvvv invalid-channel report, -vvvvv data preview which reads all data)'))

Show metadata and channel information for a GWF file.

Verbosity levels:

(default)  File summary: GPS range, frame/channel counts, compression

-v         Add channel listing with types (adc/proc/sim)

-vv        Add per-frame table

-vvv       Add per-channel detail (sample rate, dtype, units, samples,
           validity)

-vvvv      Add invalid-channel report (scans dataValid across all
           frames)

-vvvvv     Add per-frame data preview (reads and decompresses ALL
           data; may be slow for large files)

Examples: gwframe inspect data.gwf gwframe inspect -v data.gwf gwframe inspect -vvv data.gwf gwframe inspect -vvvvv data.gwf

Source code in gwframe/cli.py

@app.command()
def inspect(
    input_path: Path = typer.Argument(
        ...,
        help="GWF file to inspect",
        exists=True,
    ),
    verbose: int = typer.Option(
        0,
        "--verbose",
        "-v",
        count=True,
        help="Increase verbosity (-v channels, -vv frames, -vvv per-channel "
        "detail, -vvvv invalid-channel report, -vvvvv data preview which "
        "reads all data)",
    ),
):
    """
    Show metadata and channel information for a GWF file.

    Verbosity levels:

        (default)  File summary: GPS range, frame/channel counts, compression

        -v         Add channel listing with types (adc/proc/sim)

        -vv        Add per-frame table

        -vvv       Add per-channel detail (sample rate, dtype, units, samples,
                   validity)

        -vvvv      Add invalid-channel report (scans dataValid across all
                   frames)

        -vvvvv     Add per-frame data preview (reads and decompresses ALL
                   data; may be slow for large files)

    Examples:
        gwframe inspect data.gwf
        gwframe inspect -v data.gwf
        gwframe inspect -vvv data.gwf
        gwframe inspect -vvvvv data.gwf
    """
    try:
        info = get_info(str(input_path))
        channels_by_type = get_channels_by_type(str(input_path))
    except (OSError, RuntimeError) as e:
        console.print(f"[red]Error: {e}[/red]")
        raise typer.Exit(1) from e

    # Compression name
    try:
        comp_name = Compression(info.compression).name
    except ValueError:
        comp_name = str(info.compression)

    # GPS range
    if info.frames:
        first = info.frames[0]
        last = info.frames[-1]
        gps_start = first.start
        gps_end = last.start + last.duration
        total_duration = gps_end - gps_start
        frame_duration = first.duration
    else:
        gps_start = gps_end = total_duration = frame_duration = 0.0

    # Type counts
    type_parts = []
    for ch_type in ("adc", "proc", "sim"):
        count = len(channels_by_type[ch_type])
        if count:
            type_parts.append(f"{count} {ch_type}")
    type_summary = ", ".join(type_parts) if type_parts else "none"

    # --- Level 0: File summary ---
    console.print()

    title = Text(input_path.name, style="bold")
    title.append(f"  ({_format_file_size(input_path)})", style="dim")
    console.print(title)

    spec_str = f"v{info.frame_spec}" if info.frame_spec is not None else "unknown"
    console.print(
        f"  [cyan]Frames:[/cyan] {info.num_frames:_} "
        f"x {frame_duration:g}s    "
        f"[cyan]Channels:[/cyan] {len(info.channels):_} ({type_summary})    "
        f"[cyan]Compression:[/cyan] {comp_name}    "
        f"[cyan]Frame spec:[/cyan] {spec_str}"
    )

    if info.num_frames > 0:
        console.print(
            f"  [cyan]GPS:[/cyan] [{_format_gps(gps_start)}, "
            f"{_format_gps(gps_end)})    "
            f"[cyan]Duration:[/cyan] {total_duration:g}s"
        )

    if verbose < 1:
        console.print()
        return

    # --- Level 1-2: Simple channel listing (replaced by detail table at level 3) ---
    if verbose < 3:
        console.print()
        console.rule("[bold]Channels[/bold]", style="dim")

        channel_table = Table(box=box.SIMPLE_HEAD, show_edge=False, pad_edge=False)
        channel_table.add_column("Channel", no_wrap=True)
        channel_table.add_column("Type", style="green")

        for ch_type in ("adc", "proc", "sim"):
            for ch in channels_by_type[ch_type]:
                channel_table.add_row(ch, ch_type)

        console.print(channel_table)

    # --- Level 3: Per-channel detail (replaces simple listing) ---
    invalid_channels: dict[str, dict[int, int]] = {}
    if verbose >= 3:
        console.print()
        console.rule("[bold]Channels[/bold]", style="dim")

        try:
            details = get_channel_details(str(input_path))
            # All-frames dataValid scan (header reads only); also drives
            # the -vvvv invalid-channel report
            invalid_channels = get_invalid_channels(str(input_path))
        except (OSError, RuntimeError) as e:
            console.print(f"[red]Error reading channel details: {e}[/red]")
            raise typer.Exit(1) from e

        detail_table = Table(box=box.SIMPLE_HEAD, show_edge=False, pad_edge=False)
        detail_table.add_column("Channel", no_wrap=True)
        detail_table.add_column("Type", style="green")
        detail_table.add_column("Dtype", justify="right")
        detail_table.add_column("Rate (Hz)", justify="right")
        detail_table.add_column("Samples", justify="right")
        detail_table.add_column("Unit", no_wrap=True, overflow="ellipsis", max_width=12)
        detail_table.add_column("Valid", justify="center")

        for ch_info in details:
            # Proc/sim channels carry no dataValid flag: always valid
            if ch_info.type == "adc" and ch_info.name in invalid_channels:
                valid_cell = "no"
                row_style = "red"
            else:
                valid_cell = "yes"
                row_style = None
            detail_table.add_row(
                ch_info.name,
                ch_info.type,
                ch_info.dtype_name,
                f"{ch_info.sample_rate:g}",
                f"{ch_info.n_samples:_}",
                ch_info.unit or "-",
                valid_cell,
                style=row_style,
            )

        console.print(detail_table)

    if verbose < 2:
        console.print()
        return

    # --- Level 2+: Per-frame table ---
    console.print()
    console.rule("[bold]Frames[/bold]", style="dim")

    frame_table = Table(box=box.SIMPLE_HEAD, show_edge=False, pad_edge=False)
    frame_table.add_column("#", justify="right", style="dim")
    frame_table.add_column("Name")
    frame_table.add_column("GPS Start", justify="right")
    frame_table.add_column("Duration (s)", justify="right")
    frame_table.add_column("Run", justify="right")
    frame_table.add_column("Frame #", justify="right")

    # Show all frames if <= 30, otherwise show first/last 5
    frames = info.frames
    show_all = len(frames) <= 30
    frames_to_show = frames if show_all else frames[:5] + frames[-5:]

    for i, frame in enumerate(frames_to_show):
        if not show_all and i == 5:
            frame_table.add_row(
                "...",
                f"({len(frames) - 10} more)",
                "",
                "",
                "",
                "",
                style="dim",
            )

        frame_table.add_row(
            str(frame.index),
            frame.name,
            _format_gps(frame.start),
            f"{frame.duration:g}",
            str(frame.run),
            str(frame.frame_number),
        )

    console.print(frame_table)

    # --- Level 4: Invalid-channel report (all-frames dataValid scan) ---
    if verbose >= 4:
        console.print()
        console.rule("[bold]Invalid channels[/bold]", style="dim")

        if not channels_by_type["adc"]:
            console.print(
                "  [dim]No ADC channels (proc and sim channels carry no "
                "dataValid flag and are always valid)[/dim]"
            )
        elif not invalid_channels:
            console.print("  [green]All ADC channels valid in all frames[/green]")
        else:
            invalid_table = Table(box=box.SIMPLE_HEAD, show_edge=False, pad_edge=False)
            invalid_table.add_column("Channel", no_wrap=True, style="red")
            invalid_table.add_column("Frames", justify="right")
            invalid_table.add_column("dataValid", justify="right")

            for ch_name, frame_flags in invalid_channels.items():
                values = sorted(set(frame_flags.values()))
                invalid_table.add_row(
                    ch_name,
                    _format_frame_indices(sorted(frame_flags), info.num_frames),
                    ", ".join(f"0x{value:04x}" for value in values),
                )

            console.print(invalid_table)

    # --- Level 5: Per-frame data preview ---
    if verbose >= 5:
        console.print()
        console.rule("[bold]Data preview[/bold]", style="dim")

        try:
            with FrameReader(str(input_path)) as reader:
                indices = (
                    list(range(len(frames)))
                    if show_all
                    else [*range(5), *range(len(frames) - 5, len(frames))]
                )
                for position, frame_index in enumerate(indices):
                    if not show_all and position == 5:
                        console.print(
                            f"  [dim]... ({len(frames) - 10} frames omitted) ...[/dim]"
                        )
                    frame = frames[frame_index]
                    console.print()
                    console.print(
                        f"[bold]Frame {frame_index}[/bold]  "
                        f"[dim]{frame.name}  "
                        f"GPS [{_format_gps(frame.start)}, "
                        f"{_format_gps(frame.start + frame.duration)})[/dim]"
                    )

                    data = reader.read(
                        None, frame_index=frame_index, allow_invalid=True
                    )
                    if not data:
                        console.print("  [dim](no channels)[/dim]")
                        continue

                    preview_table = Table(
                        box=box.SIMPLE_HEAD, show_edge=False, pad_edge=False
                    )
                    preview_table.add_column("Channel", no_wrap=True)
                    preview_table.add_column("Type", style="green")
                    preview_table.add_column("Dtype", justify="right")
                    preview_table.add_column("Rate (Hz)", justify="right")
                    preview_table.add_column("Samples", justify="right")
                    preview_table.add_column("Valid", justify="center")
                    preview_table.add_column("Data", overflow="fold")

                    for ch_type in ("adc", "proc", "sim"):
                        for name in channels_by_type[ch_type]:
                            ts = data.get(name)
                            if ts is None:
                                continue
                            masked = ts.mask is not None and ts.mask.any()
                            valid_cell = "no" if masked else "yes"
                            preview_table.add_row(
                                name,
                                ch_type,
                                ts.dtype.name,
                                f"{ts.sample_rate:g}",
                                f"{len(ts.array):_}",
                                valid_cell,
                                escape(_format_array_preview(ts.array)),
                                style="red" if masked else None,
                            )

                    console.print(preview_table)
        except (OSError, RuntimeError) as e:
            console.print(f"[red]Error reading data: {e}[/red]")
            raise typer.Exit(1) from e

    console.print()

main

main()

Main entry point for the CLI.

Source code in gwframe/cli.py

def main():
    """Main entry point for the CLI."""
    app()

parse_channel_map

parse_channel_map(items: list[str]) -> dict[str, str]

Parse OLD=>NEW mappings, exiting with an error on a malformed entry.

Source code in gwframe/cli.py

def parse_channel_map(items: list[str]) -> dict[str, str]:
    """Parse OLD=>NEW mappings, exiting with an error on a malformed entry."""
    mapping = {}
    for item in items:
        if "=>" not in item:
            console.print(
                f"[red]Error: Invalid mapping format '{item}'. Expected OLD=>NEW[/red]"
            )
            raise typer.Exit(1)
        old, new = item.split("=>", 1)
        mapping[old.strip()] = new.strip()
    return mapping

parse_compression

parse_compression(name: str) -> Compression

Convert a compression name to the enum, exiting with an error if unknown.

Source code in gwframe/cli.py

def parse_compression(name: str) -> Compression:
    """Convert a compression name to the enum, exiting with an error if unknown."""
    try:
        return Compression[name.upper()]
    except KeyError:
        console.print(
            f"[red]Error: Invalid compression type '{name}'. "
            f"Valid options: {', '.join(c.name for c in Compression)}[/red]"
        )
        raise typer.Exit(1) from None

recompress

recompress(input_paths: list[Path] = Argument(..., help='Input GWF file(s) or directory/directories to process', exists=True), output_dir: Path | None = Option(None, '--output-dir', '-o', help='Output directory or file for processed files'), compression: str = Option('ZERO_SUPPRESS_OTHERWISE_GZIP', '--compression', '-c', case_sensitive=False, help='Compression type (e.g., RAW, GZIP, DIFF_GZIP)'), level: int = Option(6, '--level', '-l', help='Compression level (0-9)', min=0, max=9), in_place: bool = Option(False, '--in-place', '-i', help='Modify files in place instead of creating new ones'), recursive: bool = Option(False, '--recursive', '-r', help='Recurse into subdirectories when processing directories'))

Rewrite frame files with different compression settings.

Accepts files or directories.

Examples: gwframe recompress input.gwf -o output.gwf -c GZIP -l 9 gwframe recompress input.gwf --in-place -c GZIP -l 9 gwframe recompress data/ -o output/ -c RAW

Source code in gwframe/cli.py

@app.command()
def recompress(
    input_paths: list[Path] = typer.Argument(
        ...,
        help="Input GWF file(s) or directory/directories to process",
        exists=True,
    ),
    output_dir: Path | None = typer.Option(
        None,
        "--output-dir",
        "-o",
        help="Output directory or file for processed files",
    ),
    compression: str = typer.Option(
        "ZERO_SUPPRESS_OTHERWISE_GZIP",
        "--compression",
        "-c",
        case_sensitive=False,
        help="Compression type (e.g., RAW, GZIP, DIFF_GZIP)",
    ),
    level: int = typer.Option(
        6,
        "--level",
        "-l",
        help="Compression level (0-9)",
        min=0,
        max=9,
    ),
    in_place: bool = typer.Option(
        False,
        "--in-place",
        "-i",
        help="Modify files in place instead of creating new ones",
    ),
    recursive: bool = typer.Option(
        False,
        "--recursive",
        "-r",
        help="Recurse into subdirectories when processing directories",
    ),
):
    """
    Rewrite frame files with different compression settings.

    Accepts files or directories.

    Examples:
        gwframe recompress input.gwf -o output.gwf -c GZIP -l 9
        gwframe recompress input.gwf --in-place -c GZIP -l 9
        gwframe recompress data/ -o output/ -c RAW
    """
    # Expand paths to files
    input_files = expand_paths(input_paths, recursive=recursive)

    if not input_files:
        console.print("[red]Error: No files found matching criteria[/red]")
        raise typer.Exit(1)

    # Validate output options
    validate_output_options(output_dir, in_place)

    compression_enum = parse_compression(compression)

    console.print(
        f"[cyan]Recompressing {len(input_files)} file(s) with "
        f"{compression_enum.name} (level {level})...[/cyan]"
    )

    try:
        with frame_progress("Recompressing") as progress:
            output_files = call_operation(
                operations.recompress_frames,
                input_files,
                output_dir,
                in_place,
                compression=compression_enum,
                compression_level=level,
                progress=progress,
            )
        if in_place:
            console.print(
                f"[green]Modified {len(output_files)} file(s) in place[/green]"
            )
        else:
            console.print(
                f"[green]Wrote {len(output_files)} file(s) to {output_dir}[/green]"
            )
    except (ValueError, OSError) as e:
        console.print(f"[red]Error: {e}[/red]")
        raise typer.Exit(1) from e

rename

rename(input_paths: list[Path] = Argument(..., help='Input GWF file(s) or directory/directories to process', exists=True), output_dir: Path | None = Option(None, '--output-dir', '-o', help='Output directory or file for processed files'), channel_map: list[str] = Option(..., '--map', '-m', help='Channel mapping in format OLD=>NEW (can be specified multiple times)'), in_place: bool = Option(False, '--in-place', '-i', help='Modify files in place instead of creating new ones'), recursive: bool = Option(False, '--recursive', '-r', help='Recurse into subdirectories when processing directories'))

Rename channels in frame files.

Accepts files or directories.

Examples: gwframe rename input.gwf -o output.gwf -m "L1:OLD_CHAN=>L1:NEW_CHAN" gwframe rename input.gwf -o output/ -m "L1:OLD_CHAN=>L1:NEW_CHAN" gwframe rename input.gwf --in-place -m "L1:OLD_CHAN=>L1:NEW_CHAN" gwframe rename data/ -o output/ -m "L1:OLD_CHAN=>L1:NEW_CHAN" gwframe rename data/*.gwf -o output/ -m "L1:CHAN1=>L1:NEW1" -m "L1:CHAN2=>L1:NEW2"

Source code in gwframe/cli.py

@app.command()
def rename(
    input_paths: list[Path] = typer.Argument(
        ...,
        help="Input GWF file(s) or directory/directories to process",
        exists=True,
    ),
    output_dir: Path | None = typer.Option(
        None,
        "--output-dir",
        "-o",
        help="Output directory or file for processed files",
    ),
    channel_map: list[str] = typer.Option(
        ...,
        "--map",
        "-m",
        help="Channel mapping in format OLD=>NEW (can be specified multiple times)",
    ),
    in_place: bool = typer.Option(
        False,
        "--in-place",
        "-i",
        help="Modify files in place instead of creating new ones",
    ),
    recursive: bool = typer.Option(
        False,
        "--recursive",
        "-r",
        help="Recurse into subdirectories when processing directories",
    ),
):
    """
    Rename channels in frame files.

    Accepts files or directories.

    Examples:
        gwframe rename input.gwf -o output.gwf -m "L1:OLD_CHAN=>L1:NEW_CHAN"
        gwframe rename input.gwf -o output/ -m "L1:OLD_CHAN=>L1:NEW_CHAN"
        gwframe rename input.gwf --in-place -m "L1:OLD_CHAN=>L1:NEW_CHAN"
        gwframe rename data/ -o output/ -m "L1:OLD_CHAN=>L1:NEW_CHAN"
        gwframe rename data/*.gwf -o output/ -m "L1:CHAN1=>L1:NEW1" -m "L1:CHAN2=>L1:NEW2"
    """  # noqa: E501
    # Expand paths to files
    input_files = expand_paths(input_paths, recursive=recursive)

    if not input_files:
        console.print("[red]Error: No files found matching criteria[/red]")
        raise typer.Exit(1)

    mapping = parse_channel_map(channel_map)

    # Validate output options
    validate_output_options(output_dir, in_place)

    console.print(f"[cyan]Renaming channels in {len(input_files)} file(s)...[/cyan]")

    try:
        with frame_progress("Renaming") as progress:
            output_files = call_operation(
                operations.rename_channels,
                input_files,
                output_dir,
                in_place,
                channel_map=mapping,
                progress=progress,
            )

        if in_place:
            console.print(
                f"[green]Modified {len(output_files)} file(s) in place[/green]"
            )
        else:
            console.print(
                f"[green]Wrote {len(output_files)} file(s) to {output_dir}[/green]"
            )
    except (ValueError, OSError) as e:
        console.print(f"[red]Error: {e}[/red]")
        raise typer.Exit(1) from e

replace

replace(base_paths: list[Path] = Argument(..., help='Base GWF file(s) or directory/directories', exists=True), update_paths: list[Path] = Option(..., '--update', '-u', help='GWF file(s) or directory/directories containing updated channel data', exists=True), output_dir: Path = Option(..., '--output-dir', '-o', help='Output directory for processed files'), channels: list[str] | None = Option(None, '--channel', '-c', help='Channel(s) to replace (if not specified, replaces all)'), recursive: bool = Option(False, '--recursive', '-r', help='Recurse into subdirectories when processing directories'))

Replace data in channels with updated versions from another frame file.

Accepts files or directories.

Examples: gwframe replace base.gwf --update updated.gwf -o output/ -c L1:STRAIN gwframe replace base_dir/ --update update_dir/ -o output/ gwframe replace data/.gwf --update updates/.gwf -o output/ --recursive

Source code in gwframe/cli.py

@app.command()
def replace(
    base_paths: list[Path] = typer.Argument(
        ...,
        help="Base GWF file(s) or directory/directories",
        exists=True,
    ),
    update_paths: list[Path] = typer.Option(
        ...,
        "--update",
        "-u",
        help="GWF file(s) or directory/directories containing updated channel data",
        exists=True,
    ),
    output_dir: Path = typer.Option(
        ...,
        "--output-dir",
        "-o",
        help="Output directory for processed files",
    ),
    channels: list[str] | None = typer.Option(
        None,
        "--channel",
        "-c",
        help="Channel(s) to replace (if not specified, replaces all)",
    ),
    recursive: bool = typer.Option(
        False,
        "--recursive",
        "-r",
        help="Recurse into subdirectories when processing directories",
    ),
):
    """
    Replace data in channels with updated versions from another frame file.

    Accepts files or directories.

    Examples:
        gwframe replace base.gwf --update updated.gwf -o output/ -c L1:STRAIN
        gwframe replace base_dir/ --update update_dir/ -o output/
        gwframe replace data/*.gwf --update updates/*.gwf -o output/ --recursive
    """
    # Expand paths to files
    base_files = expand_paths(base_paths, recursive=recursive)
    update_files = expand_paths(update_paths, recursive=recursive)

    if not base_files:
        console.print("[red]Error: No base files found matching criteria[/red]")
        raise typer.Exit(1)

    if not update_files:
        console.print("[red]Error: No update files found matching criteria[/red]")
        raise typer.Exit(1)

    console.print(f"[cyan]Replacing channels in {len(base_files)} file(s)...[/cyan]")

    try:
        with frame_progress("Replacing") as progress:
            output_files = operations.replace_channels(
                [str(f) for f in base_files],
                [str(f) for f in update_files],
                str(output_dir),
                channels,
                progress=progress,
            )
        console.print(
            f"[green]Wrote {len(output_files)} file(s) to {output_dir}[/green]"
        )
    except (ValueError, OSError) as e:
        console.print(f"[red]Error: {e}[/red]")
        raise typer.Exit(1) from e

resize

resize(input_paths: list[Path] = Argument(..., help='Input GWF file(s) or directory/directories to process', exists=True), output_dir: Path | None = Option(None, '--output-dir', '-o', help='Output directory or file for processed files'), duration: float = Option(..., '--duration', '-d', help='Target frame duration in seconds'), in_place: bool = Option(False, '--in-place', '-i', help='Modify files in place instead of creating new ones'), recursive: bool = Option(False, '--recursive', '-r', help='Recurse into subdirectories when processing directories'))

Resize frames to a different duration (e.g., 64s to 4s).

Accepts files or directories.

Examples: gwframe resize input.gwf -o output.gwf -d 4.0 gwframe resize input.gwf --in-place -d 4.0 gwframe resize data/ -o output/ -d 4.0

Source code in gwframe/cli.py

@app.command()
def resize(
    input_paths: list[Path] = typer.Argument(
        ...,
        help="Input GWF file(s) or directory/directories to process",
        exists=True,
    ),
    output_dir: Path | None = typer.Option(
        None,
        "--output-dir",
        "-o",
        help="Output directory or file for processed files",
    ),
    duration: float = typer.Option(
        ...,
        "--duration",
        "-d",
        help="Target frame duration in seconds",
    ),
    in_place: bool = typer.Option(
        False,
        "--in-place",
        "-i",
        help="Modify files in place instead of creating new ones",
    ),
    recursive: bool = typer.Option(
        False,
        "--recursive",
        "-r",
        help="Recurse into subdirectories when processing directories",
    ),
):
    """
    Resize frames to a different duration (e.g., 64s to 4s).

    Accepts files or directories.

    Examples:
        gwframe resize input.gwf -o output.gwf -d 4.0
        gwframe resize input.gwf --in-place -d 4.0
        gwframe resize data/ -o output/ -d 4.0
    """
    # Expand paths to files
    input_files = expand_paths(input_paths, recursive=recursive)

    if not input_files:
        console.print("[red]Error: No files found matching criteria[/red]")
        raise typer.Exit(1)

    # Validate output options
    validate_output_options(output_dir, in_place)

    console.print(f"[cyan]Resizing frames to {duration}s duration...[/cyan]")

    try:
        with frame_progress("Resizing") as progress:
            output_files = call_operation(
                operations.resize_frames,
                input_files,
                output_dir,
                in_place,
                target_duration=duration,
                progress=progress,
            )
        if in_place:
            console.print(
                f"[green]Modified {len(output_files)} file(s) in place[/green]"
            )
        else:
            console.print(
                f"[green]Wrote {len(output_files)} file(s) to {output_dir}[/green]"
            )
    except (ValueError, OSError) as e:
        console.print(f"[red]Error: {e}[/red]")
        raise typer.Exit(1) from e

select

select(input_paths: list[Path] = Argument(..., help='Input GWF file(s) or directory/directories to process', exists=True), output_dir: Path | None = Option(None, '--output-dir', '-o', help='Output directory or file for processed files'), channels: list[str] = Option(..., '--channel', '-c', help='Channel(s) to keep (can be specified multiple times)'), in_place: bool = Option(False, '--in-place', '-i', help='Modify files in place instead of creating new ones'), recursive: bool = Option(False, '--recursive', '-r', help='Recurse into subdirectories when processing directories'))

Keep only specified channels in frame files, removing all others.

Accepts files or directories.

Examples: gwframe select input.gwf -o output.gwf -c L1:STRAIN gwframe select input.gwf --in-place -c L1:STRAIN gwframe select data/ -o output/ -c L1:CHAN1 -c L1:CHAN2

Source code in gwframe/cli.py

@app.command()
def select(
    input_paths: list[Path] = typer.Argument(
        ...,
        help="Input GWF file(s) or directory/directories to process",
        exists=True,
    ),
    output_dir: Path | None = typer.Option(
        None,
        "--output-dir",
        "-o",
        help="Output directory or file for processed files",
    ),
    channels: list[str] = typer.Option(
        ...,
        "--channel",
        "-c",
        help="Channel(s) to keep (can be specified multiple times)",
    ),
    in_place: bool = typer.Option(
        False,
        "--in-place",
        "-i",
        help="Modify files in place instead of creating new ones",
    ),
    recursive: bool = typer.Option(
        False,
        "--recursive",
        "-r",
        help="Recurse into subdirectories when processing directories",
    ),
):
    """
    Keep only specified channels in frame files, removing all others.

    Accepts files or directories.

    Examples:
        gwframe select input.gwf -o output.gwf -c L1:STRAIN
        gwframe select input.gwf --in-place -c L1:STRAIN
        gwframe select data/ -o output/ -c L1:CHAN1 -c L1:CHAN2
    """
    # Expand paths to files
    input_files = expand_paths(input_paths, recursive=recursive)

    if not input_files:
        console.print("[red]Error: No files found matching criteria[/red]")
        raise typer.Exit(1)

    # Validate output options
    validate_output_options(output_dir, in_place)

    console.print(
        f"[cyan]Selecting {len(channels)} channel(s) from "
        f"{len(input_files)} file(s)...[/cyan]"
    )

    try:
        with frame_progress("Selecting") as progress:
            output_files = call_operation(
                operations.select_channels,
                input_files,
                output_dir,
                in_place,
                channels_to_select=channels,
                progress=progress,
            )
        if in_place:
            console.print(
                f"[green]Modified {len(output_files)} file(s) in place[/green]"
            )
        else:
            console.print(
                f"[green]Wrote {len(output_files)} file(s) to {output_dir}[/green]"
            )
    except (ValueError, OSError) as e:
        console.print(f"[red]Error: {e}[/red]")
        raise typer.Exit(1) from e

transform

transform(input_paths: list[Path] = Argument(..., help='Input GWF file(s) or directory/directories to process', exists=True), output_dir: Path = Option(..., '--output-dir', '-o', help='Output directory for processed files'), recursive: bool = Option(False, '--recursive', '-r', help='Recurse into subdirectories when processing directories'), compression: str = Option('ZERO_SUPPRESS_OTHERWISE_GZIP', '--compression', case_sensitive=False, help='Compression type for the output (e.g., RAW, GZIP, DIFF_GZIP)'), level: int = Option(6, '--level', help='Compression level (0-9)', min=0, max=9), file_duration: float | None = Option(None, '--file-duration', help='Re-chunk the output into files of this many seconds, reading all inputs in GPS order as one stream; a gap in time starts a new file. Output files are named PREFIX-GPS-DURATION.gwf'), prefix: str | None = Option(None, '--prefix', help='OBSERVATORY-DESCRIPTION prefix for re-chunked file names (default: taken from the first input file)'), dry_run: bool = Option(False, '--dry-run', help='Show the plan and preflight warnings without writing anything'), strict: bool = Option(False, '--strict', help='Treat preflight warnings (a channel a stage names but the input lacks) as errors: exit 1 and write nothing'))

Apply a chain of stages to frame files in a single read/write pass.

Input paths and the options above come first, followed by any number of stages, each with its own options. Stages run left to right and take the same options as the standalone commands of the same name (minus input, output and in-place). With no stages the files are simply rewritten, which recompresses or re-chunks them.

A stage chain can also be kept in a recipe file, one or more stages per line with # comments, and spliced in with @FILE. Use --strict when running a recipe you rely on, so a typo in a channel name fails the run instead of producing a dataset that differs from the recipe.

Examples: gwframe transform raw/ -o curated/ rename -m "L1:OLD=>L1:NEW" drop -c L1:JUNK gwframe transform raw/ -o curated/ --compression GZIP --level 9 resize -d 4 gwframe transform raw/ -o archive/ --file-duration 4096 --prefix L-L1_CURATED gwframe transform raw/ -o curated/ --dry-run @curated.recipe

Source code in gwframe/cli.py

@transform_app.callback()
def transform(
    input_paths: list[Path] = typer.Argument(
        ...,
        help="Input GWF file(s) or directory/directories to process",
        exists=True,
    ),
    output_dir: Path = typer.Option(
        ...,
        "--output-dir",
        "-o",
        help="Output directory for processed files",
    ),
    recursive: bool = typer.Option(
        False,
        "--recursive",
        "-r",
        help="Recurse into subdirectories when processing directories",
    ),
    compression: str = typer.Option(
        "ZERO_SUPPRESS_OTHERWISE_GZIP",
        "--compression",
        case_sensitive=False,
        help="Compression type for the output (e.g., RAW, GZIP, DIFF_GZIP)",
    ),
    level: int = typer.Option(
        6,
        "--level",
        help="Compression level (0-9)",
        min=0,
        max=9,
    ),
    file_duration: float | None = typer.Option(
        None,
        "--file-duration",
        help="Re-chunk the output into files of this many seconds, reading "
        "all inputs in GPS order as one stream; a gap in time starts a "
        "new file. Output files are named PREFIX-GPS-DURATION.gwf",
    ),
    prefix: str | None = typer.Option(
        None,
        "--prefix",
        help="OBSERVATORY-DESCRIPTION prefix for re-chunked file names "
        "(default: taken from the first input file)",
    ),
    dry_run: bool = typer.Option(
        False,
        "--dry-run",
        help="Show the plan and preflight warnings without writing anything",
    ),
    strict: bool = typer.Option(
        False,
        "--strict",
        help="Treat preflight warnings (a channel a stage names but the input "
        "lacks) as errors: exit 1 and write nothing",
    ),
):
    """
    Apply a chain of stages to frame files in a single read/write pass.

    Input paths and the options above come first, followed by any number of
    stages, each with its own options. Stages run left to right and take the
    same options as the standalone commands of the same name (minus input,
    output and in-place). With no stages the files are simply rewritten,
    which recompresses or re-chunks them.

    A stage chain can also be kept in a recipe file, one or more stages per
    line with # comments, and spliced in with @FILE. Use --strict when
    running a recipe you rely on, so a typo in a channel name fails the run
    instead of producing a dataset that differs from the recipe.

    Examples:
        gwframe transform raw/ -o curated/ rename -m "L1:OLD=>L1:NEW" drop -c L1:JUNK
        gwframe transform raw/ -o curated/ --compression GZIP --level 9 resize -d 4
        gwframe transform raw/ -o archive/ --file-duration 4096 --prefix L-L1_CURATED
        gwframe transform raw/ -o curated/ --dry-run @curated.recipe
    """

transform_drop

transform_drop(channels: list[str] = Option(..., '--channel', '-c', help='Channel(s) to drop (can be specified multiple times)')) -> Stage

Remove specified channels.

Source code in gwframe/cli.py

@transform_app.command("drop")
def transform_drop(
    channels: list[str] = typer.Option(
        ...,
        "--channel",
        "-c",
        help="Channel(s) to drop (can be specified multiple times)",
    ),
) -> Stage:
    """Remove specified channels."""
    return DropStage(channels)

transform_fill_gaps

transform_fill_gaps(fill_value: float = Option(0.0, '--fill-value', '-f', help="Sample value for the inserted frames (cast to each channel's dtype)"), invalid: bool = Option(False, '--invalid', help='Flag the inserted data as invalid: sets the dataValid flag on ADC channels (proc/sim channels have no such flag and are left unflagged)')) -> Stage

Fill gaps between consecutive frames with constant-valued frames.

Makes the stream contiguous, like GStreamer's audiorate element: frames with the previous frame's channels and duration are inserted to cover each gap. Nothing is inserted before the first frame; overlapping frames are an error. With --invalid, consumers can tell the inserted data from real data on ADC channels.

Source code in gwframe/cli.py

@transform_app.command("fill-gaps")
def transform_fill_gaps(
    fill_value: float = typer.Option(
        0.0,
        "--fill-value",
        "-f",
        help="Sample value for the inserted frames (cast to each channel's dtype)",
    ),
    invalid: bool = typer.Option(
        False,
        "--invalid",
        help="Flag the inserted data as invalid: sets the dataValid flag on ADC "
        "channels (proc/sim channels have no such flag and are left unflagged)",
    ),
) -> Stage:
    """
    Fill gaps between consecutive frames with constant-valued frames.

    Makes the stream contiguous, like GStreamer's audiorate element: frames
    with the previous frame's channels and duration are inserted to cover
    each gap. Nothing is inserted before the first frame; overlapping frames
    are an error. With --invalid, consumers can tell the inserted data from
    real data on ADC channels.
    """
    return FillGapsStage(fill_value, invalid=invalid)

transform_impute

transform_impute(replace_value: float = Option(float('nan'), '--replace-value', '-r', help='Value to replace (default: NaN)'), fill_value: float = Option(0.0, '--fill-value', '-f', help='Value to use for replacement (will be cast to appropriate dtype)'), channels: list[str] | None = Option(None, '--channel', '-c', help='Channel(s) to impute (can be specified multiple times)')) -> Stage

Replace specific values in channel data with a fill value.

Source code in gwframe/cli.py

@transform_app.command("impute")
def transform_impute(
    replace_value: float = typer.Option(
        float("nan"),
        "--replace-value",
        "-r",
        help="Value to replace (default: NaN)",
    ),
    fill_value: float = typer.Option(
        0.0,
        "--fill-value",
        "-f",
        help="Value to use for replacement (will be cast to appropriate dtype)",
    ),
    channels: list[str] | None = typer.Option(
        None,
        "--channel",
        "-c",
        help="Channel(s) to impute (can be specified multiple times)",
    ),
) -> Stage:
    """Replace specific values in channel data with a fill value."""
    return ImputeStage(replace_value, fill_value, channels)

transform_rename

transform_rename(channel_map: list[str] = Option(..., '--map', '-m', help='Channel mapping in format OLD=>NEW (can be specified multiple times)')) -> Stage

Rename channels.

Source code in gwframe/cli.py

@transform_app.command("rename")
def transform_rename(
    channel_map: list[str] = typer.Option(
        ...,
        "--map",
        "-m",
        help="Channel mapping in format OLD=>NEW (can be specified multiple times)",
    ),
) -> Stage:
    """Rename channels."""
    return RenameStage(parse_channel_map(channel_map))

transform_resize

transform_resize(duration: float = Option(..., '--duration', '-d', help='Target frame duration in seconds')) -> Stage

Resize frames to a different duration, splitting or merging as needed.

Source code in gwframe/cli.py

@transform_app.command("resize")
def transform_resize(
    duration: float = typer.Option(
        ...,
        "--duration",
        "-d",
        help="Target frame duration in seconds",
    ),
) -> Stage:
    """Resize frames to a different duration, splitting or merging as needed."""
    try:
        return ResizeStage(duration)
    except ValueError as e:
        console.print(f"[red]Error: {e}[/red]")
        raise typer.Exit(1) from e

transform_select

transform_select(channels: list[str] = Option(..., '--channel', '-c', help='Channel(s) to keep (can be specified multiple times)')) -> Stage

Keep only specified channels, removing all others.

Source code in gwframe/cli.py

@transform_app.command("select")
def transform_select(
    channels: list[str] = typer.Option(
        ...,
        "--channel",
        "-c",
        help="Channel(s) to keep (can be specified multiple times)",
    ),
) -> Stage:
    """Keep only specified channels, removing all others."""
    return SelectStage(channels)

validate

validate(path1: Path = Argument(..., help='First GWF file or directory', exists=True), path2: Path = Argument(..., help='Second GWF file or directory', exists=True), channels: list[str] | None = Option(None, '--channel', '-c', help='Compare only the specified channel(s) (can specify multiple times)'), common_channels: bool = Option(False, '--common-channels', help='Compare only channels present on both sides'), common_time_spans: bool = Option(False, '--common-time-spans', help='Compare only time spans present on both sides (skip unmatched files and frames, including files deleted mid-run)'), common: bool = Option(False, '--common', help='Shorthand for --common-channels --common-time-spans'), ignore_channel_type: bool = Option(False, '--ignore-channel-type', help='Do not report channel type (adc/proc/sim) differences'), metadata_only: bool = Option(False, '--metadata-only', help='Compare structure and metadata only; skip sample data (faster)'), rtol: float = Option(0.0, '--rtol', help='Relative tolerance for float data (default: exact comparison)'), atol: float = Option(0.0, '--atol', help='Absolute tolerance for float data (default: exact comparison)'), recursive: bool = Option(False, '--recursive', '-r', help='Search directories recursively for GWF files'), verbose: int = Option(0, '--verbose', '-v', count=True, help=f'Show all differences instead of the first {_MAX_DIFF_ROWS} per pair'))

Check two GWF files or directories for consistency.

Compares channel sets, frame structure (count and GPS spans), per-channel metadata (type, dtype, sample rate, sample count, unit), and sample data. In directories, frames are paired by GPS span regardless of file boundaries, and results are grouped per pair of files that shared frames.

Exit codes: 0 = consistent, 1 = differences found, 2 = usage or read error.

Examples: gwframe validate a.gwf b.gwf gwframe validate dir1/ dir2/ --common gwframe validate a.gwf b.gwf -c L1:CHAN1 --metadata-only gwframe validate a.gwf b.gwf --atol 1e-9

Source code in gwframe/cli.py

@app.command()
def validate(
    path1: Path = typer.Argument(
        ...,
        help="First GWF file or directory",
        exists=True,
    ),
    path2: Path = typer.Argument(
        ...,
        help="Second GWF file or directory",
        exists=True,
    ),
    channels: list[str] | None = typer.Option(
        None,
        "--channel",
        "-c",
        help="Compare only the specified channel(s) (can specify multiple times)",
    ),
    common_channels: bool = typer.Option(
        False,
        "--common-channels",
        help="Compare only channels present on both sides",
    ),
    common_time_spans: bool = typer.Option(
        False,
        "--common-time-spans",
        help="Compare only time spans present on both sides (skip unmatched "
        "files and frames, including files deleted mid-run)",
    ),
    common: bool = typer.Option(
        False,
        "--common",
        help="Shorthand for --common-channels --common-time-spans",
    ),
    ignore_channel_type: bool = typer.Option(
        False,
        "--ignore-channel-type",
        help="Do not report channel type (adc/proc/sim) differences",
    ),
    metadata_only: bool = typer.Option(
        False,
        "--metadata-only",
        help="Compare structure and metadata only; skip sample data (faster)",
    ),
    rtol: float = typer.Option(
        0.0,
        "--rtol",
        help="Relative tolerance for float data (default: exact comparison)",
    ),
    atol: float = typer.Option(
        0.0,
        "--atol",
        help="Absolute tolerance for float data (default: exact comparison)",
    ),
    recursive: bool = typer.Option(
        False,
        "--recursive",
        "-r",
        help="Search directories recursively for GWF files",
    ),
    verbose: int = typer.Option(
        0,
        "--verbose",
        "-v",
        count=True,
        help=f"Show all differences instead of the first {_MAX_DIFF_ROWS} per pair",
    ),
):
    """
    Check two GWF files or directories for consistency.

    Compares channel sets, frame structure (count and GPS spans), per-channel
    metadata (type, dtype, sample rate, sample count, unit), and sample data.
    In directories, frames are paired by GPS span regardless of file
    boundaries, and results are grouped per pair of files that shared frames.

    Exit codes: 0 = consistent, 1 = differences found, 2 = usage or read error.

    Examples:
        gwframe validate a.gwf b.gwf
        gwframe validate dir1/ dir2/ --common
        gwframe validate a.gwf b.gwf -c L1:CHAN1 --metadata-only
        gwframe validate a.gwf b.gwf --atol 1e-9
    """
    if metadata_only and (rtol != 0.0 or atol != 0.0):
        console.print(
            "[red]Error: --metadata-only cannot be combined with --rtol/--atol[/red]"
        )
        raise typer.Exit(2)

    try:
        result = compare_paths(
            path1,
            path2,
            recursive=recursive,
            channels=channels or None,
            common_channels=common_channels or common,
            common_time_spans=common_time_spans or common,
            ignore_channel_type=ignore_channel_type,
            metadata_only=metadata_only,
            rtol=rtol,
            atol=atol,
        )
    except (ValueError, OSError) as e:
        console.print(f"[red]Error: {e}[/red]")
        raise typer.Exit(2) from e

    console.print()
    for file_result in result.file_results:
        left = str(file_result.left) if file_result.left else "-"
        right = str(file_result.right) if file_result.right else "-"
        if file_result.consistent:
            console.print(f"[green]✓[/green] {left} <-> {right}: consistent")
            continue

        n_diffs = len(file_result.differences)
        console.print(
            f"[red]✗[/red] {left} <-> {right}: "
            f"{n_diffs} difference{'s' if n_diffs != 1 else ''}"
        )

        diff_table = Table(box=box.SIMPLE_HEAD, show_edge=False, pad_edge=False)
        diff_table.add_column("Category", style="cyan", no_wrap=True)
        diff_table.add_column("Frame", justify="right")
        diff_table.add_column("Channel", no_wrap=True)
        diff_table.add_column("Field", no_wrap=True)
        diff_table.add_column("Left", overflow="fold", max_width=24)
        diff_table.add_column("Right", overflow="fold", max_width=24)
        diff_table.add_column("Detail", overflow="fold")

        ordered = sorted(
            file_result.differences,
            key=lambda d: _CATEGORY_ORDER.get(d.category, len(_CATEGORY_ORDER)),
        )
        shown = ordered if verbose else ordered[:_MAX_DIFF_ROWS]
        for diff in shown:
            diff_table.add_row(
                diff.category,
                str(diff.frame_index) if diff.frame_index is not None else "-",
                diff.channel or "-",
                diff.field,
                diff.left,
                diff.right,
                diff.detail,
            )
        if len(ordered) > len(shown):
            diff_table.add_row(
                "...",
                "",
                f"({len(ordered) - len(shown)} more, use -v)",
                "",
                "",
                "",
                "",
                style="dim",
            )

        console.print(diff_table)
        console.print()

    n_pairs = len(result.file_results)
    n_channels = sum(r.channels_compared for r in result.file_results)
    n_frames = sum(r.frames_compared for r in result.file_results)
    summary = (
        f"Compared {n_pairs} file pair{'s' if n_pairs != 1 else ''} "
        f"({n_frames} frame{'s' if n_frames != 1 else ''}, "
        f"{n_channels} channel comparison{'s' if n_channels != 1 else ''})"
    )
    if result.consistent:
        console.print(f"[green]{summary}: consistent[/green]")
        console.print()
        return

    console.print(
        f"[red]{summary}: {result.num_differences} "
        f"difference{'s' if result.num_differences != 1 else ''} found[/red]"
    )
    console.print()
    raise typer.Exit(1)

validate_output_options

validate_output_options(output: Path | None, in_place: bool) -> None

Validate that output options are correctly specified.

Source code in gwframe/cli.py

def validate_output_options(output: Path | None, in_place: bool) -> None:
    """Validate that output options are correctly specified."""
    if in_place and output is not None:
        console.print(
            "[red]Error: --in-place and --output-dir are mutually exclusive[/red]"
        )
        raise typer.Exit(1)

    if not in_place and output is None:
        console.print(
            "[red]Error: Either --in-place or --output-dir must be specified[/red]"
        )
        raise typer.Exit(1)