Skip to content

aimbat

AIMBAT (Automated and Interactive Measurement of Body wave Arrival Times) is an open-source software package for efficiently measuring teleseismic body wave arrival times for large seismic arrays (Lou et al., 2012). It is based on a widely used method called MCCC (Multi-Channel Cross-Correlation) developed by VanDecar and Crosson (1990). The package is automated in the sense of initially aligning seismograms for MCCC which is achieved by an ICCS (Iterative Cross Correlation and Stack) algorithm. Meanwhile, a graphical user interface is built to perform seismogram quality control interactively. Therefore, user processing time is reduced while valuable input from a user's expertise is retained. As a byproduct, SAC (Goldstein et al., 2003) plotting and phase picking functionalities are replicated and enhanced.

Modules:

Name Description
aimbat_types

Custom types used in AIMBAT.
app

AIMBAT command line interface entrypoint for all other commands.
cli
core
db

Module to define the AIMBAT project file and create the database engine.
io

Functions to read and write data files used with AIMBAT
logger

Logging setup.
models

Models used in AIMBAT.
utils

Utils used in AIMBAT.

aimbat_types

Custom types used in AIMBAT.

Type Aliases:

Name Description
EventParameterBool

TypeAlias for [AimbatEvent][aimbat.lib.models.AimbatEvent] attributes with bool values.
EventParameterFloat

TypeAlias for [AimbatEvent][aimbat.lib.models.AimbatEvent] attributes with float values.
EventParameterTimedelta

TypeAlias for [AimbatEvent][aimbat.lib.models.AimbatEvent] attributes with [Timedelta][pandas.Timedelta] values.

Classes:

Name Description
DataType

Valid AIMBAT data types.
EventParameter

[AimbatEvent][aimbat.lib.models.AimbatEvent] enum class for typing.
SeismogramParameter

[AimbatSeismograParameters][aimbat.lib.models.AimbatSeismogramParameters] enum class for typing.

EventParameterBool 

EventParameterBool = Literal[COMPLETED, BANDPASS_APPLY]

TypeAlias for [AimbatEvent][aimbat.lib.models.AimbatEvent] attributes with bool values.

EventParameterFloat 

EventParameterFloat = Literal[
    MIN_CCNORM, BANDPASS_FMIN, BANDPASS_FMAX
]

TypeAlias for [AimbatEvent][aimbat.lib.models.AimbatEvent] attributes with float values.

EventParameterTimedelta 

EventParameterTimedelta = Literal[WINDOW_PRE, WINDOW_POST]

TypeAlias for [AimbatEvent][aimbat.lib.models.AimbatEvent] attributes with [Timedelta][pandas.Timedelta] values.

DataType

Bases: StrEnum

Valid AIMBAT data types.

Source code in src/aimbat/aimbat_types/_data.py 
 8
 9
10
11
class DataType(StrEnum):
    """Valid AIMBAT data types."""

    SAC = auto()

EventParameter

Bases: StrEnum

[AimbatEvent][aimbat.lib.models.AimbatEvent] enum class for typing.

This enum class is used for typing, cli args etc. The attributes must be the same as in the [AimbatEvent][aimbat.lib.models.AimbatEvent] model.

Source code in src/aimbat/aimbat_types/_event.py 
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
class EventParameter(StrEnum):
    """[`AimbatEvent`][aimbat.lib.models.AimbatEvent] enum class for typing.

    This enum class is used for typing, cli args etc. The attributes must be
    the same as in the [`AimbatEvent`][aimbat.lib.models.AimbatEvent] model.
    """

    COMPLETED = auto()
    MIN_CCNORM = auto()
    WINDOW_PRE = auto()
    WINDOW_POST = auto()
    BANDPASS_APPLY = auto()
    BANDPASS_FMIN = auto()
    BANDPASS_FMAX = auto()

SeismogramParameter

Bases: StrEnum

[AimbatSeismograParameters][aimbat.lib.models.AimbatSeismogramParameters] enum class for typing.

This enum class is used for typing, cli args etc. The attributes must be the same as in the [AimbatParameters][aimbat.lib.models.AimbatParameters] model.

Source code in src/aimbat/aimbat_types/_seismogram.py 
11
12
13
14
15
16
17
18
19
20
class SeismogramParameter(StrEnum):
    """[`AimbatSeismograParameters`][aimbat.lib.models.AimbatSeismogramParameters] enum class for typing.

    This enum class is used for typing, cli args etc. The attributes must be
    the same as in the [`AimbatParameters`][aimbat.lib.models.AimbatParameters] model.
    """

    SELECT = auto()
    FLIP = auto()
    T1 = auto()

app

AIMBAT command line interface entrypoint for all other commands.

This is the main command line interface for AIMBAT. It must be executed with a command (as specified below) to actually do anything. Help for individual commands is available by typing aimbat COMMAND --help.

cli

Modules:

Name Description
common

Common parameters and functions for the AIMBAT CLI.
data

Manage seismogram files in an AIMBAT project.
event

View and manage events in the AIMBAT project.
iccs

ICCS processing tools.
project

Manage AIMBAT projects.
seismogram

View and manage seismograms in the AIMBAT project.
snapshot

View and manage snapshots.
station

View and manage stations.
utils

Utilities for AIMBAT.

common

Common parameters and functions for the AIMBAT CLI.

Classes:

Name Description
CliHints

Hints for error messages.
GlobalParameters
TableParameters

Functions:

Name Description
simple_exception

Decorator to handle exceptions and print them to the console.

CliHints dataclass

Hints for error messages.

Source code in src/aimbat/cli/common.py 
39
40
41
42
43
44
@dataclass(frozen=True)
class CliHints:
    """Hints for error messages."""

    ACTIVATE_EVENT = "Hint: activate an event with `aimbat event activate <EVENT_ID>`."
    LIST_EVENTS = "Hint: view available events with `aimbat event list`."

GlobalParameters dataclass

Attributes:

Name Type Description
debug bool

Run in debugging mode.
use_qt bool

Use pyqtgraph instead of matplotlib for plots (where applicable).
Source code in src/aimbat/cli/common.py 
13
14
15
16
17
18
19
20
21
22
23
24
@Parameter(name="*")
@dataclass
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"
debug class-attribute instance-attribute 
debug: bool = False

Run in debugging mode.

use_qt class-attribute instance-attribute 
use_qt: bool = False

Use pyqtgraph instead of matplotlib for plots (where applicable).

TableParameters dataclass

Attributes:

Name Type Description
short bool

Shorten UUIDs and format data.
Source code in src/aimbat/cli/common.py 
27
28
29
30
31
@Parameter(name="*")
@dataclass
class TableParameters:
    short: bool = True
    "Shorten UUIDs and format data."
short class-attribute instance-attribute 
short: bool = True

Shorten UUIDs and format data.

simple_exception 

simple_exception(func: F) -> F

Decorator to handle exceptions and print them to the console.

Using this decorator prints only the exception to the console without traceback, and then exits. In debugging mode this decorator returns the callable unchanged.

Source code in src/aimbat/cli/common.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def simple_exception[F: Callable[..., Any]](func: F) -> F:
    """Decorator to handle exceptions and print them to the console.

    Using this decorator prints only the exception to the console without
    traceback, and then exits. In debugging mode this decorator returns the
    callable unchanged.
    """
    from functools import wraps
    from rich.console import Console
    from rich.panel import Panel
    import sys

    @wraps(func)
    def wrapper(*args: Any, **kwargs: Any) -> Any:
        if settings.log_level in ("TRACE", "DEBUG"):
            return func(*args, **kwargs)
        try:
            return func(*args, **kwargs)
        except Exception as e:
            console = Console()
            panel = Panel(
                f"{e}",
                title="Error",
                title_align="left",
                border_style="red",
                expand=True,
            )
            console.print(panel)
            sys.exit(1)

    return wrapper  # type: ignore

data

Manage seismogram files in an AIMBAT project.

Functions:

Name Description
cli_data_add

Add or update data files in the AIMBAT project.
cli_data_dump

Dump the contents of the AIMBAT data table to json.
cli_data_list

Print information on the data stored in AIMBAT.

cli_data_add 

cli_data_add(
    seismogram_files: Annotated[
        list[Path],
        Parameter(
            name=files,
            consume_multiple=True,
            validator=Path(exists=True),
        ),
    ],
    *,
    filetype: DataType = SAC,
    show_progress_bar: Annotated[
        bool, Parameter(name=progress)
    ] = True,
    global_parameters: GlobalParameters | None = None
) -> None

Add or update data files in the AIMBAT project.

Parameters:

Name Type Description Default
seismogram_files Annotated[list[Path], Parameter(name=files, consume_multiple=True, validator=Path(exists=True))]

Seismogram files to be added.

 required
filetype DataType

Specify type of seismogram file.

 SAC
show_progress_bar Annotated[bool, Parameter(name=progress)]

Display progress bar.

 True
Source code in src/aimbat/cli/data.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
@app.command(name="add")
@simple_exception
def cli_data_add(
    seismogram_files: Annotated[
        list[Path],
        Parameter(
            name="files", consume_multiple=True, validator=validators.Path(exists=True)
        ),
    ],
    *,
    filetype: DataType = DataType.SAC,
    show_progress_bar: Annotated[bool, Parameter(name="progress")] = True,
    global_parameters: GlobalParameters | None = None,
) -> None:
    """Add or update data files in the AIMBAT project.

    Args:
        seismogram_files: Seismogram files to be added.
        filetype: Specify type of seismogram file.
        show_progress_bar: Display progress bar.
    """
    from aimbat.db import engine
    from aimbat.core import add_files_to_project

    global_parameters = global_parameters or GlobalParameters()

    disable_progress_bar = not show_progress_bar

    with Session(engine) as session:
        add_files_to_project(
            session,
            seismogram_files,
            filetype,
            disable_progress_bar,
        )

cli_data_dump 

cli_data_dump(
    *, global_parameters: GlobalParameters | None = None
) -> None

Dump the contents of the AIMBAT data table to json.

Source code in src/aimbat/cli/data.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
@app.command(name="dump")
@simple_exception
def cli_data_dump(
    *,
    global_parameters: GlobalParameters | None = None,
) -> None:
    """Dump the contents of the AIMBAT data table to json."""
    from aimbat.db import engine
    from aimbat.core import dump_data_table

    global_parameters = global_parameters or GlobalParameters()

    with Session(engine) as session:
        dump_data_table(session)

cli_data_list 

cli_data_list(
    *,
    all_events: Annotated[
        bool, Parameter(name=all)
    ] = False,
    table_parameters: TableParameters | None = None,
    global_parameters: GlobalParameters | None = None
) -> None

Print information on the data stored in AIMBAT.

Parameters:

Name Type Description Default
all_events Annotated[bool, Parameter(name=all)]

Select data for all events.

 False
Source code in src/aimbat/cli/data.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
@app.command(name="list")
@simple_exception
def cli_data_list(
    *,
    all_events: Annotated[bool, Parameter(name="all")] = False,
    table_parameters: TableParameters | None = None,
    global_parameters: GlobalParameters | None = None,
) -> None:
    """Print information on the data stored in AIMBAT.

    Args:
        all_events: Select data for all events.
    """
    from aimbat.db import engine
    from aimbat.core import print_data_table

    table_parameters = table_parameters or TableParameters()
    global_parameters = global_parameters or GlobalParameters()

    with Session(engine) as session:
        print_data_table(session, table_parameters.short, all_events)

event

View and manage events in the AIMBAT project.

Functions:

Name Description
cli_event_activate

Select the event to be active for Processing.
cli_event_delete

Delete existing event.
cli_event_dump

Dump the contents of the AIMBAT event table to json.
cli_event_list

Print information on the events stored in AIMBAT.
cli_event_parameter_get

Get parameter value for the active event.
cli_event_parameter_set

Set parameter value for the active event.

cli_event_activate 

cli_event_activate(
    event_id: Annotated[UUID | str, Parameter(name=id)],
    *,
    global_parameters: GlobalParameters | None = None
) -> None

Select the event to be active for Processing.

Parameters:

Name Type Description Default
event_id Annotated[UUID | str, Parameter(name=id)]

Event ID number.

 required
Source code in src/aimbat/cli/event.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
@app.command(name="activate")
@simple_exception
def cli_event_activate(
    event_id: Annotated[uuid.UUID | str, Parameter(name="id")],
    *,
    global_parameters: GlobalParameters | None = None,
) -> None:
    """Select the event to be active for Processing.

    Args:
        event_id: Event ID number.
    """
    from aimbat.core import set_active_event_by_id
    from aimbat.db import engine

    global_parameters = global_parameters or GlobalParameters()

    with Session(engine) as session:
        if not isinstance(event_id, uuid.UUID):
            event_id = string_to_event_uuid(session, event_id)
        set_active_event_by_id(session, event_id)

cli_event_delete 

cli_event_delete(
    event_id: Annotated[UUID | str, Parameter(name=id)],
    *,
    global_parameters: GlobalParameters | None = None
) -> None

Delete existing event.

Parameters:

Name Type Description Default
event_id Annotated[UUID | str, Parameter(name=id)]

Event ID.

 required
Source code in src/aimbat/cli/event.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
@app.command(name="delete")
@simple_exception
def cli_event_delete(
    event_id: Annotated[uuid.UUID | str, Parameter(name="id")],
    *,
    global_parameters: GlobalParameters | None = None,
) -> None:
    """Delete existing event.

    Args:
        event_id: Event ID.
    """
    from aimbat.db import engine
    from aimbat.core import delete_event_by_id

    global_parameters = global_parameters or GlobalParameters()

    with Session(engine) as session:
        if not isinstance(event_id, uuid.UUID):
            event_id = string_to_event_uuid(session, event_id)
        delete_event_by_id(session, event_id)

cli_event_dump 

cli_event_dump(
    *, global_parameters: GlobalParameters | None = None
) -> None

Dump the contents of the AIMBAT event table to json.

Source code in src/aimbat/cli/event.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
@app.command(name="dump")
@simple_exception
def cli_event_dump(
    *,
    global_parameters: GlobalParameters | None = None,
) -> None:
    """Dump the contents of the AIMBAT event table to json."""
    from aimbat.db import engine
    from aimbat.core import dump_event_table

    global_parameters = global_parameters or GlobalParameters()

    with Session(engine) as session:
        dump_event_table(session)

cli_event_list 

cli_event_list(
    *,
    table_parameters: TableParameters | None = None,
    global_parameters: GlobalParameters | None = None
) -> None

Print information on the events stored in AIMBAT.

Source code in src/aimbat/cli/event.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
@app.command(name="list")
@simple_exception
def cli_event_list(
    *,
    table_parameters: TableParameters | None = None,
    global_parameters: GlobalParameters | None = None,
) -> None:
    """Print information on the events stored in AIMBAT."""
    from aimbat.db import engine
    from aimbat.core import print_event_table

    table_parameters = table_parameters or TableParameters()
    global_parameters = global_parameters or GlobalParameters()

    with Session(engine) as session:
        print_event_table(session, table_parameters.short)

cli_event_parameter_get 

cli_event_parameter_get(
    name: EventParameter,
    *,
    global_parameters: GlobalParameters | None = None
) -> None

Get parameter value for the active event.

Parameters:

Name Type Description Default
name EventParameter

Event parameter name.

 required
Source code in src/aimbat/cli/event.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
@app.command(name="get")
@simple_exception
def cli_event_parameter_get(
    name: EventParameter,
    *,
    global_parameters: GlobalParameters | None = None,
) -> None:
    """Get parameter value for the active event.

    Args:
        name: Event parameter name.
    """

    from aimbat.db import engine
    from aimbat.core import get_event_parameter
    from sqlmodel import Session

    global_parameters = global_parameters or GlobalParameters()

    with Session(engine) as session:
        value = get_event_parameter(session, name)
        if isinstance(value, Timedelta):
            print(f"{value.total_seconds()}s")
        else:
            print(value)

cli_event_parameter_set 

cli_event_parameter_set(
    name: EventParameter,
    value: Timedelta | str,
    *,
    global_parameters: GlobalParameters | None = None
) -> None

Set parameter value for the active event.

Parameters:

Name Type Description Default
name EventParameter

Event parameter name.

 required
value Timedelta | str

Event parameter value.

 required
Source code in src/aimbat/cli/event.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
@app.command(name="set")
@simple_exception
def cli_event_parameter_set(
    name: EventParameter,
    value: Timedelta | str,
    *,
    global_parameters: GlobalParameters | None = None,
) -> None:
    """Set parameter value for the active event.

    Args:
        name: Event parameter name.
        value: Event parameter value.
    """
    from aimbat.db import engine
    from aimbat.core import set_event_parameter
    from sqlmodel import Session

    global_parameters = global_parameters or GlobalParameters()

    with Session(engine) as session:
        set_event_parameter(session, name, value)

iccs

ICCS processing tools.

Launches various processing tools related to ICCS.

Classes:

Name Description
IccsPlotParameters

Functions:

Name Description
cli_iccs_plot_seismograms

Plot the ICCS seismograms of the active event as an image.
cli_iccs_plot_stack

Plot the ICCS stack of the active event.
cli_iccs_run

Run the ICCS algorithm.
cli_iccs_update_min_ccnorm

Pick a new minimum cross-correlation norm for auto-selection.
cli_iccs_update_pick

Pick a new arrival time.
cli_iccs_update_timewindow

Pick a new time window.

IccsPlotParameters dataclass

Attributes:

Name Type Description
all bool

Include all seismograms in the plot, even if not used in stack.
context bool

Plot seismograms with extra context instead of the short tapered ones used for cross-correlation.
Source code in src/aimbat/cli/iccs.py 
12
13
14
15
16
17
18
@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."
all class-attribute instance-attribute 
all: bool = False

Include all seismograms in the plot, even if not used in stack.

context class-attribute instance-attribute 
context: bool = True

Plot seismograms with extra context instead of the short tapered ones used for cross-correlation.

cli_iccs_plot_seismograms 

cli_iccs_plot_seismograms(
    *,
    iccs_parameters: IccsPlotParameters | None = None,
    global_parameters: GlobalParameters | None = None
) -> None

Plot the ICCS seismograms of the active event as an image.

Source code in src/aimbat/cli/iccs.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
@plot.command(name="image")
@simple_exception
def cli_iccs_plot_seismograms(
    *,
    iccs_parameters: IccsPlotParameters | None = None,
    global_parameters: GlobalParameters | None = None,
) -> None:
    """Plot the ICCS seismograms of the active event as an image."""
    from aimbat.db import engine
    from aimbat.core import create_iccs_instance, plot_seismograms
    from sqlmodel import Session

    iccs_parameters = iccs_parameters or IccsPlotParameters()
    global_parameters = global_parameters or GlobalParameters()

    with Session(engine) as session:
        iccs = create_iccs_instance(session)
        plot_seismograms(iccs, iccs_parameters.context, iccs_parameters.all)

cli_iccs_plot_stack 

cli_iccs_plot_stack(
    *,
    iccs_parameters: IccsPlotParameters | None = None,
    global_parameters: GlobalParameters | None = None
) -> None

Plot the ICCS stack of the active event.

Source code in src/aimbat/cli/iccs.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
@plot.command(name="stack")
@simple_exception
def cli_iccs_plot_stack(
    *,
    iccs_parameters: IccsPlotParameters | None = None,
    global_parameters: GlobalParameters | None = None,
) -> None:
    """Plot the ICCS stack of the active event."""
    from aimbat.db import engine
    from aimbat.core import create_iccs_instance, plot_stack
    from sqlmodel import Session

    iccs_parameters = iccs_parameters or IccsPlotParameters()
    global_parameters = global_parameters or GlobalParameters()

    with Session(engine) as session:
        iccs = create_iccs_instance(session)
        plot_stack(iccs, iccs_parameters.context, iccs_parameters.all)

cli_iccs_run 

cli_iccs_run(
    *,
    autoflip: bool = False,
    autoselect: bool = False,
    global_parameters: GlobalParameters | None = None
) -> None

Run the ICCS algorithm.

Parameters:

Name Type Description Default
autoflip bool

Whether to automatically flip seismograms (multiply data by -1).

 False
autoselect bool

Whether to automatically de-select seismograms.

 False
Source code in src/aimbat/cli/iccs.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
@app.command(name="run")
@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)

cli_iccs_update_min_ccnorm 

cli_iccs_update_min_ccnorm(
    *,
    iccs_parameters: IccsPlotParameters | None = None,
    global_parameters: GlobalParameters | None = None
) -> None

Pick a new minimum cross-correlation norm for auto-selection.

Source code in src/aimbat/cli/iccs.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
@update.command(name="ccnorm")
@simple_exception
def cli_iccs_update_min_ccnorm(
    *,
    iccs_parameters: IccsPlotParameters | None = None,
    global_parameters: GlobalParameters | None = None,
) -> None:
    """Pick a new minimum cross-correlation norm for auto-selection."""
    from aimbat.db import engine
    from aimbat.core import create_iccs_instance, update_min_ccnorm
    from sqlmodel import Session

    iccs_parameters = iccs_parameters or IccsPlotParameters()
    global_parameters = global_parameters or GlobalParameters()

    with Session(engine) as session:
        iccs = create_iccs_instance(session)
        update_min_ccnorm(session, iccs, iccs_parameters.context, iccs_parameters.all)

cli_iccs_update_pick 

cli_iccs_update_pick(
    *,
    iccs_parameters: IccsPlotParameters | None = None,
    use_seismogram_image: Annotated[
        bool, Parameter(name=img)
    ] = False,
    global_parameters: GlobalParameters | None = None
) -> None

Pick a new arrival time.

Parameters:

Name Type Description Default
use_seismogram_image Annotated[bool, Parameter(name=img)]

Use the seismogram image to update pick.

 False
Source code in src/aimbat/cli/iccs.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
@update.command(name="pick")
@simple_exception
def cli_iccs_update_pick(
    *,
    iccs_parameters: IccsPlotParameters | None = None,
    use_seismogram_image: Annotated[bool, Parameter(name="img")] = False,
    global_parameters: GlobalParameters | None = None,
) -> None:
    """Pick a new arrival time.

    Args:
        use_seismogram_image: Use the seismogram image to update pick.
    """
    from aimbat.db import engine
    from aimbat.core import create_iccs_instance, update_pick
    from sqlmodel import Session

    iccs_parameters = iccs_parameters or IccsPlotParameters()
    global_parameters = global_parameters or GlobalParameters()

    with Session(engine) as session:
        iccs = create_iccs_instance(session)
        update_pick(
            session,
            iccs,
            iccs_parameters.context,
            iccs_parameters.all,
            use_seismogram_image,
        )

cli_iccs_update_timewindow 

cli_iccs_update_timewindow(
    *,
    iccs_parameters: IccsPlotParameters | None = None,
    use_seismogram_image: Annotated[
        bool, Parameter(name=img)
    ] = False,
    global_parameters: GlobalParameters | None = None
) -> None

Pick a new time window.

Parameters:

Name Type Description Default
use_seismogram_image Annotated[bool, Parameter(name=img)]

Use the seismogram image to pick the time window.

 False
Source code in src/aimbat/cli/iccs.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
@update.command(name="window")
@simple_exception
def cli_iccs_update_timewindow(
    *,
    iccs_parameters: IccsPlotParameters | None = None,
    use_seismogram_image: Annotated[bool, Parameter(name="img")] = False,
    global_parameters: GlobalParameters | None = None,
) -> None:
    """Pick a new time window.

    Args:
        use_seismogram_image: Use the seismogram image to pick the time window.
    """
    from aimbat.db import engine
    from aimbat.core import create_iccs_instance, update_timewindow
    from sqlmodel import Session

    iccs_parameters = iccs_parameters or IccsPlotParameters()
    global_parameters = global_parameters or GlobalParameters()

    with Session(engine) as session:
        iccs = create_iccs_instance(session)
        update_timewindow(
            session,
            iccs,
            iccs_parameters.context,
            iccs_parameters.all,
            use_seismogram_image,
        )

project

Manage AIMBAT projects.

This command manages projects. By default, the project consists of a file called aimbat.db in the current working directory. All aimbat commands must be executed from the same directory. The location (and name) of the project file may also be specified by setting the AIMBAT_PROJECT environment variable to the desired filename. Alternatively, aimbat can be executed with a database url directly.

Functions:

Name Description
cli_project_create

Create new AIMBAT project.
cli_project_delete

Delete project (note: this does not delete seismogram files).
cli_project_info

Show information on an exisiting project.

cli_project_create 

cli_project_create(
    *, global_parameters: GlobalParameters | None = None
) -> None

Create new AIMBAT project.

Source code in src/aimbat/cli/project.py 
18
19
20
21
22
23
24
25
26
27
@app.command(name="create")
@simple_exception
def cli_project_create(*, global_parameters: GlobalParameters | None = None) -> None:
    """Create new AIMBAT project."""
    from aimbat.db import engine
    from aimbat.core import create_project

    global_parameters = global_parameters or GlobalParameters()

    create_project(engine)

cli_project_delete 

cli_project_delete(
    *, global_parameters: GlobalParameters | None = None
) -> None

Delete project (note: this does not delete seismogram files).

Source code in src/aimbat/cli/project.py 
30
31
32
33
34
35
36
37
38
39
@app.command(name="delete")
@simple_exception
def cli_project_delete(*, global_parameters: GlobalParameters | None = None) -> None:
    """Delete project (note: this does *not* delete seismogram files)."""
    from aimbat.db import engine
    from aimbat.core import delete_project

    global_parameters = global_parameters or GlobalParameters()

    delete_project(engine)

cli_project_info 

cli_project_info(
    *, global_parameters: GlobalParameters | None = None
) -> None

Show information on an exisiting project.

Source code in src/aimbat/cli/project.py 
42
43
44
45
46
47
48
49
50
51
52
@app.command(name="info")
@simple_exception
def cli_project_info(*, global_parameters: GlobalParameters | None = None) -> None:
    """Show information on an exisiting project."""

    from aimbat.db import engine
    from aimbat.core import print_project_info

    global_parameters = global_parameters or GlobalParameters()

    print_project_info(engine)

seismogram

View and manage seismograms in the AIMBAT project.

Functions:

Name Description
cli_seismogram_delete

Delete existing seismogram.
cli_seismogram_dump

Dump the contents of the AIMBAT seismogram table to json.
cli_seismogram_get

Get the value of a processing parameter.
cli_seismogram_list

Print information on the seismograms in the active event.
cli_seismogram_plot

Plot seismograms for the active event.
cli_seismogram_set

Set value of a processing parameter.

cli_seismogram_delete 

cli_seismogram_delete(
    seismogram_id: Annotated[
        UUID | str, Parameter(name=id)
    ],
    *,
    global_parameters: GlobalParameters | None = None
) -> None

Delete existing seismogram.

Parameters:

Name Type Description Default
seismogram_id Annotated[UUID | str, Parameter(name=id)]

Seismogram ID.

 required
Source code in src/aimbat/cli/seismogram.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
@app.command(name="delete")
@simple_exception
def cli_seismogram_delete(
    seismogram_id: Annotated[uuid.UUID | str, Parameter(name="id")],
    *,
    global_parameters: GlobalParameters | None = None,
) -> None:
    """Delete existing seismogram.

    Args:
        seismogram_id: Seismogram ID.
    """
    from aimbat.utils import string_to_uuid
    from aimbat.db import engine
    from aimbat.models import AimbatSeismogram
    from aimbat.core import delete_seismogram_by_id
    from sqlmodel import Session

    global_parameters = global_parameters or GlobalParameters()

    with Session(engine) as session:
        if not isinstance(seismogram_id, uuid.UUID):
            seismogram_id = string_to_uuid(session, seismogram_id, AimbatSeismogram)
        delete_seismogram_by_id(session, seismogram_id)

cli_seismogram_dump 

cli_seismogram_dump(
    *, global_parameters: GlobalParameters | None = None
) -> None

Dump the contents of the AIMBAT seismogram table to json.

Source code in src/aimbat/cli/seismogram.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
@app.command(name="dump")
@simple_exception
def cli_seismogram_dump(
    *,
    global_parameters: GlobalParameters | None = None,
) -> None:
    """Dump the contents of the AIMBAT seismogram table to json."""
    from aimbat.db import engine
    from aimbat.core import dump_seismogram_table
    from sqlmodel import Session

    global_parameters = global_parameters or GlobalParameters()

    with Session(engine) as session:
        dump_seismogram_table(session)

cli_seismogram_get 

cli_seismogram_get(
    seismogram_id: Annotated[
        UUID | str, Parameter(name=id)
    ],
    name: SeismogramParameter,
    *,
    global_parameters: GlobalParameters | None = None
) -> None

Get the value of a processing parameter.

Parameters:

Name Type Description Default
seismogram_id Annotated[UUID | str, Parameter(name=id)]

Seismogram ID number.

 required
name SeismogramParameter

Name of the seismogram parameter.

 required
Source code in src/aimbat/cli/seismogram.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
@app.command(name="get")
@simple_exception
def cli_seismogram_get(
    seismogram_id: Annotated[uuid.UUID | str, Parameter(name="id")],
    name: SeismogramParameter,
    *,
    global_parameters: GlobalParameters | None = None,
) -> None:
    """Get the value of a processing parameter.

    Args:
        seismogram_id: Seismogram ID number.
        name: Name of the seismogram parameter.
    """
    from aimbat.utils import string_to_uuid
    from aimbat.db import engine
    from aimbat.models import AimbatSeismogram
    from aimbat.core import get_seismogram_parameter_by_id
    from sqlmodel import Session

    global_parameters = global_parameters or GlobalParameters()

    with Session(engine) as session:
        if not isinstance(seismogram_id, uuid.UUID):
            seismogram_id = string_to_uuid(session, seismogram_id, AimbatSeismogram)
        print(get_seismogram_parameter_by_id(session, seismogram_id, name))

cli_seismogram_list 

cli_seismogram_list(
    *,
    all_events: Annotated[bool, Parameter(all)] = False,
    table_parameters: TableParameters | None = None,
    global_parameters: GlobalParameters | None = None
) -> None

Print information on the seismograms in the active event.

Parameters:

Name Type Description Default
all_events Annotated[bool, Parameter(all)]

Select seismograms for all events.

 False
Source code in src/aimbat/cli/seismogram.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
@app.command(name="list")
@simple_exception
def cli_seismogram_list(
    *,
    all_events: Annotated[bool, Parameter("all")] = False,
    table_parameters: TableParameters | None = None,
    global_parameters: GlobalParameters | None = None,
) -> None:
    """Print information on the seismograms in the active event.

    Args:
        all_events: Select seismograms for all events.
    """
    from aimbat.db import engine
    from aimbat.core import print_seismogram_table
    from sqlmodel import Session

    table_parameters = table_parameters or TableParameters()
    global_parameters = global_parameters or GlobalParameters()

    with Session(engine) as session:
        print_seismogram_table(session, table_parameters.short, all_events)

cli_seismogram_plot 

cli_seismogram_plot(
    *, global_parameters: GlobalParameters | None = None
) -> None

Plot seismograms for the active event.

Source code in src/aimbat/cli/seismogram.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
@app.command(name="plot")
@simple_exception
def cli_seismogram_plot(*, global_parameters: GlobalParameters | None = None) -> None:
    """Plot seismograms for the active event."""
    from aimbat.db import engine
    from aimbat.core import plot_all_seismograms
    from sqlmodel import Session
    import pyqtgraph as pg  # type: ignore

    global_parameters = global_parameters or GlobalParameters()

    use_qt = global_parameters.use_qt

    if use_qt:
        pg.mkQApp()

    with Session(engine) as session:
        plot_all_seismograms(session, use_qt)

    if use_qt:
        pg.exec()

cli_seismogram_set 

cli_seismogram_set(
    seismogram_id: Annotated[
        UUID | str, Parameter(name=id)
    ],
    name: SeismogramParameter,
    value: str,
    *,
    global_parameters: GlobalParameters | None = None
) -> None

Set value of a processing parameter.

Parameters:

Name Type Description Default
seismogram_id Annotated[UUID | str, Parameter(name=id)]

Seismogram ID number.

 required
name SeismogramParameter

Name of the seismogram parameter.

 required
value str

Value of the seismogram parameter.

 required
Source code in src/aimbat/cli/seismogram.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
@app.command(name="set")
@simple_exception
def cli_seismogram_set(
    seismogram_id: Annotated[uuid.UUID | str, Parameter(name="id")],
    name: SeismogramParameter,
    value: str,
    *,
    global_parameters: GlobalParameters | None = None,
) -> None:
    """Set value of a processing parameter.

    Args:
        seismogram_id: Seismogram ID number.
        name: Name of the seismogram parameter.
        value: Value of the seismogram parameter.
    """
    from aimbat.utils import string_to_uuid
    from aimbat.db import engine
    from aimbat.models import AimbatSeismogram
    from aimbat.core import set_seismogram_parameter_by_id
    from sqlmodel import Session

    global_parameters = global_parameters or GlobalParameters()

    with Session(engine) as session:
        if not isinstance(seismogram_id, uuid.UUID):
            seismogram_id = string_to_uuid(session, seismogram_id, AimbatSeismogram)
        set_seismogram_parameter_by_id(session, seismogram_id, name, value)

snapshot

View and manage snapshots.

Functions:

Name Description
cli_snapshop_delete

Delete existing snapshot.
cli_snapshot_create

Create new snapshot.
cli_snapshot_list

Print information on the snapshots for the active event.
cli_snapshot_rollback

Rollback to snapshot.

cli_snapshop_delete 

cli_snapshop_delete(
    snapshot_id: Annotated[UUID | str, Parameter(name=id)],
    *,
    global_parameters: GlobalParameters | None = None
) -> None

Delete existing snapshot.

Parameters:

Name Type Description Default
snapshot_id Annotated[UUID | str, Parameter(name=id)]

Snapshot ID Number.

 required
Source code in src/aimbat/cli/snapshot.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
@app.command(name="delete")
@simple_exception
def cli_snapshop_delete(
    snapshot_id: Annotated[uuid.UUID | str, Parameter(name="id")],
    *,
    global_parameters: GlobalParameters | None = None,
) -> None:
    """Delete existing snapshot.

    Args:
        snapshot_id: Snapshot ID Number.
    """
    from aimbat.db import engine
    from aimbat.utils import string_to_uuid
    from aimbat.models import AimbatSnapshot
    from aimbat.core import delete_snapshot_by_id
    from sqlmodel import Session

    global_parameters = global_parameters or GlobalParameters()

    with Session(engine) as session:
        if not isinstance(snapshot_id, uuid.UUID):
            snapshot_id = string_to_uuid(session, snapshot_id, AimbatSnapshot)
        delete_snapshot_by_id(session, snapshot_id)

cli_snapshot_create 

cli_snapshot_create(
    comment: str | None = None,
    *,
    global_parameters: GlobalParameters | None = None
) -> None

Create new snapshot.

Parameters:

Name Type Description Default
comment str | None

Create snapshot with optional comment.

 None
Source code in src/aimbat/cli/snapshot.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
@app.command(name="create")
@simple_exception
def cli_snapshot_create(
    comment: str | None = None, *, global_parameters: GlobalParameters | None = None
) -> None:
    """Create new snapshot.

    Args:
        comment: Create snapshot with optional comment.
    """
    from aimbat.db import engine
    from aimbat.core import create_snapshot
    from sqlmodel import Session

    global_parameters = global_parameters or GlobalParameters()

    with Session(engine) as session:
        create_snapshot(session, comment)

cli_snapshot_list 

cli_snapshot_list(
    *,
    all_events: Annotated[bool, Parameter(all)] = False,
    table_parameters: TableParameters | None = None,
    global_parameters: GlobalParameters | None = None
) -> None

Print information on the snapshots for the active event.

Parameters:

Name Type Description Default
all_events Annotated[bool, Parameter(all)]

Select snapshots for all events.

 False
Source code in src/aimbat/cli/snapshot.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
@app.command(name="list")
@simple_exception
def cli_snapshot_list(
    *,
    all_events: Annotated[bool, Parameter("all")] = False,
    table_parameters: TableParameters | None = None,
    global_parameters: GlobalParameters | None = None,
) -> None:
    """Print information on the snapshots for the active event.

    Args:
        all_events: Select snapshots for all events.
    """
    from aimbat.db import engine
    from aimbat.core import print_snapshot_table
    from sqlmodel import Session

    table_parameters = table_parameters or TableParameters()
    global_parameters = global_parameters or GlobalParameters()

    with Session(engine) as session:
        print_snapshot_table(session, table_parameters.short, all_events)

cli_snapshot_rollback 

cli_snapshot_rollback(
    snapshot_id: Annotated[UUID | str, Parameter(name=id)],
    *,
    global_paramaters: GlobalParameters | None = None
) -> None

Rollback to snapshot.

Parameters:

Name Type Description Default
snapshot_id Annotated[UUID | str, Parameter(name=id)]

Snapshot ID Number.

 required
Source code in src/aimbat/cli/snapshot.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
@app.command(name="rollback")
@simple_exception
def cli_snapshot_rollback(
    snapshot_id: Annotated[uuid.UUID | str, Parameter(name="id")],
    *,
    global_paramaters: GlobalParameters | None = None,
) -> None:
    """Rollback to snapshot.

    Args:
        snapshot_id: Snapshot ID Number.
    """
    from aimbat.utils import string_to_uuid
    from aimbat.db import engine
    from aimbat.models import AimbatSnapshot
    from aimbat.core import rollback_to_snapshot_by_id
    from sqlmodel import Session

    global_paramaters = global_paramaters or GlobalParameters()

    with Session(engine) as session:
        if not isinstance(snapshot_id, uuid.UUID):
            snapshot_id = string_to_uuid(session, snapshot_id, AimbatSnapshot)
        rollback_to_snapshot_by_id(session, snapshot_id)

station

View and manage stations.

Functions:

Name Description
cli_station_delete

Delete existing station.
cli_station_dump

Dump the contents of the AIMBAT station table to json.
cli_station_list

Print information on the stations used in the active event.

cli_station_delete 

cli_station_delete(
    station_id: Annotated[UUID | str, Parameter(name=id)],
    *,
    global_parameters: GlobalParameters | None = None
) -> None

Delete existing station.

Parameters:

Name Type Description Default
station_id Annotated[UUID | str, Parameter(name=id)]

Station ID.

 required
Source code in src/aimbat/cli/station.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
@app.command(name="delete")
@simple_exception
def cli_station_delete(
    station_id: Annotated[uuid.UUID | str, Parameter(name="id")],
    *,
    global_parameters: GlobalParameters | None = None,
) -> None:
    """Delete existing station.

    Args:
        station_id: Station ID.
    """
    from aimbat.db import engine
    from aimbat.utils import string_to_uuid
    from aimbat.core import delete_station_by_id
    from aimbat.models import AimbatStation
    from sqlmodel import Session

    global_parameters = global_parameters or GlobalParameters()

    with Session(engine) as session:
        if not isinstance(station_id, uuid.UUID):
            station_id = string_to_uuid(session, station_id, AimbatStation)
        delete_station_by_id(session, station_id)

cli_station_dump 

cli_station_dump(
    *, global_parameters: GlobalParameters | None = None
) -> None

Dump the contents of the AIMBAT station table to json.

Source code in src/aimbat/cli/station.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
@app.command(name="dump")
@simple_exception
def cli_station_dump(
    *,
    global_parameters: GlobalParameters | None = None,
) -> None:
    """Dump the contents of the AIMBAT station table to json."""

    from aimbat.db import engine
    from aimbat.core import dump_station_table
    from sqlmodel import Session

    global_parameters = global_parameters or GlobalParameters()

    with Session(engine) as session:
        dump_station_table(session)

cli_station_list 

cli_station_list(
    *,
    all_events: Annotated[
        bool, Parameter(name=all)
    ] = False,
    table_parameters: TableParameters | None = None,
    global_parameters: GlobalParameters | None = None
) -> None

Print information on the stations used in the active event.

Parameters:

Name Type Description Default
all_events Annotated[bool, Parameter(name=all)]

Select stations for all events.

 False
Source code in src/aimbat/cli/station.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
@app.command(name="list")
@simple_exception
def cli_station_list(
    *,
    all_events: Annotated[bool, Parameter(name="all")] = False,
    table_parameters: TableParameters | None = None,
    global_parameters: GlobalParameters | None = None,
) -> None:
    """Print information on the stations used in the active event.

    Args:
        all_events: Select stations for all events.
    """
    from aimbat.db import engine
    from aimbat.core import print_station_table
    from sqlmodel import Session

    table_parameters = table_parameters or TableParameters()
    global_parameters = global_parameters or GlobalParameters()

    with Session(engine) as session:
        print_station_table(session, table_parameters.short, all_events)

utils

Utilities for AIMBAT.

Modules:

Name Description
app

Utilities for AIMBAT.
sampledata

Download or delete AIMBAT sample data.

app

Utilities for AIMBAT.

The utils subcommand contains useful tools that are not strictly part of an AIMBAT workflow.

Functions:

Name Description
cli_checkdata

Check if there are any problems with SAC files before adding them to a project.
cli_checkdata 
cli_checkdata(
    sacfiles: Annotated[
        list[Path],
        Parameter(name=data, consume_multiple=True),
    ],
    *,
    common: GlobalParameters | None = None
) -> None

Check if there are any problems with SAC files before adding them to a project.

Parameters:

Name Type Description Default
sacfiles Annotated[list[Path], Parameter(name=data, consume_multiple=True)]

One or more SAC files.

 required
Source code in src/aimbat/cli/utils/app.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
@app.command(name="checkdata")
def cli_checkdata(
    sacfiles: Annotated[list[Path], Parameter(name="data", consume_multiple=True)],
    *,
    common: GlobalParameters | None = None,
) -> None:
    """Check if there are any problems with SAC files before adding them to a project.

    Args:
        sacfiles: One or more SAC files.
    """

    common = common or GlobalParameters()

    _run_checks(sacfiles)

sampledata

Download or delete AIMBAT sample data.

The sampledata subcommand manages an example dataset that can be used for testing or learning how to use AIMBAT.

The sample data source url can be viewed or changed via aimbat default <list/set> sampledata_src. Likewise the sample data destination folder be viewed or changed via aimbat default <list/set> sampledata_dir.

Functions:

Name Description
sampledata_cli_delete

Recursively delete sample data directory.
sampledata_cli_download

Download AIMBAT sample data.
sampledata_cli_delete 
sampledata_cli_delete(
    *, global_parameters: GlobalParameters | None = None
) -> None

Recursively delete sample data directory.

Source code in src/aimbat/cli/utils/sampledata.py 
37
38
39
40
41
42
43
44
45
@app.command(name="delete")
@simple_exception
def sampledata_cli_delete(*, global_parameters: GlobalParameters | None = None) -> None:
    """Recursively delete sample data directory."""
    from aimbat.utils import delete_sampledata

    global_parameters = global_parameters or GlobalParameters()

    delete_sampledata()
sampledata_cli_download 
sampledata_cli_download(
    *,
    force: bool = False,
    global_parameters: GlobalParameters | None = None
) -> None

Download AIMBAT sample data.

Downloads an example dataset to the directory specified in the sampledata_dir AIMBAT default variable.

Parameters:

Name Type Description Default
force bool

Delete the download directory and re-download."

 False
Source code in src/aimbat/cli/utils/sampledata.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
@app.command(name="download")
@simple_exception
def sampledata_cli_download(
    *, force: bool = False, global_parameters: GlobalParameters | None = None
) -> None:
    """Download AIMBAT sample data.

    Downloads an example dataset to the directory specified in the
    `sampledata_dir` AIMBAT default variable.

    Args:
        force: Delete the download directory and re-download."
    """
    from aimbat.utils import download_sampledata

    global_parameters = global_parameters or GlobalParameters()

    download_sampledata(force)

core

Functions:

Name Description
add_files_to_project

Add files to the AIMBAT database.
create_iccs_instance

Create an ICCS instance for the active event.
create_project

Create a new AIMBAT project.
create_snapshot

Create a snapshot of the AIMBAT processing parameters.
delete_event

Delete an AimbatEvent from the database.
delete_event_by_id

Delete an AimbatEvent from the database by ID.
delete_project

Delete the AIMBAT project.
delete_seismogram

Delete an AimbatSeismogram from the database.
delete_seismogram_by_id

Delete an AimbatSeismogram from the database by ID.
delete_snapshot

Delete an AIMBAT parameter snapshot.
delete_snapshot_by_id

Delete an AIMBAT parameter snapshot.
delete_station

Delete an AimbatStation from the database.
delete_station_by_id

Delete an AimbatStation from the database by ID.
dump_data_table

Dump the table data to json.
dump_event_table

Dump the table data to json.
dump_seismogram_table

Dump the table data to json.
dump_station_table

Dump the table data to json.
get_active_event

Return the currently active event (i.e. the one being processed).
get_completed_events

Get the events marked as completed.
get_data_for_active_event

Returns the AimbatFiles belonging to the active event.
get_event_parameter

Get event parameter value for the active event.
get_events_using_station

Get all events that use a particular station.
get_seismogram_parameter

Get parameter value from an AimbatSeismogram instance.
get_seismogram_parameter_by_id

Get parameter value from an AimbatSeismogram by ID.
get_selected_seismograms

Get the selected seismograms for the active avent.
get_snapshots

Get the snapshots for the active avent.
get_stations_in_event

Get the stations for a particular event.
plot_all_seismograms

Plot all seismograms for a particular event ordered by great circle distance.
plot_seismograms

Plot the ICCS seismograms as an image.
plot_stack

Plot the ICCS stack.
print_data_table

Print a pretty table with AIMBAT data.
print_event_table

Print a pretty table with AIMBAT events.
print_project_info

Show AIMBAT project information.
print_seismogram_table

Prints a pretty table with AIMBAT seismograms.
print_snapshot_table

Print a pretty table with AIMBAT snapshots.
print_station_table

Prints a pretty table with AIMBAT stations.
rollback_to_snapshot

Rollback to an AIMBAT parameters snapshot.
rollback_to_snapshot_by_id

Rollback to an AIMBAT parameters snapshot.
run_iccs

Run ICCS algorithm.
set_active_event

Set the active event (i.e. the one being processed).
set_active_event_by_id

Set the currently selected event (i.e. the one being processed) by its ID.
set_event_parameter

Set event parameter value for the active event.
set_seismogram_parameter

Set parameter value for an AimbatSeismogram instance.
set_seismogram_parameter_by_id

Set parameter value for an AimbatSeismogram by ID.
update_min_ccnorm

Update the minimum cross correlation coefficient for the active event.
update_pick

Update the pick for the active event.
update_timewindow

Update the time window for the active event.

add_files_to_project 

add_files_to_project(
    session: Session,
    datasources: Sequence[str | PathLike],
    datatype: DataType,
    disable_progress_bar: bool = True,
) -> None

Add files to the AIMBAT database.

Parameters:

Name Type Description Default
datasources Sequence[str | PathLike]

List of data sources to add.

 required
datatype DataType

Type of data.

 required
disable_progress_bar bool

Do not display progress bar.

 True
Source code in src/aimbat/core/_data.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
def add_files_to_project(
    session: Session,
    datasources: Sequence[str | os.PathLike],
    datatype: DataType,
    disable_progress_bar: bool = True,
) -> None:
    """Add files to the AIMBAT database.

    Args:
        datasources: List of data sources to add.
        datatype: Type of data.
        disable_progress_bar: Do not display progress bar.
    """

    logger.info(f"Adding {len(datasources)} {datatype} files to project.")

    for datasource in track(
        sequence=datasources,
        description="Adding files ...",
        disable=disable_progress_bar,
    ):
        aimbat_station = _create_station(session, datasource, datatype)
        aimbat_event = _create_event(session, datasource, datatype)
        aimbat_seismogram = _create_seismogram(session, datasource, datatype)

        # TODO: perhaps adding potentially updated station and event information should be optional?
        aimbat_seismogram.station = aimbat_station
        aimbat_seismogram.event = aimbat_event

        # Create AimbatDataSource instance with relationship to AimbatSeismogram
        select_aimbat_data_source = select(AimbatDataSource).where(
            AimbatDataSource.sourcename == str(datasource)
        )
        aimbat_data_source = session.exec(select_aimbat_data_source).one_or_none()
        if aimbat_data_source is None:
            logger.debug(f"Adding data source {datasource} to project.")
            aimbat_data_source_create = AimbatDataSourceCreate(
                sourcename=str(datasource), datatype=datatype
            )
            aimbat_data_source = AimbatDataSource.model_validate(
                aimbat_data_source_create, update={"seismogram": aimbat_seismogram}
            )

        else:
            logger.debug(
                f"Using existing data source {datasource} instead of adding new one."
            )
            aimbat_data_source.seismogram = aimbat_seismogram
        session.add(aimbat_data_source)

    session.commit()

create_iccs_instance 

create_iccs_instance(session: Session) -> ICCS

Create an ICCS instance for the active event.

Parameters:

Name Type Description Default
session Session

Database session.

 required

Returns:

Type Description
ICCS

ICCS instance.
Source code in src/aimbat/core/_iccs.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def create_iccs_instance(session: Session) -> ICCS:
    """Create an ICCS instance for the active event.

    Args:
        session: Database session.

    Returns:
        ICCS instance.
    """

    logger.info("Creating ICCS instance for active event.")

    active_event = get_active_event(session)

    return ICCS(
        seismograms=active_event.seismograms,
        window_pre=active_event.parameters.window_pre,
        window_post=active_event.parameters.window_post,
        bandpass_apply=active_event.parameters.bandpass_apply,
        bandpass_fmin=active_event.parameters.bandpass_fmin,
        bandpass_fmax=active_event.parameters.bandpass_fmax,
        min_ccnorm=active_event.parameters.min_ccnorm,
        context_width=settings.context_width,
    )

create_project 

create_project(engine: Engine) -> None

Create a new AIMBAT project.

Source code in src/aimbat/core/_project.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def create_project(engine: Engine) -> None:
    """Create a new AIMBAT project."""

    # import this to create tables below
    import aimbat.models  # noqa: F401

    logger.info(f"Creating new project in {engine=}.")

    if _project_exists(engine):
        raise RuntimeError(
            f"Unable to create a new project: project already exists in {engine=}!"
        )

    logger.debug("Creating database tables and loading defaults.")

    SQLModel.metadata.create_all(engine)
    if engine.driver == "pysqlite":
        with engine.connect() as connection:
            connection.execute(text("PRAGMA foreign_keys=ON"))  # for SQLite only

    # This trigger ensures that only one event can be active at a time
    with engine.connect() as connection:
        connection.execute(text("""CREATE TRIGGER single_active_event
        BEFORE UPDATE ON aimbatevent
        FOR EACH ROW
        WHEN NEW.active = TRUE
        BEGIN
            UPDATE aimbatevent SET active = NULL
        WHERE active = TRUE AND id != NEW.id;
        END;
    """))

create_snapshot 

create_snapshot(
    session: Session, comment: str | None = None
) -> None

Create a snapshot of the AIMBAT processing parameters.

Parameters:

Name Type Description Default
session Session

Database session.

 required
comment str | None

Optional comment.

 None
Source code in src/aimbat/core/_snapshot.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def create_snapshot(session: Session, comment: str | None = None) -> None:
    """Create a snapshot of the AIMBAT processing parameters.

    Args:
        session: Database session.
        comment: Optional comment.
    """
    active_aimbat_event = get_active_event(session)

    logger.info(
        f"Creating snapshot for event with id={active_aimbat_event.id} with {comment=}."
    )

    event_parameters_snapshot = AimbatEventParametersSnapshot.model_validate(
        active_aimbat_event.parameters,
        update={
            "id": uuid.uuid4(),  # we don't want to carry over the id from the active parameters
            "parameters_id": active_aimbat_event.parameters.id,
        },
    )
    logger.debug(
        f"Adding event parameters snapshot with id={event_parameters_snapshot.id} to snapshot."
    )

    seismogram_parameter_snapshots = []
    for aimbat_seismogram in active_aimbat_event.seismograms:
        seismogram_parameter_snapshot = AimbatSeismogramParametersSnapshot.model_validate(
            aimbat_seismogram.parameters,
            update={
                "id": uuid.uuid4(),  # we don't want to carry over the id from the active parameters
                "seismogram_parameters_id": aimbat_seismogram.parameters.id,
            },
        )
        logger.debug(
            f"Adding seismogram parameters snapshot with id={seismogram_parameter_snapshot.id} to snapshot."
        )
        seismogram_parameter_snapshots.append(seismogram_parameter_snapshot)

    aimbat_snapshot = AimbatSnapshot(
        event=active_aimbat_event,
        event_parameters_snapshot=event_parameters_snapshot,
        seismogram_parameters_snapshots=seismogram_parameter_snapshots,
        comment=comment,
    )
    session.add(aimbat_snapshot)
    session.commit()

delete_event 

delete_event(session: Session, event: AimbatEvent) -> None

Delete an AimbatEvent from the database.

Parameters:

Name Type Description Default
session Session

Database session.

 required
event AimbatEvent

Event to delete.

 required
Source code in src/aimbat/core/_event.py 
71
72
73
74
75
76
77
78
79
80
81
82
def delete_event(session: Session, event: AimbatEvent) -> None:
    """Delete an AimbatEvent from the database.

    Args:
        session: Database session.
        event: Event to delete.
    """

    logger.info(f"Deleting event {event.id}.")

    session.delete(event)
    session.commit()

delete_event_by_id 

delete_event_by_id(
    session: Session, event_id: UUID
) -> None

Delete an AimbatEvent from the database by ID.

Parameters:

Name Type Description Default
session Session

Database session.

 required
event_id UUID

Event ID.

 required

Raises:

Type Description
NoResultFound

If no AimbatEvent is found with the given ID.
Source code in src/aimbat/core/_event.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def delete_event_by_id(session: Session, event_id: UUID) -> None:
    """Delete an AimbatEvent from the database by ID.

    Args:
        session: Database session.
        event_id: Event ID.

    Raises:
        NoResultFound: If no AimbatEvent is found with the given ID.
    """

    logger.debug(f"Getting event with id={event_id}.")

    event = session.get(AimbatEvent, event_id)
    if event is None:
        raise NoResultFound(
            f"Unable to find event using id: {event_id}. {HINTS.LIST_EVENTS}"
        )
    delete_event(session, event)

delete_project 

delete_project(engine: Engine) -> None

Delete the AIMBAT project.

Raises:

Type Description
RuntimeError

If unable to delete project.
Source code in src/aimbat/core/_project.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def delete_project(engine: Engine) -> None:
    """Delete the AIMBAT project.

    Raises:
        RuntimeError: If unable to delete project.
    """

    logger.info(f"Deleting project in {engine=}.")

    if _project_exists(engine):
        if engine.driver == "pysqlite":
            database = engine.url.database
            engine.dispose()
            if database == ":memory:":
                logger.info("Running database in memory, nothing to delete.")
                return
            elif database:
                project_path = Path(database)
                logger.info(f"Deleting project file: {project_path=}")
                project_path.unlink()
                return
    raise RuntimeError("Unable to find/delete project.")

delete_seismogram 

delete_seismogram(
    session: Session, seismogram: AimbatSeismogram
) -> None

Delete an AimbatSeismogram from the database.

Parameters:

Name Type Description Default
session Session

Database session.

 required
seismogram AimbatSeismogram

Seismogram to delete.

 required
Source code in src/aimbat/core/_seismogram.py 
70
71
72
73
74
75
76
77
78
79
80
81
def delete_seismogram(session: Session, seismogram: AimbatSeismogram) -> None:
    """Delete an AimbatSeismogram from the database.

    Args:
        session: Database session.
        seismogram: Seismogram to delete.
    """

    logger.info(f"Deleting seismogram {seismogram.id}.")

    session.delete(seismogram)
    session.commit()

delete_seismogram_by_id 

delete_seismogram_by_id(
    session: Session, seismogram_id: UUID
) -> None

Delete an AimbatSeismogram from the database by ID.

Parameters:

Name Type Description Default
session Session

Database session.

 required
seismogram_id UUID

Seismogram ID.

 required

Raises:

Type Description
NoResultFound

If no AimbatSeismogram is found with the given ID.
Source code in src/aimbat/core/_seismogram.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def delete_seismogram_by_id(session: Session, seismogram_id: uuid.UUID) -> None:
    """Delete an AimbatSeismogram from the database by ID.

    Args:
        session: Database session.
        seismogram_id: Seismogram ID.

    Raises:
        NoResultFound: If no AimbatSeismogram is found with the given ID.
    """

    logger.debug(f"Getting seismogram with id={seismogram_id}.")

    seismogram = session.get(AimbatSeismogram, seismogram_id)
    if seismogram is None:
        raise NoResultFound(f"No AimbatSeismogram found with {seismogram_id=}")
    delete_seismogram(session, seismogram)

delete_snapshot 

delete_snapshot(
    session: Session, snapshot: AimbatSnapshot
) -> None

Delete an AIMBAT parameter snapshot.

Parameters:

Name Type Description Default
session Session

Database session.

 required
snapshot AimbatSnapshot

Snapshot.

 required
Source code in src/aimbat/core/_snapshot.py 
156
157
158
159
160
161
162
163
164
165
166
167
def delete_snapshot(session: Session, snapshot: AimbatSnapshot) -> None:
    """Delete an AIMBAT parameter snapshot.

    Args:
        session: Database session.
        snapshot: Snapshot.
    """

    logger.info(f"Deleting snapshot {snapshot.id}.")

    session.delete(snapshot)
    session.commit()

delete_snapshot_by_id 

delete_snapshot_by_id(
    session: Session, snapshot_id: UUID
) -> None

Delete an AIMBAT parameter snapshot.

Parameters:

Name Type Description Default
session Session

Database session.

 required
snapshot_id UUID

Snapshot id.

 required
Source code in src/aimbat/core/_snapshot.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def delete_snapshot_by_id(session: Session, snapshot_id: uuid.UUID) -> None:
    """Delete an AIMBAT parameter snapshot.

    Args:
        session: Database session.
        snapshot_id: Snapshot id.
    """

    logger.debug(f"Searching for snapshot with id {snapshot_id}.")

    snapshot = session.get(AimbatSnapshot, snapshot_id)

    if snapshot is None:
        raise ValueError(
            f"Unable to delete snapshot: snapshot with id={snapshot_id} not found."
        )

    delete_snapshot(session, snapshot)

delete_station 

delete_station(
    session: Session, station: AimbatStation
) -> None

Delete an AimbatStation from the database.

Parameters:

Name Type Description Default
session Session

Database session.

 required
station AimbatStation

Station to delete.

 required
Source code in src/aimbat/core/_station.py 
44
45
46
47
48
49
50
51
52
53
54
55
def delete_station(session: Session, station: AimbatStation) -> None:
    """Delete an AimbatStation from the database.

    Args:
        session: Database session.
        station: Station to delete.
    """

    logger.info(f"Deleting station {station.id}.")

    session.delete(station)
    session.commit()

delete_station_by_id 

delete_station_by_id(
    session: Session, station_id: UUID
) -> None

Delete an AimbatStation from the database by ID.

Parameters:

Name Type Description Default
session Session

Database session.

 required
station_id UUID

Station ID.

 required

Raises:

Type Description
NoResultFound

If no AimbatStation is found with the given ID.
Source code in src/aimbat/core/_station.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def delete_station_by_id(session: Session, station_id: uuid.UUID) -> None:
    """Delete an AimbatStation from the database by ID.

    Args:
        session: Database session.
        station_id: Station ID.

    Raises:
        NoResultFound: If no AimbatStation is found with the given ID.
    """

    logger.debug(f"Getting station with id={station_id}.")

    station = session.get(AimbatStation, station_id)
    if station is None:
        raise NoResultFound(f"No AimbatStation found with {station_id=}")
    delete_station(session, station)

dump_data_table 

dump_data_table(session: Session) -> None

Dump the table data to json.

Source code in src/aimbat/core/_data.py 
244
245
246
247
248
249
250
def dump_data_table(session: Session) -> None:
    """Dump the table data to json."""

    logger.info("Dumping AIMBAT datasources table to json.")

    aimbat_data_sources = session.exec(select(AimbatDataSource)).all()
    dump_to_json(aimbat_data_sources)

dump_event_table 

dump_event_table(session: Session) -> None

Dump the table data to json.

Source code in src/aimbat/core/_event.py 
298
299
300
301
302
303
304
def dump_event_table(session: Session) -> None:
    """Dump the table data to json."""

    logger.info("Dumping AIMBAT event table to json.")

    aimbat_events = session.exec(select(AimbatEvent)).all()
    dump_to_json(aimbat_events)

dump_seismogram_table 

dump_seismogram_table(session: Session) -> None

Dump the table data to json.

Source code in src/aimbat/core/_seismogram.py 
349
350
351
352
353
354
355
def dump_seismogram_table(session: Session) -> None:
    """Dump the table data to json."""

    logger.info("Dumping AIMBAT seismogram table to json.")

    aimbat_seismograms = session.exec(select(AimbatSeismogram)).all()
    dump_to_json(aimbat_seismograms)

dump_station_table 

dump_station_table(session: Session) -> None

Dump the table data to json.

Source code in src/aimbat/core/_station.py 
177
178
179
180
181
182
183
def dump_station_table(session: Session) -> None:
    """Dump the table data to json."""

    logger.info("Dumping AIMBAT station table to json.")

    aimbat_stations = session.exec(select(AimbatStation)).all()
    dump_to_json(aimbat_stations)

get_active_event 

get_active_event(session: Session) -> AimbatEvent

Return the currently active event (i.e. the one being processed).

Parameters:

Name Type Description Default
session Session

SQL session.

 required

Returns:

Type Description
AimbatEvent

Active Event

Raises NoResultFound: When no event is active.

Source code in src/aimbat/utils/_active_event.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def get_active_event(session: Session) -> AimbatEvent:
    """
    Return the currently active event (i.e. the one being processed).

    Args:
        session: SQL session.

    Returns:
        Active Event

    Raises
        NoResultFound: When no event is active.
    """

    logger.debug("Attempting to determine active event.")

    select_active_event = select(AimbatEvent).where(AimbatEvent.active == 1)

    # NOTE: While there technically can be no active event in the database,
    # we typically don't really want to go beyond this point when that is the
    # case. Hence we call `one` rather than `one_or_none`.
    try:
        active_event = session.exec(select_active_event).one()
    except NoResultFound:
        raise NoResultFound(f"No active event found. {HINTS.ACTIVATE_EVENT}")

    logger.debug(f"Active event: {active_event.id}")

    return active_event

get_completed_events 

get_completed_events(
    session: Session,
) -> Sequence[AimbatEvent]

Get the events marked as completed.

Parameters:

Name Type Description Default
session Session

SQL session.

 required
Source code in src/aimbat/core/_event.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def get_completed_events(session: Session) -> Sequence[AimbatEvent]:
    """Get the events marked as completed.

    Args:
        session: SQL session.
    """

    select_completed_events = (
        select(AimbatEvent)
        .join(AimbatEventParameters)
        .where(AimbatEventParameters.completed == 1)
    )

    return session.exec(select_completed_events).all()

get_data_for_active_event 

get_data_for_active_event(
    session: Session,
) -> Sequence[AimbatDataSource]

Returns the AimbatFiles belonging to the active event.

Parameters:

Name Type Description Default
session Session

Database session.

 required

Returns:

Type Description
Sequence[AimbatDataSource]

List of AimbatFiles.
Source code in src/aimbat/core/_data.py 
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
def get_data_for_active_event(session: Session) -> Sequence[AimbatDataSource]:
    """Returns the AimbatFiles belonging to the active event.

    Args:
        session: Database session.

    Returns:
        List of AimbatFiles.
    """

    logger.info("Getting aimbatfiles in active event.")

    select_files = (
        select(AimbatDataSource)
        .join(AimbatSeismogram)
        .join(AimbatEvent)
        .where(AimbatEvent.active == 1)
    )
    return session.exec(select_files).all()

get_event_parameter 

get_event_parameter(
    session: Session, name: EventParameter
) -> Timedelta | bool | float

Get event parameter value for the active event.

Parameters:

Name Type Description Default
session Session

Database session.

 required
name EventParameter

Name of the parameter.

 required
Source code in src/aimbat/core/_event.py 
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
def get_event_parameter(
    session: Session, name: EventParameter
) -> Timedelta | bool | float:
    """Get event parameter value for the active event.

    Args:
        session: Database session.
        name: Name of the parameter.
    """

    active_event = get_active_event(session)

    logger.info(f"Getting {name=} value for {active_event=}.")

    return getattr(active_event.parameters, name)

get_events_using_station 

get_events_using_station(
    session: Session, station: AimbatStation
) -> Sequence[AimbatEvent]

Get all events that use a particular station.

Parameters:

Name Type Description Default
session Session

Database session.

 required
station AimbatStation

Station to return events for.

 required

Returns: Events that use the station.

Source code in src/aimbat/core/_event.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def get_events_using_station(
    session: Session, station: AimbatStation
) -> Sequence[AimbatEvent]:
    """Get all events that use a particular station.

    Args:
        session: Database session.
        station: Station to return events for.

    Returns: Events that use the station.
    """

    logger.info(f"Getting events for station: {station.id}.")

    select_events = (
        select(AimbatEvent)
        .join(AimbatSeismogram)
        .join(AimbatStation)
        .where(AimbatStation.id == station.id)
    )

    events = session.exec(select_events).all()

    logger.debug(f"Found {len(events)}.")

    return events

get_seismogram_parameter 

get_seismogram_parameter(
    seismogram: AimbatSeismogram, name: SeismogramParameter
) -> bool | Timestamp

Get parameter value from an AimbatSeismogram instance.

Parameters:

Name Type Description Default
seismogram AimbatSeismogram

Seismogram.

 required
name SeismogramParameter

Name of the parameter value to return.

 required

Returns:

Type Description
bool | Timestamp

Seismogram parameter value.
Source code in src/aimbat/core/_seismogram.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def get_seismogram_parameter(
    seismogram: AimbatSeismogram, name: SeismogramParameter
) -> bool | Timestamp:
    """Get parameter value from an AimbatSeismogram instance.

    Args:
        seismogram: Seismogram.
        name: Name of the parameter value to return.

    Returns:
        Seismogram parameter value.
    """

    logger.info(f"Getting seismogram parameter {name=} value for {seismogram=}.")

    return getattr(seismogram.parameters, name)

get_seismogram_parameter_by_id 

get_seismogram_parameter_by_id(
    session: Session,
    seismogram_id: UUID,
    name: SeismogramParameter,
) -> bool | Timestamp

Get parameter value from an AimbatSeismogram by ID.

Parameters:

Name Type Description Default
session Session

Database session.

 required
seismogram_id UUID

Seismogram ID.

 required
name SeismogramParameter

Name of the parameter value to return.

 required

Returns:

Type Description
bool | Timestamp

Seismogram parameter value.

Raises:

Type Description
ValueError

If no AimbatSeismogram is found with the given ID.
Source code in src/aimbat/core/_seismogram.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def get_seismogram_parameter_by_id(
    session: Session, seismogram_id: uuid.UUID, name: SeismogramParameter
) -> bool | Timestamp:
    """Get parameter value from an AimbatSeismogram by ID.

    Args:
        session: Database session.
        seismogram_id: Seismogram ID.
        name: Name of the parameter value to return.

    Returns:
        Seismogram parameter value.

    Raises:
        ValueError: If no AimbatSeismogram is found with the given ID.
    """

    logger.info(f"Getting seismogram {name=} for seismogram with id={seismogram_id}.")

    aimbat_seismogram = session.get(AimbatSeismogram, seismogram_id)

    if aimbat_seismogram is None:
        raise ValueError(f"No AimbatSeismogram found with {seismogram_id=}")

    return get_seismogram_parameter(aimbat_seismogram, name)

get_selected_seismograms 

get_selected_seismograms(
    session: Session, all_events: bool = False
) -> Sequence[AimbatSeismogram]

Get the selected seismograms for the active avent.

Parameters:

Name Type Description Default
session Session

Database session.

 required
all_events bool

Get the selected seismograms for all events.

 False

Returns: Selected seismograms.

Source code in src/aimbat/core/_seismogram.py 
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
def get_selected_seismograms(
    session: Session, all_events: bool = False
) -> Sequence[AimbatSeismogram]:
    """Get the selected seismograms for the active avent.

    Args:
        session: Database session.
        all_events: Get the selected seismograms for all events.

    Returns: Selected seismograms.
    """

    logger.info("Getting selected AIMBAT seismograms.")

    if all_events is True:
        logger.debug("Selecting seismograms for all events.")
        select_seismograms = (
            select(AimbatSeismogram)
            .join(AimbatSeismogramParameters)
            .where(AimbatSeismogramParameters.select == 1)
        )
    else:
        logger.debug("Selecting seismograms for active event only.")
        select_seismograms = (
            select(AimbatSeismogram)
            .join(AimbatSeismogramParameters)
            .join(AimbatEvent)
            .where(AimbatSeismogramParameters.select == 1)
            .where(AimbatEvent.active == 1)
        )

    seismograms = session.exec(select_seismograms).all()

    logger.debug(f"Found {len(seismograms)} selected seismograms.")

    return seismograms

get_snapshots 

get_snapshots(
    session: Session, all_events: bool = False
) -> Sequence[AimbatSnapshot]

Get the snapshots for the active avent.

Parameters:

Name Type Description Default
session Session

Database session.

 required
all_events bool

Get the selected snapshots for all events.

 False

Returns: Snapshots.

Source code in src/aimbat/core/_snapshot.py 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
def get_snapshots(
    session: Session, all_events: bool = False
) -> Sequence[AimbatSnapshot]:
    """Get the snapshots for the active avent.

    Args:
        session: Database session.
        all_events: Get the selected snapshots for all events.

    Returns: Snapshots.
    """

    logger.info("Getting AIMBAT snapshots.")

    if all_events:
        logger.debug("Getting snapshots for all events.")
        return session.exec(select(AimbatSnapshot)).all()

    logger.debug("Getting snapshots for active event.")
    select_active_event_snapshots = (
        select(AimbatSnapshot)
        .join(AimbatEventParametersSnapshot)
        .join(AimbatEventParameters)
        .join(AimbatEvent)
        .where(AimbatEvent.active == 1)
    )
    return session.exec(select_active_event_snapshots).all()

get_stations_in_event 

get_stations_in_event(
    session: Session, event: AimbatEvent
) -> Sequence[AimbatStation]

Get the stations for a particular event.

Parameters:

Name Type Description Default
session Session

Database session.

 required
event AimbatEvent

Event to return stations for.

 required

Returns: Stations in event.

Source code in src/aimbat/core/_station.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def get_stations_in_event(
    session: Session, event: AimbatEvent
) -> Sequence[AimbatStation]:
    """Get the stations for a particular event.

    Args:
        session: Database session.
        event: Event to return stations for.

    Returns: Stations in event.
    """

    logger.info(f"Getting stations for event: {event.id}.")

    select_stations = (
        select(AimbatStation)
        .join(AimbatSeismogram)
        .join(AimbatEvent)
        .where(AimbatEvent.id == event.id)
    )

    stations = session.exec(select_stations).all()

    logger.debug(f"Found {len(stations)}.")

    return stations

plot_all_seismograms 

plot_all_seismograms(
    session: Session, use_qt: bool = False
) -> Figure

Plot all seismograms for a particular event ordered by great circle distance.

Parameters:

Name Type Description Default
use_qt bool

Plot with pqtgraph instead of pyplot

 False
Source code in src/aimbat/core/_seismogram.py 
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
def plot_all_seismograms(session: Session, use_qt: bool = False) -> Figure:
    """Plot all seismograms for a particular event ordered by great circle distance.

    Args:
        use_qt: Plot with pqtgraph instead of pyplot
    """

    active_event = get_active_event(session)

    if active_event is None:
        raise RuntimeError("No active event set.")

    seismograms = active_event.seismograms

    if len(seismograms) == 0:
        raise RuntimeError("No seismograms found in active event.")

    distance_dict = {
        seismogram.id: distance(seismogram.station, seismogram.event) / 1000
        for seismogram in seismograms
    }
    distance_min = min(distance_dict.values())
    distance_max = max(distance_dict.values())
    scaling_factor = (distance_max - distance_min) / len(seismograms) * 5

    title = seismograms[0].event.time.strftime("Event %Y-%m-%d %H:%M:%S")
    xlabel = "Time of day"
    ylabel = "Epicentral distance [km]"

    plot_widget = None
    if use_qt:
        plot_widget = pg.plot(title=title)
        axis = pg.DateAxisItem()
        plot_widget.setAxisItems({"bottom": axis})
        plot_widget.setLabel("bottom", xlabel)
        plot_widget.setLabel("left", ylabel)
    else:
        fig, ax = plt.subplots()

    for seismogram in seismograms:
        clone = clone_to_mini(MiniSeismogram, seismogram)
        detrend(clone)
        normalize(clone)
        plot_data = clone.data * scaling_factor + distance_dict[seismogram.id]
        if use_qt and plot_widget is not None:
            times = unix_time_array(clone)
            plot_widget.plot(times, plot_data)
        else:
            times = time_array(clone)
            ax.plot(
                times,
                plot_data,
                scalex=True,
                scaley=True,
            )
    if not use_qt:
        plt.xlabel(xlabel=xlabel)
        plt.ylabel(ylabel=ylabel)
        plt.gcf().autofmt_xdate()
        fmt = mdates.DateFormatter("%H:%M:%S")
        plt.gca().xaxis.set_major_formatter(fmt)
        plt.title(title)
        plt.show()
    return fig

plot_seismograms 

plot_seismograms(
    iccs: ICCS, context: bool, all: bool
) -> None

Plot the ICCS seismograms as an image.

Parameters:

Name Type Description Default
iccs ICCS

ICCS instance.

 required
context bool

Whether to use seismograms with extra context.

 required
all bool

Whether to plot all seismograms.

 required
Source code in src/aimbat/core/_iccs.py 
83
84
85
86
87
88
89
90
91
92
93
94
def plot_seismograms(iccs: ICCS, context: bool, all: bool) -> None:
    """Plot the ICCS seismograms as an image.

    Args:
        iccs: ICCS instance.
        context: Whether to use seismograms with extra context.
        all: Whether to plot all seismograms.
    """

    logger.info("Plotting ICCS seismograms for active event.")

    _plot_seismograms(iccs, context, all)

plot_stack 

plot_stack(iccs: ICCS, context: bool, all: bool) -> None

Plot the ICCS stack.

Parameters:

Name Type Description Default
iccs ICCS

ICCS instance.

 required
context bool

Whether to use seismograms with extra context.

 required
all bool

Whether to plot all seismograms.

 required
Source code in src/aimbat/core/_iccs.py 
70
71
72
73
74
75
76
77
78
79
80
def plot_stack(iccs: ICCS, context: bool, all: bool) -> None:
    """Plot the ICCS stack.

    Args:
        iccs: ICCS instance.
        context: Whether to use seismograms with extra context.
        all: Whether to plot all seismograms.
    """

    logger.info("Plotting ICCS stack for active event.")
    _plot_stack(iccs, context, all)

print_data_table 

print_data_table(
    session: Session, short: bool, all_events: bool = False
) -> None

Print a pretty table with AIMBAT data.

Parameters:

Name Type Description Default
short bool

Shorten UUIDs and format data.

 required
all_events bool

Print all files instead of limiting to the active event.

 False
Source code in src/aimbat/core/_data.py 
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
def print_data_table(session: Session, short: bool, all_events: bool = False) -> None:
    """Print a pretty table with AIMBAT data.

    Args:
        short: Shorten UUIDs and format data.
        all_events: Print all files instead of limiting to the active event.
    """

    logger.info("Printing AIMBAT data table.")

    if all_events:
        aimbat_data_sources = session.exec(select(AimbatDataSource)).all()
        title = "AIMBAT data for all events"
    else:
        active_event = get_active_event(session)
        aimbat_data_sources = get_data_for_active_event(session)
        time = (
            active_event.time.strftime("%Y-%m-%d %H:%M:%S")
            if short
            else active_event.time
        )
        id = uuid_shortener(session, active_event) if short else active_event.id
        title = f"AIMBAT data for event {time} (ID={id})"

    logger.debug(f"Found {len(aimbat_data_sources)} files in total.")

    rows = [
        [
            uuid_shortener(session, a) if short else str(a.id),
            str(a.datatype),
            str(a.sourcename),
            (uuid_shortener(session, a.seismogram) if short else str(a.seismogram.id)),
        ]
        for a in aimbat_data_sources
    ]

    table = make_table(title=title)

    table.add_column(
        "ID (shortened)" if short else "ID",
        justify="center",
        style=TABLE_STYLING.id,
        no_wrap=True,
    )
    table.add_column("Datatype", justify="center", style=TABLE_STYLING.mine)
    table.add_column("Filename", justify="left", style=TABLE_STYLING.mine, no_wrap=True)
    table.add_column(
        "Seismogram ID", justify="center", style=TABLE_STYLING.linked, no_wrap=True
    )

    for row in rows:
        table.add_row(*row)

    console = Console()
    console.print(table)

print_event_table 

print_event_table(
    session: Session, short: bool = True
) -> None

Print a pretty table with AIMBAT events.

Parameters:

Name Type Description Default
short bool

Shorten and format the output to be more human-readable.

 True
Source code in src/aimbat/core/_event.py 
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
def print_event_table(session: Session, short: bool = True) -> None:
    """Print a pretty table with AIMBAT events.

    Args:
        short: Shorten and format the output to be more human-readable.
    """

    logger.info("Printing AIMBAT events table.")

    table = make_table(title="AIMBAT Events")
    table.add_column(
        "ID (shortened)" if short else "ID",
        justify="center",
        style=TABLE_STYLING.id,
        no_wrap=True,
    )
    table.add_column("Active", justify="center", style=TABLE_STYLING.mine, no_wrap=True)
    table.add_column(
        "Date & Time", justify="center", style=TABLE_STYLING.mine, no_wrap=True
    )
    table.add_column("Latitude", justify="center", style=TABLE_STYLING.mine)
    table.add_column("Longitude", justify="center", style=TABLE_STYLING.mine)
    table.add_column("Depth", justify="center", style=TABLE_STYLING.mine)
    table.add_column("Completed", justify="center", style=TABLE_STYLING.parameters)
    table.add_column("# Seismograms", justify="center", style=TABLE_STYLING.linked)
    table.add_column("# Stations", justify="center", style=TABLE_STYLING.linked)

    for event in session.exec(select(AimbatEvent)).all():
        logger.debug(f"Adding event with id={event.id} to the table.")
        table.add_row(
            uuid_shortener(session, event) if short else str(event.id),
            TABLE_STYLING.bool_formatter(event.active),
            TABLE_STYLING.timestamp_formatter(event.time, short),
            f"{event.latitude:.3f}" if short else str(event.latitude),
            f"{event.longitude:.3f}" if short else str(event.longitude),
            f"{event.depth:.0f}" if short else str(event.depth),
            TABLE_STYLING.bool_formatter(event.parameters.completed),
            str(len(event.seismograms)),
            str(len(station.get_stations_in_event(session, event))),
        )

    console = Console()
    console.print(table)

print_project_info 

print_project_info(engine: Engine) -> None

Show AIMBAT project information.

Raises:

Type Description
RuntimeError

If no project found.
Source code in src/aimbat/core/_project.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
def print_project_info(engine: Engine) -> None:
    """Show AIMBAT project information.

    Raises:
        RuntimeError: If no project found.
    """

    logger.info("Printing project info.")

    if not _project_exists(engine):
        raise RuntimeError(
            'No AIMBAT project found. Try running "aimbat project create" first.'
        )

    with Session(engine) as session:
        grid = Table.grid(expand=False)
        grid.add_column()
        grid.add_column(justify="left")
        if engine.driver == "pysqlite":
            project = str(engine.url.database)
            grid.add_row("AIMBAT Project File: ", project)

        events = len(session.exec(select(AimbatEvent)).all())
        completed_events = len(event.get_completed_events(session))
        stations = len(session.exec(select(AimbatStation)).all())
        seismograms = len(session.exec(select(AimbatSeismogram)).all())
        selected_seismograms = len(
            seismogram.get_selected_seismograms(session, all_events=True)
        )

        grid.add_row(
            "Number of Events (total/completed): ",
            f"({events}/{completed_events})",
        )

        try:
            active_event = get_active_event(session)
            active_event_id = active_event.id
            active_stations = len(station.get_stations_in_event(session, active_event))
            seismograms_in_event = len(active_event.seismograms)
            selected_seismograms_in_event = len(
                seismogram.get_selected_seismograms(session)
            )
        except NoResultFound:
            active_event_id = None
            active_stations = None
            seismograms_in_event = None
            selected_seismograms_in_event = None
        grid.add_row("Active Event ID: ", f"{active_event_id}")
        grid.add_row(
            "Number of Stations in Project (total/active event): ",
            f"({stations}/{active_stations})",
        )

        grid.add_row(
            "Number of Seismograms in Project (total/selected): ",
            f"({seismograms}/{selected_seismograms})",
        )
        grid.add_row(
            "Number of Seismograms in Active Event (total/selected): ",
            f"({seismograms_in_event}/{selected_seismograms_in_event})",
        )

        console = Console()
        console.print(
            Panel(grid, title="Project Info", title_align="left", border_style="dim")
        )

print_seismogram_table 

print_seismogram_table(
    session: Session, short: bool, all_events: bool = False
) -> None

Prints a pretty table with AIMBAT seismograms.

Parameters:

Name Type Description Default
short bool

Shorten and format the output to be more human-readable.

 required
all_events bool

Print seismograms for all events.

 False
Source code in src/aimbat/core/_seismogram.py 
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
def print_seismogram_table(
    session: Session, short: bool, all_events: bool = False
) -> None:
    """Prints a pretty table with AIMBAT seismograms.

    Args:
        short: Shorten and format the output to be more human-readable.
        all_events: Print seismograms for all events.
    """

    logger.info("Printing AIMBAT seismogram table.")

    title = "AIMBAT seismograms for all events"
    seismograms = None

    if all_events:
        logger.debug("Selecting seismograms for all events.")
        seismograms = session.exec(select(AimbatSeismogram)).all()
    else:
        logger.debug("Selecting seismograms for active event only.")
        active_event = get_active_event(session)
        seismograms = active_event.seismograms
        if short:
            title = f"AIMBAT seismograms for event {active_event.time.strftime('%Y-%m-%d %H:%M:%S')} (ID={event.uuid_shortener(session, active_event)})"
        else:
            title = f"AIMBAT seismograms for event {active_event.time} (ID={active_event.id})"

    logger.debug(f"Found {len(seismograms)} seismograms for the table.")

    table = make_table(title=title)
    table.add_column(
        "ID (shortened)" if short else "ID",
        justify="center",
        style=TABLE_STYLING.id,
        no_wrap=True,
    )
    table.add_column(
        "Selected", justify="center", style=TABLE_STYLING.mine, no_wrap=True
    )
    table.add_column("NPTS", justify="center", style=TABLE_STYLING.mine, no_wrap=True)
    table.add_column("Delta", justify="center", style=TABLE_STYLING.mine, no_wrap=True)
    table.add_column(
        "Data ID", justify="center", style=TABLE_STYLING.linked, no_wrap=True
    )
    table.add_column("Station ID", justify="center", style=TABLE_STYLING.linked)
    table.add_column("Station Name", justify="center", style=TABLE_STYLING.linked)
    if all_events:
        table.add_column("Event ID", justify="center", style=TABLE_STYLING.linked)

    for seismogram in seismograms:
        logger.debug(f"Adding seismogram with ID {seismogram.id} to the table.")
        row = [
            (uuid_shortener(session, seismogram) if short else str(seismogram.id)),
            TABLE_STYLING.bool_formatter(seismogram.parameters.select),
            str(len(seismogram)),
            str(seismogram.delta.total_seconds()),
            (
                uuid_shortener(session, seismogram.datasource)
                if short
                else str(seismogram.datasource.id)
            ),
            (
                uuid_shortener(session, seismogram.station)
                if short
                else str(seismogram.station.id)
            ),
            f"{seismogram.station.name} - {seismogram.station.network}",
        ]

        if all_events:
            row.append(
                uuid_shortener(session, seismogram.event)
                if short
                else str(seismogram.event.id)
            )
        table.add_row(*row)

    console = Console()
    console.print(table)

print_snapshot_table 

print_snapshot_table(
    session: Session, short: bool, all_events: bool
) -> None

Print a pretty table with AIMBAT snapshots.

Parameters:

Name Type Description Default
short bool

Shorten and format the output to be more human-readable.

 required
all_events bool

Print all snapshots instead of limiting to the active event.

 required
Source code in src/aimbat/core/_snapshot.py 
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
def print_snapshot_table(session: Session, short: bool, all_events: bool) -> None:
    """Print a pretty table with AIMBAT snapshots.

    Args:
        short: Shorten and format the output to be more human-readable.
        all_events: Print all snapshots instead of limiting to the active event.
    """

    logger.info("Printing AIMBAT snapshots table.")

    title = "AIMBAT snapshots for all events"

    snapshots = get_snapshots(session, all_events)
    logger.debug(f"Found {len(snapshots)} snapshots for the table.")

    if not all_events:
        active_event = get_active_event(session)
        if short:
            title = f"AIMBAT snapshots for event {active_event.time.strftime('%Y-%m-%d %H:%M:%S')} (ID={uuid_shortener(session, active_event)})"
        else:
            title = (
                f"AIMBAT snapshots for event {active_event.time} (ID={active_event.id})"
            )

    table = make_table(title=title)

    table.add_column(
        "ID (shortened)" if short else "ID",
        justify="center",
        style=TABLE_STYLING.id,
        no_wrap=True,
    )
    table.add_column(
        "Date & Time", justify="center", style=TABLE_STYLING.mine, no_wrap=True
    )
    table.add_column("Comment", justify="center", style=TABLE_STYLING.mine)
    table.add_column("# Seismograms", justify="center", style=TABLE_STYLING.linked)
    if all_events:
        table.add_column("Event ID", justify="center", style=TABLE_STYLING.linked)

    for snapshot in snapshots:
        logger.debug(f"Adding snapshot with id={snapshot.id} to the table.")
        row = [
            (uuid_shortener(session, snapshot) if short else str(snapshot.id)),
            TABLE_STYLING.timestamp_formatter(snapshot.date, short),
            str(snapshot.comment),
            str(len(snapshot.seismogram_parameters_snapshots)),
        ]
        if all_events:
            aimbat_event = snapshot.event
            row.append(
                uuid_shortener(session, aimbat_event) if short else str(aimbat_event.id)
            )
        table.add_row(*row)

    console = Console()
    console.print(table)

print_station_table 

print_station_table(
    session: Session, short: bool, all_events: bool = False
) -> None

Prints a pretty table with AIMBAT stations.

Parameters:

Name Type Description Default
session Session

Database session.

 required
short bool

Shorten and format the output to be more human-readable.

 required
all_events bool

Print stations for all events.

 False
Source code in src/aimbat/core/_station.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def print_station_table(
    session: Session, short: bool, all_events: bool = False
) -> None:
    """Prints a pretty table with AIMBAT stations.

    Args:
        session: Database session.
        short: Shorten and format the output to be more human-readable.
        all_events: Print stations for all events.
    """

    logger.info("Printing station table.")

    title = "AIMBAT stations for all events"
    aimbat_stations = None

    if all_events:
        logger.debug("Selecting all AIMBAT stations.")
        aimbat_stations = session.exec(select(AimbatStation)).all()
    else:
        logger.debug("Selecting AIMBAT stations for active event.")
        active_event = get_active_event(session)
        aimbat_stations = get_stations_in_event(session, active_event)
        if short:
            title = f"AIMBAT stations for event {active_event.time.strftime('%Y-%m-%d %H:%M:%S')} (ID={uuid_shortener(session, active_event)})"
        else:
            title = (
                f"AIMBAT stations for event {active_event.time} (ID={active_event.id})"
            )
    logger.debug("Found {len(aimbat_stations)} stations for the table.")

    table = make_table(title=title)

    table.add_column(
        "ID (shortened)" if short else "ID",
        justify="center",
        style=TABLE_STYLING.id,
        no_wrap=True,
    )
    table.add_column(
        "Name & Network", justify="center", style=TABLE_STYLING.mine, no_wrap=True
    )
    table.add_column("Channel", justify="center", style=TABLE_STYLING.mine)
    table.add_column("Location", justify="center", style=TABLE_STYLING.mine)
    table.add_column("Latitude", justify="center", style=TABLE_STYLING.mine)
    table.add_column("Longitude", justify="center", style=TABLE_STYLING.mine)
    table.add_column("Elevation", justify="center", style=TABLE_STYLING.mine)
    if all_events:
        table.add_column("# Seismograms", justify="center", style=TABLE_STYLING.linked)
        table.add_column("# Events", justify="center", style=TABLE_STYLING.linked)

    for aimbat_station in aimbat_stations:
        logger.debug(f"Adding {aimbat_station.name} to the table.")
        row = [
            (
                uuid_shortener(session, aimbat_station)
                if short
                else str(aimbat_station.id)
            ),
            f"{aimbat_station.name} - {aimbat_station.network}",
            f"{aimbat_station.channel}",
            f"{aimbat_station.location}",
            (
                f"{aimbat_station.latitude:.3f}"
                if short
                else str(aimbat_station.latitude)
            ),
            (
                f"{aimbat_station.longitude:.3f}"
                if short
                else str(aimbat_station.longitude)
            ),
            (
                f"{aimbat_station.elevation:.0f}"
                if short
                else str(aimbat_station.elevation)
            ),
        ]
        if all_events:
            row.extend(
                [
                    str(len(aimbat_station.seismograms)),
                    str(len({i.event_id for i in aimbat_station.seismograms})),
                ]
            )
        table.add_row(*row)

    console = Console()
    console.print(table)

rollback_to_snapshot 

rollback_to_snapshot(
    session: Session, snapshot: AimbatSnapshot
) -> None

Rollback to an AIMBAT parameters snapshot.

Parameters:

Name Type Description Default
snapshot AimbatSnapshot

Snapshot.

 required
Source code in src/aimbat/core/_snapshot.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def rollback_to_snapshot(session: Session, snapshot: AimbatSnapshot) -> None:
    """Rollback to an AIMBAT parameters snapshot.

    Args:
        snapshot: Snapshot.
    """

    logger.info(f"Rolling back to snapshot with id={snapshot.id}.")

    # create object with just the parameters
    rollback_event_parameters = AimbatEventParametersBase.model_validate(
        snapshot.event_parameters_snapshot
    )
    logger.debug(
        f"Using event parameters snapshot with id={snapshot.event_parameters_snapshot.id} for rollback."
    )
    current_event_parameters = snapshot.event.parameters
    # setting attributes explicitly brings them into the session
    for k, v in rollback_event_parameters.model_dump().items():
        setattr(current_event_parameters, k, v)

    session.add(current_event_parameters)

    for seismogram_parameters_snapshot in snapshot.seismogram_parameters_snapshots:
        # create object with just the parameters
        rollback_seismogram_parameters = AimbatSeismogramParametersBase.model_validate(
            seismogram_parameters_snapshot
        )
        logger.debug(
            f"Using seismogram parameters snapshot with id={seismogram_parameters_snapshot.id} for rollback."
        )
        # setting attributes explicitly brings them into the session
        current_seismogram_parameters = seismogram_parameters_snapshot.parameters
        for parameter, value in rollback_seismogram_parameters.model_dump().items():
            setattr(current_seismogram_parameters, parameter, value)
        session.add(current_seismogram_parameters)

    session.commit()

rollback_to_snapshot_by_id 

rollback_to_snapshot_by_id(
    session: Session, snapshot_id: UUID
) -> None

Rollback to an AIMBAT parameters snapshot.

Parameters:

Name Type Description Default
session Session

Database session.

 required
snapshot_id UUID

Snapshot id.

 required
Source code in src/aimbat/core/_snapshot.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def rollback_to_snapshot_by_id(session: Session, snapshot_id: uuid.UUID) -> None:
    """Rollback to an AIMBAT parameters snapshot.

    Args:
        session: Database session.
        snapshot_id: Snapshot id.
    """

    logger.info(f"Deleting snapshot with id={snapshot_id}.")

    snapshot = session.get(AimbatSnapshot, snapshot_id)

    if snapshot is None:
        raise ValueError(
            f"Unable to delete snapshot: snapshot with id={snapshot_id} not found."
        )

    rollback_to_snapshot(session, snapshot)

run_iccs 

run_iccs(
    session: Session,
    iccs: ICCS,
    autoflip: bool,
    autoselect: bool,
) -> None

Run ICCS algorithm.

Parameters:

Name Type Description Default
session Session

Database session.

 required
iccs ICCS

ICCS instance.

 required
autoflip bool

Whether to automatically flip seismograms.

 required
autoselect bool

Whether to automatically select seismograms.

 required
Source code in src/aimbat/core/_iccs.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def run_iccs(session: Session, iccs: ICCS, autoflip: bool, autoselect: bool) -> None:
    """Run ICCS algorithm.

    Args:
        session: Database session.
        iccs: ICCS instance.
        autoflip: Whether to automatically flip seismograms.
        autoselect: Whether to automatically select seismograms.
    """

    logger.info(f"Running ICCS with {autoflip=}, {autoselect=}.")

    results = iccs(autoflip=autoflip, autoselect=autoselect)
    logger.info(f"ICCS {results = }")
    session.commit()

set_active_event 

set_active_event(
    session: Session, event: AimbatEvent
) -> None

Set the active event (i.e. the one being processed).

Parameters:

Name Type Description Default
session Session

SQL session.

 required
event AimbatEvent

AIMBAT Event to set as active.

 required
Source code in src/aimbat/core/_event.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def set_active_event(session: Session, event: AimbatEvent) -> None:
    """
    Set the active event (i.e. the one being processed).

    Args:
        session: SQL session.
        event: AIMBAT Event to set as active.
    """

    logger.info(f"Activating {event=}")

    event.active = True
    session.add(event)
    session.commit()

set_active_event_by_id 

set_active_event_by_id(
    session: Session, event_id: UUID
) -> None

Set the currently selected event (i.e. the one being processed) by its ID.

Parameters:

Name Type Description Default
session Session

SQL session.

 required
event_id UUID

ID of AIMBAT Event to set as active one.

 required

Raises:

Type Description
ValueError

If no event with the given ID is found.
Source code in src/aimbat/core/_event.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def set_active_event_by_id(session: Session, event_id: UUID) -> None:
    """
    Set the currently selected event (i.e. the one being processed) by its ID.

    Args:
        session: SQL session.
        event_id: ID of AIMBAT Event to set as active one.

    Raises:
        ValueError: If no event with the given ID is found.
    """
    logger.info(f"Setting active event to event with id={event_id}.")

    if event_id not in session.exec(select(AimbatEvent.id)).all():
        raise ValueError(
            f"No AimbatEvent found with id: {event_id}. {HINTS.LIST_EVENTS}"
        )

    aimbat_event = session.exec(
        select(AimbatEvent).where(AimbatEvent.id == event_id)
    ).one()
    set_active_event(session, aimbat_event)

set_event_parameter 

set_event_parameter(
    session: Session,
    name: EventParameter,
    value: Timedelta | bool | float | str,
) -> None

Set event parameter value for the active event.

Parameters:

Name Type Description Default
session Session

Database session.

 required
name EventParameter

Name of the parameter.

 required
value Timedelta | bool | float | str

Value to set.

 required
Source code in src/aimbat/core/_event.py 
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
def set_event_parameter(
    session: Session, name: EventParameter, value: Timedelta | bool | float | str
) -> None:
    """Set event parameter value for the active event.

    Args:
        session: Database session.
        name: Name of the parameter.
        value: Value to set.
    """

    active_event = get_active_event(session)

    logger.info(f"Setting {name=} to {value} for {active_event=}.")

    parameters = AimbatEventParametersBase.model_validate(
        active_event.parameters, update={name: value}
    )
    setattr(active_event.parameters, name, getattr(parameters, name))
    session.add(active_event)
    session.commit()

set_seismogram_parameter 

set_seismogram_parameter(
    session: Session,
    seismogram: AimbatSeismogram,
    name: SeismogramParameter,
    value: Timestamp | bool | str,
) -> None

Set parameter value for an AimbatSeismogram instance.

Parameters:

Name Type Description Default
session Session

Database session

 required
seismogram AimbatSeismogram

Seismogram to set parameter for.

 required
name SeismogramParameter

Name of the parameter.

 required
value Timestamp | bool | str

Value to set parameter to.

 required
Source code in src/aimbat/core/_seismogram.py 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
def set_seismogram_parameter(
    session: Session,
    seismogram: AimbatSeismogram,
    name: SeismogramParameter,
    value: Timestamp | bool | str,
) -> None:
    """Set parameter value for an AimbatSeismogram instance.

    Args:
        session: Database session
        seismogram: Seismogram to set parameter for.
        name: Name of the parameter.
        value: Value to set parameter to.

    """

    logger.info(f"Setting seismogram {name=} to {value=} in {seismogram=}.")

    parameters = AimbatSeismogramParametersBase.model_validate(
        seismogram.parameters, update={name: value}
    )
    setattr(seismogram.parameters, name, getattr(parameters, name))
    session.add(seismogram)
    session.commit()

set_seismogram_parameter_by_id 

set_seismogram_parameter_by_id(
    session: Session,
    seismogram_id: UUID,
    name: SeismogramParameter,
    value: Timestamp | bool | str,
) -> None

Set parameter value for an AimbatSeismogram by ID.

Parameters:

Name Type Description Default
session Session

Database session

 required
seismogram_id UUID

Seismogram id.

 required
name SeismogramParameter

Name of the parameter.

 required
value Timestamp | bool | str

Value to set.

 required

Raises:

Type Description
ValueError

If no AimbatSeismogram is found with the given ID.
Source code in src/aimbat/core/_seismogram.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def set_seismogram_parameter_by_id(
    session: Session,
    seismogram_id: uuid.UUID,
    name: SeismogramParameter,
    value: Timestamp | bool | str,
) -> None:
    """Set parameter value for an AimbatSeismogram by ID.

    Args:
        session: Database session
        seismogram_id: Seismogram id.
        name: Name of the parameter.
        value: Value to set.

    Raises:
        ValueError: If no AimbatSeismogram is found with the given ID.
    """

    logger.info(
        f"Setting seismogram {name=} to {value=} for seismogram with id={seismogram_id}."
    )

    aimbat_seismogram = session.get(AimbatSeismogram, seismogram_id)

    if aimbat_seismogram is None:
        raise ValueError(f"No AimbatSeismogram found with {seismogram_id=}")

    set_seismogram_parameter(session, aimbat_seismogram, name, value)

update_min_ccnorm 

update_min_ccnorm(
    session: Session, iccs: ICCS, context: bool, all: bool
) -> None

Update the minimum cross correlation coefficient for the active event.

Parameters:

Name Type Description Default
iccs ICCS

ICCS instance.

 required
context bool

Whether to use seismograms with extra context.

 required
all bool

Whether to plot all seismograms.

 required
Source code in src/aimbat/core/_iccs.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
def update_min_ccnorm(session: Session, iccs: ICCS, context: bool, all: bool) -> None:
    """Update the minimum cross correlation coefficient for the active event.

    Args:
        iccs: ICCS instance.
        context: Whether to use seismograms with extra context.
        all: Whether to plot all seismograms.
    """

    logger.info("Updating minimum cross correlation coefficient for active event.")

    logger.debug(f"Current {iccs.min_ccnorm = }.")
    _update_min_ccnorm(iccs, context, all)
    logger.debug(f"Updated {iccs.min_ccnorm = }.")

    active_event = get_active_event(session)
    active_event.parameters.min_ccnorm = float(iccs.min_ccnorm)
    session.commit()

update_pick 

update_pick(
    session: Session,
    iccs: ICCS,
    context: bool,
    all: bool,
    use_seismogram_image: bool,
) -> None

Update the pick for the active event.

Parameters:

Name Type Description Default
iccs ICCS

ICCS instance.

 required
context bool

Whether to use seismograms with extra context.

 required
all bool

Whether to plot all seismograms.

 required
use_seismogram_image bool

Whether to use the seismogram image to update pick.

 required
Source code in src/aimbat/core/_iccs.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def update_pick(
    session: Session, iccs: ICCS, context: bool, all: bool, use_seismogram_image: bool
) -> None:
    """Update the pick for the active event.

    Args:
        iccs: ICCS instance.
        context: Whether to use seismograms with extra context.
        all: Whether to plot all seismograms.
        use_seismogram_image: Whether to use the seismogram image to update pick.
    """

    logger.info("Updating pick for active event.")

    _update_pick(iccs, context, all, use_seismogram_image)
    session.commit()

update_timewindow 

update_timewindow(
    session: Session,
    iccs: ICCS,
    context: bool,
    all: bool,
    use_seismogram_image: bool,
) -> None

Update the time window for the active event.

Parameters:

Name Type Description Default
iccs ICCS

ICCS instance.

 required
context bool

Whether to use seismograms with extra context.

 required
all bool

Whether to plot all seismograms.

 required
use_seismogram_image bool

Whether to use the seismogram image to update pick.

 required
Source code in src/aimbat/core/_iccs.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
def update_timewindow(
    session: Session, iccs: ICCS, context: bool, all: bool, use_seismogram_image: bool
) -> None:
    """Update the time window for the active event.

    Args:
        iccs: ICCS instance.
        context: Whether to use seismograms with extra context.
        all: Whether to plot all seismograms.
        use_seismogram_image: Whether to use the seismogram image to update pick.
    """

    logger.info("Updating time window for active event.")

    logger.debug(f"Current {iccs.window_pre = }, {iccs.window_post = }.")
    _update_timewindow(iccs, context, all, use_seismogram_image)
    logger.debug(f"Updated {iccs.window_pre = }, {iccs.window_post = }.")

    active_event = get_active_event(session)
    active_event.parameters.window_pre = iccs.window_pre
    active_event.parameters.window_post = iccs.window_post
    session.commit()

db

Module to define the AIMBAT project file and create the database engine.

Attributes:

Name Type Description
engine

AIMBAT database engine.

engine module-attribute 

engine = create_engine(url=db_url, echo=False)

AIMBAT database engine.

io

Functions to read and write data files used with AIMBAT

Functions:

Name Description
create_event

Read event data from a data source and create an AimbatEvent.
create_seismogram

Read seismogram data from a data source and create an AimbatSeismogram.
create_station

Read station data from a data source and create an AimbatStation.
read_seismogram_data

Read seismogram data from a data source.
write_seismogram_data

Write seismogram data to a data source.

create_event 

create_event(
    datasource: str | PathLike, datatype: DataType
) -> AimbatEvent

Read event data from a data source and create an AimbatEvent.

Parameters:

Name Type Description Default
datasource str | PathLike

Name of the data source.

 required
datatype DataType

AIMBAT compatible datatype.

 required

Returns:

Type Description
AimbatEvent

AimbatEvent instance.

Raises:

Type Description
NotImplementedError

If the datatype is not supported.
Source code in src/aimbat/io/_base.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def create_event(datasource: str | PathLike, datatype: DataType) -> AimbatEvent:
    """Read event data from a data source and create an AimbatEvent.

    Args:
        datasource: Name of the data source.
        datatype: AIMBAT compatible datatype.

    Returns:
        AimbatEvent instance.

    Raises:
        NotImplementedError: If the datatype is not supported.
    """

    logger.debug(f"Creating AimbatEvent from {datasource}.")

    event_creator_fn = event_creator.get(datatype)
    if event_creator_fn is None:
        raise NotImplementedError(
            f"I don't know how to create an AimbatEvent from {datatype}."
        )
    return event_creator_fn(datasource)

create_seismogram 

create_seismogram(
    datasource: str | PathLike, datatype: DataType
) -> AimbatSeismogram

Read seismogram data from a data source and create an AimbatSeismogram.

Parameters:

Name Type Description Default
datasource str | PathLike

Name of the data source.

 required
datatype DataType

AIMBAT compatible datatype.

 required

Returns:

Type Description
AimbatSeismogram

AimbatSeismogram instance.

Raises:

Type Description
NotImplementedError

If the datatype is not supported.
Source code in src/aimbat/io/_base.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def create_seismogram(
    datasource: str | PathLike, datatype: DataType
) -> AimbatSeismogram:
    """Read seismogram data from a data source and create an AimbatSeismogram.

    Args:
        datasource: Name of the data source.
        datatype: AIMBAT compatible datatype.

    Returns:
        AimbatSeismogram instance.

    Raises:
        NotImplementedError: If the datatype is not supported.
    """

    logger.debug(f"Creating AimbatSeismogram from {datasource}.")

    station_creator_fn = seismogram_creator.get(datatype)
    if station_creator_fn is None:
        raise NotImplementedError(
            f"I don't know how to create an AimbatSeismgoram from {datatype}."
        )
    return station_creator_fn(datasource)

create_station 

create_station(
    datasource: str | PathLike, datatype: DataType
) -> AimbatStation

Read station data from a data source and create an AimbatStation.

Parameters:

Name Type Description Default
datasource str | PathLike

Name of the data source.

 required
datatype DataType

AIMBAT compatible datatype.

 required

Returns:

Type Description
AimbatStation

AimbatStation instance.

Raises:

Type Description
NotImplementedError

If the datatype is not supported.
Source code in src/aimbat/io/_base.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
def create_station(datasource: str | PathLike, datatype: DataType) -> AimbatStation:
    """Read station data from a data source and create an AimbatStation.

    Args:
        datasource: Name of the data source.
        datatype: AIMBAT compatible datatype.

    Returns:
        AimbatStation instance.

    Raises:
        NotImplementedError: If the datatype is not supported.
    """

    logger.debug(f"Creating AimbatStation from {datasource}.")

    station_creator_fn = station_creator.get(datatype)
    if station_creator_fn is None:
        raise NotImplementedError(
            f"I don't know how to create an AimbatStation from {datatype}."
        )
    return station_creator_fn(datasource)

read_seismogram_data 

read_seismogram_data(
    datasource: str | PathLike, datatype: DataType
) -> NDArray[float64]

Read seismogram data from a data source.

Parameters:

Name Type Description Default
datasource str | PathLike

Name of the data source.

 required
datatype DataType

AIMBAT compatible filetype.

 required

Returns:

Type Description
NDArray[float64]

Seismogram data.

Raises:

Type Description
NotImplementedError

If the datatype is not supported.
Source code in src/aimbat/io/_base.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def read_seismogram_data(
    datasource: str | PathLike, datatype: DataType
) -> npt.NDArray[np.float64]:
    """Read seismogram data from a data source.

    Args:
        datasource: Name of the data source.
        datatype: AIMBAT compatible filetype.

    Returns:
        Seismogram data.

    Raises:
        NotImplementedError: If the datatype is not supported.
    """

    logger.debug(f"Reading seismogram data from {datasource}.")

    data_reader_fn = seismogram_data_reader.get(datatype)
    if data_reader_fn is None:
        raise NotImplementedError(f"I don't know how to read data of type {datatype}.")
    return data_reader_fn(datasource)

write_seismogram_data 

write_seismogram_data(
    datasource: str | PathLike,
    datatype: DataType,
    data: NDArray[float64],
) -> None

Write seismogram data to a data source.

Parameters:

Name Type Description Default
datasource str | PathLike

Name of the data source.

 required
datatype DataType

AIMBAT compatible filetype.

 required
data NDArray[float64]

Seismogram data

 required

Raises:

Type Description
NotImplementedError

If the datatype is not supported.
Source code in src/aimbat/io/_base.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def write_seismogram_data(
    datasource: str | PathLike,
    datatype: DataType,
    data: npt.NDArray[np.float64],
) -> None:
    """Write seismogram data to a data source.

    Args:
        datasource: Name of the data source.
        datatype: AIMBAT compatible filetype.
        data: Seismogram data

    Raises:
        NotImplementedError: If the datatype is not supported.
    """

    logger.debug(f"Writing seismogram data to {datasource}.")

    data_writer_fn = seismogram_data_writer.get(datatype)
    if data_writer_fn is None:
        raise NotImplementedError(
            f"I don't know how to write data to file of type {datatype}"
        )
    data_writer_fn(datasource, data)

logger

Logging setup.

Functions:

Name Description
configure_logging

Reconfigure loguru sinks based on current settings.

configure_logging 

configure_logging() -> None

Reconfigure loguru sinks based on current settings.

Source code in src/aimbat/logger.py 
 7
 8
 9
10
def configure_logging() -> None:
    """Reconfigure loguru sinks based on current settings."""
    logger.remove()
    logger.add(settings.logfile, rotation="100 MB", level=settings.log_level)

models

Models used in AIMBAT.

Type Aliases:

Name Description
AimbatTypes

Union of all AIMBAT models that exist in the database.

Classes:

Name Description
AimbatDataSource

Class to store data source information.
AimbatDataSourceCreate

Class to store data source information.
AimbatEvent

Store event information.
AimbatEventParameters

Processing parameters common to all seismograms of a particular event.
AimbatEventParametersBase

Base class that defines the event parameters used in AIMBAT.
AimbatEventParametersSnapshot

Event parameter snapshot.
AimbatSeismogram

Class to store seismogram data
AimbatSeismogramParameters

Class to store ICCS processing parameters of a single seismogram.
AimbatSeismogramParametersBase

Base class that defines the seismogram parameters used in AIMBAT.
AimbatSeismogramParametersSnapshot

Class to store a snapshot of ICCS processing parameters of a single seismogram.
AimbatSnapshot

Class to store AIMBAT snapshots.
AimbatStation

Class to store station information.

AimbatTypes 

AimbatTypes = (
    AimbatDataSource
    | AimbatStation
    | AimbatEvent
    | AimbatEventParameters
    | AimbatSeismogram
    | AimbatSeismogramParameters
    | AimbatSnapshot
    | AimbatEventParametersSnapshot
    | AimbatSeismogramParametersSnapshot
)

Union of all AIMBAT models that exist in the database.

AimbatDataSource

Bases: SQLModel

Class to store data source information.

Source code in src/aimbat/models/_models.py 
49
50
51
52
53
54
55
56
57
58
class AimbatDataSource(SQLModel, table=True):
    """Class to store data source information."""

    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    sourcename: str
    datatype: DataType
    seismogram_id: uuid.UUID = Field(
        default=None, foreign_key="aimbatseismogram.id", ondelete="CASCADE"
    )
    seismogram: "AimbatSeismogram" = Relationship(back_populates="datasource")

AimbatDataSourceCreate

Bases: SQLModel

Class to store data source information.

Source code in src/aimbat/models/_models.py 
42
43
44
45
46
class AimbatDataSourceCreate(SQLModel):
    """Class to store data source information."""

    sourcename: str | os.PathLike = Field(unique=True)
    datatype: DataType = DataType.SAC

AimbatEvent

Bases: SQLModel

Store event information.

Attributes:

Name Type Description
active bool | None

Indicates if an event is the active event.
depth float | None

Event depth.
id UUID

Unique ID.
latitude float

Event latitude.
longitude float

Event longitude.
parameters AimbatEventParameters

Event parameters.
seismograms list[AimbatSeismogram]

List of seismograms of this event.
snapshots list[AimbatSnapshot]

List of snapshots.
time PydanticTimestamp

Event time.
Source code in src/aimbat/models/_models.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
class AimbatEvent(SQLModel, table=True):
    """Store event information."""

    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    "Unique ID."

    active: bool | None = Field(default=None, unique=True)
    "Indicates if an event is the active event."

    time: PydanticTimestamp = Field(
        unique=True, sa_type=SAPandasTimestamp, allow_mutation=False
    )
    "Event time."

    latitude: float
    "Event latitude."

    longitude: float
    "Event longitude."

    depth: float | None = None
    "Event depth."

    seismograms: list["AimbatSeismogram"] = Relationship(
        back_populates="event", cascade_delete=True
    )
    "List of seismograms of this event."

    parameters: "AimbatEventParameters" = Relationship(
        back_populates="event", cascade_delete=True
    )
    "Event parameters."

    snapshots: list["AimbatSnapshot"] = Relationship(
        back_populates="event", cascade_delete=True
    )
    "List of snapshots."

active class-attribute instance-attribute 

active: bool | None = Field(default=None, unique=True)

Indicates if an event is the active event.

depth class-attribute instance-attribute 

depth: float | None = None

Event depth.

id class-attribute instance-attribute 

id: UUID = Field(default_factory=uuid4, primary_key=True)

Unique ID.

latitude instance-attribute 

latitude: float

Event latitude.

longitude instance-attribute 

longitude: float

Event longitude.

parameters class-attribute instance-attribute 

parameters: AimbatEventParameters = Relationship(
    back_populates="event", cascade_delete=True
)

Event parameters.

seismograms class-attribute instance-attribute 

seismograms: list[AimbatSeismogram] = Relationship(
    back_populates="event", cascade_delete=True
)

List of seismograms of this event.

snapshots class-attribute instance-attribute 

snapshots: list[AimbatSnapshot] = Relationship(
    back_populates="event", cascade_delete=True
)

List of snapshots.

time class-attribute instance-attribute 

time: PydanticTimestamp = Field(
    unique=True,
    sa_type=SAPandasTimestamp,
    allow_mutation=False,
)

Event time.

AimbatEventParameters

Bases: AimbatEventParametersBase, EventParametersValidatorMixin

Processing parameters common to all seismograms of a particular event.

Attributes:

Name Type Description
bandpass_apply bool

Whether to apply bandpass filter to seismograms.
bandpass_fmax float

Maximum frequency for bandpass filter (ignored if bandpass_apply is False).
bandpass_fmin float

Minimum frequency for bandpass filter (ignored if bandpass_apply is False).
completed bool

Mark an event as completed.
event AimbatEvent

Event these parameters are associated with.
event_id UUID

Event ID these parameters are associated with.
id UUID

Unique ID.
min_ccnorm float

Minimum cross-correlation used when automatically de-selecting seismograms.
snapshots list[AimbatEventParametersSnapshot]

Snapshots these parameters are associated with.
window_post PydanticPositiveTimedelta

Post-pick window length.
window_pre PydanticNegativeTimedelta

Pre-pick window length.
Source code in src/aimbat/models/_models.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
class AimbatEventParameters(
    AimbatEventParametersBase, EventParametersValidatorMixin, table=True
):
    """Processing parameters common to all seismograms of a particular event."""

    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    "Unique ID."

    event_id: uuid.UUID = Field(
        default=None, foreign_key="aimbatevent.id", ondelete="CASCADE"
    )
    "Event ID these parameters are associated with."

    event: AimbatEvent = Relationship(back_populates="parameters")
    "Event these parameters are associated with."

    snapshots: list["AimbatEventParametersSnapshot"] = Relationship(
        back_populates="parameters", cascade_delete=True
    )
    "Snapshots these parameters are associated with."

bandpass_apply class-attribute instance-attribute 

bandpass_apply: bool = Field(
    default_factory=lambda: bandpass_apply
)

Whether to apply bandpass filter to seismograms.

bandpass_fmax class-attribute instance-attribute 

bandpass_fmax: float = Field(
    default_factory=lambda: bandpass_fmax, gt=0
)

Maximum frequency for bandpass filter (ignored if bandpass_apply is False).

bandpass_fmin class-attribute instance-attribute 

bandpass_fmin: float = Field(
    default_factory=lambda: bandpass_fmin, ge=0
)

Minimum frequency for bandpass filter (ignored if bandpass_apply is False).

completed class-attribute instance-attribute 

completed: bool = False

Mark an event as completed.

event class-attribute instance-attribute 

event: AimbatEvent = Relationship(
    back_populates="parameters"
)

Event these parameters are associated with.

event_id class-attribute instance-attribute 

event_id: UUID = Field(
    default=None,
    foreign_key="aimbatevent.id",
    ondelete="CASCADE",
)

Event ID these parameters are associated with.

id class-attribute instance-attribute 

id: UUID = Field(default_factory=uuid4, primary_key=True)

Unique ID.

min_ccnorm class-attribute instance-attribute 

min_ccnorm: float = Field(
    ge=0.0, le=1.0, default_factory=lambda: min_ccnorm
)

Minimum cross-correlation used when automatically de-selecting seismograms.

snapshots class-attribute instance-attribute 

snapshots: list[AimbatEventParametersSnapshot] = (
    Relationship(
        back_populates="parameters", cascade_delete=True
    )
)

Snapshots these parameters are associated with.

window_post class-attribute instance-attribute 

window_post: PydanticPositiveTimedelta = Field(
    sa_type=SAPandasTimedelta,
    default_factory=lambda: window_post,
)

Post-pick window length.

window_pre class-attribute instance-attribute 

window_pre: PydanticNegativeTimedelta = Field(
    sa_type=SAPandasTimedelta,
    default_factory=lambda: window_pre,
)

Pre-pick window length.

AimbatEventParametersBase

Bases: SQLModel

Base class that defines the event parameters used in AIMBAT.

This class serves as a base that is inherited by the actual classes that create the database tables. The attributes in this class correspond exactl to the AIMBAT event parameters.

Attributes:

Name Type Description
bandpass_apply bool

Whether to apply bandpass filter to seismograms.
bandpass_fmax float

Maximum frequency for bandpass filter (ignored if bandpass_apply is False).
bandpass_fmin float

Minimum frequency for bandpass filter (ignored if bandpass_apply is False).
completed bool

Mark an event as completed.
min_ccnorm float

Minimum cross-correlation used when automatically de-selecting seismograms.
window_post PydanticPositiveTimedelta

Post-pick window length.
window_pre PydanticNegativeTimedelta

Pre-pick window length.
Source code in src/aimbat/models/_models.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
class AimbatEventParametersBase(SQLModel):
    """Base class that defines the event parameters used in AIMBAT.

    This class serves as a base that is inherited by the actual
    classes that create the database tables. The attributes in
    this class correspond exactl to the AIMBAT event parameters.
    """

    completed: bool = False
    "Mark an event as completed."

    min_ccnorm: float = Field(
        ge=0.0, le=1.0, default_factory=lambda: settings.min_ccnorm
    )
    "Minimum cross-correlation used when automatically de-selecting seismograms."

    window_pre: PydanticNegativeTimedelta = Field(
        sa_type=SAPandasTimedelta, default_factory=lambda: settings.window_pre
    )
    "Pre-pick window length."

    window_post: PydanticPositiveTimedelta = Field(
        sa_type=SAPandasTimedelta, default_factory=lambda: settings.window_post
    )
    "Post-pick window length."

    bandpass_apply: bool = Field(default_factory=lambda: settings.bandpass_apply)
    "Whether to apply bandpass filter to seismograms."

    bandpass_fmin: float = Field(default_factory=lambda: settings.bandpass_fmin, ge=0)
    "Minimum frequency for bandpass filter (ignored if `bandpass_apply` is False)."

    bandpass_fmax: float = Field(default_factory=lambda: settings.bandpass_fmax, gt=0)
    "Maximum frequency for bandpass filter (ignored if `bandpass_apply` is False)."

bandpass_apply class-attribute instance-attribute 

bandpass_apply: bool = Field(
    default_factory=lambda: bandpass_apply
)

Whether to apply bandpass filter to seismograms.

bandpass_fmax class-attribute instance-attribute 

bandpass_fmax: float = Field(
    default_factory=lambda: bandpass_fmax, gt=0
)

Maximum frequency for bandpass filter (ignored if bandpass_apply is False).

bandpass_fmin class-attribute instance-attribute 

bandpass_fmin: float = Field(
    default_factory=lambda: bandpass_fmin, ge=0
)

Minimum frequency for bandpass filter (ignored if bandpass_apply is False).

completed class-attribute instance-attribute 

completed: bool = False

Mark an event as completed.

min_ccnorm class-attribute instance-attribute 

min_ccnorm: float = Field(
    ge=0.0, le=1.0, default_factory=lambda: min_ccnorm
)

Minimum cross-correlation used when automatically de-selecting seismograms.

window_post class-attribute instance-attribute 

window_post: PydanticPositiveTimedelta = Field(
    sa_type=SAPandasTimedelta,
    default_factory=lambda: window_post,
)

Post-pick window length.

window_pre class-attribute instance-attribute 

window_pre: PydanticNegativeTimedelta = Field(
    sa_type=SAPandasTimedelta,
    default_factory=lambda: window_pre,
)

Pre-pick window length.

AimbatEventParametersSnapshot

Bases: AimbatEventParametersBase

Event parameter snapshot.

Attributes:

Name Type Description
bandpass_apply bool

Whether to apply bandpass filter to seismograms.
bandpass_fmax float

Maximum frequency for bandpass filter (ignored if bandpass_apply is False).
bandpass_fmin float

Minimum frequency for bandpass filter (ignored if bandpass_apply is False).
completed bool

Mark an event as completed.
min_ccnorm float

Minimum cross-correlation used when automatically de-selecting seismograms.
window_post PydanticPositiveTimedelta

Post-pick window length.
window_pre PydanticNegativeTimedelta

Pre-pick window length.
Source code in src/aimbat/models/_models.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
class AimbatEventParametersSnapshot(AimbatEventParametersBase, table=True):
    """Event parameter snapshot."""

    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    snapshot_id: uuid.UUID = Field(
        default=None, foreign_key="aimbatsnapshot.id", ondelete="CASCADE"
    )
    snapshot: "AimbatSnapshot" = Relationship(
        back_populates="event_parameters_snapshot"
    )
    parameters: AimbatEventParameters = Relationship(back_populates="snapshots")
    parameters_id: uuid.UUID = Field(
        default=None, foreign_key="aimbateventparameters.id", ondelete="CASCADE"
    )

bandpass_apply class-attribute instance-attribute 

bandpass_apply: bool = Field(
    default_factory=lambda: bandpass_apply
)

Whether to apply bandpass filter to seismograms.

bandpass_fmax class-attribute instance-attribute 

bandpass_fmax: float = Field(
    default_factory=lambda: bandpass_fmax, gt=0
)

Maximum frequency for bandpass filter (ignored if bandpass_apply is False).

bandpass_fmin class-attribute instance-attribute 

bandpass_fmin: float = Field(
    default_factory=lambda: bandpass_fmin, ge=0
)

Minimum frequency for bandpass filter (ignored if bandpass_apply is False).

completed class-attribute instance-attribute 

completed: bool = False

Mark an event as completed.

min_ccnorm class-attribute instance-attribute 

min_ccnorm: float = Field(
    ge=0.0, le=1.0, default_factory=lambda: min_ccnorm
)

Minimum cross-correlation used when automatically de-selecting seismograms.

window_post class-attribute instance-attribute 

window_post: PydanticPositiveTimedelta = Field(
    sa_type=SAPandasTimedelta,
    default_factory=lambda: window_post,
)

Post-pick window length.

window_pre class-attribute instance-attribute 

window_pre: PydanticNegativeTimedelta = Field(
    sa_type=SAPandasTimedelta,
    default_factory=lambda: window_pre,
)

Pre-pick window length.

AimbatSeismogram

Bases: SQLModel

Class to store seismogram data

Attributes:

Name Type Description
begin_time PydanticTimestamp

Begin time of seismogram.
delta PydanticPositiveTimedelta

Sampling interval.
id UUID

Unique ID.
t0 PydanticTimestamp

Initial pick.
Source code in src/aimbat/models/_models.py 
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
class AimbatSeismogram(SQLModel, table=True):
    """Class to store seismogram data"""

    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    "Unique ID."

    begin_time: PydanticTimestamp = Field(sa_type=SAPandasTimestamp)
    "Begin time of seismogram."

    delta: PydanticPositiveTimedelta = Field(sa_type=SAPandasTimedelta)
    "Sampling interval."

    t0: PydanticTimestamp = Field(sa_type=SAPandasTimestamp)
    "Initial pick."

    datasource: AimbatDataSource = Relationship(
        back_populates="seismogram", cascade_delete=True
    )
    station_id: uuid.UUID = Field(
        default=None, foreign_key="aimbatstation.id", ondelete="CASCADE"
    )
    station: AimbatStation = Relationship(back_populates="seismograms")
    event_id: uuid.UUID = Field(
        default=None, foreign_key="aimbatevent.id", ondelete="CASCADE"
    )
    event: AimbatEvent = Relationship(back_populates="seismograms")
    parameters: "AimbatSeismogramParameters" = Relationship(
        back_populates="seismogram",
        cascade_delete=True,
    )

    def __len__(self) -> int:
        return np.size(self.data)

    @property
    def end_time(self) -> Timestamp:
        if len(self) == 0:
            return self.begin_time
        return self.begin_time + self.delta * (len(self) - 1)

    if TYPE_CHECKING:
        flip: bool
        select: bool
        t1: Timestamp | None
        data: np.ndarray
    else:

        @property
        def flip(self) -> bool:
            return self.parameters.flip

        @flip.setter
        def flip(self, value: bool) -> None:
            self.parameters.flip = value

        @property
        def select(self) -> bool:
            return self.parameters.select

        @select.setter
        def select(self, value: bool) -> None:
            self.parameters.select = value

        @property
        def t1(self) -> Timestamp | None:
            return self.parameters.t1

        @t1.setter
        def t1(self, value: Timestamp | None) -> None:
            self.parameters.t1 = value

        @property
        def data(self) -> np.ndarray:
            if self.datasource is None:
                raise ValueError("Expected a valid datasource name, got None.")
            return read_seismogram_data(
                self.datasource.sourcename, self.datasource.datatype
            )

        @data.setter
        def data(self, value: np.ndarray) -> None:
            if self.datasource is None:
                raise ValueError("Expected a valid datasource name, got None.")
            write_seismogram_data(
                self.datasource.sourcename, self.datasource.datatype, value
            )

begin_time class-attribute instance-attribute 

begin_time: PydanticTimestamp = Field(
    sa_type=SAPandasTimestamp
)

Begin time of seismogram.

delta class-attribute instance-attribute 

delta: PydanticPositiveTimedelta = Field(
    sa_type=SAPandasTimedelta
)

Sampling interval.

id class-attribute instance-attribute 

id: UUID = Field(default_factory=uuid4, primary_key=True)

Unique ID.

t0 class-attribute instance-attribute 

t0: PydanticTimestamp = Field(sa_type=SAPandasTimestamp)

Initial pick.

AimbatSeismogramParameters

Bases: AimbatSeismogramParametersBase

Class to store ICCS processing parameters of a single seismogram.

Attributes:

Name Type Description
flip bool

Whether or not the seismogram should be flipped.
select bool

Whether or not this seismogram should be used for processing.
t1 PydanticTimestamp | None

Working pick.
Source code in src/aimbat/models/_models.py 
315
316
317
318
319
320
321
322
323
324
325
class AimbatSeismogramParameters(AimbatSeismogramParametersBase, table=True):
    """Class to store ICCS processing parameters of a single seismogram."""

    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    seismogram_id: uuid.UUID = Field(
        default=None, foreign_key="aimbatseismogram.id", ondelete="CASCADE"
    )
    seismogram: AimbatSeismogram = Relationship(back_populates="parameters")
    snapshots: list["AimbatSeismogramParametersSnapshot"] = Relationship(
        back_populates="parameters", cascade_delete=True
    )

flip class-attribute instance-attribute 

flip: bool = False

Whether or not the seismogram should be flipped.

select class-attribute instance-attribute 

select: bool = True

Whether or not this seismogram should be used for processing.

t1 class-attribute instance-attribute 

t1: PydanticTimestamp | None = Field(
    default=None, sa_type=SAPandasTimestamp
)

Working pick.

This pick serves as working as well as output pick. It is changed by:

  1. Picking the phase arrival in the stack.
  2. Running ICCS.
  3. Running MCCC.

AimbatSeismogramParametersBase

Bases: SQLModel

Base class that defines the seismogram parameters used in AIMBAT.

Attributes:

Name Type Description
flip bool

Whether or not the seismogram should be flipped.
select bool

Whether or not this seismogram should be used for processing.
t1 PydanticTimestamp | None

Working pick.
Source code in src/aimbat/models/_models.py 
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
class AimbatSeismogramParametersBase(SQLModel):
    """Base class that defines the seismogram parameters used in AIMBAT."""

    flip: bool = False
    "Whether or not the seismogram should be flipped."

    select: bool = True
    "Whether or not this seismogram should be used for processing."

    t1: PydanticTimestamp | None = Field(default=None, sa_type=SAPandasTimestamp)
    """Working pick.

    This pick serves as working as well as output pick. It is changed by:

    1. Picking the phase arrival in the stack.
    2. Running ICCS.
    3. Running MCCC.
    """

flip class-attribute instance-attribute 

flip: bool = False

Whether or not the seismogram should be flipped.

select class-attribute instance-attribute 

select: bool = True

Whether or not this seismogram should be used for processing.

t1 class-attribute instance-attribute 

t1: PydanticTimestamp | None = Field(
    default=None, sa_type=SAPandasTimestamp
)

Working pick.

This pick serves as working as well as output pick. It is changed by:

  1. Picking the phase arrival in the stack.
  2. Running ICCS.
  3. Running MCCC.

AimbatSeismogramParametersSnapshot

Bases: AimbatSeismogramParametersBase

Class to store a snapshot of ICCS processing parameters of a single seismogram.

Attributes:

Name Type Description
flip bool

Whether or not the seismogram should be flipped.
select bool

Whether or not this seismogram should be used for processing.
t1 PydanticTimestamp | None

Working pick.
Source code in src/aimbat/models/_models.py 
328
329
330
331
332
333
334
335
336
337
338
339
340
341
class AimbatSeismogramParametersSnapshot(AimbatSeismogramParametersBase, table=True):
    """Class to store a snapshot of ICCS processing parameters of a single seismogram."""

    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    seismogram_parameters_id: uuid.UUID = Field(
        foreign_key="aimbatseismogramparameters.id", ondelete="CASCADE"
    )
    parameters: AimbatSeismogramParameters = Relationship(back_populates="snapshots")
    snapshot_id: uuid.UUID = Field(
        default=None, foreign_key="aimbatsnapshot.id", ondelete="CASCADE"
    )
    snapshot: "AimbatSnapshot" = Relationship(
        back_populates="seismogram_parameters_snapshots"
    )

flip class-attribute instance-attribute 

flip: bool = False

Whether or not the seismogram should be flipped.

select class-attribute instance-attribute 

select: bool = True

Whether or not this seismogram should be used for processing.

t1 class-attribute instance-attribute 

t1: PydanticTimestamp | None = Field(
    default=None, sa_type=SAPandasTimestamp
)

Working pick.

This pick serves as working as well as output pick. It is changed by:

  1. Picking the phase arrival in the stack.
  2. Running ICCS.
  3. Running MCCC.

AimbatSnapshot

Bases: SQLModel

Class to store AIMBAT snapshots.

The AimbatSnapshot class does not actually save any parameter data. It is used to keep track of the AimbatEventParametersSnapshot and AimbatSeismogramParametersSnapshot instances.

Attributes:

Name Type Description
event AimbatEvent

Event this snapshot is associated with.
event_id UUID

Event ID this snapshot is associated with.
Source code in src/aimbat/models/_models.py 
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
class AimbatSnapshot(SQLModel, table=True):
    """Class to store AIMBAT snapshots.

    The AimbatSnapshot class does not actually save any parameter data.
    It is used to keep track of the AimbatEventParametersSnapshot and
    AimbatSeismogramParametersSnapshot instances.
    """

    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    date: PydanticTimestamp = Field(
        default_factory=lambda: Timestamp.now(tz=timezone.utc),
        unique=True,
        allow_mutation=False,
        sa_type=SAPandasTimestamp,
    )
    comment: str | None = None
    event_parameters_snapshot: AimbatEventParametersSnapshot = Relationship(
        back_populates="snapshot", cascade_delete=True
    )
    seismogram_parameters_snapshots: list[AimbatSeismogramParametersSnapshot] = (
        Relationship(back_populates="snapshot", cascade_delete=True)
    )

    event_id: uuid.UUID = Field(
        default=None, foreign_key="aimbatevent.id", ondelete="CASCADE"
    )
    "Event ID this snapshot is associated with."

    event: AimbatEvent = Relationship(back_populates="snapshots")
    "Event this snapshot is associated with."

event class-attribute instance-attribute 

event: AimbatEvent = Relationship(
    back_populates="snapshots"
)

Event this snapshot is associated with.

event_id class-attribute instance-attribute 

event_id: UUID = Field(
    default=None,
    foreign_key="aimbatevent.id",
    ondelete="CASCADE",
)

Event ID this snapshot is associated with.

AimbatStation

Bases: SQLModel

Class to store station information.

Attributes:

Name Type Description
channel str

Channel code.
elevation float | None

Station elevation.
id UUID

Unique ID.
latitude float

Station latitude
location str

Location ID.
longitude float

Station longitude
name str

Station name.
network str

Network name.
seismograms list[AimbatSeismogram]

Seismograms recorded at this station.
Source code in src/aimbat/models/_models.py 
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
class AimbatStation(SQLModel, table=True):
    """Class to store station information."""

    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
    "Unique ID."

    name: str = Field(allow_mutation=False)
    "Station name."

    network: str = Field(allow_mutation=False)
    "Network name."

    location: str = Field(allow_mutation=False)
    "Location ID."

    channel: str = Field(allow_mutation=False)
    "Channel code."

    latitude: float
    "Station latitude"

    longitude: float
    "Station longitude"

    elevation: float | None = None
    "Station elevation."

    seismograms: list["AimbatSeismogram"] = Relationship(
        back_populates="station", cascade_delete=True
    )
    "Seismograms recorded at this station."

channel class-attribute instance-attribute 

channel: str = Field(allow_mutation=False)

Channel code.

elevation class-attribute instance-attribute 

elevation: float | None = None

Station elevation.

id class-attribute instance-attribute 

id: UUID = Field(default_factory=uuid4, primary_key=True)

Unique ID.

latitude instance-attribute 

latitude: float

Station latitude

location class-attribute instance-attribute 

location: str = Field(allow_mutation=False)

Location ID.

longitude instance-attribute 

longitude: float

Station longitude

name class-attribute instance-attribute 

name: str = Field(allow_mutation=False)

Station name.

network class-attribute instance-attribute 

network: str = Field(allow_mutation=False)

Network name.

seismograms class-attribute instance-attribute 

seismograms: list[AimbatSeismogram] = Relationship(
    back_populates="station", cascade_delete=True
)

Seismograms recorded at this station.

utils

Utils used in AIMBAT.

Classes:

Name Description
TableStyling

This class is to set the colour of the table columns and elements.

Functions:

Name Description
delete_sampledata

Delete sample data.
download_sampledata

Download sample data.
dump_to_json

Dump a sequence of AimbatTypes to a JSON string and print it.
get_active_event

Return the currently active event (i.e. the one being processed).
run_checks

Run all checks on one or more SAC files.
string_to_uuid

Determine a UUID from a string containing the first few characters.

TableStyling dataclass

This class is to set the colour of the table columns and elements.

Source code in src/aimbat/utils/_style.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
@dataclass(frozen=True)
class TableStyling:
    """This class is to set the colour of the table columns and elements."""

    id: str = "bright_blue"
    mine: str = "cyan"
    linked: str = "magenta"
    parameters: str = "green"

    @staticmethod
    def bool_formatter(true_or_false: bool | Any) -> str:
        if true_or_false is True:
            return "[bold green]:heavy_check_mark:[/]"
        elif true_or_false is False:
            return "[bold red]:heavy_multiplication_x:[/]"
        return true_or_false

    @staticmethod
    def timestamp_formatter(dt: Timestamp, short: bool) -> str:
        if short:
            return dt.strftime("%Y-%m-%d [light_sea_green]%H:%M:%S[/]")
        return str(dt)

delete_sampledata 

delete_sampledata() -> None

Delete sample data.

Source code in src/aimbat/utils/_sampledata.py 
12
13
14
15
16
17
def delete_sampledata() -> None:
    """Delete sample data."""

    logger.info(f"Deleting sample data in {settings.sampledata_dir}.")

    shutil.rmtree(settings.sampledata_dir)

download_sampledata 

download_sampledata(force: bool = False) -> None

Download sample data.

Source code in src/aimbat/utils/_sampledata.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def download_sampledata(force: bool = False) -> None:
    """Download sample data."""

    logger.info(
        f"Downloading sample data from {settings.sampledata_src} to {settings.sampledata_dir}."
    )

    if (
        settings.sampledata_dir.exists()
        and len(os.listdir(settings.sampledata_dir)) != 0
    ):
        if force is True:
            delete_sampledata()
        else:
            raise FileExistsError(
                f"The directory {settings.sampledata_dir} already exists and is non-empty."
            )

    with urlopen(settings.sampledata_src) as zipresp:
        with ZipFile(BytesIO(zipresp.read())) as zfile:
            zfile.extractall(settings.sampledata_dir)

dump_to_json 

dump_to_json(aimbat_data: Sequence[AimbatTypes]) -> None

Dump a sequence of AimbatTypes to a JSON string and print it.

Parameters:

Name Type Description Default
aimbat_data Sequence[AimbatTypes]

A sequence of AimbatTypes to dump to JSON.

 required
Source code in src/aimbat/utils/_json.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
def dump_to_json(aimbat_data: Sequence[AimbatTypes]) -> None:
    """Dump a sequence of AimbatTypes to a JSON string and print it.

    Args:
        aimbat_data: A sequence of AimbatTypes to dump to JSON.
    """

    class CustomEncoder(json.JSONEncoder):
        def default(self, o: Any) -> str | Any:
            if isinstance(o, uuid.UUID):
                return str(o)
            if isinstance(o, Timestamp):
                return o.isoformat()
            if isinstance(o, Timedelta):
                return o.total_seconds()
            return super().default(o)

    json_str = json.dumps(
        [r.model_dump(mode="python") for r in aimbat_data],
        cls=CustomEncoder,
        indent=4,
    )
    print(json_str)

get_active_event 

get_active_event(session: Session) -> AimbatEvent

Return the currently active event (i.e. the one being processed).

Parameters:

Name Type Description Default
session Session

SQL session.

 required

Returns:

Type Description
AimbatEvent

Active Event

Raises NoResultFound: When no event is active.

Source code in src/aimbat/utils/_active_event.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def get_active_event(session: Session) -> AimbatEvent:
    """
    Return the currently active event (i.e. the one being processed).

    Args:
        session: SQL session.

    Returns:
        Active Event

    Raises
        NoResultFound: When no event is active.
    """

    logger.debug("Attempting to determine active event.")

    select_active_event = select(AimbatEvent).where(AimbatEvent.active == 1)

    # NOTE: While there technically can be no active event in the database,
    # we typically don't really want to go beyond this point when that is the
    # case. Hence we call `one` rather than `one_or_none`.
    try:
        active_event = session.exec(select_active_event).one()
    except NoResultFound:
        raise NoResultFound(f"No active event found. {HINTS.ACTIVATE_EVENT}")

    logger.debug(f"Active event: {active_event.id}")

    return active_event

run_checks 

run_checks(sacfiles: list[Path]) -> None

Run all checks on one or more SAC files.

Parameters:

Name Type Description Default
sacfiles list[Path]

SAC files to test.

 required
Source code in src/aimbat/utils/_checkdata.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def run_checks(sacfiles: list[Path]) -> None:
    """Run all checks on one or more SAC files.

    Args:
        sacfiles: SAC files to test.
    """

    logger.info("Running all checks.")

    from pysmo.classes import SAC

    def checkmark() -> None:
        print("\N{CHECK MARK}", end="")

    def crossmark() -> None:
        print("\N{BALLOT X}", end="")

    all_issues = dict()

    for sacfile in sacfiles:
        issues = list()
        my_sac = SAC.from_file(str(sacfile))
        print(f"\n{sacfile}: ", end="")

        station_issues = checkdata_station(my_sac.station)
        if len(station_issues) == 0:
            checkmark()
        else:
            issues.extend(station_issues)
            crossmark()

        event_issues = checkdata_event(my_sac.event)
        if len(event_issues) == 0:
            checkmark()
        else:
            issues.extend(event_issues)
            crossmark()

        seismogram_issues = checkdata_seismogram(my_sac.seismogram)
        if len(seismogram_issues) == 0:
            checkmark()
        else:
            issues.extend(seismogram_issues)
            crossmark()

        if len(issues) > 0:
            all_issues[sacfile] = issues

    if len(all_issues) == 0:
        print("\n\nNo issues found!")
        return

    print("\n\nPlease fix the following issues before proceeding:")
    for sacfile, issues in all_issues.items():
        print(f"\n file: {sacfile}:")
        for issue in issues:
            print(f"  - {issue}")

string_to_uuid 

string_to_uuid(
    session: Session,
    id: str,
    aimbat_class: type[AimbatTypes],
    custom_error: str | None = None,
) -> UUID

Determine a UUID from a string containing the first few characters.

Parameters:

Name Type Description Default
session Session

Database session.

 required
id str

Input string to find UUID for.

 required
aimbat_class type[AimbatTypes]

Aimbat class to use to find UUID.

 required
custom_error str | None

Overrides the default error message.

 None

Returns:

Type Description
UUID

The full UUID.

Raises:

Type Description
ValueError

If the UUID could not be determined.
Source code in src/aimbat/utils/_uuid.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def string_to_uuid(
    session: Session,
    id: str,
    aimbat_class: type[AimbatTypes],
    custom_error: str | None = None,
) -> UUID:
    """Determine a UUID from a string containing the first few characters.

    Args:
        session: Database session.
        id: Input string to find UUID for.
        aimbat_class: Aimbat class to use to find UUID.
        custom_error: Overrides the default error message.

    Returns:
        The full UUID.

    Raises:
        ValueError: If the UUID could not be determined.
    """
    uuid_set = {
        u for u in session.exec(select(aimbat_class.id)).all() if str(u).startswith(id)
    }
    if len(uuid_set) == 1:
        return uuid_set.pop()
    if len(uuid_set) == 0:
        raise ValueError(
            custom_error or f"Unable to find {aimbat_class.__name__} using id: {id}."
        )
    raise ValueError(f"Found more than one {aimbat_class.__name__} using id: {id}")