aimbat._cli

@app.command(name="iccs")
@handle_issues
def cli_iccs_run(
    event_id: Annotated[UUID, event_parameter()],
    *,
    autoselect: Annotated[
        bool,
        Parameter(
            name="autoselect",
            help=(
                "Whether to automatically de-select seismograms whose "
                + "cross-correlation with the stack falls below `min_cc`, and re-select"
                + " them if the cross-correlation later exceeds `min_cc`."
            ),
        ),
    ] = False,
    autoflip: Annotated[
        bool,
        Parameter(
            name="autoflip",
            help=(
                "Whether to automatically flip seismograms (multiply data by -1) when"
                + " the cross-correlation is negative."
            ),
        ),
    ] = False,
    _: DebugParameter = DebugParameter(),
) -> None:
    """Run the ICCS algorithm to align seismograms for an event.

    Iteratively cross-correlates seismograms against a running stack to refine
    arrival time picks (`t1`). If `t1` is not yet set, `t0` is used as the
    starting point.
    """
    from sqlmodel import Session

    from aimbat.core import create_iccs_instance, resolve_event, run_iccs
    from aimbat.db import engine

    with Session(engine) as session:
        event = resolve_event(session, event_id)
        iccs = create_iccs_instance(session, event).iccs
        run_iccs(session, event, iccs, autoflip, autoselect)

@app.command(name="mccc")
@handle_issues
def cli_mccc_run(
    event_id: Annotated[UUID, event_parameter()],
    *,
    all_seismograms: Annotated[
        bool,
        Parameter(
            name="all",
            help=(
                "Include all seismograms of an event in MCCC processing, not just the"
                + " currently selected ones."
            ),
        ),
    ] = False,
    _: DebugParameter = DebugParameter(),
) -> None:
    """Run the MCCC algorithm to refine arrival time picks for an event.

    Multi-channel cross-correlation simultaneously determines the optimal time
    shifts for all seismograms. Results are stored in `t1`.
    """
    from sqlmodel import Session

    from aimbat.core import create_iccs_instance, resolve_event, run_mccc
    from aimbat.db import engine

    with Session(engine) as session:
        event = resolve_event(session, event_id)
        iccs = create_iccs_instance(session, event).iccs
        run_mccc(session, event, iccs, all_seismograms)

def causal() -> Parameter:
    """Return a cyclopts `Parameter` for switching between causal and zero-phase filtering."""
    return Parameter(
        name="causal",
        help="Use causal (single-pass) instead of zero-phase filtering.",
        negative="--zero-phase",
    )

def confirm_or_abort(message: str, *, yes: bool) -> None:
    """Prompt for confirmation before a destructive action, then abort if declined.

    Args:
        message: The Yes/No question to display.
        yes: If True, skip the prompt (for scripting/non-interactive use).

    Raises:
        SystemExit: With status 0, if the user declines or the prompt cannot
            be read (e.g. no interactive terminal is attached).
    """
    if yes:
        return

    import sys

    from rich.console import Console
    from rich.prompt import Confirm

    console = Console(stderr=True)
    try:
        confirmed = Confirm.ask(message, console=console, default=False)
    except (EOFError, KeyboardInterrupt):
        confirmed = False

    if not confirmed:
        console.print("Aborted.")
        sys.exit(0)

def event_parameter(help: str | None = None) -> Parameter:
    """Return a cyclopts `Parameter` for selecting a single event by UUID or prefix.

    Args:
        help: Custom help string; falls back to a generic event UUID prompt.
    """
    from aimbat.models import AimbatEvent

    return Parameter(
        name=["event", "event-id"],
        help=help or "UUID (or unique prefix) of event to process.",
        env_var="DEFAULT_EVENT_ID",
        converter=_make_uuid_converter(AimbatEvent),
    )

def event_parameter_is_all(event_id: UUID | Literal["all"]) -> TypeIs[Literal["all"]]:
    """Return `True` if `event_id` is the literal string `"all"` (case-insensitive)."""
    if isinstance(event_id, str) and event_id.lower() == "all":
        return True
    return False

def event_parameter_with_all(help: str | None = None) -> Parameter:
    """Return a cyclopts `Parameter` for selecting an event or the literal `"all"`.

    Args:
        help: Custom help string; falls back to a generic prompt.
    """
    from aimbat.models import AimbatEvent

    return Parameter(
        name=["event", "event-id"],
        help=help
        or '"all" for all events, or UUID (or unique prefix) of event to process.',
        env_var="DEFAULT_EVENT_ID",
        converter=_make_uuid_converter(AimbatEvent, allow_all=True),
        show_choices=False,
    )

def event_table_title(
    session: "Session", event: "AimbatEvent", *, subject: str, raw: bool
) -> str:
    """Title for a table of records belonging to one event.

    Args:
        session: Database session.
        event: The event the records belong to.
        subject: Plural noun for what the table lists, e.g. `"seismograms"`.
        raw: If `True`, show the full event time and UUID instead of the
            rounded time and short ID.
    """
    if raw:
        return f"AIMBAT {subject} for event {event.time} (ID={event.id})"
    from aimbat.utils import uuid_shortener

    time = event.time.strftime("%Y-%m-%d %H:%M:%S")
    return f"AIMBAT {subject} for event {time} (ID={uuid_shortener(session, event)})"

def handle_issues[F: Callable[..., Any]](func: F) -> F:
    """Decorator that reports exceptions to the console and exits cleanly.

    Exceptions raised by the wrapped command are printed (without traceback)
    in a red panel, then the process exits with status 1. Any
    `aimbat.core.SchemaStaleWarning` raised during the call is always
    promoted to an error first, so a stale database schema is reported
    through the same red-panel path as any other failure, regardless of
    `AIMBAT_STRICT_SCHEMA_CHECK`.

    In debugging mode (`AIMBAT_LOG_LEVEL=DEBUG` or `TRACE`), the schema
    staleness promotion still applies, but exceptions are no longer caught
    and rendered as a panel; they propagate as a normal Python traceback
    instead.
    """
    from functools import wraps

    @wraps(func)
    def wrapper(*args: Any, **kwargs: Any) -> Any:
        return run_reporting_issues(lambda: func(*args, **kwargs))

    return wrapper  # type: ignore[return-value]

def id_label(
    session: "Session", model: type["AimbatTypes"], record_id: UUID | str, *, raw: bool
) -> str:
    """Render a record's ID the way a table title shows it.

    Args:
        session: Database session.
        model: Model the ID belongs to, used to shorten it unambiguously.
        record_id: The ID to render.
        raw: If `True`, show the full UUID instead of its short prefix.
    """
    from aimbat.utils import uuid_shortener

    if raw:
        return str(record_id)
    return uuid_shortener(session, model, str_uuid=str(record_id))

def id_parameter(model_class: type, help: str = "") -> Parameter:
    """Return a cyclopts `Parameter` for selecting a record by UUID or unique prefix.

    Args:
        model_class: AIMBAT model class used to resolve short UUID prefixes.
        help: Custom help string; falls back to a generic UUID prompt if empty.
    """
    return Parameter(
        name="id",
        help=help or "UUID (or any unique prefix).",
        converter=_make_uuid_converter(model_class),
    )

def json_to_table(
    data: dict[str, Any] | list[dict[str, Any]],
    model: type[BaseModel],
    title: str | None = None,
    raw: bool = False,
    col_specs: dict[str, RichColSpec] | None = None,
    column_order: list[str] | None = None,
    key_header: str = "Property",
    value_header: str = "Value",
) -> None:
    """Print a JSON dict or list of dicts as a rich table driven by a Pydantic model.

    Args:
        data: A single row (dict) or list of rows to display.
        model: Pydantic model whose field metadata drives column configuration.
        title: Optional table title.
        raw: If `True`, ignore `RichColSpec` metadata and render using only
            type-based heuristics. Useful for a quick unformatted view.
        col_specs: Optional per-field overrides. Each entry is merged on top of
            the spec derived from the model field, so only the attributes that
            differ need to be set. Ignored when `raw=True`.
        column_order: Optional list of field names that should appear first, in
            that order. Fields not listed appear after in model-declaration order.
        key_header: Header for the property-name column in vertical (dict) tables.
        value_header: Header for the value column in vertical (dict) tables.
    """
    console = Console()
    table = Table(title=title)

    data_list = [data] if isinstance(data, dict) else data
    if not data_list:
        console.print(table)
        return

    # Build specs from model field metadata (skipped in raw mode).
    specs: dict[str, RichColSpec] = {}
    if not raw:
        for name, field in model.model_fields.items():
            spec: RichColSpec | None = None
            # Pydantic Field uses json_schema_extra; SQLModel Field uses schema_extra
            # which spreads its keys directly into _attributes_set on the FieldInfo.
            for source in (
                field.json_schema_extra,
                getattr(field, "_attributes_set", None),
            ):
                if isinstance(source, dict):
                    candidate = source.get("rich")
                    if isinstance(candidate, RichColSpec):
                        spec = candidate
                        break
            specs[name] = spec if spec is not None else RichColSpec()

        # Apply caller-supplied overrides. model_fields_set tracks which fields
        # were explicitly provided, so only those replace the model-derived spec.
        if col_specs:
            for name, override in col_specs.items():
                base = specs.get(name, RichColSpec())
                specs[name] = base.model_copy(
                    update={k: getattr(override, k) for k in override.model_fields_set}
                )

    def _ordered(names: list[str]) -> list[str]:
        if not column_order:
            return names
        ordered = [n for n in column_order if n in names]
        rest = [n for n in names if n not in set(column_order)]
        return ordered + rest

    def _fmt_val(name: str, val: Any) -> str:
        if raw:
            return "" if val is None else str(val)
        if val is None or val is NaT:
            return _MISSING_MARKER
        spec = specs.get(name)
        if spec and spec.formatter:
            return spec.formatter(val)
        if isinstance(val, bool):
            return fmt_bool(val)
        if isinstance(val, float):
            return fmt_float(val)
        low_key = name.lower()
        if (
            isinstance(val, str)
            and val.strip()
            and any(k in low_key for k in ("time", "date", "modified"))
        ):
            try:
                dt = to_datetime(val)
                return fmt_timestamp(dt)
            except (ValueError, TypeError):
                return str(val)
        return str(val)

    field_names = _ordered(list(model.model_fields.keys()))

    if isinstance(data, dict):
        # Vertical table
        table.add_column(key_header, style="cyan")
        table.add_column(value_header)
        for name in field_names:
            if name not in data:
                continue
            spec = specs.get(name)
            if spec and spec.display_title:
                label = spec.display_title
            else:
                field = model.model_fields[name]
                label = field.title if field.title else name
            table.add_row(label, _fmt_val(name, data[name]))
    else:
        # Horizontal table: restrict to fields present in data.
        data_keys: set[str] = set()
        for item in data_list:
            data_keys.update(item.keys())
        visible_fields = [n for n in field_names if n in data_keys]

        for name in visible_fields:
            spec = specs.get(name)
            field = model.model_fields[name]
            header = (
                spec.display_title
                if spec and spec.display_title
                else field.title or name
            )
            kwargs: dict[str, Any] = {}
            default_justify = _justify_for_annotation(field.annotation)
            if default_justify:
                kwargs["justify"] = default_justify
            if spec:
                if spec.justify:
                    kwargs["justify"] = spec.justify
                if spec.style:
                    kwargs["style"] = spec.style
                if spec.no_wrap is not None:
                    kwargs["no_wrap"] = spec.no_wrap
                if spec.highlight is not None:
                    kwargs["highlight"] = spec.highlight
            table.add_column(header, **kwargs)

        for item in data:
            table.add_row(*[_fmt_val(n, item.get(n)) for n in visible_fields])

    console.print(table)

def make_note_app(
    target: "NoteTarget", model: type["AimbatTypes"], parameter: Parameter
) -> App:
    """Build the `note` sub-app (`read` and `edit`) for one kind of record.

    Args:
        target: The kind of record the notes belong to, also the noun used in
            the commands' help text.
        model: Model class the note is attached to.
        parameter: cyclopts parameter that selects the record to act on.
    """
    note_app = App(
        name="note", help=f"Read and edit {target} notes.", help_format="markdown"
    )
    article = "an" if target.startswith(("a", "e", "i", "o", "u")) else "a"

    def read(
        record_id: Annotated[UUID, parameter],
        *,
        _: DebugParameter = DebugParameter(),
    ) -> None:
        from rich.console import Console
        from rich.markdown import Markdown
        from sqlmodel import Session

        from aimbat.core import get_note_content
        from aimbat.db import engine

        with Session(engine) as session:
            content = get_note_content(
                session, target, _resolve_record(session, model, record_id)
            )

        Console().print(Markdown(content) if content else "(no note)")

    def edit(
        record_id: Annotated[UUID, parameter],
        *,
        _: DebugParameter = DebugParameter(),
    ) -> None:
        from sqlmodel import Session

        from aimbat.core import get_note_content, save_note
        from aimbat.db import engine

        with Session(engine) as session:
            original = get_note_content(
                session, target, _resolve_record(session, model, record_id)
            )

        updated = open_in_editor(original)

        if updated != original:
            with Session(engine) as session:
                raced = save_note(
                    session, target, record_id, updated, expected_previous=original
                )
            if raced:
                print_warning(
                    "Note changed elsewhere while the editor was open; your edit has"
                    + " overwritten that change."
                )

    read.__doc__ = (
        f"Display the note attached to {article} {target}, rendered as Markdown."
    )
    edit.__doc__ = _EDIT_HELP.format(target=target)

    note_app.command(handle_issues(read), name="read")
    note_app.command(handle_issues(edit), name="edit")

    return note_app

def open_in_editor(initial_content: str) -> str:
    """Write `initial_content` to a temporary Markdown file, open it in `$EDITOR`.

    Returns the (possibly updated) content after the editor exits. The
    temporary file is not removed until then, since Windows does not allow a
    second process to open a file that is already open.

    The editor command is taken from `$EDITOR` or `$VISUAL`. If neither is
    set, `notepad` is used on Windows and `vi` elsewhere. To use a GUI editor
    that does not block by default (e.g. VS Code), set `EDITOR="code --wait"`.
    """
    import os
    import shlex
    import subprocess
    import tempfile
    import time

    editor = os.environ.get("EDITOR") or os.environ.get("VISUAL")
    if not editor:
        editor = "notepad" if sys.platform == "win32" else "vi"

    with tempfile.NamedTemporaryFile(
        mode="w",
        suffix=".md",
        delete=False,
        encoding="utf-8",
    ) as tmp:
        tmp.write(initial_content)
        tmp_path = tmp.name

    try:
        started = time.monotonic()
        result = subprocess.run([*shlex.split(editor), tmp_path], check=False)
        elapsed = time.monotonic() - started
        if result.returncode != 0:
            from aimbat.logger import logger

            message = (
                f"Editor '{editor}' exited with code {result.returncode}; discarding "
                + "changes."
            )
            logger.warning(message)
            print_warning(message)
            return initial_content
        with open(tmp_path, encoding="utf-8") as f:
            content = f.read()
        if content == initial_content and elapsed < 1:
            from aimbat.logger import logger

            message = (
                f"Editor '{editor}' returned in under a second with no changes - if "
                + "it launched a separate window that is still open (a non-blocking "
                + "GUI editor), any edits made there will be lost once that window "
                + "closes, since this temporary file is already gone. Use a "
                + 'wait-for-close flag, e.g. EDITOR="code --wait".'
            )
            logger.warning(message)
            print_warning(message)
        return content
    finally:
        os.unlink(tmp_path)

def print_error_panel(e: Exception) -> None:
    """Print an exception to the console in a red panel."""
    from rich.console import Console
    from rich.panel import Panel

    from aimbat.utils import exception_message

    console = Console(stderr=True)
    panel = Panel(
        exception_message(e)
        + "\n\n(run with --debug or AIMBAT_LOG_LEVEL=DEBUG for a full traceback)",
        title="Error",
        title_align="left",
        border_style="red",
        expand=True,
    )
    console.print(panel)

def print_json_dump(dump: TableDump, *, by_alias: bool) -> None:
    """Print a table dump as JSON, in its own database session.

    Args:
        dump: The `core.dump_*_table` function to call.
        by_alias: Whether to key the records by field alias.
    """
    from rich import print_json
    from sqlmodel import Session

    from aimbat.db import engine

    with Session(engine) as session:
        print_json(data=dump(session, by_alias=by_alias))

def print_quality_table(
    session: "Session",
    dump: TableDump,
    *,
    title: str,
    filter_id: UUID | None,
    id_field: str,
    id_columns: dict[str, type["AimbatTypes"]],
    raw: bool,
) -> None:
    """Print one of the aggregated quality tables.

    Args:
        session: Database session.
        dump: The `core.dump_*_quality_table` function to call.
        title: Table title.
        filter_id: Restrict the table to this record, or `None` for all of them.
        id_field: Name of `dump`'s filter argument, e.g. `"event_id"`. Also the
            column left out when the table covers a single record, since every
            row would repeat it.
        id_columns: The ID columns to shorten, mapped to the model each belongs to.
        raw: If `True`, render the table unformatted.
    """
    from aimbat.models import SeismogramQualityStats

    def shorten(model: type["AimbatTypes"]) -> Callable[[Any], str]:
        return lambda value: id_label(session, model, value, raw=False)

    col_specs = {
        column: RichColSpec(formatter=shorten(model))
        for column, model in id_columns.items()
    }

    json_to_table(
        data=dump(
            session,
            **{id_field: filter_id},
            exclude=None if filter_id is None else {id_field},
        ),
        model=SeismogramQualityStats,
        title=title,
        raw=raw,
        col_specs=col_specs,
    )

def print_warning(message: object) -> None:
    """Print a non-fatal warning message to the console, styled yellow."""
    from rich.console import Console
    from rich.text import Text

    # A `Text` object applies styling structurally rather than via markup
    # tags, so a `[`/`]` in the message (e.g. a future revision id, built
    # from database content) can't be misinterpreted as Rich markup.
    Console(stderr=True).print(Text(str(message), style="yellow"))