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_ytd_by_year(
        self: Self,
        reference_date: datetime | None = None,
    ) -> list[tuple[str, float]]:
        """Return Jan 1 → reference_date total for each year with records, oldest first.

        reference_date defaults to today (local time). Records whose calendar
        month/day falls after reference_date's month/day are excluded so all
        years are compared over the same portion of the calendar.
        """
        if reference_date is None:
            reference_date = datetime.now(tz=LOCAL_TZ)
        cutoff = (reference_date.month, reference_date.day)
        year_totals: dict[str, float] = {}
        for row in self.db_connection.execute("SELECT date, rain FROM rain_daily ORDER BY date ASC"):
            recorded_date = datetime.fromtimestamp(row[0], tz=LOCAL_TZ)
            if (recorded_date.month, recorded_date.day) <= cutoff:
                year_label = str(recorded_date.year)
                year_totals[year_label] = year_totals.get(year_label, 0.0) + row[1]
        return sorted(year_totals.items())

    def get_monthly_comparison(
        self: Self,
        num_years: int,
        reference_date: datetime | None = None,
    ) -> tuple[list[str], dict[str, list[float]]]:
        """Return monthly totals for current year and num_years prior, Jan through current month.

        reference_date defaults to today (local time). Returns (month_labels, series) where
        series maps year_label to a list of monthly totals. Only years with at least one
        record in the window are included. Series is ordered oldest year first.
        """
        if reference_date is None:
            reference_date = datetime.now(tz=LOCAL_TZ)
        months_count = reference_date.month
        current_year = reference_date.year
        labels = [datetime(current_year, month_num, 1).strftime("%b") for month_num in range(1, months_count + 1)]
        target_years = list(range(current_year - num_years, current_year + 1))
        year_month_totals: dict[int, dict[int, float]] = {year: {} for year in target_years}
        for row in self.db_connection.execute("SELECT date, rain FROM rain_daily ORDER BY date ASC"):
            recorded_date = datetime.fromtimestamp(row[0], tz=LOCAL_TZ)
            if recorded_date.year in year_month_totals and recorded_date.month <= months_count:
                month_bucket = year_month_totals[recorded_date.year]
                month_num = recorded_date.month
                month_bucket[month_num] = month_bucket.get(month_num, 0.0) + row[1]
        series: dict[str, list[float]] = {}
        for year in target_years:
            monthly_values = [year_month_totals[year].get(month_num, 0.0) for month_num in range(1, months_count + 1)]
            if any(value > 0 for value in monthly_values):
                series[str(year)] = monthly_values
        return labels, series

    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_monthly_comparison(num_years, reference_date=None)

Return monthly totals for current year and num_years prior, Jan through current month.

reference_date defaults to today (local time). Returns (month_labels, series) where series maps year_label to a list of monthly totals. Only years with at least one record in the window are included. Series is ordered oldest year first.

Source code in src/rainlog/db_helpers.py

def get_monthly_comparison(
    self: Self,
    num_years: int,
    reference_date: datetime | None = None,
) -> tuple[list[str], dict[str, list[float]]]:
    """Return monthly totals for current year and num_years prior, Jan through current month.

    reference_date defaults to today (local time). Returns (month_labels, series) where
    series maps year_label to a list of monthly totals. Only years with at least one
    record in the window are included. Series is ordered oldest year first.
    """
    if reference_date is None:
        reference_date = datetime.now(tz=LOCAL_TZ)
    months_count = reference_date.month
    current_year = reference_date.year
    labels = [datetime(current_year, month_num, 1).strftime("%b") for month_num in range(1, months_count + 1)]
    target_years = list(range(current_year - num_years, current_year + 1))
    year_month_totals: dict[int, dict[int, float]] = {year: {} for year in target_years}
    for row in self.db_connection.execute("SELECT date, rain FROM rain_daily ORDER BY date ASC"):
        recorded_date = datetime.fromtimestamp(row[0], tz=LOCAL_TZ)
        if recorded_date.year in year_month_totals and recorded_date.month <= months_count:
            month_bucket = year_month_totals[recorded_date.year]
            month_num = recorded_date.month
            month_bucket[month_num] = month_bucket.get(month_num, 0.0) + row[1]
    series: dict[str, list[float]] = {}
    for year in target_years:
        monthly_values = [year_month_totals[year].get(month_num, 0.0) for month_num in range(1, months_count + 1)]
        if any(value > 0 for value in monthly_values):
            series[str(year)] = monthly_values
    return labels, series

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

get_ytd_by_year(reference_date=None)

Return Jan 1 → reference_date total for each year with records, oldest first.

reference_date defaults to today (local time). Records whose calendar month/day falls after reference_date's month/day are excluded so all years are compared over the same portion of the calendar.

Source code in src/rainlog/db_helpers.py

def get_ytd_by_year(
    self: Self,
    reference_date: datetime | None = None,
) -> list[tuple[str, float]]:
    """Return Jan 1 → reference_date total for each year with records, oldest first.

    reference_date defaults to today (local time). Records whose calendar
    month/day falls after reference_date's month/day are excluded so all
    years are compared over the same portion of the calendar.
    """
    if reference_date is None:
        reference_date = datetime.now(tz=LOCAL_TZ)
    cutoff = (reference_date.month, reference_date.day)
    year_totals: dict[str, float] = {}
    for row in self.db_connection.execute("SELECT date, rain FROM rain_daily ORDER BY date ASC"):
        recorded_date = datetime.fromtimestamp(row[0], tz=LOCAL_TZ)
        if (recorded_date.month, recorded_date.day) <= cutoff:
            year_label = str(recorded_date.year)
            year_totals[year_label] = year_totals.get(year_label, 0.0) + row[1]
    return sorted(year_totals.items())

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.

AddRainModal

Bases: ModalScreen[AddRainResult | None]

Modal form for adding a new rain record.

Source code in src/rainlog/tui.py

class AddRainModal(ModalScreen[AddRainResult | None]):
    """Modal form for adding a new rain record."""

    DEFAULT_CSS = """
    AddRainModal {
        align: center middle;
    }
    AddRainModal > Vertical {
        width: 44;
        height: auto;
        padding: 1 2;
        border: thick $primary;
    }
    AddRainModal .field-error {
        border: solid red;
    }
    """

    BINDINGS: ClassVar[list[Binding]] = [
        Binding("escape", "cancel", "Cancel"),
        Binding("ctrl+s", "submit", "Save"),
    ]

    def compose(self) -> ComposeResult:
        """Render the add-rain form."""
        today = datetime.now(tz=LOCAL_TZ).strftime("%Y-%m-%d")
        with Vertical():
            yield Label("Add Rain Record")
            yield Label("Date (YYYY-MM-DD)")
            yield Input(value=today, id="date_input")
            yield Label("Amount (mm)")
            yield Input(placeholder="0.0", id="amount_input")
            yield Label("Back-fill zeros to last record")
            yield Switch(id="backfill_switch", value=True)
            yield Button("Save", id="save_btn", variant="primary")

    def on_button_pressed(self, event: Button.Pressed) -> None:
        """Forward Save button press to submit action."""
        if event.button.id == "save_btn":
            self.action_submit()

    def action_cancel(self) -> None:
        """Dismiss without saving."""
        self.dismiss(None)

    def action_submit(self) -> None:
        """Validate inputs and dismiss with result, or mark invalid fields."""
        date_input = self.query_one("#date_input", Input)
        amount_input = self.query_one("#amount_input", Input)
        backfill_switch = self.query_one("#backfill_switch", Switch)

        date_input.remove_class("field-error")
        amount_input.remove_class("field-error")

        parsed_date = _parse_date_input(date_input.value)
        parsed_amount = _parse_amount_input(amount_input.value)

        if parsed_date is None:
            date_input.add_class("field-error")
            return
        if parsed_amount is None:
            amount_input.add_class("field-error")
            return

        self.dismiss(AddRainResult(date=parsed_date, amount=parsed_amount, backfill=backfill_switch.value))

action_cancel()

Dismiss without saving.

Source code in src/rainlog/tui.py

def action_cancel(self) -> None:
    """Dismiss without saving."""
    self.dismiss(None)

action_submit()

Validate inputs and dismiss with result, or mark invalid fields.

Source code in src/rainlog/tui.py

def action_submit(self) -> None:
    """Validate inputs and dismiss with result, or mark invalid fields."""
    date_input = self.query_one("#date_input", Input)
    amount_input = self.query_one("#amount_input", Input)
    backfill_switch = self.query_one("#backfill_switch", Switch)

    date_input.remove_class("field-error")
    amount_input.remove_class("field-error")

    parsed_date = _parse_date_input(date_input.value)
    parsed_amount = _parse_amount_input(amount_input.value)

    if parsed_date is None:
        date_input.add_class("field-error")
        return
    if parsed_amount is None:
        amount_input.add_class("field-error")
        return

    self.dismiss(AddRainResult(date=parsed_date, amount=parsed_amount, backfill=backfill_switch.value))

compose()

Render the add-rain form.

Source code in src/rainlog/tui.py

def compose(self) -> ComposeResult:
    """Render the add-rain form."""
    today = datetime.now(tz=LOCAL_TZ).strftime("%Y-%m-%d")
    with Vertical():
        yield Label("Add Rain Record")
        yield Label("Date (YYYY-MM-DD)")
        yield Input(value=today, id="date_input")
        yield Label("Amount (mm)")
        yield Input(placeholder="0.0", id="amount_input")
        yield Label("Back-fill zeros to last record")
        yield Switch(id="backfill_switch", value=True)
        yield Button("Save", id="save_btn", variant="primary")

on_button_pressed(event)

Forward Save button press to submit action.

Source code in src/rainlog/tui.py

def on_button_pressed(self, event: Button.Pressed) -> None:
    """Forward Save button press to submit action."""
    if event.button.id == "save_btn":
        self.action_submit()

AddRainResult dataclass

Payload returned by AddRainModal on successful submission.

Source code in src/rainlog/tui.py

@dataclass
class AddRainResult:
    """Payload returned by AddRainModal on successful submission."""

    date: datetime
    amount: float
    backfill: bool

BarChartWidget

Bases: Widget

Vertical bar chart rendered with Unicode block characters.

Source code in src/rainlog/tui.py

class BarChartWidget(Widget):
    """Vertical bar chart rendered with Unicode block characters."""

    DEFAULT_CSS = """
    BarChartWidget {
        width: 1fr;
        height: 1fr;
    }
    """

    def __init__(self) -> None:
        """Initialise with empty dataset."""
        super().__init__()
        self._data: list[tuple[str, float]] = []
        self._tentative_labels: set[str] = set()
        self._selected_index: int | None = None

    def set_data(
        self,
        data: list[tuple[str, float]],
        tentative_labels: set[str],
        selected_index: int | None = None,
    ) -> None:
        """Replace chart data and trigger a repaint."""
        self._data = data
        self._tentative_labels = tentative_labels
        self._selected_index = selected_index
        self.refresh()

    def _append_chart_rows(
        self,
        result: Text,
        bar_entries: list[tuple[int, str]],
        bar_width: int,
        chart_height: int,
    ) -> None:
        """Append one character row per chart row to result.

        bar_entries is a list of (height, color) per bar.
        """
        for row in range(chart_height, 0, -1):
            for index, (bar_height, color) in enumerate(bar_entries):
                if index > 0:
                    result.append(" ")
                if bar_height >= row:
                    result.append("█" * bar_width, style=color)
                else:
                    result.append(" " * bar_width)
            result.append("\n")

    def _append_value_row(
        self,
        result: Text,
        values: list[float],
        colors: list[str],
        bar_width: int,
    ) -> None:
        """Append a row showing each bar's rainfall amount in its bar colour."""
        for index, (value, color) in enumerate(zip(values, colors, strict=True)):
            if index > 0:
                result.append(" ")
            result.append(_format_rain_value(value, bar_width), style=color)
        result.append("\n")

    def _compute_colors(
        self,
        values: list[float],
        tentative_flags: list[bool],
        max_value: float,
    ) -> tuple[list[str], list[str]]:
        """Return (bar_colors, value_colors) for each bar.

        Selected bar is white; tentative bars use the amber palette; others use
        the blue palette. Value colors use a brightness floor (max_intensity=0.5)
        so dark bars remain legible on black terminals.
        """
        bar_colors: list[str] = []
        value_colors: list[str] = []
        for index, (value, flag) in enumerate(zip(values, tentative_flags, strict=True)):
            if index == self._selected_index:
                bar_colors.append("rgb(255,255,255)")
                value_colors.append("rgb(255,255,255)")
            elif flag:
                bar_colors.append(_tentative_bar_color(value, max_value))
                value_colors.append(_tentative_bar_color(value, max_value))
            else:
                bar_colors.append(_bar_color(value, max_value))
                value_colors.append(_bar_color(value, max_value, max_intensity=0.5))
        return bar_colors, value_colors

    def render(self) -> RenderableType:
        """Draw bars scaled to the current widget height and width, coloured by rain intensity."""
        if not self._data:
            return Text("No data")

        chart_height = max(1, self.size.height - 3)
        labels = [label for label, _ in self._data]
        values = [rain for _, rain in self._data]

        bar_count = len(values)
        bar_width = max(1, (self.size.width - 4) // bar_count - 1)

        heights = calculate_bar_heights(values, chart_height)
        max_value = max(values)
        tentative_flags = [label in self._tentative_labels for label in labels]
        colors, value_colors = self._compute_colors(values, tentative_flags, max_value)

        bar_entries = list(zip(heights, colors, strict=True))
        result = Text()
        self._append_chart_rows(result, bar_entries, bar_width, chart_height)
        self._append_value_row(result, values, value_colors, bar_width)
        label_line = " ".join(label[-bar_width:].ljust(bar_width) for label in labels)
        result.append(label_line)

        return result

__init__()

Initialise with empty dataset.

Source code in src/rainlog/tui.py

def __init__(self) -> None:
    """Initialise with empty dataset."""
    super().__init__()
    self._data: list[tuple[str, float]] = []
    self._tentative_labels: set[str] = set()
    self._selected_index: int | None = None

render()

Draw bars scaled to the current widget height and width, coloured by rain intensity.

Source code in src/rainlog/tui.py

def render(self) -> RenderableType:
    """Draw bars scaled to the current widget height and width, coloured by rain intensity."""
    if not self._data:
        return Text("No data")

    chart_height = max(1, self.size.height - 3)
    labels = [label for label, _ in self._data]
    values = [rain for _, rain in self._data]

    bar_count = len(values)
    bar_width = max(1, (self.size.width - 4) // bar_count - 1)

    heights = calculate_bar_heights(values, chart_height)
    max_value = max(values)
    tentative_flags = [label in self._tentative_labels for label in labels]
    colors, value_colors = self._compute_colors(values, tentative_flags, max_value)

    bar_entries = list(zip(heights, colors, strict=True))
    result = Text()
    self._append_chart_rows(result, bar_entries, bar_width, chart_height)
    self._append_value_row(result, values, value_colors, bar_width)
    label_line = " ".join(label[-bar_width:].ljust(bar_width) for label in labels)
    result.append(label_line)

    return result

set_data(data, tentative_labels, selected_index=None)

Replace chart data and trigger a repaint.

Source code in src/rainlog/tui.py

def set_data(
    self,
    data: list[tuple[str, float]],
    tentative_labels: set[str],
    selected_index: int | None = None,
) -> None:
    """Replace chart data and trigger a repaint."""
    self._data = data
    self._tentative_labels = tentative_labels
    self._selected_index = selected_index
    self.refresh()

ChartMode

Bases: str, Enum

Chart display mode: rainfall amounts, soil moisture index, or year-over-year comparison.

Source code in src/rainlog/tui.py

class ChartMode(str, Enum):
    """Chart display mode: rainfall amounts, soil moisture index, or year-over-year comparison."""

    rainfall = "rainfall"
    moisture = "moisture"
    comparison = "comparison"

ComparisonChartWidget

Bases: Widget

Bar chart showing the same monthly period across multiple years side by side.

Source code in src/rainlog/tui.py

class ComparisonChartWidget(Widget):
    """Bar chart showing the same monthly period across multiple years side by side."""

    DEFAULT_CSS = """
    ComparisonChartWidget {
        width: 1fr;
        height: 1fr;
    }
    """

    def __init__(self) -> None:
        """Initialise with empty data."""
        super().__init__()
        self._labels: list[str] = []
        self._series: dict[str, list[float]] = {}

    def set_data(
        self,
        labels: list[str],
        series: dict[str, list[float]],
    ) -> None:
        """Replace chart data and trigger a repaint."""
        self._labels = labels
        self._series = series
        self.refresh()

    def _build_layout(self) -> _ComparisonLayout:
        """Compute bar geometry and normalised heights for the current widget size."""
        years = list(self._series.keys())
        num_years = len(years)
        months_count = len(self._labels)
        chart_height = max(1, self.size.height - 4)

        available_width = max(1, self.size.width - 4)
        bar_width = max(
            1,
            (available_width - (months_count - 1) * 2 - months_count * (num_years - 1)) // (months_count * num_years),
        )

        all_values = [value for year_values in self._series.values() for value in year_values]
        global_max = max(all_values) if any(value > 0 for value in all_values) else 1.0
        heights_per_year: dict[str, list[int]] = {
            year: [round(value / global_max * chart_height) for value in self._series[year]] for year in years
        }
        year_colors = dict(zip(years, _comparison_year_colors(num_years), strict=True))
        return _ComparisonLayout(years, bar_width, chart_height, heights_per_year, year_colors)

    def _append_cluster(
        self,
        result: Text,
        layout: _ComparisonLayout,
        month_index: int,
        row: int,
    ) -> None:
        """Append one month-cluster of bars for a single chart row."""
        for year_index, year in enumerate(layout.years):
            if year_index > 0:
                result.append(" ")  # 1-char intra-cluster gap
            bar_height = layout.heights_per_year[year][month_index]
            color = layout.year_colors[year]
            if bar_height >= row:
                result.append("█" * layout.bar_width, style=color)
            else:
                result.append(" " * layout.bar_width)

    def _append_chart_rows(self, result: Text, layout: _ComparisonLayout, months_count: int) -> None:
        """Append bar rows (top to bottom) to result."""
        for row in range(layout.chart_height, 0, -1):
            for month_index in range(months_count):
                if month_index > 0:
                    result.append("  ")  # 2-char inter-cluster gap
                self._append_cluster(result, layout, month_index, row)
            result.append("\n")

    def _append_labels_and_legend(
        self,
        result: Text,
        years: list[str],
        year_colors: dict[str, str],
        cluster_width: int,
    ) -> None:
        """Append the month-label row and year-legend row to result."""
        for month_index, label in enumerate(self._labels):
            if month_index > 0:
                result.append("  ")
            result.append(label[-cluster_width:].ljust(cluster_width))
        result.append("\n")
        for year_index, year in enumerate(years):
            if year_index > 0:
                result.append("  ")
            result.append("■", style=year_colors[year])
            result.append(f" {year}")

    def render(self) -> RenderableType:
        """Draw grouped bars: one cluster per month, one bar per year."""
        if not self._series or not self._labels:
            return Text("No comparison data")

        months_count = len(self._labels)
        layout = self._build_layout()
        result = Text()
        self._append_chart_rows(result, layout, months_count)

        cluster_width = len(layout.years) * layout.bar_width + (len(layout.years) - 1)
        self._append_labels_and_legend(result, layout.years, layout.year_colors, cluster_width)

        return result

__init__()

Initialise with empty data.

Source code in src/rainlog/tui.py

def __init__(self) -> None:
    """Initialise with empty data."""
    super().__init__()
    self._labels: list[str] = []
    self._series: dict[str, list[float]] = {}

render()

Draw grouped bars: one cluster per month, one bar per year.

Source code in src/rainlog/tui.py

def render(self) -> RenderableType:
    """Draw grouped bars: one cluster per month, one bar per year."""
    if not self._series or not self._labels:
        return Text("No comparison data")

    months_count = len(self._labels)
    layout = self._build_layout()
    result = Text()
    self._append_chart_rows(result, layout, months_count)

    cluster_width = len(layout.years) * layout.bar_width + (len(layout.years) - 1)
    self._append_labels_and_legend(result, layout.years, layout.year_colors, cluster_width)

    return result

set_data(labels, series)

Replace chart data and trigger a repaint.

Source code in src/rainlog/tui.py

def set_data(
    self,
    labels: list[str],
    series: dict[str, list[float]],
) -> None:
    """Replace chart data and trigger a repaint."""
    self._labels = labels
    self._series = series
    self.refresh()

EditRainModal

Bases: ModalScreen[EditRainResult | None]

Modal form for editing an existing rain record.

Source code in src/rainlog/tui.py

class EditRainModal(ModalScreen[EditRainResult | None]):
    """Modal form for editing an existing rain record."""

    DEFAULT_CSS = """
    EditRainModal {
        align: center middle;
    }
    EditRainModal > Vertical {
        width: 44;
        height: auto;
        padding: 1 2;
        border: thick $primary;
    }
    EditRainModal .field-error {
        border: solid red;
    }
    """

    BINDINGS: ClassVar[list[Binding]] = [
        Binding("escape", "cancel", "Cancel"),
        Binding("ctrl+s", "submit", "Save"),
    ]

    def __init__(self, prefill_date: str = "", prefill_amount: str = "") -> None:
        """Initialise with optional pre-filled values from a selected bar."""
        super().__init__()
        self._prefill_date = prefill_date
        self._prefill_amount = prefill_amount

    def compose(self) -> ComposeResult:
        """Render the edit-rain form."""
        with Vertical():
            yield Label("Edit Rain Record")
            yield Label("Date (YYYY-MM-DD)")
            yield Input(value=self._prefill_date, placeholder="YYYY-MM-DD", id="date_input")
            yield Label("Amount (mm)")
            yield Input(value=self._prefill_amount, placeholder="0.0", id="amount_input")
            yield Button("Save", id="save_btn", variant="primary")

    def on_button_pressed(self, event: Button.Pressed) -> None:
        """Forward Save button press to submit action."""
        if event.button.id == "save_btn":
            self.action_submit()

    def action_cancel(self) -> None:
        """Dismiss without saving."""
        self.dismiss(None)

    def action_submit(self) -> None:
        """Validate and dismiss with result, or mark invalid fields."""
        date_input = self.query_one("#date_input", Input)
        amount_input = self.query_one("#amount_input", Input)

        date_input.remove_class("field-error")
        amount_input.remove_class("field-error")

        parsed_date = _parse_date_input(date_input.value)
        parsed_amount = _parse_amount_input(amount_input.value)

        if parsed_date is None:
            date_input.add_class("field-error")
            return
        if parsed_amount is None:
            amount_input.add_class("field-error")
            return

        self.dismiss(EditRainResult(date=parsed_date, amount=parsed_amount))

__init__(prefill_date='', prefill_amount='')

Initialise with optional pre-filled values from a selected bar.

Source code in src/rainlog/tui.py

def __init__(self, prefill_date: str = "", prefill_amount: str = "") -> None:
    """Initialise with optional pre-filled values from a selected bar."""
    super().__init__()
    self._prefill_date = prefill_date
    self._prefill_amount = prefill_amount

action_cancel()

Dismiss without saving.

Source code in src/rainlog/tui.py

def action_cancel(self) -> None:
    """Dismiss without saving."""
    self.dismiss(None)

action_submit()

Validate and dismiss with result, or mark invalid fields.

Source code in src/rainlog/tui.py

def action_submit(self) -> None:
    """Validate and dismiss with result, or mark invalid fields."""
    date_input = self.query_one("#date_input", Input)
    amount_input = self.query_one("#amount_input", Input)

    date_input.remove_class("field-error")
    amount_input.remove_class("field-error")

    parsed_date = _parse_date_input(date_input.value)
    parsed_amount = _parse_amount_input(amount_input.value)

    if parsed_date is None:
        date_input.add_class("field-error")
        return
    if parsed_amount is None:
        amount_input.add_class("field-error")
        return

    self.dismiss(EditRainResult(date=parsed_date, amount=parsed_amount))

compose()

Render the edit-rain form.

Source code in src/rainlog/tui.py

def compose(self) -> ComposeResult:
    """Render the edit-rain form."""
    with Vertical():
        yield Label("Edit Rain Record")
        yield Label("Date (YYYY-MM-DD)")
        yield Input(value=self._prefill_date, placeholder="YYYY-MM-DD", id="date_input")
        yield Label("Amount (mm)")
        yield Input(value=self._prefill_amount, placeholder="0.0", id="amount_input")
        yield Button("Save", id="save_btn", variant="primary")

on_button_pressed(event)

Forward Save button press to submit action.

Source code in src/rainlog/tui.py

def on_button_pressed(self, event: Button.Pressed) -> None:
    """Forward Save button press to submit action."""
    if event.button.id == "save_btn":
        self.action_submit()

EditRainResult dataclass

Payload returned by EditRainModal on successful submission.

Source code in src/rainlog/tui.py

@dataclass
class EditRainResult:
    """Payload returned by EditRainModal on successful submission."""

    date: datetime
    amount: float

RainTuiApp

Bases: App[None]

Interactive TUI for browsing rain history.

Source code in src/rainlog/tui.py

class RainTuiApp(App[None]):
    """Interactive TUI for browsing rain history."""

    BINDINGS: ClassVar[list[Binding]] = [
        Binding("left", "scroll_back", "Scroll back"),
        Binding("right", "scroll_forward", "Scroll fwd"),
        Binding("g", "cycle_group", "Cycle group"),
        Binding("+", "increase_size", "More bars"),
        Binding("-", "decrease_size", "Fewer bars"),
        Binding("a", "reset_auto_bars", "Auto bars"),
        Binding("n", "open_add_modal", "Add record"),
        Binding("e", "open_edit_modal", "Edit record"),
        Binding("s", "toggle_select_mode", "Select bar"),
        Binding("m", "toggle_chart_mode", "Cycle mode"),
        Binding("escape", "exit_select_mode", "Exit select"),
        Binding("q", "quit", "Quit"),
    ]

    def __init__(self, database: Database) -> None:
        """Initialise app with a database connection."""
        super().__init__()
        self._database = database
        self._group = GraphGrouping.daily
        self._bar_mode: str = "auto"
        self._manual_bar_count: int = 30
        self._offset = 0
        self._select_mode: bool = False
        self._selected_index: int | None = None
        self._chart_mode: ChartMode = ChartMode.rainfall
        self._num_comparison_years: int = 4

    @property
    def _bar_count(self) -> int:
        """Current bar count: auto-computed from chart width, or stored manual value."""
        if self._bar_mode == "auto":
            chart_width = self.query_one(BarChartWidget).size.width
            return _compute_auto_bar_count(chart_width)
        return self._manual_bar_count

    def compose(self) -> ComposeResult:
        """Build the two-column layout."""
        with Horizontal():
            yield BarChartWidget()
            yield ComparisonChartWidget()
            yield StatsPanel()
        yield Footer()

    def on_mount(self) -> None:
        """Load initial data after the UI is ready."""
        self.call_after_refresh(self._refresh_data)

    def on_resize(self) -> None:
        """Recompute bar count on terminal resize when in auto mode."""
        self.call_after_refresh(self._refresh_data)

    def check_action(self, action: str, parameters: tuple[object, ...]) -> bool | None:  # noqa: ARG002
        """Conditionally disable/hide bindings based on app state."""
        if self._chart_mode == ChartMode.comparison:
            if action in ("scroll_back", "scroll_forward", "toggle_select_mode", "open_edit_modal", "reset_auto_bars"):
                return False
        if action == "toggle_select_mode":
            return self._group == GraphGrouping.daily and self._chart_mode == ChartMode.rainfall
        if action == "exit_select_mode":
            return self._select_mode
        return True

    def _apply_auto_bar_count(self) -> None:
        """Sync manual bar count cache with auto-computed width; no-op in manual mode."""
        if self._bar_mode == "auto":
            chart_width = self.query_one(BarChartWidget).size.width
            self._manual_bar_count = _compute_auto_bar_count(chart_width)

    def _merge_tentative_entries(
        self,
        data: list[tuple[str, float]],
    ) -> tuple[list[tuple[str, float]], set[str]]:
        """Prepend synthetic entries for any gap since the last DB record.

        In rainfall mode, synthetic entries carry 0.0 rain. In moisture mode, synthetic
        entries show the decaying moisture index for each day since the last record.
        Returns (merged_data, tentative_labels). Returns (data, empty set) unchanged
        when scrolled past the present, when the DB is empty, or when it is up to date.
        """
        if self._offset != 0:
            return data, set()
        last_date = self._database.get_most_recent_date()
        if last_date is None:
            return data, set()
        today = datetime.now(tz=LOCAL_TZ)
        if today.date() <= last_date.date():
            return data, set()

        if self._chart_mode == ChartMode.moisture:
            last_moisture = data[0][1] if data else 0.0
            synthetic = _compute_tentative_moisture_entries(
                last_date=last_date,
                today=today,
                last_moisture=last_moisture,
                group=self._group,
            )
        else:
            synthetic = _compute_tentative_entries(last_date, today, self._group)

        tentative_labels = {label for label, _ in synthetic}
        real_labels = {label for label, _ in data}
        entries_to_prepend = [(label, value) for label, value in synthetic if label not in real_labels]
        return (entries_to_prepend + data)[: self._bar_count], tentative_labels

    def _decay_current_index_to_today(
        self,
        current_index: float,
        data: list[tuple[str, float]],
        tentative_labels: set[str],
        decay: float = 0.85,
    ) -> float:
        """Decay current_index to today when the most-recent bar is a real (non-tentative) entry.

        When viewing weekly/monthly/yearly groupings and today falls in the same period as
        the last DB record, _merge_tentative_entries adds no tentative entries. The bar shows
        the end-of-period moisture value, which may be several days stale. This method applies
        the remaining decay so the stats panel "Current index" always reflects today's estimate.
        Only active when offset is 0 (viewing the present window).
        """
        if not data or data[0][0] in tentative_labels or self._offset != 0:
            return current_index
        last_date = self._database.get_most_recent_date()
        if last_date is None:
            return current_index
        today = datetime.now(tz=LOCAL_TZ)
        elapsed_days = (today.date() - last_date.date()).days
        if elapsed_days > 0:
            current_index *= decay**elapsed_days
        return current_index

    def _refresh_comparison_monthly(
        self,
        bar_widget: BarChartWidget,
        comparison_widget: ComparisonChartWidget,
        stats_panel: StatsPanel,
        streak: tuple[str, int],
    ) -> None:
        """Push monthly-comparison data to widgets and stats panel."""
        bar_widget.display = False
        comparison_widget.display = True
        today = datetime.now(tz=LOCAL_TZ)
        current_year_label = str(today.year)
        labels, series = self._database.get_monthly_comparison(num_years=self._num_comparison_years)
        comparison_widget.set_data(labels, series)
        current_year_values = series.get(current_year_label, [])
        current_ytd = sum(current_year_values)
        prior_series = {year: values for year, values in series.items() if year != current_year_label}
        wetter_years = sum(1 for values in prior_series.values() if sum(values) > current_ytd)
        current_month_label = labels[-1] if labels else None
        current_month_total = current_year_values[-1] if current_year_values else None
        avg_current_month = (
            sum(values[-1] for values in prior_series.values()) / len(prior_series) if prior_series and labels else None
        )
        stats_panel.update_comparison_stats(
            ytd_year=today.year,
            ytd_total=current_ytd,
            wetter_years=wetter_years,
            total_years=len(prior_series),
            streak=streak,
            current_month_label=current_month_label,
            current_month_total=current_month_total,
            avg_current_month=avg_current_month,
        )

    def _refresh_comparison_yearly(
        self,
        bar_widget: BarChartWidget,
        comparison_widget: ComparisonChartWidget,
        stats_panel: StatsPanel,
        streak: tuple[str, int],
    ) -> None:
        """Push yearly-comparison (YTD) data to widgets and stats panel."""
        comparison_widget.display = False
        bar_widget.display = True
        today = datetime.now(tz=LOCAL_TZ)
        current_year_label = str(today.year)
        ytd_data = self._database.get_ytd_by_year()
        current_ytd = next((total for label, total in ytd_data if label == current_year_label), 0.0)
        prior_years = [(label, total) for label, total in ytd_data if label != current_year_label]
        wetter_years = sum(1 for _, total in prior_years if total > current_ytd)
        bar_widget.set_data(ytd_data, set(), None)
        stats_panel.update_comparison_stats(
            ytd_year=today.year,
            ytd_total=current_ytd,
            wetter_years=wetter_years,
            total_years=len(prior_years),
            streak=streak,
        )

    def _refresh_moisture(
        self,
        bar_widget: BarChartWidget,
        stats_panel: StatsPanel,
        streak: tuple[str, int],
    ) -> None:
        """Push moisture-index data to bar widget and stats panel."""
        data = self._database.get_moisture_index(
            history_size=self._bar_count,
            group=self._group,
            offset=self._offset,
        )
        data, tentative_labels = self._merge_tentative_entries(data)
        current_index = data[0][1] if data else 0.0
        current_index = self._decay_current_index_to_today(current_index, data, tentative_labels)
        period_average = sum(moisture for _, moisture in data) / len(data) if data else 0.0
        bar_widget.set_data(data, tentative_labels, self._selected_index)
        stats_panel.update_stats(
            period_total=current_index,
            daily_average=period_average,
            streak=streak,
            selected_entry=None,
            chart_mode=self._chart_mode,
            group=self._group,
        )

    def _refresh_rainfall(
        self,
        bar_widget: BarChartWidget,
        stats_panel: StatsPanel,
        streak: tuple[str, int],
    ) -> None:
        """Push rainfall data to bar widget and stats panel."""
        data = self._database.get_rain(
            history_size=self._bar_count,
            group=self._group,
            offset=self._offset,
        )
        data, tentative_labels = self._merge_tentative_entries(data)
        period_total = sum(rain for _, rain in data)
        total_days = len(data) * DAYS_PER_GROUP[self._group]
        daily_average = period_total / total_days if total_days > 0 else 0.0
        selected_entry: tuple[str, float] | None = None
        if self._selected_index is not None and self._selected_index < len(data):
            selected_entry = data[self._selected_index]
        bar_widget.set_data(data, tentative_labels, self._selected_index)
        stats_panel.update_stats(
            period_total=period_total,
            daily_average=daily_average,
            streak=streak,
            selected_entry=selected_entry,
            chart_mode=self._chart_mode,
            group=self._group,
        )

    def _refresh_data(self) -> None:
        """Re-query the DB and push results to chart widget and stats panel."""
        streak = self._database.get_current_streak()
        bar_widget = self.query_one(BarChartWidget)
        comparison_widget = self.query_one(ComparisonChartWidget)
        stats_panel = self.query_one(StatsPanel)

        if self._chart_mode == ChartMode.comparison:
            if self._group == GraphGrouping.monthly:
                self._refresh_comparison_monthly(bar_widget, comparison_widget, stats_panel, streak)
            else:
                self._refresh_comparison_yearly(bar_widget, comparison_widget, stats_panel, streak)
            return

        comparison_widget.display = False
        bar_widget.display = True

        if self._chart_mode == ChartMode.moisture:
            self._refresh_moisture(bar_widget, stats_panel, streak)
        else:
            self._refresh_rainfall(bar_widget, stats_panel, streak)

    def action_scroll_back(self) -> None:
        """In select mode: move cursor left (newer bar). Otherwise: scroll history back."""
        if self._select_mode:
            if self._selected_index is not None:
                self._selected_index = max(0, self._selected_index - 1)
            self._refresh_data()
            return
        self._offset += 1
        self._refresh_data()

    def action_scroll_forward(self) -> None:
        """In select mode: move cursor right (older bar). Otherwise: scroll history forward."""
        if self._select_mode:
            if self._selected_index is not None:
                bar_count = len(self.query_one(BarChartWidget)._data)
                self._selected_index = min(bar_count - 1, self._selected_index + 1)
            self._refresh_data()
            return
        if self._offset > 0:
            self._offset -= 1
            self._refresh_data()

    def action_cycle_group(self) -> None:
        """Cycle grouping; reset comparison mode if new grouping doesn't support it."""
        self._select_mode = False
        self._selected_index = None
        current_group_index = GROUPING_CYCLE.index(self._group)
        self._group = GROUPING_CYCLE[(current_group_index + 1) % len(GROUPING_CYCLE)]
        if self._group in (GraphGrouping.daily, GraphGrouping.weekly):
            if self._chart_mode == ChartMode.comparison:
                self._chart_mode = ChartMode.rainfall
        self._offset = 0
        self._refresh_data()

    def action_toggle_select_mode(self) -> None:
        """Enter or exit bar-selection mode (daily grouping only)."""
        self._select_mode = not self._select_mode
        self._selected_index = 0 if self._select_mode else None
        self._refresh_data()

    def action_exit_select_mode(self) -> None:
        """Exit select mode and clear the cursor."""
        self._select_mode = False
        self._selected_index = None
        self._refresh_data()

    def action_toggle_chart_mode(self) -> None:
        """Cycle chart mode: Rainfall → Moisture → Comparison (monthly/yearly only)."""
        available_modes = [ChartMode.rainfall, ChartMode.moisture]
        if self._group in (GraphGrouping.monthly, GraphGrouping.yearly):
            available_modes.append(ChartMode.comparison)
        current_mode_index = available_modes.index(self._chart_mode) if self._chart_mode in available_modes else 0
        self._chart_mode = available_modes[(current_mode_index + 1) % len(available_modes)]
        if self._chart_mode != ChartMode.rainfall:
            self._select_mode = False
            self._selected_index = None
        self._refresh_data()

    def action_increase_size(self) -> None:
        """In comparison mode: add one comparison year (up to available years). Otherwise: add one bar."""
        if self._chart_mode == ChartMode.comparison:
            current_year_str = str(datetime.now(tz=LOCAL_TZ).year)
            if self._group == GraphGrouping.yearly:
                available_years = sum(1 for label, _ in self._database.get_ytd_by_year() if label != current_year_str)
            else:
                _, probe = self._database.get_monthly_comparison(num_years=self._num_comparison_years + 1)
                available_years = sum(1 for k in probe if k != current_year_str)
            if self._num_comparison_years < available_years:
                self._num_comparison_years += 1
                self._refresh_data()
            return
        current_count = self._bar_count
        self._bar_mode = "manual"
        self._manual_bar_count = min(365, current_count + 1)
        self._refresh_data()

    def action_decrease_size(self) -> None:
        """In comparison mode: remove one comparison year (min 1). Otherwise: remove one bar."""
        if self._chart_mode == ChartMode.comparison:
            self._num_comparison_years = max(1, self._num_comparison_years - 1)
            self._refresh_data()
            return
        current_count = self._bar_count
        self._bar_mode = "manual"
        self._manual_bar_count = max(7, current_count - 1)
        self._refresh_data()

    def action_reset_auto_bars(self) -> None:
        """Switch back to auto bar count and recalculate."""
        self._bar_mode = "auto"
        self._apply_auto_bar_count()
        self._refresh_data()

    def action_open_add_modal(self) -> None:
        """Open the add-rain modal."""
        self.push_screen(AddRainModal(), self._handle_add_rain_result)

    def action_open_edit_modal(self) -> None:
        """Open edit modal; pre-populate from selected bar if in select mode."""
        prefill_date = ""
        prefill_amount = ""
        if self._select_mode and self._selected_index is not None:
            data = self.query_one(BarChartWidget)._data
            if self._selected_index < len(data):
                label, amount = data[self._selected_index]
                prefill_date = label  # label is YYYY-MM-DD in daily grouping
                prefill_amount = f"{amount:.1f}"
        self.push_screen(
            EditRainModal(prefill_date=prefill_date, prefill_amount=prefill_amount),
            self._handle_edit_rain_result,
        )

    def _handle_edit_rain_result(self, result: EditRainResult | None) -> None:
        """Update the DB record and refresh."""
        if result is None:
            return
        rain_period_end = result.date.replace(hour=9, minute=0, second=0, microsecond=0)
        self._database.update_rain_record(date=rain_period_end, amount=result.amount)
        self._refresh_data()

    def _handle_add_rain_result(self, result: AddRainResult | None) -> None:
        """Write the new record to the DB and refresh."""
        if result is None:
            return
        rain_period_end = result.date.replace(hour=9, minute=0, second=0, microsecond=0)
        earliest_before_insert = self._database.get_earliest_date()
        try:
            self._database.add_rain_record(date=rain_period_end, amount=result.amount)
        except sqlite3.IntegrityError:
            self.notify(
                f"A record for {rain_period_end.strftime('%Y-%m-%d')} already exists — use edit (e) to update it.",
                severity="error",
            )
            return
        if result.backfill and earliest_before_insert is not None:
            back_fill_date = rain_period_end - timedelta(days=1)
            while not isinstance(self._database.get_single_day_rain(date=back_fill_date), float):
                if back_fill_date < earliest_before_insert:
                    break
                self._database.add_rain_record(date=back_fill_date, amount=0.0)
                back_fill_date = back_fill_date - timedelta(days=1)
        self._refresh_data()

__init__(database)

Initialise app with a database connection.

Source code in src/rainlog/tui.py

def __init__(self, database: Database) -> None:
    """Initialise app with a database connection."""
    super().__init__()
    self._database = database
    self._group = GraphGrouping.daily
    self._bar_mode: str = "auto"
    self._manual_bar_count: int = 30
    self._offset = 0
    self._select_mode: bool = False
    self._selected_index: int | None = None
    self._chart_mode: ChartMode = ChartMode.rainfall
    self._num_comparison_years: int = 4

action_cycle_group()

Cycle grouping; reset comparison mode if new grouping doesn't support it.

Source code in src/rainlog/tui.py

def action_cycle_group(self) -> None:
    """Cycle grouping; reset comparison mode if new grouping doesn't support it."""
    self._select_mode = False
    self._selected_index = None
    current_group_index = GROUPING_CYCLE.index(self._group)
    self._group = GROUPING_CYCLE[(current_group_index + 1) % len(GROUPING_CYCLE)]
    if self._group in (GraphGrouping.daily, GraphGrouping.weekly):
        if self._chart_mode == ChartMode.comparison:
            self._chart_mode = ChartMode.rainfall
    self._offset = 0
    self._refresh_data()

action_decrease_size()

In comparison mode: remove one comparison year (min 1). Otherwise: remove one bar.

Source code in src/rainlog/tui.py

def action_decrease_size(self) -> None:
    """In comparison mode: remove one comparison year (min 1). Otherwise: remove one bar."""
    if self._chart_mode == ChartMode.comparison:
        self._num_comparison_years = max(1, self._num_comparison_years - 1)
        self._refresh_data()
        return
    current_count = self._bar_count
    self._bar_mode = "manual"
    self._manual_bar_count = max(7, current_count - 1)
    self._refresh_data()

action_exit_select_mode()

Exit select mode and clear the cursor.

Source code in src/rainlog/tui.py

def action_exit_select_mode(self) -> None:
    """Exit select mode and clear the cursor."""
    self._select_mode = False
    self._selected_index = None
    self._refresh_data()

action_increase_size()

In comparison mode: add one comparison year (up to available years). Otherwise: add one bar.

Source code in src/rainlog/tui.py

def action_increase_size(self) -> None:
    """In comparison mode: add one comparison year (up to available years). Otherwise: add one bar."""
    if self._chart_mode == ChartMode.comparison:
        current_year_str = str(datetime.now(tz=LOCAL_TZ).year)
        if self._group == GraphGrouping.yearly:
            available_years = sum(1 for label, _ in self._database.get_ytd_by_year() if label != current_year_str)
        else:
            _, probe = self._database.get_monthly_comparison(num_years=self._num_comparison_years + 1)
            available_years = sum(1 for k in probe if k != current_year_str)
        if self._num_comparison_years < available_years:
            self._num_comparison_years += 1
            self._refresh_data()
        return
    current_count = self._bar_count
    self._bar_mode = "manual"
    self._manual_bar_count = min(365, current_count + 1)
    self._refresh_data()

action_open_add_modal()

Open the add-rain modal.

Source code in src/rainlog/tui.py

def action_open_add_modal(self) -> None:
    """Open the add-rain modal."""
    self.push_screen(AddRainModal(), self._handle_add_rain_result)

action_open_edit_modal()

Open edit modal; pre-populate from selected bar if in select mode.

Source code in src/rainlog/tui.py

def action_open_edit_modal(self) -> None:
    """Open edit modal; pre-populate from selected bar if in select mode."""
    prefill_date = ""
    prefill_amount = ""
    if self._select_mode and self._selected_index is not None:
        data = self.query_one(BarChartWidget)._data
        if self._selected_index < len(data):
            label, amount = data[self._selected_index]
            prefill_date = label  # label is YYYY-MM-DD in daily grouping
            prefill_amount = f"{amount:.1f}"
    self.push_screen(
        EditRainModal(prefill_date=prefill_date, prefill_amount=prefill_amount),
        self._handle_edit_rain_result,
    )

action_reset_auto_bars()

Switch back to auto bar count and recalculate.

Source code in src/rainlog/tui.py

def action_reset_auto_bars(self) -> None:
    """Switch back to auto bar count and recalculate."""
    self._bar_mode = "auto"
    self._apply_auto_bar_count()
    self._refresh_data()

action_scroll_back()

In select mode: move cursor left (newer bar). Otherwise: scroll history back.

Source code in src/rainlog/tui.py

def action_scroll_back(self) -> None:
    """In select mode: move cursor left (newer bar). Otherwise: scroll history back."""
    if self._select_mode:
        if self._selected_index is not None:
            self._selected_index = max(0, self._selected_index - 1)
        self._refresh_data()
        return
    self._offset += 1
    self._refresh_data()

action_scroll_forward()

In select mode: move cursor right (older bar). Otherwise: scroll history forward.

Source code in src/rainlog/tui.py

def action_scroll_forward(self) -> None:
    """In select mode: move cursor right (older bar). Otherwise: scroll history forward."""
    if self._select_mode:
        if self._selected_index is not None:
            bar_count = len(self.query_one(BarChartWidget)._data)
            self._selected_index = min(bar_count - 1, self._selected_index + 1)
        self._refresh_data()
        return
    if self._offset > 0:
        self._offset -= 1
        self._refresh_data()

action_toggle_chart_mode()

Cycle chart mode: Rainfall → Moisture → Comparison (monthly/yearly only).

Source code in src/rainlog/tui.py

def action_toggle_chart_mode(self) -> None:
    """Cycle chart mode: Rainfall → Moisture → Comparison (monthly/yearly only)."""
    available_modes = [ChartMode.rainfall, ChartMode.moisture]
    if self._group in (GraphGrouping.monthly, GraphGrouping.yearly):
        available_modes.append(ChartMode.comparison)
    current_mode_index = available_modes.index(self._chart_mode) if self._chart_mode in available_modes else 0
    self._chart_mode = available_modes[(current_mode_index + 1) % len(available_modes)]
    if self._chart_mode != ChartMode.rainfall:
        self._select_mode = False
        self._selected_index = None
    self._refresh_data()

action_toggle_select_mode()

Enter or exit bar-selection mode (daily grouping only).

Source code in src/rainlog/tui.py

def action_toggle_select_mode(self) -> None:
    """Enter or exit bar-selection mode (daily grouping only)."""
    self._select_mode = not self._select_mode
    self._selected_index = 0 if self._select_mode else None
    self._refresh_data()

check_action(action, parameters)

Conditionally disable/hide bindings based on app state.

Source code in src/rainlog/tui.py

def check_action(self, action: str, parameters: tuple[object, ...]) -> bool | None:  # noqa: ARG002
    """Conditionally disable/hide bindings based on app state."""
    if self._chart_mode == ChartMode.comparison:
        if action in ("scroll_back", "scroll_forward", "toggle_select_mode", "open_edit_modal", "reset_auto_bars"):
            return False
    if action == "toggle_select_mode":
        return self._group == GraphGrouping.daily and self._chart_mode == ChartMode.rainfall
    if action == "exit_select_mode":
        return self._select_mode
    return True

compose()

Build the two-column layout.

Source code in src/rainlog/tui.py

def compose(self) -> ComposeResult:
    """Build the two-column layout."""
    with Horizontal():
        yield BarChartWidget()
        yield ComparisonChartWidget()
        yield StatsPanel()
    yield Footer()

on_mount()

Load initial data after the UI is ready.

Source code in src/rainlog/tui.py

def on_mount(self) -> None:
    """Load initial data after the UI is ready."""
    self.call_after_refresh(self._refresh_data)

on_resize()

Recompute bar count on terminal resize when in auto mode.

Source code in src/rainlog/tui.py

def on_resize(self) -> None:
    """Recompute bar count on terminal resize when in auto mode."""
    self.call_after_refresh(self._refresh_data)

StatsPanel

Bases: Widget

Sidebar showing period total, daily average, and current streak.

Source code in src/rainlog/tui.py

class StatsPanel(Widget):
    """Sidebar showing period total, daily average, and current streak."""

    DEFAULT_CSS = """
    StatsPanel {
        width: auto;
        height: 1fr;
        padding: 1 2;
    }
    """

    def __init__(self) -> None:
        """Initialise with zero stats."""
        super().__init__()
        self._period_total = 0.0
        self._daily_average = 0.0
        self._streak: tuple[str, int] = ("dry", 0)
        self._selected_entry: tuple[str, float] | None = None
        self._chart_mode: ChartMode = ChartMode.rainfall
        self._group: GraphGrouping = GraphGrouping.daily
        self._ytd_year: int = 0
        self._ytd_total: float = 0.0
        self._wetter_years: int = 0
        self._total_years: int = 0
        self._current_month_label: str | None = None
        self._current_month_total: float | None = None
        self._avg_current_month: float | None = None

    def update_stats(  # noqa: PLR0913
        self,
        period_total: float,
        daily_average: float,
        streak: tuple[str, int],
        *,
        selected_entry: tuple[str, float] | None = None,
        chart_mode: ChartMode = ChartMode.rainfall,
        group: GraphGrouping = GraphGrouping.daily,
    ) -> None:
        """Replace all stats and trigger a repaint."""
        self._period_total = period_total
        self._daily_average = daily_average
        self._streak = streak
        self._selected_entry = selected_entry
        self._chart_mode = chart_mode
        self._group = group
        self.refresh()

    def update_comparison_stats(  # noqa: PLR0913
        self,
        ytd_year: int,
        ytd_total: float,
        wetter_years: int,
        total_years: int,
        streak: tuple[str, int],
        *,
        current_month_label: str | None = None,
        current_month_total: float | None = None,
        avg_current_month: float | None = None,
    ) -> None:
        """Store comparison-mode stats and trigger repaint."""
        self._chart_mode = ChartMode.comparison
        self._ytd_year = ytd_year
        self._ytd_total = ytd_total
        self._wetter_years = wetter_years
        self._total_years = total_years
        self._streak = streak
        self._current_month_label = current_month_label
        self._current_month_total = current_month_total
        self._avg_current_month = avg_current_month
        self.refresh()

    def _render_comparison_stats(self, result: Text) -> None:
        """Append comparison-mode lines to result."""
        years_current_is_wetter_than = self._total_years - self._wetter_years
        compare_word = "wetter" if years_current_is_wetter_than >= self._wetter_years else "drier"
        result.append(f"{self._ytd_year} YTD\n", style="bold")
        result.append(f"  {self._ytd_total:.1f} mm\n\n")
        if self._total_years > 0:
            result.append(f"  {compare_word} than\n  {years_current_is_wetter_than} of {self._total_years} years\n\n")
        if self._current_month_label is not None and self._current_month_total is not None:
            result.append(f"{self._current_month_label}\n", style="bold")
            result.append(f"  {self._current_month_total:.1f} mm\n")
            if self._avg_current_month is not None:
                result.append(f"  avg: {self._avg_current_month:.1f} mm\n\n")
        streak_type, streak_count = self._streak
        result.append(f"Streak\n  {streak_count} {streak_type} days")

    def render(self) -> RenderableType:
        """Render mode/grouping header then mode-appropriate stats."""
        streak_type, streak_count = self._streak
        result = Text()

        if self._chart_mode == ChartMode.moisture:
            mode_label = "Moisture"
        elif self._chart_mode == ChartMode.comparison:
            mode_label = "Comparison"
        else:
            mode_label = "Rain"
        group_label = self._group.value.capitalize()
        result.append(f"Mode:  {mode_label}\n", style="dim")
        result.append(f"Group: {group_label}\n\n", style="dim")

        if self._selected_entry is not None:
            label, amount = self._selected_entry
            result.append("Selected\n", style="bold")
            result.append(f"  {label}  {amount:.1f} mm\n\n")

        if self._chart_mode == ChartMode.comparison:
            self._render_comparison_stats(result)
        elif self._chart_mode == ChartMode.moisture:
            result.append(
                f"Current index\n"
                f"  {self._period_total:.1f} mm\n\n"
                f"Period average\n"
                f"  {self._daily_average:.1f} mm\n\n"
                f"Streak\n"
                f"  {streak_count} {streak_type} days"
            )
        else:
            result.append(
                f"Period total\n"
                f"  {self._period_total:.1f} mm\n\n"
                f"Daily average\n"
                f"  {self._daily_average:.1f} mm\n\n"
                f"Streak\n"
                f"  {streak_count} {streak_type} days"
            )

        return result

__init__()

Initialise with zero stats.

Source code in src/rainlog/tui.py

def __init__(self) -> None:
    """Initialise with zero stats."""
    super().__init__()
    self._period_total = 0.0
    self._daily_average = 0.0
    self._streak: tuple[str, int] = ("dry", 0)
    self._selected_entry: tuple[str, float] | None = None
    self._chart_mode: ChartMode = ChartMode.rainfall
    self._group: GraphGrouping = GraphGrouping.daily
    self._ytd_year: int = 0
    self._ytd_total: float = 0.0
    self._wetter_years: int = 0
    self._total_years: int = 0
    self._current_month_label: str | None = None
    self._current_month_total: float | None = None
    self._avg_current_month: float | None = None

render()

Render mode/grouping header then mode-appropriate stats.

Source code in src/rainlog/tui.py

def render(self) -> RenderableType:
    """Render mode/grouping header then mode-appropriate stats."""
    streak_type, streak_count = self._streak
    result = Text()

    if self._chart_mode == ChartMode.moisture:
        mode_label = "Moisture"
    elif self._chart_mode == ChartMode.comparison:
        mode_label = "Comparison"
    else:
        mode_label = "Rain"
    group_label = self._group.value.capitalize()
    result.append(f"Mode:  {mode_label}\n", style="dim")
    result.append(f"Group: {group_label}\n\n", style="dim")

    if self._selected_entry is not None:
        label, amount = self._selected_entry
        result.append("Selected\n", style="bold")
        result.append(f"  {label}  {amount:.1f} mm\n\n")

    if self._chart_mode == ChartMode.comparison:
        self._render_comparison_stats(result)
    elif self._chart_mode == ChartMode.moisture:
        result.append(
            f"Current index\n"
            f"  {self._period_total:.1f} mm\n\n"
            f"Period average\n"
            f"  {self._daily_average:.1f} mm\n\n"
            f"Streak\n"
            f"  {streak_count} {streak_type} days"
        )
    else:
        result.append(
            f"Period total\n"
            f"  {self._period_total:.1f} mm\n\n"
            f"Daily average\n"
            f"  {self._daily_average:.1f} mm\n\n"
            f"Streak\n"
            f"  {streak_count} {streak_type} days"
        )

    return result

update_comparison_stats(ytd_year, ytd_total, wetter_years, total_years, streak, *, current_month_label=None, current_month_total=None, avg_current_month=None)

Store comparison-mode stats and trigger repaint.

Source code in src/rainlog/tui.py

def update_comparison_stats(  # noqa: PLR0913
    self,
    ytd_year: int,
    ytd_total: float,
    wetter_years: int,
    total_years: int,
    streak: tuple[str, int],
    *,
    current_month_label: str | None = None,
    current_month_total: float | None = None,
    avg_current_month: float | None = None,
) -> None:
    """Store comparison-mode stats and trigger repaint."""
    self._chart_mode = ChartMode.comparison
    self._ytd_year = ytd_year
    self._ytd_total = ytd_total
    self._wetter_years = wetter_years
    self._total_years = total_years
    self._streak = streak
    self._current_month_label = current_month_label
    self._current_month_total = current_month_total
    self._avg_current_month = avg_current_month
    self.refresh()

update_stats(period_total, daily_average, streak, *, selected_entry=None, chart_mode=ChartMode.rainfall, group=GraphGrouping.daily)

Replace all stats and trigger a repaint.

Source code in src/rainlog/tui.py

def update_stats(  # noqa: PLR0913
    self,
    period_total: float,
    daily_average: float,
    streak: tuple[str, int],
    *,
    selected_entry: tuple[str, float] | None = None,
    chart_mode: ChartMode = ChartMode.rainfall,
    group: GraphGrouping = GraphGrouping.daily,
) -> None:
    """Replace all stats and trigger a repaint."""
    self._period_total = period_total
    self._daily_average = daily_average
    self._streak = streak
    self._selected_entry = selected_entry
    self._chart_mode = chart_mode
    self._group = group
    self.refresh()

calculate_bar_heights(values, max_height)

Scale a list of rain values to bar heights in terminal character rows.

Source code in src/rainlog/tui.py

def calculate_bar_heights(values: list[float], max_height: int) -> list[int]:
    """Scale a list of rain values to bar heights in terminal character rows."""
    if not values or max(values) == 0:
        return [0] * len(values)
    max_value = max(values)
    return [round(value / max_value * max_height) for value in values]

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

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_ytd_by_year(
        self: Self,
        reference_date: datetime | None = None,
    ) -> list[tuple[str, float]]:
        """Return Jan 1 → reference_date total for each year with records, oldest first.

        reference_date defaults to today (local time). Records whose calendar
        month/day falls after reference_date's month/day are excluded so all
        years are compared over the same portion of the calendar.
        """
        if reference_date is None:
            reference_date = datetime.now(tz=LOCAL_TZ)
        cutoff = (reference_date.month, reference_date.day)
        year_totals: dict[str, float] = {}
        for row in self.db_connection.execute("SELECT date, rain FROM rain_daily ORDER BY date ASC"):
            recorded_date = datetime.fromtimestamp(row[0], tz=LOCAL_TZ)
            if (recorded_date.month, recorded_date.day) <= cutoff:
                year_label = str(recorded_date.year)
                year_totals[year_label] = year_totals.get(year_label, 0.0) + row[1]
        return sorted(year_totals.items())

    def get_monthly_comparison(
        self: Self,
        num_years: int,
        reference_date: datetime | None = None,
    ) -> tuple[list[str], dict[str, list[float]]]:
        """Return monthly totals for current year and num_years prior, Jan through current month.

        reference_date defaults to today (local time). Returns (month_labels, series) where
        series maps year_label to a list of monthly totals. Only years with at least one
        record in the window are included. Series is ordered oldest year first.
        """
        if reference_date is None:
            reference_date = datetime.now(tz=LOCAL_TZ)
        months_count = reference_date.month
        current_year = reference_date.year
        labels = [datetime(current_year, month_num, 1).strftime("%b") for month_num in range(1, months_count + 1)]
        target_years = list(range(current_year - num_years, current_year + 1))
        year_month_totals: dict[int, dict[int, float]] = {year: {} for year in target_years}
        for row in self.db_connection.execute("SELECT date, rain FROM rain_daily ORDER BY date ASC"):
            recorded_date = datetime.fromtimestamp(row[0], tz=LOCAL_TZ)
            if recorded_date.year in year_month_totals and recorded_date.month <= months_count:
                month_bucket = year_month_totals[recorded_date.year]
                month_num = recorded_date.month
                month_bucket[month_num] = month_bucket.get(month_num, 0.0) + row[1]
        series: dict[str, list[float]] = {}
        for year in target_years:
            monthly_values = [year_month_totals[year].get(month_num, 0.0) for month_num in range(1, months_count + 1)]
            if any(value > 0 for value in monthly_values):
                series[str(year)] = monthly_values
        return labels, series

    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_monthly_comparison(num_years, reference_date=None)

Return monthly totals for current year and num_years prior, Jan through current month.

reference_date defaults to today (local time). Returns (month_labels, series) where series maps year_label to a list of monthly totals. Only years with at least one record in the window are included. Series is ordered oldest year first.

Source code in src/rainlog/db_helpers.py

def get_monthly_comparison(
    self: Self,
    num_years: int,
    reference_date: datetime | None = None,
) -> tuple[list[str], dict[str, list[float]]]:
    """Return monthly totals for current year and num_years prior, Jan through current month.

    reference_date defaults to today (local time). Returns (month_labels, series) where
    series maps year_label to a list of monthly totals. Only years with at least one
    record in the window are included. Series is ordered oldest year first.
    """
    if reference_date is None:
        reference_date = datetime.now(tz=LOCAL_TZ)
    months_count = reference_date.month
    current_year = reference_date.year
    labels = [datetime(current_year, month_num, 1).strftime("%b") for month_num in range(1, months_count + 1)]
    target_years = list(range(current_year - num_years, current_year + 1))
    year_month_totals: dict[int, dict[int, float]] = {year: {} for year in target_years}
    for row in self.db_connection.execute("SELECT date, rain FROM rain_daily ORDER BY date ASC"):
        recorded_date = datetime.fromtimestamp(row[0], tz=LOCAL_TZ)
        if recorded_date.year in year_month_totals and recorded_date.month <= months_count:
            month_bucket = year_month_totals[recorded_date.year]
            month_num = recorded_date.month
            month_bucket[month_num] = month_bucket.get(month_num, 0.0) + row[1]
    series: dict[str, list[float]] = {}
    for year in target_years:
        monthly_values = [year_month_totals[year].get(month_num, 0.0) for month_num in range(1, months_count + 1)]
        if any(value > 0 for value in monthly_values):
            series[str(year)] = monthly_values
    return labels, series

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

get_ytd_by_year(reference_date=None)

Return Jan 1 → reference_date total for each year with records, oldest first.

reference_date defaults to today (local time). Records whose calendar month/day falls after reference_date's month/day are excluded so all years are compared over the same portion of the calendar.

Source code in src/rainlog/db_helpers.py

def get_ytd_by_year(
    self: Self,
    reference_date: datetime | None = None,
) -> list[tuple[str, float]]:
    """Return Jan 1 → reference_date total for each year with records, oldest first.

    reference_date defaults to today (local time). Records whose calendar
    month/day falls after reference_date's month/day are excluded so all
    years are compared over the same portion of the calendar.
    """
    if reference_date is None:
        reference_date = datetime.now(tz=LOCAL_TZ)
    cutoff = (reference_date.month, reference_date.day)
    year_totals: dict[str, float] = {}
    for row in self.db_connection.execute("SELECT date, rain FROM rain_daily ORDER BY date ASC"):
        recorded_date = datetime.fromtimestamp(row[0], tz=LOCAL_TZ)
        if (recorded_date.month, recorded_date.day) <= cutoff:
            year_label = str(recorded_date.year)
            year_totals[year_label] = year_totals.get(year_label, 0.0) + row[1]
    return sorted(year_totals.items())

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"