Skip to content

aimbat._types

Custom types used in AIMBAT.

Three modules are re-exported from here:

  • _enumsEventParameter and SeismogramParameter StrEnums for CLI arg typing and validation.
  • _pydanticPydanticTimestamp, PydanticTimedelta, and constrained variants (PydanticPositiveTimedelta, PydanticNegativeTimedelta) for use in Pydantic models.
  • _sqlalchemySAPandasTimestamp and SAPandasTimedelta SQLAlchemy TypeDecorator classes for storing pandas.Timestamp and pandas.Timedelta values in the database.

Classes:

Name Description
EventParameter

AimbatEvent enum for CLI arg typing.

SAPandasTimedelta

SQLAlchemy TypeDecorator for pandas.Timedelta.

SAPandasTimestamp

SQLAlchemy TypeDecorator for pandas.Timestamp.

SeismogramParameter

AimbatSeismogramParameters enum for CLI arg typing.

EventParameter

Bases: StrEnum

AimbatEvent enum for CLI arg typing.

Source code in src/aimbat/_types/_enums.py
class EventParameter(StrEnum):
    """[`AimbatEvent`][aimbat.models.AimbatEvent] enum for CLI arg typing."""

    COMPLETED = auto()
    MIN_CC = auto()
    WINDOW_PRE = auto()
    WINDOW_POST = auto()
    RAMP_WIDTH = auto()
    BANDPASS_APPLY = auto()
    BANDPASS_FMIN = auto()
    BANDPASS_FMAX = auto()
    MCCC_DAMP = auto()
    MCCC_MIN_CC = auto()

SAPandasTimedelta

Bases: TypeDecorator

SQLAlchemy TypeDecorator for pandas.Timedelta. Stores duration as an integer of nanoseconds for maximum precision.

Source code in src/aimbat/_types/_sqlalchemy.py
class SAPandasTimedelta(TypeDecorator):
    """
    SQLAlchemy TypeDecorator for pandas.Timedelta.
    Stores duration as an integer of nanoseconds for maximum precision.
    """

    impl = BigInteger
    cache_ok = True

    def process_bind_param(self, value: Any, dialect: Dialect) -> int | None:
        if pd.isnull(value):
            return None

        td = value if isinstance(value, Timedelta) else Timedelta(value)
        # Explicit int cast for safety with some SQL drivers
        return int(td.value)

    def process_result_value(self, value: Any, dialect: Dialect) -> Timedelta | None:
        if value is None:
            return None
        # Construct pd.Timedelta from the nanosecond integer
        return Timedelta(value).as_unit("ns")

SAPandasTimestamp

Bases: TypeDecorator

SQLAlchemy TypeDecorator for pandas.Timestamp. Ensures timezone-aware UTC storage in a DateTime column.

Source code in src/aimbat/_types/_sqlalchemy.py
class SAPandasTimestamp(TypeDecorator):
    """
    SQLAlchemy TypeDecorator for pandas.Timestamp.
    Ensures timezone-aware UTC storage in a DateTime column.
    """

    impl = DateTime(timezone=True)
    cache_ok = True

    def process_bind_param(self, value: Any, dialect: Dialect) -> datetime | None:
        if pd.isnull(value):
            return None

        ts = value if isinstance(value, Timestamp) else Timestamp(value)

        # If naive (no TZ), localize to UTC. If aware, convert to UTC.
        if ts.tzinfo is None:
            ts = ts.tz_localize(timezone.utc)
        else:
            ts = ts.tz_convert(timezone.utc)

        # Truncate to microseconds: datetime lacks nanosecond precision.
        return ts.floor("us").to_pydatetime()

    def process_result_value(self, value: Any, dialect: Dialect) -> Timestamp | None:
        if value is None:
            return None

        ts = Timestamp(value)
        # Ensure the returned pandas object is always UTC aware
        if ts.tzinfo is None:
            return ts.tz_localize(timezone.utc)
        return ts.tz_convert(timezone.utc)

SeismogramParameter

Bases: StrEnum

AimbatSeismogramParameters enum for CLI arg typing.

Source code in src/aimbat/_types/_enums.py
class SeismogramParameter(StrEnum):
    """[`AimbatSeismogramParameters`][aimbat.models.AimbatSeismogramParameters] enum for CLI arg typing."""

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