Skip to content

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 
23
24
25
26
27
28
29
@Parameter(name="*")  # Flatten the namespace; i.e. option will be "--db-dir" instead of "--common.db-dir"
@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

add(reading, date=None, back_fill=False, common=None)

Add rain data to database.

Paramters

date: datetime Date to record rain for reading: float Rain reading common : Common, optional Shared options such as --db-path.

Source code in src/rainlog/cli_commands.py 
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
@app.command()
def add(
    reading: float,
    date: datetime | None = None,
    back_fill: bool = False,
    common: Common | None = None,
) -> None:
    """Add rain data to database.

    Paramters
    ---------
    date: datetime
        Date to record rain for
    reading: float
        Rain reading
    common : Common, optional
            Shared options such as --db-path.
    """
    if common is None:
        common = Common()
    if date is None:  # Cyclopts will have populated this with datetime.now()
        date = datetime.now()  # Fallback, though typically not needed
    rain_period_end = date.replace(hour=9, minute=0, second=0, microsecond=0)
    with Database(common.db_dir) as history:
        history.add_rain_record(date=rain_period_end, amount=reading)
        print(f"We had {reading} mm in the 24 hours ending at {rain_period_end}.")

        back_fill_date = rain_period_end - timedelta(days=1)
        while back_fill and not isinstance(history.get_single_day_rain(date=back_fill_date), float):
            history.add_rain_record(date=back_fill_date, amount=0.0)
            back_fill_date = back_fill_date - timedelta(days=1)
            print(f"Back filled 0.0 mm in the 24 hours ending at {back_fill_date}.")

calculate_interval(max_value)

Calculate order of magnitude.

Source code in src/rainlog/cli_commands.py 
250
251
252
253
254
255
256
257
258
259
260
def calculate_interval(max_value):
    """Calculate order of magnitude."""
    exponent = math.floor(math.log10(max_value))
    magnitude = 10**exponent

    # Find the best interval (1, 2, 5, 10) * magnitude
    for candidate in [1, 2, 5, 10]:
        interval = candidate * magnitude
        if max_value / interval <= 10:  # Aim for ~5-10 ticks
            return interval
    return 10 * magnitude  # Fallback

change(reading, date=None, common=None)

Update rain data in the database.

Paramters

date: datetime Date to record rain for reading: float Rain reading common : Common, optional Shared options such as --db-path.

Source code in src/rainlog/cli_commands.py 
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
@app.command()
def change(
    reading: float,
    date: datetime | None = None,
    common: Common | None = None,
) -> None:
    """Update rain data in the database.

    Paramters
    ---------
    date: datetime
        Date to record rain for
    reading: float
        Rain reading
    common : Common, optional
            Shared options such as --db-path.
    """
    if common is None:
        common = Common()
    if date is None:  # Cyclopts will have populated this with datetime.now()
        date = datetime.now()  # Fallback, though typically not needed
    rain_period_end = date.replace(hour=9, minute=0, second=0, microsecond=0)
    with Database(common.db_dir) as history:
        history.update_rain_record(date=rain_period_end, amount=reading)
    print(f"Update rain for {rain_period_end} to {reading} mm.")

default_datetime()

Register a default function for datetime parameters.

Source code in src/rainlog/cli_commands.py 
35
36
37
38
@app.default
def default_datetime() -> datetime:
    """Register a default function for datetime parameters."""
    return datetime.now()

graph(size=30, group=GraphGrouping.daily, common=None)

Retrieve historic rain data from DB and show it as a graph.

Parameters

size: int Number of graph elements to show group: GraphGrouping Grouping for graph common : Common, optional Shared options such as --db-path.

Source code in src/rainlog/cli_commands.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
@app.command()
def graph(
    size: int = 30,
    group: GraphGrouping = GraphGrouping.daily,
    common: Common | None = None,
) -> None:
    """Retrieve historic rain data from DB and show it as a graph.

    Parameters
    ----------
    size: int
        Number of graph elements to show
    group: GraphGrouping
        Grouping for graph
    common : Common, optional
            Shared options such as --db-path.

    """
    if common is None:
        common = Common()
    with Database(common.db_dir) as history:
        data_list = history.get_rain(history_size=size, group=group)
    dates, rains = zip(*data_list, strict=False)
    date_list = list(dates)
    rain_list = list(rains)
    max_y_tick = max(rain_list)

    y_ticks_interval = calculate_interval(max_y_tick)
    y_ticks = [y_ticks_interval * i for i in range(int(max_y_tick / y_ticks_interval) + 1)]
    y_ticks.append(max_y_tick)

    plotext.simple_bar(date_list, rain_list)
    plotext.ylabel(label="mm")
    plotext.xlabel(label="Date")
    plotext.yticks(ticks=y_ticks)
    plotext.title("Recent Rain")
    plotext.show()

migrate(db_dir=Path('.'))

Migrate MyWeather.sqlite from db_dir to the default XDG location as rainlog.sqlite.

Parameters

db_dir : Path Directory containing the existing MyWeather.sqlite to migrate (default: current directory).

Source code in src/rainlog/cli_commands.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
@app.command()
def migrate(db_dir: Path = Path(".")) -> None:
    """Migrate MyWeather.sqlite from db_dir to the default XDG location as rainlog.sqlite.

    Parameters
    ----------
    db_dir : Path
        Directory containing the existing MyWeather.sqlite to migrate (default: current directory).

    """
    source = db_dir / LEGACY_DB_FILE_NAME
    destination = DEFAULT_DB_DIR / DEFAULT_DB_FILE_NAME
    if not source.exists():
        raise SystemExit(f"No database found at {source}")
    if destination.exists():
        raise SystemExit(f"Database already exists at {destination} — aborting to avoid overwrite")
    _migrate_db(source, destination)
    print(f"Migrated {source} → {destination}")

rainy_days(size=30, group=GraphGrouping.daily, common=None)

Retrieve historic rain data from DB and show it as a graph.

Parameters

size: int Number of graph elements to show group: GraphGrouping Grouping for graph common : Common, optional Shared options such as --db-path.

Source code in src/rainlog/cli_commands.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
@app.command()
def rainy_days(
    size: int = 30,
    group: GraphGrouping = GraphGrouping.daily,
    common: Common | None = None,
) -> None:
    """Retrieve historic rain data from DB and show it as a graph.

    Parameters
    ----------
    size: int
        Number of graph elements to show
    group: GraphGrouping
        Grouping for graph
    common : Common, optional
            Shared options such as --db-path.

    """
    if common is None:
        common = Common()
    with Database(common.db_dir) as history:
        data_list = history.get_rainy_days(history_size=size, group=group)
    date_list: list[str] = []
    rain_list: list[float] = []
    max_y_tick = 0.0
    for day in data_list:
        date_list.append(day[0])
        rain_list.append(day[1])
        max_y_tick = max(max_y_tick, day[1])

    y_ticks: list[float] = []
    y_ticks_interval = 5.0
    for i in range(int(max_y_tick / y_ticks_interval) + 1):
        y_ticks.append(y_ticks_interval * i)
    y_ticks.append(max_y_tick)

    plotext.simple_bar(date_list, rain_list)
    plotext.yfrequency(frequency=5.0)
    plotext.ylabel(label="mm")
    plotext.xlabel(label="Date")
    plotext.yticks(ticks=y_ticks)
    plotext.title("Recent Rain")
    plotext.show()

tui(common=None)

Launch the interactive TUI for browsing rain history.

Source code in src/rainlog/cli_commands.py 
240
241
242
243
244
245
246
247
@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()

weewx_import(weewx_db=Path('./weewx.sdb'), common=None)

Export daily rain from weewx database into the DB for this project.

Parameters

weewx_db: Path Full path to the weewx sqlite database file. common : Common, optional Shared options such as --db-path.

Source code in src/rainlog/cli_commands.py 
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
@app.command()
def weewx_import(
    weewx_db: Path = Path("./weewx.sdb"),
    common: Common | None = None,
) -> None:
    """Export daily rain from weewx database into the DB for this project.

    Parameters
    ----------
    weewx_db: Path
        Full path to the weewx sqlite database file.
    common : Common, optional
            Shared options such as --db-path.

    """
    if common is None:
        common = Common()
    with Database(common.db_dir) as mydb:
        mydb.import_weewx(weewx_db_file=weewx_db)

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 
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
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_rainy_days(
        self: Self,
        history_size: int,
        group: GraphGrouping,
    ) -> list[tuple[str, float]]:
        """Get 'history_size' number of rain records."""
        return_list: list[tuple[str, float]] = []
        group_id = None
        group_sum = 0.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),
            )

            day_rainy = 1 if row[1] > 0 else 0

            if not group_id:
                group_id = current_group_id

            if current_group_id == group_id:
                group_sum += day_rainy
            else:
                return_list.append((group_id, group_sum))
                group_id = current_group_id
                group_sum = day_rainy

            if len(return_list) >= history_size:
                break

        if len(return_list) < history_size and group_id:
            return_list.append((group_id, group_sum))

        return return_list

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

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

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

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

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

        return (streak_type, streak_count)

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

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

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

        group_id: str = group_date.strftime(format_for_grouping)

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

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

        return group_id

    def import_weewx(self: Self, weewx_db_file: Path) -> None:
        """Import data from a weewx database file."""
        print(f"Importing data from {weewx_db_file} to {self.db_connection}")

        weewx_rain_in_mm: list[tuple[int, float]] = []

        with sqlite3.connect(database=weewx_db_file) as weewx_db:
            for row in weewx_db.execute("SELECT dateTime, sum FROM archive_day_rain"):
                weewx_rain_in_mm.append(
                    (row[0], row[1] * 25.4),
                )

        self.db_connection.executemany(
            "INSERT INTO rain_daily (date, rain) VALUES (?,?)",
            weewx_rain_in_mm,
        )

        self.db_connection.commit()

__enter__()

Return self to support use as a context manager.

Source code in src/rainlog/db_helpers.py 
39
40
41
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 
43
44
45
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 
27
28
29
30
31
32
33
34
35
36
37
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 
47
48
49
50
51
52
53
54
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 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
def get_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 
189
190
191
192
193
194
195
196
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_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 
180
181
182
183
184
185
186
187
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 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
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_rainy_days(history_size, group)

Get 'history_size' number of rain records.

Source code in src/rainlog/db_helpers.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def get_rainy_days(
    self: Self,
    history_size: int,
    group: GraphGrouping,
) -> list[tuple[str, float]]:
    """Get 'history_size' number of rain records."""
    return_list: list[tuple[str, float]] = []
    group_id = None
    group_sum = 0.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),
        )

        day_rainy = 1 if row[1] > 0 else 0

        if not group_id:
            group_id = current_group_id

        if current_group_id == group_id:
            group_sum += day_rainy
        else:
            return_list.append((group_id, group_sum))
            group_id = current_group_id
            group_sum = day_rainy

        if len(return_list) >= history_size:
            break

    if len(return_list) < history_size and group_id:
        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 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
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

import_weewx(weewx_db_file)

Import data from a weewx database file.

Source code in src/rainlog/db_helpers.py 
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
def import_weewx(self: Self, weewx_db_file: Path) -> None:
    """Import data from a weewx database file."""
    print(f"Importing data from {weewx_db_file} to {self.db_connection}")

    weewx_rain_in_mm: list[tuple[int, float]] = []

    with sqlite3.connect(database=weewx_db_file) as weewx_db:
        for row in weewx_db.execute("SELECT dateTime, sum FROM archive_day_rain"):
            weewx_rain_in_mm.append(
                (row[0], row[1] * 25.4),
            )

    self.db_connection.executemany(
        "INSERT INTO rain_daily (date, rain) VALUES (?,?)",
        weewx_rain_in_mm,
    )

    self.db_connection.commit()

update_rain_record(date, amount)

Update a record / measurement of rain in the DB.

Source code in src/rainlog/db_helpers.py 
56
57
58
59
60
61
62
63
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 
14
15
16
17
18
19
20
21
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 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
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")
            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 
120
121
122
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 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
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 
102
103
104
105
106
107
108
109
110
111
112
113
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")
        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 
115
116
117
118
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 
70
71
72
73
74
75
76
@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 
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
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 
293
294
295
296
297
298
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 
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
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 
300
301
302
303
304
305
306
307
308
309
310
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()

EditRainModal

Bases: ModalScreen[EditRainResult | None]

Modal form for editing an existing rain record.

Source code in src/rainlog/tui.py 
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
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 
177
178
179
180
181
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 
198
199
200
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 
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
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 
183
184
185
186
187
188
189
190
191
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 
193
194
195
196
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 
146
147
148
149
150
151
@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 
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
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("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

    @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 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 action == "toggle_select_mode":
            return self._group == GraphGrouping.daily
        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 zero-rain entries for any gap since the last DB 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()
        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, rain) for label, rain in synthetic if label not in real_labels]
        return (entries_to_prepend + data)[: self._bar_count], tentative_labels

    def _refresh_data(self) -> None:
        """Re-query the DB and push results to both panels."""
        data = self._database.get_rain(
            history_size=self._bar_count,
            group=self._group,
            offset=self._offset,
        )
        streak = self._database.get_current_streak()
        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]

        self.query_one(BarChartWidget).set_data(data, tentative_labels, self._selected_index)
        self.query_one(StatsPanel).update_stats(period_total, daily_average, streak, selected_entry)

    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; exit select mode if active."""
        self._select_mode = False
        self._selected_index = None
        current_index = GROUPING_CYCLE.index(self._group)
        self._group = GROUPING_CYCLE[(current_index + 1) % len(GROUPING_CYCLE)]
        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_increase_size(self) -> None:
        """Switch to manual mode and add one bar (max 365)."""
        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:
        """Switch to manual mode and remove one bar (min 7)."""
        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 
469
470
471
472
473
474
475
476
477
478
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

action_cycle_group()

Cycle grouping; exit select mode if active.

Source code in src/rainlog/tui.py 
583
584
585
586
587
588
589
590
def action_cycle_group(self) -> None:
    """Cycle grouping; exit select mode if active."""
    self._select_mode = False
    self._selected_index = None
    current_index = GROUPING_CYCLE.index(self._group)
    self._group = GROUPING_CYCLE[(current_index + 1) % len(GROUPING_CYCLE)]
    self._offset = 0
    self._refresh_data()

action_decrease_size()

Switch to manual mode and remove one bar (min 7).

Source code in src/rainlog/tui.py 
611
612
613
614
615
616
def action_decrease_size(self) -> None:
    """Switch to manual mode and remove one bar (min 7)."""
    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 
598
599
600
601
602
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()

Switch to manual mode and add one bar (max 365).

Source code in src/rainlog/tui.py 
604
605
606
607
608
609
def action_increase_size(self) -> None:
    """Switch to manual mode and add one bar (max 365)."""
    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 
624
625
626
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 
628
629
630
631
632
633
634
635
636
637
638
639
640
641
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 
618
619
620
621
622
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 
561
562
563
564
565
566
567
568
569
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 
571
572
573
574
575
576
577
578
579
580
581
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_select_mode()

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

Source code in src/rainlog/tui.py 
592
593
594
595
596
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 
503
504
505
506
507
508
509
def check_action(self, action: str, parameters: tuple[object, ...]) -> bool | None:  # noqa: ARG002
    """Conditionally disable/hide bindings based on app state."""
    if action == "toggle_select_mode":
        return self._group == GraphGrouping.daily
    if action == "exit_select_mode":
        return self._select_mode
    return True

compose()

Build the two-column layout.

Source code in src/rainlog/tui.py 
488
489
490
491
492
493
def compose(self) -> ComposeResult:
    """Build the two-column layout."""
    with Horizontal():
        yield BarChartWidget()
        yield StatsPanel()
    yield Footer()

on_mount()

Load initial data after the UI is ready.

Source code in src/rainlog/tui.py 
495
496
497
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 
499
500
501
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 
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
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

    def update_stats(
        self,
        period_total: float,
        daily_average: float,
        streak: tuple[str, int],
        selected_entry: tuple[str, float] | None = None,
    ) -> 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.refresh()

    def render(self) -> RenderableType:
        """Render stats, prepending selected-entry info when in select mode."""
        streak_type, streak_count = self._streak
        result = Text()
        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")
        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 
411
412
413
414
415
416
417
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

render()

Render stats, prepending selected-entry info when in select mode.

Source code in src/rainlog/tui.py 
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
def render(self) -> RenderableType:
    """Render stats, prepending selected-entry info when in select mode."""
    streak_type, streak_count = self._streak
    result = Text()
    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")
    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_stats(period_total, daily_average, streak, selected_entry=None)

Replace all stats and trigger a repaint.

Source code in src/rainlog/tui.py 
419
420
421
422
423
424
425
426
427
428
429
430
431
def update_stats(
    self,
    period_total: float,
    daily_average: float,
    streak: tuple[str, int],
    selected_entry: tuple[str, float] | None = None,
) -> 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.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 
223
224
225
226
227
228
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 
23
24
25
26
27
28
29
@Parameter(name="*")  # Flatten the namespace; i.e. option will be "--db-dir" instead of "--common.db-dir"
@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

add(reading, date=None, back_fill=False, common=None)

Add rain data to database.

Paramters

date: datetime Date to record rain for reading: float Rain reading common : Common, optional Shared options such as --db-path.

Source code in src/rainlog/cli_commands.py 
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
@app.command()
def add(
    reading: float,
    date: datetime | None = None,
    back_fill: bool = False,
    common: Common | None = None,
) -> None:
    """Add rain data to database.

    Paramters
    ---------
    date: datetime
        Date to record rain for
    reading: float
        Rain reading
    common : Common, optional
            Shared options such as --db-path.
    """
    if common is None:
        common = Common()
    if date is None:  # Cyclopts will have populated this with datetime.now()
        date = datetime.now()  # Fallback, though typically not needed
    rain_period_end = date.replace(hour=9, minute=0, second=0, microsecond=0)
    with Database(common.db_dir) as history:
        history.add_rain_record(date=rain_period_end, amount=reading)
        print(f"We had {reading} mm in the 24 hours ending at {rain_period_end}.")

        back_fill_date = rain_period_end - timedelta(days=1)
        while back_fill and not isinstance(history.get_single_day_rain(date=back_fill_date), float):
            history.add_rain_record(date=back_fill_date, amount=0.0)
            back_fill_date = back_fill_date - timedelta(days=1)
            print(f"Back filled 0.0 mm in the 24 hours ending at {back_fill_date}.")

calculate_interval(max_value)

Calculate order of magnitude.

Source code in src/rainlog/cli_commands.py 
250
251
252
253
254
255
256
257
258
259
260
def calculate_interval(max_value):
    """Calculate order of magnitude."""
    exponent = math.floor(math.log10(max_value))
    magnitude = 10**exponent

    # Find the best interval (1, 2, 5, 10) * magnitude
    for candidate in [1, 2, 5, 10]:
        interval = candidate * magnitude
        if max_value / interval <= 10:  # Aim for ~5-10 ticks
            return interval
    return 10 * magnitude  # Fallback

change(reading, date=None, common=None)

Update rain data in the database.

Paramters

date: datetime Date to record rain for reading: float Rain reading common : Common, optional Shared options such as --db-path.

Source code in src/rainlog/cli_commands.py 
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
@app.command()
def change(
    reading: float,
    date: datetime | None = None,
    common: Common | None = None,
) -> None:
    """Update rain data in the database.

    Paramters
    ---------
    date: datetime
        Date to record rain for
    reading: float
        Rain reading
    common : Common, optional
            Shared options such as --db-path.
    """
    if common is None:
        common = Common()
    if date is None:  # Cyclopts will have populated this with datetime.now()
        date = datetime.now()  # Fallback, though typically not needed
    rain_period_end = date.replace(hour=9, minute=0, second=0, microsecond=0)
    with Database(common.db_dir) as history:
        history.update_rain_record(date=rain_period_end, amount=reading)
    print(f"Update rain for {rain_period_end} to {reading} mm.")

default_datetime()

Register a default function for datetime parameters.

Source code in src/rainlog/cli_commands.py 
35
36
37
38
@app.default
def default_datetime() -> datetime:
    """Register a default function for datetime parameters."""
    return datetime.now()

graph(size=30, group=GraphGrouping.daily, common=None)

Retrieve historic rain data from DB and show it as a graph.

Parameters

size: int Number of graph elements to show group: GraphGrouping Grouping for graph common : Common, optional Shared options such as --db-path.

Source code in src/rainlog/cli_commands.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
@app.command()
def graph(
    size: int = 30,
    group: GraphGrouping = GraphGrouping.daily,
    common: Common | None = None,
) -> None:
    """Retrieve historic rain data from DB and show it as a graph.

    Parameters
    ----------
    size: int
        Number of graph elements to show
    group: GraphGrouping
        Grouping for graph
    common : Common, optional
            Shared options such as --db-path.

    """
    if common is None:
        common = Common()
    with Database(common.db_dir) as history:
        data_list = history.get_rain(history_size=size, group=group)
    dates, rains = zip(*data_list, strict=False)
    date_list = list(dates)
    rain_list = list(rains)
    max_y_tick = max(rain_list)

    y_ticks_interval = calculate_interval(max_y_tick)
    y_ticks = [y_ticks_interval * i for i in range(int(max_y_tick / y_ticks_interval) + 1)]
    y_ticks.append(max_y_tick)

    plotext.simple_bar(date_list, rain_list)
    plotext.ylabel(label="mm")
    plotext.xlabel(label="Date")
    plotext.yticks(ticks=y_ticks)
    plotext.title("Recent Rain")
    plotext.show()

migrate(db_dir=Path('.'))

Migrate MyWeather.sqlite from db_dir to the default XDG location as rainlog.sqlite.

Parameters

db_dir : Path Directory containing the existing MyWeather.sqlite to migrate (default: current directory).

Source code in src/rainlog/cli_commands.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
@app.command()
def migrate(db_dir: Path = Path(".")) -> None:
    """Migrate MyWeather.sqlite from db_dir to the default XDG location as rainlog.sqlite.

    Parameters
    ----------
    db_dir : Path
        Directory containing the existing MyWeather.sqlite to migrate (default: current directory).

    """
    source = db_dir / LEGACY_DB_FILE_NAME
    destination = DEFAULT_DB_DIR / DEFAULT_DB_FILE_NAME
    if not source.exists():
        raise SystemExit(f"No database found at {source}")
    if destination.exists():
        raise SystemExit(f"Database already exists at {destination} — aborting to avoid overwrite")
    _migrate_db(source, destination)
    print(f"Migrated {source} → {destination}")

rainy_days(size=30, group=GraphGrouping.daily, common=None)

Retrieve historic rain data from DB and show it as a graph.

Parameters

size: int Number of graph elements to show group: GraphGrouping Grouping for graph common : Common, optional Shared options such as --db-path.

Source code in src/rainlog/cli_commands.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
@app.command()
def rainy_days(
    size: int = 30,
    group: GraphGrouping = GraphGrouping.daily,
    common: Common | None = None,
) -> None:
    """Retrieve historic rain data from DB and show it as a graph.

    Parameters
    ----------
    size: int
        Number of graph elements to show
    group: GraphGrouping
        Grouping for graph
    common : Common, optional
            Shared options such as --db-path.

    """
    if common is None:
        common = Common()
    with Database(common.db_dir) as history:
        data_list = history.get_rainy_days(history_size=size, group=group)
    date_list: list[str] = []
    rain_list: list[float] = []
    max_y_tick = 0.0
    for day in data_list:
        date_list.append(day[0])
        rain_list.append(day[1])
        max_y_tick = max(max_y_tick, day[1])

    y_ticks: list[float] = []
    y_ticks_interval = 5.0
    for i in range(int(max_y_tick / y_ticks_interval) + 1):
        y_ticks.append(y_ticks_interval * i)
    y_ticks.append(max_y_tick)

    plotext.simple_bar(date_list, rain_list)
    plotext.yfrequency(frequency=5.0)
    plotext.ylabel(label="mm")
    plotext.xlabel(label="Date")
    plotext.yticks(ticks=y_ticks)
    plotext.title("Recent Rain")
    plotext.show()

tui(common=None)

Launch the interactive TUI for browsing rain history.

Source code in src/rainlog/cli_commands.py 
240
241
242
243
244
245
246
247
@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()

weewx_import(weewx_db=Path('./weewx.sdb'), common=None)

Export daily rain from weewx database into the DB for this project.

Parameters

weewx_db: Path Full path to the weewx sqlite database file. common : Common, optional Shared options such as --db-path.

Source code in src/rainlog/cli_commands.py 
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
@app.command()
def weewx_import(
    weewx_db: Path = Path("./weewx.sdb"),
    common: Common | None = None,
) -> None:
    """Export daily rain from weewx database into the DB for this project.

    Parameters
    ----------
    weewx_db: Path
        Full path to the weewx sqlite database file.
    common : Common, optional
            Shared options such as --db-path.

    """
    if common is None:
        common = Common()
    with Database(common.db_dir) as mydb:
        mydb.import_weewx(weewx_db_file=weewx_db)

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 
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
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_rainy_days(
        self: Self,
        history_size: int,
        group: GraphGrouping,
    ) -> list[tuple[str, float]]:
        """Get 'history_size' number of rain records."""
        return_list: list[tuple[str, float]] = []
        group_id = None
        group_sum = 0.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),
            )

            day_rainy = 1 if row[1] > 0 else 0

            if not group_id:
                group_id = current_group_id

            if current_group_id == group_id:
                group_sum += day_rainy
            else:
                return_list.append((group_id, group_sum))
                group_id = current_group_id
                group_sum = day_rainy

            if len(return_list) >= history_size:
                break

        if len(return_list) < history_size and group_id:
            return_list.append((group_id, group_sum))

        return return_list

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

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

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

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

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

        return (streak_type, streak_count)

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

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

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

        group_id: str = group_date.strftime(format_for_grouping)

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

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

        return group_id

    def import_weewx(self: Self, weewx_db_file: Path) -> None:
        """Import data from a weewx database file."""
        print(f"Importing data from {weewx_db_file} to {self.db_connection}")

        weewx_rain_in_mm: list[tuple[int, float]] = []

        with sqlite3.connect(database=weewx_db_file) as weewx_db:
            for row in weewx_db.execute("SELECT dateTime, sum FROM archive_day_rain"):
                weewx_rain_in_mm.append(
                    (row[0], row[1] * 25.4),
                )

        self.db_connection.executemany(
            "INSERT INTO rain_daily (date, rain) VALUES (?,?)",
            weewx_rain_in_mm,
        )

        self.db_connection.commit()

__enter__()

Return self to support use as a context manager.

Source code in src/rainlog/db_helpers.py 
39
40
41
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 
43
44
45
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 
27
28
29
30
31
32
33
34
35
36
37
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 
47
48
49
50
51
52
53
54
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 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
def get_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 
189
190
191
192
193
194
195
196
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_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 
180
181
182
183
184
185
186
187
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 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
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_rainy_days(history_size, group)

Get 'history_size' number of rain records.

Source code in src/rainlog/db_helpers.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def get_rainy_days(
    self: Self,
    history_size: int,
    group: GraphGrouping,
) -> list[tuple[str, float]]:
    """Get 'history_size' number of rain records."""
    return_list: list[tuple[str, float]] = []
    group_id = None
    group_sum = 0.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),
        )

        day_rainy = 1 if row[1] > 0 else 0

        if not group_id:
            group_id = current_group_id

        if current_group_id == group_id:
            group_sum += day_rainy
        else:
            return_list.append((group_id, group_sum))
            group_id = current_group_id
            group_sum = day_rainy

        if len(return_list) >= history_size:
            break

    if len(return_list) < history_size and group_id:
        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 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
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

import_weewx(weewx_db_file)

Import data from a weewx database file.

Source code in src/rainlog/db_helpers.py 
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
def import_weewx(self: Self, weewx_db_file: Path) -> None:
    """Import data from a weewx database file."""
    print(f"Importing data from {weewx_db_file} to {self.db_connection}")

    weewx_rain_in_mm: list[tuple[int, float]] = []

    with sqlite3.connect(database=weewx_db_file) as weewx_db:
        for row in weewx_db.execute("SELECT dateTime, sum FROM archive_day_rain"):
            weewx_rain_in_mm.append(
                (row[0], row[1] * 25.4),
            )

    self.db_connection.executemany(
        "INSERT INTO rain_daily (date, rain) VALUES (?,?)",
        weewx_rain_in_mm,
    )

    self.db_connection.commit()

update_rain_record(date, amount)

Update a record / measurement of rain in the DB.

Source code in src/rainlog/db_helpers.py 
56
57
58
59
60
61
62
63
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 
14
15
16
17
18
19
20
21
class GraphGrouping(str, Enum):
    """Provides possible values for grouping of graphs."""

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