API reference¶

The public Python surface is small: package metadata in rainlog, the Cyclopts app in rainlog.cli_commands, and Database / GraphGrouping in rainlog.db_helpers.

Package wide variables.

`cli_commands` ¶

Main methods to interact with rain data.

`Common` `dataclass` ¶

Class for common db-path parameter.

Source code in src/rainlog/cli_commands.py

@Parameter(name="*")
@dataclass
class Common:
    """Class for common db-path parameter."""

    db_dir: Path = DEFAULT_DB_DIR
    "Path to database file"

`db_dir = DEFAULT_DB_DIR` `class-attribute` `instance-attribute` ¶

Path to database file

`tui(common=None)` ¶

Launch the interactive TUI for browsing rain history.

Source code in src/rainlog/cli_commands.py

@app.default
@app.command()
def tui(common: Common | None = None) -> None:
    """Launch the interactive TUI for browsing rain history."""
    if common is None:
        common = Common()
    with Database(common.db_dir) as database:
        rain_app = RainTuiApp(database=database)
        rain_app.run()

`db_helpers` ¶

Classes and methods around working with the history database.

`Database` ¶

Implements helper methods to add and retrieve data from database.

Source code in src/rainlog/db_helpers.py

class Database:
    """Implements helper methods to add and retrieve data from database."""

    def __init__(self: Self, db_dir: Path) -> None:
        """Create database connection. Also creates database, directory, and table(s) if they don't exist yet."""
        db_dir.mkdir(parents=True, exist_ok=True)
        self.db_connection = sqlite3.connect(database=db_dir / DEFAULT_DB_FILE_NAME)

        # Make sure DB tables exist
        self.db_connection.execute(
            "CREATE TABLE IF NOT EXISTS rain_daily (date INT NOT NULL UNIQUE PRIMARY KEY, rain REAL)"
        )

        self.db_connection.commit()

    def __enter__(self: Self) -> Self:
        """Return self to support use as a context manager."""
        return self

    def __exit__(self: Self, *_: object) -> None:
        """Close the database connection on context manager exit."""
        self.db_connection.close()

    def add_rain_record(self: Self, date: datetime, amount: float) -> None:
        """Add a record / measurement of rain to the DB."""
        self.db_connection.execute(
            "INSERT INTO rain_daily (date, rain) VALUES (?,?)",
            (date.timestamp(), amount),
        )

        self.db_connection.commit()

    def update_rain_record(self: Self, date: datetime, amount: float) -> None:
        """Update a record / measurement of rain in the DB."""
        self.db_connection.execute(
            "UPDATE rain_daily set rain = :rain WHERE date = :ts",
            {"rain": amount, "ts": date.timestamp()},
        )

        self.db_connection.commit()

    def get_single_day_rain(self: Self, date: datetime) -> float | None:
        """Return amount of rain for that day as a float or False if no rain
        record was found for that particular date.
        """
        cursor = self.db_connection.cursor()
        cursor.execute(
            "SELECT rain FROM rain_daily WHERE date = ?",
            (date.timestamp(),),
        )
        cursor_data = cursor.fetchone()
        if cursor_data:
            return float(cursor_data[0])

        return False

    def get_rain(
        self: Self,
        history_size: int,
        group: GraphGrouping,
        offset: int = 0,
    ) -> list[tuple[str, float]]:
        """Get 'history_size' number of rain records, skipping 'offset' most-recent groups."""
        return_list: list[tuple[str, float]] = []
        group_id = None
        group_sum = 0.0
        groups_skipped = 0

        for row in self.db_connection.execute("SELECT date, rain FROM rain_daily ORDER BY date DESC"):
            current_group_id = Database._determine_group(
                group=group,
                group_date=datetime.fromtimestamp(row[0]).astimezone(tz=LOCAL_TZ),
            )

            if not group_id:
                group_id = current_group_id

            if current_group_id == group_id:
                group_sum += row[1]
            else:
                if groups_skipped >= offset:
                    return_list.append((group_id, group_sum))
                else:
                    groups_skipped += 1
                group_id = current_group_id
                group_sum = row[1]

            if len(return_list) >= history_size:
                break

        if len(return_list) < history_size and group_id and groups_skipped >= offset:
            return_list.append((group_id, group_sum))

        return return_list

    def get_moisture_index(
        self: Self,
        history_size: int,
        group: GraphGrouping,
        offset: int = 0,
        decay: float = 0.85,
    ) -> list[tuple[str, float]]:
        """Compute soil moisture index via exponential decay and return paginated groups.

        Applies moisture = moisture * decay + rain sequentially over all records in
        ascending date order (initial moisture = 0). Groups results via _determine_group,
        keeping the last moisture value per group (end-of-period moisture). Returns at most
        history_size groups in descending order, skipping the offset most-recent groups.
        """
        current_moisture = 0.0
        group_moisture: dict[str, float] = {}
        previous_record_date: datetime | None = None

        for row in self.db_connection.execute("SELECT date, rain FROM rain_daily ORDER BY date ASC"):
            record_date = datetime.fromtimestamp(row[0]).astimezone(tz=LOCAL_TZ)
            if previous_record_date is not None:
                elapsed_days = (record_date.date() - previous_record_date.date()).days
                if elapsed_days > 1:
                    current_moisture *= decay ** (elapsed_days - 1)
            current_moisture = current_moisture * decay + row[1]
            previous_record_date = record_date
            group_label = Database._determine_group(
                group=group,
                group_date=record_date,
            )
            group_moisture[group_label] = current_moisture

        descending_groups = list(group_moisture.items())
        descending_groups.reverse()
        return descending_groups[offset : offset + history_size]

    def get_current_streak(self: Self) -> tuple[str, int]:
        """Return the type and length of the current consecutive wet or dry streak.

        Walks backward from the most recent record. Returns ('dry', 0) for an empty DB.
        """
        streak_type: str | None = None
        streak_count = 0

        for row in self.db_connection.execute("SELECT rain FROM rain_daily ORDER BY date DESC"):
            rain = row[0]
            row_type = "wet" if rain > 0 else "dry"

            if streak_type is None:
                streak_type = row_type
                streak_count = 1
            elif row_type == streak_type:
                streak_count += 1
            else:
                break

        if streak_type is None:
            return ("dry", 0)

        return (streak_type, streak_count)

    def get_most_recent_date(self: Self) -> datetime | None:
        """Return the datetime of the most recent rain record, or None if the DB is empty."""
        cursor = self.db_connection.cursor()
        cursor.execute("SELECT MAX(date) FROM rain_daily")
        row = cursor.fetchone()
        if row and row[0] is not None:
            return datetime.fromtimestamp(row[0]).astimezone(tz=LOCAL_TZ)
        return None

    def get_earliest_date(self: Self) -> datetime | None:
        """Return the datetime of the earliest rain record, or None if the DB is empty."""
        cursor = self.db_connection.cursor()
        cursor.execute("SELECT MIN(date) FROM rain_daily")
        row = cursor.fetchone()
        if row and row[0] is not None:
            return datetime.fromtimestamp(row[0]).astimezone(tz=LOCAL_TZ)
        return None

    @staticmethod
    def _determine_group(group: str, group_date: datetime) -> str:
        """Determine group value for grouping of data."""
        match group:
            case GraphGrouping.daily:
                format_for_grouping = "%Y-%m-%d"
            case GraphGrouping.weekly:
                format_for_grouping = "%Y-%W"
            case GraphGrouping.monthly:
                format_for_grouping = "%Y-%m"
            case GraphGrouping.yearly | GraphGrouping.annually:
                format_for_grouping = "%Y"
            case _:
                raise ValueError(f"Unrecognized value for {group=}")

        group_id: str = group_date.strftime(format_for_grouping)

        if group == "weekly":
            year_part, week_part = group_id.split("-", 1)
            group_id = f"{year_part}-{week_part}"

        if group_id is None:
            raise ValueError(f"Database._determine_group({group=}, {group_date=}) -> {group_id=}")

        return group_id

`enter()` ¶

Return self to support use as a context manager.

Source code in src/rainlog/db_helpers.py

def __enter__(self: Self) -> Self:
    """Return self to support use as a context manager."""
    return self

`exit(*_)` ¶

Close the database connection on context manager exit.

Source code in src/rainlog/db_helpers.py

def __exit__(self: Self, *_: object) -> None:
    """Close the database connection on context manager exit."""
    self.db_connection.close()

`init(db_dir)` ¶

Create database connection. Also creates database, directory, and table(s) if they don't exist yet.

Source code in src/rainlog/db_helpers.py

def __init__(self: Self, db_dir: Path) -> None:
    """Create database connection. Also creates database, directory, and table(s) if they don't exist yet."""
    db_dir.mkdir(parents=True, exist_ok=True)
    self.db_connection = sqlite3.connect(database=db_dir / DEFAULT_DB_FILE_NAME)

    # Make sure DB tables exist
    self.db_connection.execute(
        "CREATE TABLE IF NOT EXISTS rain_daily (date INT NOT NULL UNIQUE PRIMARY KEY, rain REAL)"
    )

    self.db_connection.commit()

`add_rain_record(date, amount)` ¶

Add a record / measurement of rain to the DB.

Source code in src/rainlog/db_helpers.py

def add_rain_record(self: Self, date: datetime, amount: float) -> None:
    """Add a record / measurement of rain to the DB."""
    self.db_connection.execute(
        "INSERT INTO rain_daily (date, rain) VALUES (?,?)",
        (date.timestamp(), amount),
    )

    self.db_connection.commit()

`get_current_streak()` ¶

Return the type and length of the current consecutive wet or dry streak.

Walks backward from the most recent record. Returns ('dry', 0) for an empty DB.

Source code in src/rainlog/db_helpers.py

def get_current_streak(self: Self) -> tuple[str, int]:
    """Return the type and length of the current consecutive wet or dry streak.

    Walks backward from the most recent record. Returns ('dry', 0) for an empty DB.
    """
    streak_type: str | None = None
    streak_count = 0

    for row in self.db_connection.execute("SELECT rain FROM rain_daily ORDER BY date DESC"):
        rain = row[0]
        row_type = "wet" if rain > 0 else "dry"

        if streak_type is None:
            streak_type = row_type
            streak_count = 1
        elif row_type == streak_type:
            streak_count += 1
        else:
            break

    if streak_type is None:
        return ("dry", 0)

    return (streak_type, streak_count)

`get_earliest_date()` ¶

Return the datetime of the earliest rain record, or None if the DB is empty.

Source code in src/rainlog/db_helpers.py

def get_earliest_date(self: Self) -> datetime | None:
    """Return the datetime of the earliest rain record, or None if the DB is empty."""
    cursor = self.db_connection.cursor()
    cursor.execute("SELECT MIN(date) FROM rain_daily")
    row = cursor.fetchone()
    if row and row[0] is not None:
        return datetime.fromtimestamp(row[0]).astimezone(tz=LOCAL_TZ)
    return None

`get_moisture_index(history_size, group, offset=0, decay=0.85)` ¶

Compute soil moisture index via exponential decay and return paginated groups.

Applies moisture = moisture * decay + rain sequentially over all records in ascending date order (initial moisture = 0). Groups results via _determine_group, keeping the last moisture value per group (end-of-period moisture). Returns at most history_size groups in descending order, skipping the offset most-recent groups.

Source code in src/rainlog/db_helpers.py

def get_moisture_index(
    self: Self,
    history_size: int,
    group: GraphGrouping,
    offset: int = 0,
    decay: float = 0.85,
) -> list[tuple[str, float]]:
    """Compute soil moisture index via exponential decay and return paginated groups.

    Applies moisture = moisture * decay + rain sequentially over all records in
    ascending date order (initial moisture = 0). Groups results via _determine_group,
    keeping the last moisture value per group (end-of-period moisture). Returns at most
    history_size groups in descending order, skipping the offset most-recent groups.
    """
    current_moisture = 0.0
    group_moisture: dict[str, float] = {}
    previous_record_date: datetime | None = None

    for row in self.db_connection.execute("SELECT date, rain FROM rain_daily ORDER BY date ASC"):
        record_date = datetime.fromtimestamp(row[0]).astimezone(tz=LOCAL_TZ)
        if previous_record_date is not None:
            elapsed_days = (record_date.date() - previous_record_date.date()).days
            if elapsed_days > 1:
                current_moisture *= decay ** (elapsed_days - 1)
        current_moisture = current_moisture * decay + row[1]
        previous_record_date = record_date
        group_label = Database._determine_group(
            group=group,
            group_date=record_date,
        )
        group_moisture[group_label] = current_moisture

    descending_groups = list(group_moisture.items())
    descending_groups.reverse()
    return descending_groups[offset : offset + history_size]

`get_most_recent_date()` ¶

Return the datetime of the most recent rain record, or None if the DB is empty.

Source code in src/rainlog/db_helpers.py

def get_most_recent_date(self: Self) -> datetime | None:
    """Return the datetime of the most recent rain record, or None if the DB is empty."""
    cursor = self.db_connection.cursor()
    cursor.execute("SELECT MAX(date) FROM rain_daily")
    row = cursor.fetchone()
    if row and row[0] is not None:
        return datetime.fromtimestamp(row[0]).astimezone(tz=LOCAL_TZ)
    return None

`get_rain(history_size, group, offset=0)` ¶

Get 'history_size' number of rain records, skipping 'offset' most-recent groups.

Source code in src/rainlog/db_helpers.py

def get_rain(
    self: Self,
    history_size: int,
    group: GraphGrouping,
    offset: int = 0,
) -> list[tuple[str, float]]:
    """Get 'history_size' number of rain records, skipping 'offset' most-recent groups."""
    return_list: list[tuple[str, float]] = []
    group_id = None
    group_sum = 0.0
    groups_skipped = 0

    for row in self.db_connection.execute("SELECT date, rain FROM rain_daily ORDER BY date DESC"):
        current_group_id = Database._determine_group(
            group=group,
            group_date=datetime.fromtimestamp(row[0]).astimezone(tz=LOCAL_TZ),
        )

        if not group_id:
            group_id = current_group_id

        if current_group_id == group_id:
            group_sum += row[1]
        else:
            if groups_skipped >= offset:
                return_list.append((group_id, group_sum))
            else:
                groups_skipped += 1
            group_id = current_group_id
            group_sum = row[1]

        if len(return_list) >= history_size:
            break

    if len(return_list) < history_size and group_id and groups_skipped >= offset:
        return_list.append((group_id, group_sum))

    return return_list

`get_single_day_rain(date)` ¶

Return amount of rain for that day as a float or False if no rain record was found for that particular date.

Source code in src/rainlog/db_helpers.py

def get_single_day_rain(self: Self, date: datetime) -> float | None:
    """Return amount of rain for that day as a float or False if no rain
    record was found for that particular date.
    """
    cursor = self.db_connection.cursor()
    cursor.execute(
        "SELECT rain FROM rain_daily WHERE date = ?",
        (date.timestamp(),),
    )
    cursor_data = cursor.fetchone()
    if cursor_data:
        return float(cursor_data[0])

    return False

`update_rain_record(date, amount)` ¶

Update a record / measurement of rain in the DB.

Source code in src/rainlog/db_helpers.py

def update_rain_record(self: Self, date: datetime, amount: float) -> None:
    """Update a record / measurement of rain in the DB."""
    self.db_connection.execute(
        "UPDATE rain_daily set rain = :rain WHERE date = :ts",
        {"rain": amount, "ts": date.timestamp()},
    )

    self.db_connection.commit()

`GraphGrouping` ¶

Bases: str, Enum

Provides possible values for grouping of graphs.

Source code in src/rainlog/db_helpers.py

class GraphGrouping(str, Enum):
    """Provides possible values for grouping of graphs."""

    daily = "daily"
    weekly = "weekly"
    monthly = "monthly"
    yearly = "yearly"
    annually = "annually"

`tui` ¶

Interactive TUI for browsing rain history.