Skip to content

aimbat.lib #

Modules:

Name Description
common

Common functions for AIMBAT.
data
db

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

Module to manage and view events in AIMBAT.
iccs

Processing of data for AIMBAT.
io

Functions to read and write data files used with AIMBAT
models

This module defines the "Aimbat" classes.
project
seismogram
snapshot
station
typing

Custom types used in AIMBAT.
utils

common #

Common functions for AIMBAT.

Classes:

Name Description
CliHints

Hints for error messages.
TableStyling

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

Functions:

Name Description
check_for_notebook

Check if we ware running inside a jupyter notebook.
string_to_uuid

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

CliHints dataclass #

Hints for error messages.

Source code in src/aimbat/lib/common.py 
101
102
103
104
105
106
@dataclass
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`."

TableStyling dataclass #

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

Source code in src/aimbat/lib/common.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
@dataclass
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 datetime_formatter(dt: datetime, short: bool) -> str:
        if short:
            return dt.strftime("%Y-%m-%d [light_sea_green]%H:%M:%S[/]")
        return str(dt)

check_for_notebook #

check_for_notebook() -> bool

Check if we ware running inside a jupyter notebook.

Source code in src/aimbat/lib/common.py 
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def check_for_notebook() -> bool:
    """Check if we ware running inside a jupyter notebook."""
    import IPython.core.getipython as getipython

    try:
        shell = getipython.get_ipython().__class__.__name__
        if shell == "ZMQInteractiveShell":
            return True  # Jupyter notebook or qtconsole
        elif shell == "TerminalInteractiveShell":
            return False  # Terminal running IPython
        else:
            return False  # Other type (?)
    except NameError:
        return False  # Probably standard Python interpreter

string_to_uuid #

string_to_uuid(
    session: Session,
    id: str,
    aimbat_class: type[
        AimbatDataSource
        | AimbatStation
        | AimbatEvent
        | AimbatEventParameters
        | AimbatSeismogram
        | AimbatSeismogramParameters
        | AimbatSnapshot
        | AimbatEventParametersSnapshot
        | AimbatSeismogramParametersSnapshot
    ],
    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[AimbatDataSource | AimbatStation | AimbatEvent | AimbatEventParameters | AimbatSeismogram | AimbatSeismogramParameters | AimbatSnapshot | AimbatEventParametersSnapshot | AimbatSeismogramParametersSnapshot]

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/lib/common.py 
26
27
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
def string_to_uuid(
    session: Session,
    id: str,
    aimbat_class: type[
        AimbatDataSource
        | AimbatStation
        | AimbatEvent
        | AimbatEventParameters
        | AimbatSeismogram
        | AimbatSeismogramParameters
        | AimbatSnapshot
        | AimbatEventParametersSnapshot
        | AimbatSeismogramParametersSnapshot
    ],
    custom_error: str | None = None,
) -> UUID:
    """Determine a UUID from a string containing the first few characters.

    Parameters:
        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}")

data #

Functions:

Name Description
add_files_to_project

Add files to the AIMBAT database.
dump_data_table

Dump the table data to json.
get_data_for_active_event

Returns the AimbatFiles belonging to the active event.
print_data_table

Print a pretty table with AIMBAT data.

add_files_to_project #

add_files_to_project(
    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/lib/data.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
def add_files_to_project(
    datasources: Sequence[str | os.PathLike],
    datatype: DataType,
    disable_progress_bar: bool = True,
) -> None:
    """Add files to the AIMBAT database.

    Parameters:
        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.")

    with Session(engine) as session:
        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()

dump_data_table #

dump_data_table() -> None

Dump the table data to json.

Source code in src/aimbat/lib/data.py 
241
242
243
244
245
246
247
248
def dump_data_table() -> None:
    """Dump the table data to json."""

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

    with Session(engine) as session:
        aimbat_data_sources = session.exec(select(AimbatDataSource)).all()
        dump_to_json(aimbat_data_sources)

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/lib/data.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def get_data_for_active_event(session: Session) -> Sequence[AimbatDataSource]:
    """Returns the AimbatFiles belonging to the active event.

    Parameters:
        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()

print_data_table #

print_data_table(
    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/lib/data.py 
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
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
def print_data_table(short: bool, all_events: bool = False) -> None:
    """Print a pretty table with AIMBAT data.

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

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

    with Session(engine) as session:
        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)

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.

event #

Module to manage and view events in AIMBAT.

Functions:

Name Description
delete_event

Delete an AimbatEvent from the database.
delete_event_by_id

Delete an AimbatEvent from the database by ID.
dump_event_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_event_parameter

Get event parameter value for the active event.
get_events_using_station

Get all events that use a particular station.
print_event_table

Print a pretty table with AIMBAT events.
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.

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/lib/event.py 
52
53
54
55
56
57
58
59
60
61
62
63
def delete_event(session: Session, event: AimbatEvent) -> None:
    """Delete an AimbatEvent from the database.

    Parameters:
        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/lib/event.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def delete_event_by_id(session: Session, event_id: UUID) -> None:
    """Delete an AimbatEvent from the database by ID.

    Parameters:
        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)

dump_event_table #

dump_event_table() -> None

Dump the table data to json.

Source code in src/aimbat/lib/event.py 
311
312
313
314
315
316
317
318
def dump_event_table() -> None:
    """Dump the table data to json."""

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

    with Session(engine) as session:
        aimbat_events = session.exec(select(AimbatEvent)).all()
        dump_to_json(aimbat_events)

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/lib/event.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
94
def get_active_event(session: Session) -> AimbatEvent:
    """
    Return the currently active event (i.e. the one being processed).

    Parameters:
        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/lib/event.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
def get_completed_events(session: Session) -> Sequence[AimbatEvent]:
    """Get the events marked as completed.

    Parameters:
        session: SQL session.
    """

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

    return session.exec(select_completed_events).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/lib/event.py 
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
def get_event_parameter(
    session: Session, name: EventParameter
) -> timedelta | bool | float:
    """Get event parameter value for the active event.

    Parameters:
        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/lib/event.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
def get_events_using_station(
    session: Session, station: AimbatStation
) -> Sequence[AimbatEvent]:
    """Get all events that use a particular station.

    Parameters:
        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

print_event_table #

print_event_table(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/lib/event.py 
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
296
297
298
299
300
301
302
303
304
305
306
307
308
def print_event_table(short: bool = True) -> None:
    """Print a pretty table with AIMBAT events.

    Parameters:
        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)

    with Session(engine) as session:
        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.datetime_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)

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/lib/event.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def set_active_event(session: Session, event: AimbatEvent) -> None:
    """
    Set the active event (i.e. the one being processed).

    Parameters:
        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

id #

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/lib/event.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
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.

    Parameters:
        session: SQL session.
        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/lib/event.py 
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
def set_event_parameter(
    session: Session, name: EventParameter, value: timedelta | bool | float | str
) -> None:
    """Set event parameter value for the active event.

    Parameters:
        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()

iccs #

Processing of data for AIMBAT.

Functions:

Name Description
create_iccs_instance

Create an ICCS instance for the active event.
plot_seismograms

Plot the ICCS seismograms as an image.
plot_stack

Plot the ICCS stack.
run_iccs

Run ICCS algorithm.
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.

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/lib/iccs.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
def create_iccs_instance(session: Session) -> ICCS:
    """Create an ICCS instance for the active event.

    Parameters:
        session: Database session.

    Returns:
        ICCS instance.
    """

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

    active_event = 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,
        min_ccnorm=active_event.parameters.min_ccnorm,
        context_width=settings.context_width,
    )

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/lib/iccs.py 
70
71
72
73
74
75
76
77
78
79
80
81
def plot_seismograms(iccs: ICCS, context: bool, all: bool) -> None:
    """Plot the ICCS seismograms as an image.

    Parameters:
        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/lib/iccs.py 
57
58
59
60
61
62
63
64
65
66
67
def plot_stack(iccs: ICCS, context: bool, all: bool) -> None:
    """Plot the ICCS stack.

    Parameters:
        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)

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/lib/iccs.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def run_iccs(session: Session, iccs: ICCS, autoflip: bool, autoselect: bool) -> None:
    """Run ICCS algorithm.

    Parameters:
        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()

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/lib/iccs.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def update_min_ccnorm(session: Session, iccs: ICCS, context: bool, all: bool) -> None:
    """Update the minimum cross correlation coefficient for the active event.

    Parameters:
        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 = 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/lib/iccs.py 
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def update_pick(
    session: Session, iccs: ICCS, context: bool, all: bool, use_seismogram_image: bool
) -> None:
    """Update the pick for the active event.

    Parameters:
        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/lib/iccs.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def update_timewindow(
    session: Session, iccs: ICCS, context: bool, all: bool, use_seismogram_image: bool
) -> None:
    """Update the time window for the active event.

    Parameters:
        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 = event.get_active_event(session)
    active_event.parameters.window_pre = iccs.window_pre
    active_event.parameters.window_post = iccs.window_post
    session.commit()

io #

Functions to read and write data files used with AIMBAT

Classes:

Name Description
DataType

Valid AIMBAT data types.

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.

DataType #

Bases: StrEnum


              flowchart TD
              aimbat.lib.io.DataType[DataType]

              

              click aimbat.lib.io.DataType href "" "aimbat.lib.io.DataType"

Valid AIMBAT data types.

Used by:
Source code in src/aimbat/lib/io/_io.py 
18
19
20
21
class DataType(StrEnum):
    """Valid AIMBAT data types."""

    SAC = auto()

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/lib/io/_io.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def create_event(datasource: str | PathLike, datatype: DataType) -> AimbatEvent:
    """Read event data from a data source and create an AimbatEvent.

    Parameters:
        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/lib/io/_io.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def create_seismogram(
    datasource: str | PathLike, datatype: DataType
) -> AimbatSeismogram:
    """Read seismogram data from a data source and create an AimbatSeismogram.

    Parameters:
        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/lib/io/_io.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def create_station(datasource: str | PathLike, datatype: DataType) -> AimbatStation:
    """Read station data from a data source and create an AimbatStation.

    Parameters:
        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/lib/io/_io.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def read_seismogram_data(
    datasource: str | PathLike, datatype: DataType
) -> npt.NDArray[np.float64]:
    """Read seismogram data from a data source.

    Parameters:
        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/lib/io/_io.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
def write_seismogram_data(
    datasource: str | PathLike,
    datatype: DataType,
    data: npt.NDArray[np.float64],
) -> None:
    """Write seismogram data to a data source.

    Parameters:
        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)

models #

This module defines the "Aimbat" classes.

These classes are ORMs that present data stored in a database as classes to use with python in AIMBAT.

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.

Attributes:

Name Type Description
AimbatTypes

Union of all AIMBAT models that exist in the database.

AimbatTypes module-attribute #

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

Union of all AIMBAT models that exist in the database.

AimbatDataSource #

Bases: SQLModel


              flowchart TD
              aimbat.lib.models.AimbatDataSource[AimbatDataSource]

              

              click aimbat.lib.models.AimbatDataSource href "" "aimbat.lib.models.AimbatDataSource"

Class to store data source information.

Returned by: Used by:
Source code in src/aimbat/lib/models.py 
37
38
39
40
41
42
43
44
45
46
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: io.DataType
    seismogram_id: uuid.UUID = Field(
        default=None, foreign_key="aimbatseismogram.id", ondelete="CASCADE"
    )
    seismogram: "AimbatSeismogram" = Relationship(back_populates="datasource")

AimbatDataSourceCreate #

Bases: SQLModel


              flowchart TD
              aimbat.lib.models.AimbatDataSourceCreate[AimbatDataSourceCreate]

              

              click aimbat.lib.models.AimbatDataSourceCreate href "" "aimbat.lib.models.AimbatDataSourceCreate"

Class to store data source information.

Source code in src/aimbat/lib/models.py 
30
31
32
33
34
class AimbatDataSourceCreate(SQLModel):
    """Class to store data source information."""

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

AimbatEvent #

Bases: SQLModel


              flowchart TD
              aimbat.lib.models.AimbatEvent[AimbatEvent]

              

              click aimbat.lib.models.AimbatEvent href "" "aimbat.lib.models.AimbatEvent"

Store event information.

Referenced by: Returned by: Used by:

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 datetime

Event time.
Source code in src/aimbat/lib/models.py 
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
77
78
79
80
81
82
83
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: datetime = Field(unique=True, sa_type=_DateTimeUTC, 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: datetime = Field(
    unique=True, sa_type=_DateTimeUTC, allow_mutation=False
)

Event time.

AimbatEventParameters #

Bases: AimbatEventParametersBase


              flowchart TD
              aimbat.lib.models.AimbatEventParameters[AimbatEventParameters]
              aimbat.lib.models.AimbatEventParametersBase[AimbatEventParametersBase]

                              aimbat.lib.models.AimbatEventParametersBase --> aimbat.lib.models.AimbatEventParameters
                


              click aimbat.lib.models.AimbatEventParameters href "" "aimbat.lib.models.AimbatEventParameters"
              click aimbat.lib.models.AimbatEventParametersBase href "" "aimbat.lib.models.AimbatEventParametersBase"

Processing parameters common to all seismograms of a particular event.

Returned by: Used by:

Attributes:

Name Type Description
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 timedelta

Post-pick window length.
window_pre timedelta

Pre-pick window length.
Source code in src/aimbat/lib/models.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
class AimbatEventParameters(AimbatEventParametersBase, 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."

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=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: timedelta = Field(gt=0, default=window_post)

Post-pick window length.

window_pre class-attribute instance-attribute #

window_pre: timedelta = Field(lt=0, default=window_pre)

Pre-pick window length.

AimbatEventParametersBase #

Bases: SQLModel


              flowchart TD
              aimbat.lib.models.AimbatEventParametersBase[AimbatEventParametersBase]

              

              click aimbat.lib.models.AimbatEventParametersBase href "" "aimbat.lib.models.AimbatEventParametersBase"

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.

Referenced by: Subclassed by:

Attributes:

Name Type Description
completed bool

Mark an event as completed.
min_ccnorm float

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

Post-pick window length.
window_pre timedelta

Pre-pick window length.
Source code in src/aimbat/lib/models.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
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=settings.min_ccnorm)
    "Minimum cross-correlation used when automatically de-selecting seismograms."

    window_pre: timedelta = Field(lt=0, default=settings.window_pre)
    "Pre-pick window length."

    window_post: timedelta = Field(gt=0, default=settings.window_post)
    "Post-pick window length."

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=min_ccnorm
)

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

window_post class-attribute instance-attribute #

window_post: timedelta = Field(gt=0, default=window_post)

Post-pick window length.

window_pre class-attribute instance-attribute #

window_pre: timedelta = Field(lt=0, default=window_pre)

Pre-pick window length.

AimbatEventParametersSnapshot #

Bases: AimbatEventParametersBase


              flowchart TD
              aimbat.lib.models.AimbatEventParametersSnapshot[AimbatEventParametersSnapshot]
              aimbat.lib.models.AimbatEventParametersBase[AimbatEventParametersBase]

                              aimbat.lib.models.AimbatEventParametersBase --> aimbat.lib.models.AimbatEventParametersSnapshot
                


              click aimbat.lib.models.AimbatEventParametersSnapshot href "" "aimbat.lib.models.AimbatEventParametersSnapshot"
              click aimbat.lib.models.AimbatEventParametersBase href "" "aimbat.lib.models.AimbatEventParametersBase"

Event parameter snapshot.

Returned by: Used by:

Attributes:

Name Type Description
completed bool

Mark an event as completed.
min_ccnorm float

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

Post-pick window length.
window_pre timedelta

Pre-pick window length.
Source code in src/aimbat/lib/models.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
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"
    )

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=min_ccnorm
)

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

window_post class-attribute instance-attribute #

window_post: timedelta = Field(gt=0, default=window_post)

Post-pick window length.

window_pre class-attribute instance-attribute #

window_pre: timedelta = Field(lt=0, default=window_pre)

Pre-pick window length.

AimbatSeismogram #

Bases: SQLModel


              flowchart TD
              aimbat.lib.models.AimbatSeismogram[AimbatSeismogram]

              

              click aimbat.lib.models.AimbatSeismogram href "" "aimbat.lib.models.AimbatSeismogram"

Class to store seismogram data

Returned by: Used by:

Attributes:

Name Type Description
begin_time datetime

Begin time of seismogram.
delta timedelta

Sampling interval.
id UUID

Unique ID.
t0 datetime

Initial pick.
Source code in src/aimbat/lib/models.py 
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
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
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: datetime = Field(sa_type=_DateTimeUTC)
    "Begin time of seismogram."

    delta: timedelta
    "Sampling interval."

    t0: datetime = Field(sa_type=_DateTimeUTC)
    "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,
    )

    @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) -> datetime | None:
        return self.parameters.t1

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

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

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

    @property
    def data(self) -> np.ndarray:
        if self.datasource is None:
            raise ValueError("Expected a valid datasource name, got None.")
        return io.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.")
        io.write_seismogram_data(
            self.datasource.sourcename, self.datasource.datatype, value
        )

begin_time class-attribute instance-attribute #

begin_time: datetime = Field(sa_type=_DateTimeUTC)

Begin time of seismogram.

delta instance-attribute #

delta: timedelta

Sampling interval.

id class-attribute instance-attribute #

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

Unique ID.

t0 class-attribute instance-attribute #

t0: datetime = Field(sa_type=_DateTimeUTC)

Initial pick.

AimbatSeismogramParameters #

Bases: AimbatSeismogramParametersBase


              flowchart TD
              aimbat.lib.models.AimbatSeismogramParameters[AimbatSeismogramParameters]
              aimbat.lib.models.AimbatSeismogramParametersBase[AimbatSeismogramParametersBase]

                              aimbat.lib.models.AimbatSeismogramParametersBase --> aimbat.lib.models.AimbatSeismogramParameters
                


              click aimbat.lib.models.AimbatSeismogramParameters href "" "aimbat.lib.models.AimbatSeismogramParameters"
              click aimbat.lib.models.AimbatSeismogramParametersBase href "" "aimbat.lib.models.AimbatSeismogramParametersBase"

Class to store ICCS processing parameters of a single seismogram.

Referenced by: Used by:

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 datetime | None

Working pick.
Source code in src/aimbat/lib/models.py 
277
278
279
280
281
282
283
284
285
286
287
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: datetime | None = Field(
    default=None, sa_type=_DateTimeUTC
)

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


              flowchart TD
              aimbat.lib.models.AimbatSeismogramParametersBase[AimbatSeismogramParametersBase]

              

              click aimbat.lib.models.AimbatSeismogramParametersBase href "" "aimbat.lib.models.AimbatSeismogramParametersBase"

Base class that defines the seismogram parameters used in AIMBAT.

Referenced by: Subclassed by:

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 datetime | None

Working pick.
Source code in src/aimbat/lib/models.py 
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
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: datetime | None = Field(default=None, sa_type=_DateTimeUTC)
    """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: datetime | None = Field(
    default=None, sa_type=_DateTimeUTC
)

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


              flowchart TD
              aimbat.lib.models.AimbatSeismogramParametersSnapshot[AimbatSeismogramParametersSnapshot]
              aimbat.lib.models.AimbatSeismogramParametersBase[AimbatSeismogramParametersBase]

                              aimbat.lib.models.AimbatSeismogramParametersBase --> aimbat.lib.models.AimbatSeismogramParametersSnapshot
                


              click aimbat.lib.models.AimbatSeismogramParametersSnapshot href "" "aimbat.lib.models.AimbatSeismogramParametersSnapshot"
              click aimbat.lib.models.AimbatSeismogramParametersBase href "" "aimbat.lib.models.AimbatSeismogramParametersBase"

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

Used by:

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 datetime | None

Working pick.
Source code in src/aimbat/lib/models.py 
290
291
292
293
294
295
296
297
298
299
300
301
302
303
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: datetime | None = Field(
    default=None, sa_type=_DateTimeUTC
)

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


              flowchart TD
              aimbat.lib.models.AimbatSnapshot[AimbatSnapshot]

              

              click aimbat.lib.models.AimbatSnapshot href "" "aimbat.lib.models.AimbatSnapshot"

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.

Returned by: Used by:

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/lib/models.py 
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
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: datetime = Field(
        default_factory=lambda: datetime.now(timezone.utc),
        unique=True,
        allow_mutation=False,
        sa_type=_DateTimeUTC,
    )
    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


              flowchart TD
              aimbat.lib.models.AimbatStation[AimbatStation]

              

              click aimbat.lib.models.AimbatStation href "" "aimbat.lib.models.AimbatStation"

Class to store station information.

Returned by: Used by:

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/lib/models.py 
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
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.

project #

Functions:

Name Description
create_project

Create a new AIMBAT project.
delete_project

Delete the AIMBAT project.
print_project_info

Show AIMBAT project information.

create_project #

create_project() -> None

Create a new AIMBAT project.

Source code in src/aimbat/lib/project.py 
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_project() -> None:
    """Create a new AIMBAT project."""

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

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

    if _project_exists():
        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;
    """))

delete_project #

delete_project() -> None

Delete the AIMBAT project.

Raises:

Type Description
RuntimeError

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

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

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

    if _project_exists():
        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.")

print_project_info #

print_project_info() -> None

Show AIMBAT project information.

Parameters:

Name Type Description Default

engine #

Database engine.

 required

Raises:

Type Description
RuntimeError

If no project found.
Source code in src/aimbat/lib/project.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
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() -> None:
    """Show AIMBAT project information.

    Parameters:
        engine: Database engine.

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

    logger.info("Printing project info.")

    if not _project_exists():
        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 = 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")
        )

seismogram #

Functions:

Name Description
delete_seismogram

Delete an AimbatSeismogram from the database.
delete_seismogram_by_id

Delete an AimbatSeismogram from the database by ID.
dump_seismogram_table

Dump the table data to json.
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.
plot_seismograms

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

Prints a pretty table with AIMBAT seismograms.
set_seismogram_parameter

Set parameter value for an AimbatSeismogram instance.
set_seismogram_parameter_by_id

Set parameter value for an AimbatSeismogram by ID.

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/lib/seismogram.py 
59
60
61
62
63
64
65
66
67
68
69
70
def delete_seismogram(session: Session, seismogram: AimbatSeismogram) -> None:
    """Delete an AimbatSeismogram from the database.

    Parameters:
        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/lib/seismogram.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def delete_seismogram_by_id(session: Session, seismogram_id: uuid.UUID) -> None:
    """Delete an AimbatSeismogram from the database by ID.

    Parameters:
        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)

dump_seismogram_table #

dump_seismogram_table() -> None

Dump the table data to json.

Source code in src/aimbat/lib/seismogram.py 
341
342
343
344
345
346
347
348
def dump_seismogram_table() -> None:
    """Dump the table data to json."""

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

    with Session(engine) as session:
        aimbat_seismograms = session.exec(select(AimbatSeismogram)).all()
        dump_to_json(aimbat_seismograms)

get_seismogram_parameter #

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

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 | datetime

Seismogram parameter value.
Source code in src/aimbat/lib/seismogram.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def get_seismogram_parameter(
    seismogram: AimbatSeismogram, name: SeismogramParameter
) -> bool | datetime:
    """Get parameter value from an AimbatSeismogram instance.

    Parameters:
        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 | datetime

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 | datetime

Seismogram parameter value.

Raises:

Type Description
ValueError

If no AimbatSeismogram is found with the given ID.
Source code in src/aimbat/lib/seismogram.py 
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
def get_seismogram_parameter_by_id(
    session: Session, seismogram_id: uuid.UUID, name: SeismogramParameter
) -> bool | datetime:
    """Get parameter value from an AimbatSeismogram by ID.

    Parameters:
        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/lib/seismogram.py 
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
def get_selected_seismograms(
    session: Session, all_events: bool = False
) -> Sequence[AimbatSeismogram]:
    """Get the selected seismograms for the active avent.

    Parameters:
        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

plot_seismograms #

plot_seismograms(
    use_qt: bool = False,
) -> Figure | PlotWidget

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/lib/seismogram.py 
357
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
422
423
424
425
426
def plot_seismograms(use_qt: bool = False) -> Figure | PlotWidget:
    """Plot all seismograms for a particular event ordered by great circle distance.

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

    with Session(engine) as session:
        active_event = 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:
            if check_for_notebook():
                plot_widget = PlotWidget(width=200)
            else:
                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 use_qt and isinstance(plot_widget, PlotWidget):
            return plot_widget
        elif 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

print_seismogram_table #

print_seismogram_table(
    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/lib/seismogram.py 
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
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
def print_seismogram_table(short: bool, all_events: bool = False) -> None:
    """Prints a pretty table with AIMBAT seismograms.

    Parameters:
        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

    with Session(engine) as session:
        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 = 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)

set_seismogram_parameter #

set_seismogram_parameter(
    session: Session,
    seismogram: AimbatSeismogram,
    name: SeismogramParameter,
    value: datetime | 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 #

 datetime | bool | str

Value to set parameter to.

 required
Source code in src/aimbat/lib/seismogram.py 
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
def set_seismogram_parameter(
    session: Session,
    seismogram: AimbatSeismogram,
    name: SeismogramParameter,
    value: datetime | bool | str,
) -> None:
    """Set parameter value for an AimbatSeismogram instance.

    Parameters:
        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: datetime | 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 #

 datetime | bool | str

Value to set.

 required

Raises:

Type Description
ValueError

If no AimbatSeismogram is found with the given ID.
Source code in src/aimbat/lib/seismogram.py 
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 set_seismogram_parameter_by_id(
    session: Session,
    seismogram_id: uuid.UUID,
    name: SeismogramParameter,
    value: datetime | bool | str,
) -> None:
    """Set parameter value for an AimbatSeismogram by ID.

    Parameters:
        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)

snapshot #

Functions:

Name Description
create_snapshot

Create a snapshot of the AIMBAT processing parameters.
delete_snapshot

Delete an AIMBAT parameter snapshot.
delete_snapshot_by_id

Delete an AIMBAT parameter snapshot.
get_snapshots

Get the snapshots for the active avent.
print_snapshot_table

Print a pretty table with AIMBAT snapshots.
rollback_to_snapshot

Rollback to an AIMBAT parameters snapshot.
rollback_to_snapshot_by_id

Rollback to an AIMBAT parameters snapshot.

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/lib/snapshot.py 
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
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def create_snapshot(session: Session, comment: str | None = None) -> None:
    """Create a snapshot of the AIMBAT processing parameters.

    Parameters:
        session: Database session.
        comment: Optional comment.
    """
    active_aimbat_event = 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_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/lib/snapshot.py 
148
149
150
151
152
153
154
155
156
157
158
159
def delete_snapshot(session: Session, snapshot: AimbatSnapshot) -> None:
    """Delete an AIMBAT parameter snapshot.

    Parameters:
        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/lib/snapshot.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def delete_snapshot_by_id(session: Session, snapshot_id: uuid.UUID) -> None:
    """Delete an AIMBAT parameter snapshot.

    Parameters:
        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)

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/lib/snapshot.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
def get_snapshots(
    session: Session, all_events: bool = False
) -> Sequence[AimbatSnapshot]:
    """Get the snapshots for the active avent.

    Parameters:
        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()

print_snapshot_table #

print_snapshot_table(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/lib/snapshot.py 
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
242
243
244
245
246
247
248
def print_snapshot_table(short: bool, all_events: bool) -> None:
    """Print a pretty table with AIMBAT snapshots.

    Parameters:
        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"

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

        if not all_events:
            active_event = 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.datetime_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)

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/lib/snapshot.py 
 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
def rollback_to_snapshot(session: Session, snapshot: AimbatSnapshot) -> None:
    """Rollback to an AIMBAT parameters snapshot.

    Parameters:
        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/lib/snapshot.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def rollback_to_snapshot_by_id(session: Session, snapshot_id: uuid.UUID) -> None:
    """Rollback to an AIMBAT parameters snapshot.

    Parameters:
        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)

station #

Functions:

Name Description
delete_station

Delete an AimbatStation from the database.
delete_station_by_id

Delete an AimbatStation from the database by ID.
dump_station_table

Dump the table data to json.
get_stations_in_event

Get the stations for a particular event.
print_station_table

Prints a pretty table with AIMBAT stations.

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/lib/station.py 
33
34
35
36
37
38
39
40
41
42
43
44
def delete_station(session: Session, station: AimbatStation) -> None:
    """Delete an AimbatStation from the database.

    Parameters:
        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/lib/station.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
def delete_station_by_id(session: Session, station_id: uuid.UUID) -> None:
    """Delete an AimbatStation from the database by ID.

    Parameters:
        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_station_table #

dump_station_table() -> None

Dump the table data to json.

Source code in src/aimbat/lib/station.py 
164
165
166
167
168
169
170
171
def dump_station_table() -> None:
    """Dump the table data to json."""

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

    with Session(engine) as session:
        aimbat_stations = session.exec(select(AimbatStation)).all()
        dump_to_json(aimbat_stations)

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/lib/station.py 
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
def get_stations_in_event(
    session: Session, event: AimbatEvent
) -> Sequence[AimbatStation]:
    """Get the stations for a particular event.

    Parameters:
        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

print_station_table #

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

Prints a pretty table with AIMBAT stations.

Parameters:

Name Type Description Default

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/lib/station.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 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
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
def print_station_table(short: bool, all_events: bool = False) -> None:
    """Prints a pretty table with AIMBAT stations.

    Parameters:
        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

    with Session(engine) as session:
        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 = 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)

typing #

Custom types used in AIMBAT.

Classes:

Name Description
EventParameter

AimbatEvent enum class for typing.
SeismogramParameter

AimbatSeismograParameters enum class for typing.

Attributes:

Name Type Description
EventParameterBool TypeAlias

[TypeAlias][typing.TypeAlias] for AimbatEvent attributes with [bool][] values.
EventParameterFloat TypeAlias

[TypeAlias][typing.TypeAlias] for AimbatEvent attributes with [float][] values.
EventParameterTimedelta TypeAlias

[TypeAlias][typing.TypeAlias] for AimbatEvent attributes with [timedelta][datetime.timedelta] values.

EventParameterBool module-attribute #

EventParameterBool: TypeAlias = Literal[COMPLETED]

[TypeAlias][typing.TypeAlias] for AimbatEvent attributes with [bool][] values.

EventParameterFloat module-attribute #

EventParameterFloat: TypeAlias = Literal[MIN_CCNORM]

[TypeAlias][typing.TypeAlias] for AimbatEvent attributes with [float][] values.

EventParameterTimedelta module-attribute #

EventParameterTimedelta: TypeAlias = Literal[
    WINDOW_PRE, WINDOW_POST
]

[TypeAlias][typing.TypeAlias] for AimbatEvent attributes with [timedelta][datetime.timedelta] values.

EventParameter #

Bases: StrEnum


              flowchart TD
              aimbat.lib.typing.EventParameter[EventParameter]

              

              click aimbat.lib.typing.EventParameter href "" "aimbat.lib.typing.EventParameter"

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 model.

Used by:
Source code in src/aimbat/lib/typing.py 
 7
 8
 9
10
11
12
13
14
15
16
17
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()

SeismogramParameter #

Bases: StrEnum


              flowchart TD
              aimbat.lib.typing.SeismogramParameter[SeismogramParameter]

              

              click aimbat.lib.typing.SeismogramParameter href "" "aimbat.lib.typing.SeismogramParameter"

AimbatSeismograParameters 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.

Used by:
Source code in src/aimbat/lib/typing.py 
32
33
34
35
36
37
38
39
40
41
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()

utils #

Modules:

Name Description
checkdata
sampledata

checkdata #

Functions:

Name Description
checkdata_event

Check if event information is complete.
checkdata_seismogram

Check if seismogram information is complete.
checkdata_station

Check if station information is complete.
run_checks

Run all checks on one or more SAC files.

checkdata_event #

checkdata_event(event: Event) -> list[str]

Check if event information is complete.

Parameters:

Name Type Description Default
event #
 Event

event object to test.

 required
Source code in src/aimbat/lib/utils/checkdata.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
64
65
66
67
def checkdata_event(event: Event) -> list[str]:
    """Check if event information is complete.

    Parameters:
        event: event object to test.
    """

    logger.info("Checking event information.")

    issues = list()

    try:
        assert event.latitude is not None
    except (AssertionError, Exception):
        issue = "No event latitude found in file."
        issues.append(issue)

    try:
        assert event.longitude is not None
    except (AssertionError, Exception):
        issue = "No event longitude found in file."
        issues.append(issue)

    try:
        assert event.time is not None
    except (AssertionError, Exception):
        issue = "No event time found in file."
        issues.append(issue)

    return issues

checkdata_seismogram #

checkdata_seismogram(seismogram: Seismogram) -> list[str]

Check if seismogram information is complete.

Parameters:

Name Type Description Default
seismogram #
 Seismogram

seismogram object to test.

 required
Source code in src/aimbat/lib/utils/checkdata.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def checkdata_seismogram(seismogram: Seismogram) -> list[str]:
    """Check if seismogram information is complete.

    Parameters:
        seismogram: seismogram object to test.
    """

    logger.info("Checking seismogram information.")

    issues = list()
    try:
        assert seismogram.data is not None
        assert len(seismogram.data) > 0
    except (AssertionError, Exception):
        issue = "No seismogram data found in file."
        issues.append(issue)

    return issues

checkdata_station #

checkdata_station(station: Station) -> list[str]

Check if station information is complete.

Parameters:

Name Type Description Default
station #
 Station

station object to test.

 required
Source code in src/aimbat/lib/utils/checkdata.py 
 6
 7
 8
 9
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
def checkdata_station(station: Station) -> list[str]:
    """Check if station information is complete.

    Parameters:
        station: station object to test.
    """

    logger.info("Checking station information.")

    issues = list()

    try:
        assert station.name is not None
    except (AssertionError, Exception):
        issue = "No station name found in file."
        issues.append(issue)

    try:
        assert station.latitude is not None
    except (AssertionError, Exception):
        issue = "No station latitude found in file."
        issues.append(issue)

    try:
        assert station.longitude is not None
    except (AssertionError, Exception):
        issue = "No station longitude found in file."
        issues.append(issue)

    return issues

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/lib/utils/checkdata.py 
 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
def run_checks(sacfiles: list[Path]) -> None:
    """Run all checks on one or more SAC files.

    Parameters:
        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}")

sampledata #

Functions:

Name Description
delete_sampledata

Delete sample data.
download_sampledata

Download sample data.

delete_sampledata #

delete_sampledata() -> None

Delete sample data.

Source code in src/aimbat/lib/utils/sampledata.py 
10
11
12
13
14
15
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/lib/utils/sampledata.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
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)