Skip to content

API reference

The public Python surface is small: package metadata in weather_tools, the Cyclopts app in weather_tools.cli_commands, and Database / GraphGrouping in weather_tools.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/weather_tools/cli_commands.py 
17
18
19
20
21
22
23
@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 = Path(".")
    "Path to database file"

db_dir = Path('.') 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/weather_tools/cli_commands.py 
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
@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()
    history = Database(common.db_dir)
    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)
    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/weather_tools/cli_commands.py 
201
202
203
204
205
206
207
208
209
210
211
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/weather_tools/cli_commands.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
@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
    history = Database(common.db_dir)
    rain_period_end = date.replace(hour=9, minute=0, second=0, microsecond=0)
    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/weather_tools/cli_commands.py 
29
30
31
32
@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/weather_tools/cli_commands.py 
 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
@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()
    history = Database(common.db_dir)
    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()

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/weather_tools/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
145
146
147
148
149
150
@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()
    history = Database(common.db_dir)
    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()

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/weather_tools/cli_commands.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
@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()
    mydb = Database(common.db_dir)
    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/weather_tools/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
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 and table(s) if they don't exist yet."""
        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 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,
    ) -> 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),
            )

            if not group_id:
                group_id = current_group_id

            if current_group_id == group_id:
                group_sum += row[1]
            else:
                return_list.append((group_id, group_sum))
                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:
            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

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

__init__(db_dir)

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

Source code in src/weather_tools/db_helpers.py 
27
28
29
30
31
32
33
34
35
36
def __init__(self: Self, db_dir: Path) -> None:
    """Create database connection. Also creates database and table(s) if they don't exist yet."""
    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/weather_tools/db_helpers.py 
38
39
40
41
42
43
44
45
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_rain(history_size, group)

Get 'history_size' number of rain records.

Source code in src/weather_tools/db_helpers.py 
 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
def get_rain(
    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),
        )

        if not group_id:
            group_id = current_group_id

        if current_group_id == group_id:
            group_sum += row[1]
        else:
            return_list.append((group_id, group_sum))
            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:
        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/weather_tools/db_helpers.py 
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
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/weather_tools/db_helpers.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
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/weather_tools/db_helpers.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
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/weather_tools/db_helpers.py 
47
48
49
50
51
52
53
54
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/weather_tools/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"

Main methods to interact with rain data.

Common dataclass

Class for common db-path parameter.

Source code in src/weather_tools/cli_commands.py 
17
18
19
20
21
22
23
@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 = Path(".")
    "Path to database file"

db_dir = Path('.') 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/weather_tools/cli_commands.py 
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
@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()
    history = Database(common.db_dir)
    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)
    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/weather_tools/cli_commands.py 
201
202
203
204
205
206
207
208
209
210
211
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/weather_tools/cli_commands.py 
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
@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
    history = Database(common.db_dir)
    rain_period_end = date.replace(hour=9, minute=0, second=0, microsecond=0)
    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/weather_tools/cli_commands.py 
29
30
31
32
@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/weather_tools/cli_commands.py 
 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
@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()
    history = Database(common.db_dir)
    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()

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/weather_tools/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
145
146
147
148
149
150
@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()
    history = Database(common.db_dir)
    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()

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/weather_tools/cli_commands.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
@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()
    mydb = Database(common.db_dir)
    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/weather_tools/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
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 and table(s) if they don't exist yet."""
        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 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,
    ) -> 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),
            )

            if not group_id:
                group_id = current_group_id

            if current_group_id == group_id:
                group_sum += row[1]
            else:
                return_list.append((group_id, group_sum))
                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:
            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

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

__init__(db_dir)

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

Source code in src/weather_tools/db_helpers.py 
27
28
29
30
31
32
33
34
35
36
def __init__(self: Self, db_dir: Path) -> None:
    """Create database connection. Also creates database and table(s) if they don't exist yet."""
    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/weather_tools/db_helpers.py 
38
39
40
41
42
43
44
45
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_rain(history_size, group)

Get 'history_size' number of rain records.

Source code in src/weather_tools/db_helpers.py 
 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
def get_rain(
    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),
        )

        if not group_id:
            group_id = current_group_id

        if current_group_id == group_id:
            group_sum += row[1]
        else:
            return_list.append((group_id, group_sum))
            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:
        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/weather_tools/db_helpers.py 
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
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/weather_tools/db_helpers.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
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/weather_tools/db_helpers.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
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/weather_tools/db_helpers.py 
47
48
49
50
51
52
53
54
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/weather_tools/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"