feat: add mccc

src/aimbat/_config.py

@@ -169,11 +169,16 @@ def cli_settings_list(
     *,
     pretty: bool = True,
 ) -> None:
-    """Print a table with default settings used in AIMBAT.
+    """Print a table with default settings currently in use by AIMBAT.
 
     These defaults control the default behavior of AIMBAT within a project.
-    They can be changed using environment variables of the same name, or by
-    adding a `.env` file to the current working directory.
+    Overriding these defaults can be done on a per-project basis in the
+    fllowing ways (in order of precedence):
+
+    - By using environment variables of the form `AIMBAT_{SETTING_NAME}`
+      (e.g. `AIMBAT_LOG_LEVEL=DEBUG`).
+    - Setting them in a `.env` file in the current working directory
+      (e.g. `AIMBAT_LOG_LEVEL=DEBUG` in `.env`).
 
     Args:
         pretty: Print the table in a pretty format.

src/aimbat/aimbat_types/_pydantic.py

@@ -1,9 +1,5 @@
-from aimbat._lib.validators import (
-    must_be_negative_pd_timedelta,
-    must_be_positive_pd_timedelta,
-)
 from typing import Annotated, Callable, Any, cast, ClassVar
-from pydantic import AfterValidator
+from pydantic import AfterValidator, PlainSerializer
 from pydantic_core.core_schema import CoreSchema, no_info_plain_validator_function
 from pandas import Timestamp, Timedelta
 
@@ -15,6 +11,24 @@
 ]
 
 
+def _format_timedelta(td: Timedelta) -> float:
+    return td.total_seconds()
+
+
+def _must_be_negative_pd_timedelta(v: Timedelta) -> Timedelta:
+    """Validator to ensure a Timedelta is negative."""
+    if v.total_seconds() >= 0:
+        raise ValueError(f"Duration must be negative, got {v}")
+    return v
+
+
+def _must_be_positive_pd_timedelta(v: Timedelta) -> Timedelta:
+    """Validator to ensure a Timedelta is positive."""
+    if v.total_seconds() <= 0:
+        raise ValueError(f"Duration must be positive, got {v}")
+    return v
+
+
 class _PandasBaseAnnotation[T: Timestamp | Timedelta]:
     """Base class to provide Pydantic core schema for Pandas types."""
 
@@ -46,10 +60,14 @@ class _AnnotatedTimedelta(_PandasBaseAnnotation):
 
 
 type PydanticTimestamp = Annotated[Timestamp, _AnnotatedTimestamp]
-type PydanticTimedelta = Annotated[Timedelta, _AnnotatedTimedelta]
+type PydanticTimedelta = Annotated[
+    Timedelta,
+    _AnnotatedTimedelta,
+    PlainSerializer(_format_timedelta, return_type=float),
+]
 type PydanticNegativeTimedelta = Annotated[
-    PydanticTimedelta, AfterValidator(must_be_negative_pd_timedelta)
+    PydanticTimedelta, AfterValidator(_must_be_negative_pd_timedelta)
 ]
 type PydanticPositiveTimedelta = Annotated[
-    PydanticTimedelta, AfterValidator(must_be_positive_pd_timedelta)
+    PydanticTimedelta, AfterValidator(_must_be_positive_pd_timedelta)
 ]

src/aimbat/app.py

@@ -7,18 +7,20 @@
 """
 
 from ._config import cli_settings_list
-from importlib import metadata
-from cyclopts import App
 from .cli import (
-    data,
-    event,
-    iccs,
-    project,
-    seismogram,
-    snapshot,
-    station,
-    utils,
+    _align,
+    _data,
+    _event,
+    _pick,
+    _plot,
+    _project,
+    _station,
+    _seismogram,
+    _snapshot,
+    _utils,
 )
+from importlib import metadata
+from cyclopts import App
 from rich.console import Console
 import sys
 
@@ -30,15 +32,17 @@
 console = Console()
 
 app = App(version=__version__, help=__doc__, help_format="markdown", console=console)
-app.command(data.app)
-app.command(event.app)
-app.command(iccs.app)
-app.command(project.app)
-app.command(seismogram.app)
+app.command(_align.app)
+app.command(_data.app)
+app.command(_event.app)
+app.command(_pick.app)
+app.command(_plot.app)
+app.command(_project.app)
+app.command(_station.app)
+app.command(_seismogram.app)
 app.command(cli_settings_list, name="settings")
-app.command(snapshot.app)
-app.command(station.app)
-app.command(utils.app)
+app.command(_snapshot.app)
+app.command(_utils.app)
 
 
 if __name__ == "__main__":

src/aimbat/cli/_align.py

@@ -0,0 +1,64 @@
+"""Align seismograms using ICCS or MCCC.
+
+This command aligns seismograms using either the ICCS or MCCC algorithm. Both
+commands update the pick stored in `t1`. If `t1` is `None`, `t0` is used as
+starting point instead, with the resulting pick stored in `t1`.
+"""
+
+from ._common import GlobalParameters, simple_exception
+from cyclopts import App, Parameter
+from typing import Annotated
+
+app = App(name="align", help=__doc__, help_format="markdown")
+
+
+@app.command(name="iccs")
+@simple_exception
+def cli_iccs_run(
+    *,
+    autoflip: bool = False,
+    autoselect: bool = False,
+    global_parameters: GlobalParameters | None = None,
+) -> None:
+    """Run the ICCS algorithm.
+
+    Args:
+        autoflip: Whether to automatically flip seismograms (multiply data by -1).
+        autoselect: Whether to automatically de-select seismograms.
+    """
+    from aimbat.db import engine
+    from aimbat.core import create_iccs_instance, run_iccs
+    from sqlmodel import Session
+
+    global_parameters = global_parameters or GlobalParameters()
+
+    with Session(engine) as session:
+        iccs = create_iccs_instance(session)
+        run_iccs(session, iccs, autoflip, autoselect)
+
+
+@app.command(name="mccc")
+@simple_exception
+def cli_mccc_run(
+    *,
+    all_seismograms: Annotated[bool, Parameter(name="all")] = False,
+    global_parameters: GlobalParameters | None = None,
+) -> None:
+    """Run the MCCC algorithm.
+
+    Args:
+        all_seismograms: Whether to include all seismograms in the MCCC processing, or just the selected ones.
+    """
+    from aimbat.db import engine
+    from aimbat.core import create_iccs_instance, run_mccc
+    from sqlmodel import Session
+
+    global_parameters = global_parameters or GlobalParameters()
+
+    with Session(engine) as session:
+        iccs = create_iccs_instance(session)
+        run_mccc(session, iccs, all_seismograms)
+
+
+if __name__ == "__main__":
+    app()

src/aimbat/cli/common.py → src/aimbat/cli/_common.py

@@ -16,14 +16,27 @@ class GlobalParameters:
     debug: bool = False
     "Run in debugging mode."
 
-    use_qt: bool = False
-    "Use pyqtgraph instead of matplotlib for plots (where applicable)."
-
     def __post_init__(self) -> None:
         if self.debug:
             settings.log_level = "DEBUG"
 
 
+@Parameter(name="*")
+@dataclass
+class PlotParameters:
+    use_qt: bool = False
+    "Use pyqtgraph instead of matplotlib for plots (where applicable)."
+
+
+@Parameter(name="*")
+@dataclass
+class IccsPlotParameters:
+    context: bool = True
+    "Plot seismograms with extra context instead of the short tapered ones used for cross-correlation."
+    all: bool = False
+    "Include all seismograms in the plot, even if not used in stack."
+
+
 @Parameter(name="*")
 @dataclass
 class TableParameters:

src/aimbat/cli/data.py → src/aimbat/cli/_data.py

@@ -1,6 +1,6 @@
 """Manage seismogram files in an AIMBAT project."""
 
-from aimbat.cli.common import GlobalParameters, TableParameters, simple_exception
+from ._common import GlobalParameters, TableParameters, simple_exception
 from aimbat.aimbat_types import DataType
 from sqlmodel import Session
 from cyclopts import App, Parameter, validators
@@ -78,12 +78,13 @@ def cli_data_dump(
 ) -> None:
     """Dump the contents of the AIMBAT data table to json."""
     from aimbat.db import engine
-    from aimbat.core import dump_data_table
+    from aimbat.core import dump_data_table_to_json
+    from rich import print_json
 
     global_parameters = global_parameters or GlobalParameters()
 
     with Session(engine) as session:
-        dump_data_table(session)
+        print_json(dump_data_table_to_json(session))
 
 
 if __name__ == "__main__":