ESP気象センサーの気象データ収集システム

　Raspberry Pi zero (以下ラズパイゼロ) で収集している気象データDBの外気温を集計する2つの方法を紹介します。

1. 気象データDB

1-1. テーブル定義

-- 気象センサーデバイス名テーブル
CREATE TABLE IF NOT EXISTS t_device(
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name VARCHAR UNIQUE NOT NULL
);
-- 観測データテーブル
CREATE TABLE IF NOT EXISTS t_weather(
    did INTEGER NOT NULL,
    measurement_time INTEGER NOT NULL,
    temp_out real,
    temp_in real,
    humid real,
    pressure real,
    PRIMARY KEY (did, measurement_time),
    FOREIGN KEY (did) REFERENCES t_devices (id) ON DELETE CASCADE
);

1-2. 気象データの登録内容

気象センサーデバイス名テーブル

"id","name"
1,esp8266_1

観測データテーブル
- ESP気象センサからUDPパケットで１０分ごとに受信
- [行データ] デバイスID, 測定時刻(パケット到着時刻), 外気温, 室内気温, 室内湿度, 気圧

"did","measurement_time","temp_out","temp_in","humid","pressure"
1,"2024-01-03 00:02:49",1.1,12.8,55.6,1008.7
1,"2024-01-03 00:12:34",1.1,12.7,55.7,1008.7
1,"2024-01-03 00:22:17",1.1,12.6,55.6,1008.7
1,"2024-01-03 00:32:01",1.1,12.4,55.5,1008.8
1,"2024-01-03 00:41:45",1.1,12.3,55.4,1008.9
...[中略]...
1,"2024-01-03 23:34:32",-1.3,12.0,48.2,1012.1
1,"2024-01-03 23:44:14",-1.3,11.9,48.2,1011.9
1,"2024-01-03 23:54:00",-1.3,11.8,48.2,1011.9

2. 集計方法

観測データ: 測定時刻と外気温のみを対象とする
1日分の外気温データから最低気温と最高気温を出現時刻も含めて取得する
最低気温・最高気温は複数回出現する可能性が有るので、それぞれ直近のデータ１件とする
指定した日付とその前日のデータを集計する

2-1. 集計結果をテキストデータとしてHTMLで出力

(A) SQLのウインドウ関数を使って集計結果を取得
(B) pandas で集計結果を取得

2-2. 集計結果を可視化画像に埋め込む

pandas + matplotlib で集計結果とデータのプロット画像を生成する

3. 実行環境

4. 外気温データの集計方法

4-1. 集計SQL

CTE と SQLウインドウ関数を組み合わせて下記のように定義

WITH find_records AS (
SELECT
  datetime(measurement_time, 'unixepoch', 'localtime') AS measurement_time,
  CASE
    WHEN temp_out <= min(temp_out) OVER (PARTITION BY date(measurement_time))
	THEN temp_out
  END AS min_temp_out,
  CASE
    WHEN temp_out >= max(temp_out) OVER (PARTITION BY date(measurement_time))
	THEN temp_out
  END AS max_temp_out
FROM
  t_weather tw INNER JOIN t_device td ON tw.did = td.id
WHERE
  td.name = ?
  AND (
    datetime(measurement_time, 'unixepoch', 'localtime') >= ? 
    AND 
    datetime(measurement_time, 'unixepoch', 'localtime') < ?
  )
-- 降順でソート ※最低気温と最高気温は指定範囲に複数出現するが、直近レコードを取得値とする  
ORDER BY measurement_time DESC
), min_temp_out_records AS (
--直近の最低気温 1レコード
  SELECT
     measurement_time, min_temp_out AS temp_out
  FROM
     find_records
  WHERE
     min_temp_out IS NOT NULL LIMIT 1
), max_temp_out_records AS (
--直近の最高気温 1レコード
  SELECT
     measurement_time, max_temp_out AS temp_out
  FROM
     find_records
  WHERE
     max_temp_out IS NOT NULL LIMIT 1
)
-- 最低気温
SELECT * FROM min_temp_out_records
UNION ALL
-- 最高気温
SELECT * FROM max_temp_out_records;

まず上記SQLを組み込んだ動作確認用のシェルスクリプトを作成します

get_weather_window_func.sh

#!/bin/bash

next_to_date() {
    retval=$(date -d "$1 1 days" +'%F');
    echo "$retval"
}

get_records() {
  device_name="$1";
  from_date="$2";
  eclude_to_date="$3"
cat<<-EOF | sqlite3 "${PATH_WEATHER_DB}" -csv
WITH find_records AS (
SELECT
  datetime(measurement_time, 'unixepoch', 'localtime') AS measurement_time,
  CASE
    WHEN temp_out <= min(temp_out) OVER (PARTITION BY date(measurement_time))
	THEN temp_out
  END AS min_temp_out,
  CASE
    WHEN temp_out >= max(temp_out) OVER (PARTITION BY date(measurement_time))
	THEN temp_out
  END AS max_temp_out
FROM
  t_weather tw INNER JOIN t_device td ON tw.did = td.id
WHERE
  td.name = '${device_name}'
  AND (
    datetime(measurement_time, 'unixepoch', 'localtime') >= '${from_date}' 
    AND 
    datetime(measurement_time, 'unixepoch', 'localtime') < '${eclude_to_date}'
  )
-- 降順でソート ※最低気温と最高気温は指定範囲に複数出現するが、直近レコードを取得値とする  
ORDER BY measurement_time DESC
), min_temp_out_records AS (
--直近の最低気温 1レコード
  SELECT
     measurement_time, min_temp_out AS temp_out
  FROM
     find_records
  WHERE
     min_temp_out IS NOT NULL LIMIT 1
), max_temp_out_records AS (
--直近の最高気温 1レコード
  SELECT
     measurement_time, max_temp_out AS temp_out
  FROM
     find_records
  WHERE
     max_temp_out IS NOT NULL LIMIT 1
)
SELECT * FROM min_temp_out_records
UNION ALL
SELECT * FROM max_temp_out_records;
EOF
}

# Eclude to_date
exclude_to_date=$(next_to_date "$2");

echo '"measurement_time","temp_out"'
get_records "$1" "$2" "${exclude_to_date}"

シェルスクリプトを実行しSQLが正しく動作するか確認します
※予めラスパイゼロから気象データDBファイルを実行環境にコピー

$ # SQLite3 気象テータベースパス ※ラズパイゼロからPCにコピー
$ export PATH_WEATHER_DB=~/db/weather.db
$ # [引数] デバイス名 検索日 ※ISO8601形式
$ ./get_weather_group.sh esp8266_1 2024-01-03
"measurement_time","temp_out"
"2024-01-03 23:15:03",-1.4
"2024-01-03 09:56:45",1.4
$ # 前日の集計結果
$ ./get_weather_group.sh esp8266_1 2024-01-02
"measurement_time","temp_out"
"2024-01-02 03:35:55",-6.7
"2024-01-02 14:28:25",3.5

4-1-1. SQL 集計モジュールの定義

(1) インポートと集計結果(データクラス)の定義

plot_weather/dao/windowfunc_sqlite.py

import sqlite3
from dataclasses import dataclass
from typing import List, Tuple

"""
SQLite3 気象データベースから外気温統計情報を取得するモジュール
SQL window function with CTE
"""


@dataclass
class TempOut:
    appear_time: str
    temper: float


# 集計用SQLの定義
_STAT_QUERY = """
WITH find_records AS (
SELECT
  datetime(measurement_time, 'unixepoch', 'localtime') AS measurement_time,
  CASE
    WHEN temp_out <= min(temp_out) OVER (PARTITION BY date(measurement_time))
	THEN temp_out
  END AS min_temp_out,
  CASE
    WHEN temp_out >= max(temp_out) OVER (PARTITION BY date(measurement_time))
	THEN temp_out
  END AS max_temp_out
FROM
  t_weather tw INNER JOIN t_device td ON tw.did = td.id
WHERE
  td.name = ?
  AND (
    datetime(measurement_time, 'unixepoch', 'localtime') >= ? 
    AND 
    datetime(measurement_time, 'unixepoch', 'localtime') < ?
  )
-- 降順でソート ※最低気温と最高気温は指定範囲に複数出現するが、直近レコードを取得値とする  
ORDER BY measurement_time DESC
), min_temp_out_records AS (
--直近の最低気温 1レコード
  SELECT
     measurement_time, min_temp_out AS temp_out
  FROM
     find_records
  WHERE
     min_temp_out IS NOT NULL LIMIT 1
), max_temp_out_records AS (
--直近の最高気温 1レコード
  SELECT
     measurement_time, max_temp_out AS temp_out
  FROM
     find_records
  WHERE
     max_temp_out IS NOT NULL LIMIT 1
)
SELECT * FROM min_temp_out_records
UNION ALL
SELECT * FROM max_temp_out_records
"""

(3) 集計結果取得関数の定義

関数の引数
- SQLite3 接続オブジェクト
- デバイス名, 検索開始日, 検索日終了日 (含まない) ※シェルスクリプトと同じ
検索結果
- 検索データが存在すれば２件のリストを返却
- 検索データが存在しない場合は空のリストを返却

def get_temp_out_stat(conn: sqlite3.Connection,
                      device_name: str,
                      from_date: str, exclude_to_date: str) -> List[TempOut]:
    params: Tuple[str, str, str] = (device_name, from_date, exclude_to_date,)
    result: List[TempOut] = []
    with conn:
        cursor: sqlite3.Cursor = conn.execute(_STAT_QUERY, params)
        rows: List[Tuple[str, float]] = cursor.fetchall()
        if len(rows) > 0:
            for row in rows:
                measurement_time: str = row[0]
                temp_out: float = row[1]
                rec: TempOut = TempOut(measurement_time, temp_out)
                result.append(rec)
    return result

3-2. pandas を使った集計

3-2-1. 検索SQLから pandasのDataFrameを取得するモジュール

(1) インポートと検索クエリーの定義

plot_weather/dataloader/tempout_loader_sqlite.py

import sqlite3
from typing import Tuple

import pandas as pd
from pandas.core.frame import DataFrame

"""
SQLite3 気象データの外気温ロードモジュール
"""


COL_TIME: str = "measurement_time"
COL_TEMP_OUT: str = "temp_out"


_QUERY: str = """
SELECT
   datetime(measurement_time,'unixepoch', 'localtime') as measurement_time,
   temp_out
FROM
   t_weather tw INNER JOIN t_device td ON tw.did=td.id
WHERE
   td.name = ?
   AND (
       datetime(measurement_time,'unixepoch', 'localtime') >= ?
       AND
       datetime(measurement_time,'unixepoch', 'localtime') < ?
   )
ORDER BY measurement_time DESC;
"""

(2) DataFrame取得関数の定義

関数の引数
- SQLite3 接続オブジェクト
- デバイス名, 検索開始日, 検索日終了日 (含まない)

def get_dataframe(conn: sqlite3.Connection,
                  device_name: str, from_date: str, exclude_to_date
                  ) -> DataFrame:
    params: Tuple = (device_name, from_date, exclude_to_date)
    try:
        read_df = pd.read_sql(_QUERY, conn, params=params, parse_dates=[COL_TIME])
        return read_df
    except Exception as err:
        raise err

3-2-2. DataFrameから集計結果を生成するモジュール

このモジュールの実装内容の詳細は下記 Qiita 投稿をご覧ください。
Qiita投稿 @pipito-yukio(吉田幸雄): Matplotlib 見栄えの良い凡例を作る

plot_weather/dataloader/tempout_stat.py

from dataclasses import dataclass
from datetime import datetime

import numpy as np
import pandas as pd
from pandas.core.frame import DataFrame
from pandas.core.series import Series

from plot_weather.dataloader.tempout_loader_sqlite import (
    COL_TIME, COL_TEMP_OUT
)

"""
外気温集計モジュール by pandas
"""


@dataclass
class TempOut:
    appear_time: str
    temper: float


@dataclass
class TempOutStat:
    """ 外気温統計情報 """
    # 測定日
    measurement_day: str
    # 最低外気温情報
    min: TempOut
    # 最高外気温情報
    max: TempOut


def get_temp_out_stat(df_desc: DataFrame) -> TempOutStat:
    """ 外気温の統計情報 ([最低気温|最高気温] の気温とその出現時刻) を取得する """

    def get_measurement_time(pd_timestamp: pd.Timestamp) -> str:
        py_datetime: datetime = pd_timestamp.to_pydatetime()
        # 時刻部分は "時:分"までとする
        return py_datetime.strftime("%Y-%m-%d %H:%M")

    # 外気温列
    temp_out_ser: Series = df_desc[COL_TEMP_OUT]
    # 外気温列から最低・最高・平均気温を取得
    min_temper: np.float64 = temp_out_ser.min()
    max_temper: np.float64 = temp_out_ser.max()
    # 全ての最低気温を取得する
    df_min_all: DataFrame = df_desc[temp_out_ser <= min_temper]
    # 全ての最高気温を取得する
    df_max_all: pd.DataFrame = df_desc[temp_out_ser >= max_temper]
    # それぞれ直近の１レコードのみ取得
    min_first: Series = df_min_all.iloc[0]
    max_first: Series = df_max_all.iloc[0]
    # 測定日は先頭 10桁分(年月日)
    min_measurement_time: str = get_measurement_time(min_first[COL_TIME])
    measurement_day: str = min_measurement_time[:10]
    # 最低気温情報
    min_data: TempOut = TempOut(min_measurement_time, float(min_first[COL_TEMP_OUT]))
    # 最高気温情報
    max_measurement_time: str = get_measurement_time(max_first[COL_TIME])
    max_data: TempOut = TempOut(max_measurement_time, float(max_first[COL_TEMP_OUT]))
    return TempOutStat(measurement_day, min=min_data, max=max_data)

4. pythonメインスクリプト

Webアプリで実装するのは面倒なので pythonメインアプリケーションとして実装する。

4-0. バッチ用スクリプトの共通モジュール

HTML出力用テンプレート
※1 辞書オブジェクトのキーの埋め込み (例) <td colspan="2">{find_day}</dt>
※2 テンプレートと辞書オブジェクトを合体させてHTMLを生成する

batch_common.py

import sqlite3
from datetime import datetime, timedelta


# 出力画層用HTMLテンプレート (Bootstrap5.x with CDN)
OUT_HTML = """
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<link href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.2/dist/css/bootstrap.min.css" 
 rel="stylesheet" integrity="sha384-T3c6CoIi6uLrA9TneNEoa7RxnatzjcDSCmG1MXxSR1GAsXEV/Dwwykc2MPK8M2HN"
 crossorigin="anonymous">
</head>
<body>
<div class="container">
<h1 class="text-center m-3">外気温統計情報</h1>
<div class="table-responsive-sm">
<table class="table table-sm table-bordered">
  <tbody>
  <tr>
    <th scope="row" class="table-light">当日測定値</th>
    <td colspan="2">{find_day}</td>
  </tr>
  <tr class="table-primary align-middle">
    <th scope="row" class="text-end">最低気温</th>
    <td class="align-middle text-center">{find_min_time}</td>
    <td class="text-end">{find_min_temper} ℃</td>
  </tr>
  <tr class="table-danger align-middle">
    <th scope="row" class="text-end">最高気温</th>
    <td class="align-middle text-center">{find_max_time}</td>
    <td class="text-end">{find_max_temper} ℃</td>
  </tr>
  </tbody>
</table>
</div>
<div class="table-responsive-sm">
<table class="table table-sm table-bordered">
  <tbody>
  <tr>
    <th scope="row">前日測定値</th>
    <td colspan="2">{before_day}</td>
  </tr>
  <tr class="table-primary align-middle">
    <th scope="row" class="text-end">最低気温</th>
    <td class="text-center">{before_min_time}</td>
    <td class="text-end">{before_min_temper} ℃</td>
  </tr>
  <tr class="table-danger align-middle">
    <th scope="row" class="text-end">最高気温</th>
    <td class="text-center">{before_max_time}</td>
    <td class="text-end">{before_max_temper} ℃</td>
  </tr>
  </tbody>
</table>
</div>
</div>
</body>
</html>
"""

共通関数定義
- SQLite3 データベース接続オブジェクト取得関数
- HTMLファイル保存関数
- ISO8601日付文字列の n日加減算関数
  ※ SQLの検索終了日の計算 ※検索終了日を含まない
- ISO8601日付文字列を日本語の年月日な変換する関数
  ※ (例) "2024-01-03" ⇒ "2024 年 01 月 03 日"

# (1) SQLite3 データベース接続オブジェクト取得関数
def get_connection(db_path: str, auto_commit=False, read_only=False, logger=None):
    connection: sqlite3.Connection
    try:
        if read_only:
            db_uri = "file://{}?mode=ro".format(db_path)
            connection = sqlite3.connect(db_uri, uri=True)
        else:
            connection = sqlite3.connect(db_path)
            if auto_commit:
                connection.isolation_level = None
    except sqlite3.Error as e:
        if logger is not None:
            logger.error(e)
        raise e
    return connection


# (2) HTMLファイル保存関数
def save_html(file, contents):
    with open(file, 'w') as fp:
        fp.write(contents)


# (3) ISO8601日付文字列+n日加算関数
def date_add_days(iso8601_date: str, add_days=1) -> str:
    dt: datetime = datetime.strptime(iso8601_date, "%Y-%m-%d")
    dt += timedelta(days=add_days)
    return dt.strftime("%Y-%m-%d")


# (4) ISO8601日付文字列を日本語の年月日に変換する関数
def to_title_date(curr_date: str) -> str:
    dt: datetime = datetime.strptime(curr_date, "%Y-%m-%d")
    return dt.strftime("%Y 年 %m 月 %d 日")

4-1. SQLウインドウ関数取得モジュールを使用するスクリプト

4-1-1. インポート

TempOutStatHtml_sqliteFunc.py

import argparse
import os
import sqlite3
from typing import Dict, List, Optional

from plot_weather.dao.windowfunc_sqlite import (
    get_temp_out_stat, TempOut
)
from batch_common import (
    get_connection, date_add_days, to_title_date, save_html, OUT_HTML
)

"""
外気温の当日データと前日データの統計情報をHTMLに出力
[DB] sqlite3 気象データ
[集計方法] SQL Window function
"""

# スクリプト名
script_name = os.path.basename(__file__)

4-1-2. メイン処理

入力パラメータ処理
- データベースファイルパス (例) --db-path ~/db/weather.db
- 観測デバイス名 (例) --device-name esp8266_1
- 検索日 (例) --find-date 2024-01-03

if __name__ == '__main__':
    parser: argparse.ArgumentParser = argparse.ArgumentParser()
    # データペースパス: ~/db/weather.db
    parser.add_argument("--db-path", type=str, required=True,
                        help="SQLite3 Database path.")
    # デバイス名: esp8266_1
    parser.add_argument("--device-name", type=str, required=True,
                        help="device name in t_device.")
    # 検索日: 2023-11-01
    parser.add_argument("--find-date", type=str, required=True,
                        help="ISO8601 format.")
    args: argparse.Namespace = parser.parse_args()
    # SQLite3 気象データペースファイルパス
    db_full_path: str = os.path.expanduser(args.db_path)
    # デバイス名
    device_name: str = args.device_name
    # 検索日
    find_date: str = args.find_date

データベース接続オブジェクト取得
- (1) 検索日の集計結果データを取得
  - 検索日の辞書オブジェクトのキーに対応するデータを辞書オブジェクトに設定する
- (2) 前日の集計結果データを取得
  - 前日の辞書オブジェクトのキーに対応するデータを辞書オブジェクトに設定する
- (3) HTMLテンプレートと辞書オブジェクトを合体させてHTML文字列を生成
- (4) HTML文字列をスクリプト名でファイル保存する
データベース接続オブジェクトクローズ

    html_dict: Dict = {}
    exclude_to_date: str
    conn: Optional[sqlite3.Connection] = None
    try:
        conn = get_connection(db_full_path)
        # 検索日のデータ
        exclude_to_date = date_add_days(find_date)
        stat_data: List[TempOut] = get_temp_out_stat(
            conn, device_name, find_date, exclude_to_date
        )
        find_day_min: TempOut = stat_data[0]
        find_day_max: TempOut = stat_data[1]
        print(f"today_min: {find_day_min}, today_max: {find_day_max}")
        # HTML用辞書オブジェクトに指定日データを設定する
        html_dict["find_day"] = to_title_date(find_date)
        html_dict["find_min_time"] = find_day_min.appear_time[11:16]
        html_dict["find_min_temper"] = find_day_min.temper
        html_dict["find_max_time"] = find_day_max.appear_time[11:16]
        html_dict["find_max_temper"] = find_day_max.temper

        # 前日の統計情報
        before_date: str = date_add_days(find_date, add_days=-1)
        stat_data: List[TempOut] = get_temp_out_stat(
            conn, device_name, before_date, find_date
        )
        before_day_min: TempOut = stat_data[0]
        before_day_max: TempOut = stat_data[1]
        print(f"before_min: {before_day_min}, before_max: {before_day_max}")
        # HTML用辞書オブジェクトに前日データを設定する
        html_dict["before_day"] = to_title_date(before_date)
        html_dict["before_min_time"] = before_day_min.appear_time[11:16]
        html_dict["before_min_temper"] = before_day_min.temper
        html_dict["before_max_time"] = before_day_max.appear_time[11:16]
        html_dict["before_max_temper"] = before_day_max.temper

        # HTMLを生成する
        html: str = OUT_HTML.format(**html_dict)
        script_names: List[str] = script_name.split(".")
        save_name = f"{script_names[0]}.html"
        save_path = os.path.join("output", save_name)
        print(save_path)
        save_html(save_path, html)
    except sqlite3.Error as db_err:
        print(f"type({type(db_err)}): {db_err}")
        exit(1)
    except Exception as exp:
        print(exp)
        exit(1)
    finally:
        if conn is not None:
            conn.close()

4-2. pandas の集計モジュールを使用するスクリプト

4-2-1. インポート

TempOutStatHtml_pandas.py

import argparse
import os
import sqlite3
from typing import Dict, List, Optional

from pandas.core.frame import DataFrame

from plot_weather.dataloader.tempout_loader_sqlite import (
    get_dataframe
)
from plot_weather.dataloader.tempout_stat import (
    get_temp_out_stat, TempOutStat, TempOut
)
from batch_common import (
    get_connection, date_add_days, to_title_date, save_html, OUT_HTML
)

"""
外気温の当日データと前日データの統計情報をHTMLに出力
[DB] sqlite3 気象データ
[集計方法] pandas
"""

# スクリプト名
script_name = os.path.basename(__file__)

4-2-2. メイン処理

入力パラメータ処理 ※4-1-2と同一
データベース接続オブジェクト取得
- (1) 検索日の集計結果取得処理
  - 検索SQLを実行して DataFrame (測定時刻の降順) を生成する
  - 上記 DataFrame から集計結果テータを取得する
  - 辞書オブジェクトのキーに対応するデータを辞書オブジェクトに設定する
- (2) 前日の集計結果取得処理 ※処理順は上記 (1)と同じ
- (3) HTMLテンプレートと辞書オブジェクトを合体させてHTML文字列を生成
- (4) HTML文字列をスクリプト名でファイル保存する
データベース接続オブジェクトクローズ

if __name__ == '__main__':
    # 入力パラメータ処理は割愛 4-1-2 参照
    html_dict: Dict = {}
    exclude_to_date: str
    conn: Optional[sqlite3.Connection] = None
    try:
        conn = get_connection(db_full_path)
        # 指定日の翌日 (含まない)
        exclude_to_date = date_add_days(find_date)
        df_find: DataFrame = get_dataframe(
            conn, device_name, find_date, exclude_to_date
        )
        # 外気温統計データ
        find_stat: TempOutStat = get_temp_out_stat(df_find)
        find_day_min: TempOut = find_stat.min
        find_day_max: TempOut = find_stat.max
        print(f"today_min: {find_day_min}, today_max: {find_day_max}")
        # HTML用辞書オブジェクトに指定日データを設定する
        html_dict["find_day"] = to_title_date(find_date)
        html_dict["find_min_time"] = find_day_min.appear_time[11:16]
        html_dict["find_min_temper"] = find_day_min.temper
        html_dict["find_max_time"] = find_day_max.appear_time[11:16]
        html_dict["find_max_temper"] = find_day_max.temper

        # 前日の統計情報
        before_date: str = date_add_days(find_date, add_days=-1)
        df_before: DataFrame = get_dataframe(
            conn, device_name, before_date, find_date, order_by_desc=True
        )
        before_stat: TempOutStat = get_temp_out_stat(df_before)
        before_day_min: TempOut = before_stat.min
        before_day_max: TempOut = before_stat.max
        print(f"before_min: {before_day_min}, before_max: {before_day_max}")
        # HTML用辞書オブジェクトに前日データを設定する
        html_dict["before_day"] = to_title_date(before_date)
        html_dict["before_min_time"] = before_day_min.appear_time[11:16]
        html_dict["before_min_temper"] = before_day_min.temper
        html_dict["before_max_time"] = before_day_max.appear_time[11:16]
        html_dict["before_max_temper"] = before_day_max.temper

        # HTMLを生成する
        html: str = OUT_HTML.format(**html_dict)
        script_names: List[str] = script_name.split(".")
        save_name = f"{script_names[0]}.html"
        save_path = os.path.join("output", save_name)
        print(save_path)
        save_html(save_path, html)
    except sqlite3.Error as db_err:
        print(f"type({type(db_err)}): {db_err}")
        exit(1)
    except Exception as exp:
        print(exp)
        exit(1)
    finally:
        if conn is not None:
            conn.close()

4-3. 検索日の可視化画像生成

4-3-1. 可視化画像生成モジュール

plotweather_sqlite.py

import base64
from io import BytesIO
from datetime import datetime, timedelta
from typing import Dict, List

from pandas.core.frame import DataFrame

import matplotlib.dates as mdates
from matplotlib import rcParams
from matplotlib.axes import Axes
from matplotlib.figure import Figure
from matplotlib.patches import Patch
from matplotlib.pyplot import setp
from matplotlib.legend import Legend
from matplotlib.text import Text

from plot_weather.dataloader.tempout_loader_sqlite import (
    COL_TIME, COL_TEMP_OUT
)
from plot_weather.dataloader.pandas_statistics import (
    get_temp_out_stat, TempOutStat, TempOut
)

""" 気象データの外気温プロット画像のbase64エンコードテキストデータを出力する """

# 日本語表示
rcParams['font.family'] = ["sans-serif", "monospace"]
rcParams['font.sans-serif'] = ["IPAexGothic", "Noto Sans CJK JP"]
# 固定ピッチフォント
rcParams['font.monospace'] = ["Source Han Code JP", "Noto Sans Mono CJK JP"]
# カラー定数定義
COLOR_MIN_TEMPER: str = "darkcyan"
COLOR_MAX_TEMPER: str = "orange"


def sub_graph(ax: Axes, title: str, df: DataFrame, temp_stat: TempOutStat):
    def make_patch(label: str, temp_out: TempOut, patch_color: str) -> Patch:
        """ 指定されたラベルと外気温統計の凡例を生成 """
        # 発現時刻は時分
        patch_label: str = f"{label} {temp_out.temper:4.1f} ℃ [{temp_out.appear_time[11:16]}"
        return Patch(color=patch_color, label=patch_label)

    def plot_hline(axes: Axes, temper: float, line_color: str):
        """ 指定された統計情報の外気温の横線を生成する """
        line_style_dict: Dict = {"color": line_color, "linestyle": "dashed", "linewidth": 1.}
        axes.axhline(temper, **line_style_dict)

    # グリッド線
    ax.grid(linestyle="dotted", linewidth=1.0)
    # 軸ラベルのフォントサイズを小さめに設定
    setp(ax.get_xticklabels(), fontsize=9.)
    setp(ax.get_yticklabels(), fontsize=9.)
    # x軸フォーマット: 軸ラベルは時間 (00,03,06,09,12,15,18,21,翌日の00)
    ax.xaxis.set_major_formatter(mdates.DateFormatter("%H"))
    # 外気温の最低(-20℃)と最大(40℃)
    ax.set_ylim(ymin=-20.0, ymax=40.0)
    # Y軸ラベル
    ax.set_ylabel("外気温 (℃)", fontsize=10.)

    # タイトルの測定日付
    curr_date: str = temp_stat.measurement_day
    # 最低気温と最高気温の凡例を生成
    mim_patch: Patch = make_patch("最低", temp_stat.min, COLOR_MIN_TEMPER)
    max_patch: Patch = make_patch("最高", temp_stat.max, COLOR_MAX_TEMPER)
    # 当日データx軸の範囲: 当日 00時 から 翌日 00時
    dt_min: datetime = datetime.strptime(curr_date, "%Y-%m-%d")
    dt_max: datetime = dt_min + timedelta(days=1)
    # タイトル日付
    title_date: str = dt_min.strftime("%Y 年 %m 月 %d 日")
    ax.set_xlim(xmin=dt_min, xmax=dt_max)
    # 外気温データプロット
    ax.plot(df[COL_TIME], df[COL_TEMP_OUT], color="blue", marker="")
    # 最低気温の横線
    plot_hline(ax, temp_stat.min.temper, COLOR_MIN_TEMPER)
    # 最高気温の横線
    plot_hline(ax, temp_stat.max.temper, COLOR_MAX_TEMPER)
    ax.set_title(f"【{title} データ】{title_date}")
    # 凡例の設定
    ax_legend: Legend = ax.legend(handles=[mim_patch, max_patch], fontsize=10.)
    # Patchオブジェクトのテキストラベルに日本語等倍フォントを設定する
    text: Text
    for text in ax_legend.get_texts():
        text.set_fontfamily("monospace")


def gen_plot_image(
        curr_df: DataFrame, before_df: DataFrame, phone_size: str = None) -> str:
    """
    観測データの画像を生成する
    """

    # 検索日の統計情報
    curr_stat: TempOutStat = get_temp_out_stat(curr_df)
    # 前日の統計情報
    before_stat: TempOutStat = get_temp_out_stat(before_df)

    # 端末に応じたサイズのプロット領域枠(Figure)を生成する
    fig: Figure
    if phone_size is not None and len(phone_size) > 8:
        sizes: List[str] = phone_size.split("x")
        width_pixel: int = int(sizes[0])
        height_pixel: int = int(sizes[1])
        density: float = float(sizes[2])
        # Androidスマホは pixel指定
        px: float = 1 / rcParams["figure.dpi"]  # pixel in inches
        px = px / (2.0 if density > 2.0 else density)
        fig_width_px: float = width_pixel * px
        fig_height_px: float = height_pixel * px
        fig = Figure(figsize=(fig_width_px, fig_height_px), constrained_layout=True)
    else:
        # PCブラウザはinch指定
        fig = Figure(figsize=(9.8, 6.4), constrained_layout=True)
    # ２行 (指定日データ, 前日データ) １列のサブプロット生成
    ax_temp_curr: Axes
    ax_temp_prev: Axes
    (ax_temp_curr, ax_temp_prev) = fig.subplots(nrows=2, ncols=1)

    # 1. 指定日の外気温プロット (上段)
    sub_graph(ax_temp_curr, "検索日", curr_df, curr_stat)
    # 2. 前日の外気温プロット (下段)
    sub_graph(ax_temp_prev, "前　日", before_df, before_stat)

    # 画像をバイトストリームに溜め込みそれをbase64エンコードしてレスポンスとして返す
    buf = BytesIO()
    fig.savefig(buf, format="png", bbox_inches="tight")
    data = base64.b64encode(buf.getbuffer()).decode("ascii")
    return "data:image/png;base64," + data

4-3-2. 可視化画像生成メインスクリプト

PlotTOdayImageForPhone_sqlite.py

import argparse
import os
import sqlite3

from typing import List, Optional

from pandas.core.frame import DataFrame

from plot_weather.dataloader.tempout_loader_sqlite import (
    get_dataframe
)
from plot_weather.plotter.plotterweather_sqlite import gen_plot_image
from batch_common import (
    get_connection, date_add_days, save_html
)

"""
気象センサーデータの当日データと前日データのプロット画像を取得する
スマートフォン版 ※描画領域サイズ必須
[DB] sqlite3 気象データ
"""

# スクリプト名
script_name = os.path.basename(__file__)

# 画層出力HTMLテンプレート
OUT_HTML = """
<!DOCTYPE html>
<html lang="ja">
<body>
<img src="{}"/>
</body>
</html>
"""


if __name__ == '__main__':
    parser: argparse.ArgumentParser = argparse.ArgumentParser()
    # データペースパス: ~/db/weather.db
    parser.add_argument("--db-path", type=str, required=True,
                        help="SQLite3 Database path.")
    # デバイス名: esp8266_1
    parser.add_argument("--device-name", type=str, required=True,
                        help="device name in t_device.")
    # 検索日: 2023-11-01
    parser.add_argument("--find-date", type=str, required=True,
                        help="ISO8601 format.")
    # スマートフォンの描画領域サイズ ※任意
    parser.add_argument("--phone-image-size", type=str, required=False,
                        help="スマートフォンの描画領域サイズ['幅,高さ,密度'] (例) '1064x1704x2.75'")
    args: argparse.Namespace = parser.parse_args()
    # SQLite3 気象データペースファイルパス
    db_full_path: str = os.path.expanduser(args.db_path)
    # デバイス名
    device_name: str = args.device_name
    # 検索日
    find_date: str = args.find_date
    # スマホに表示するイメージビューのサイズ
    phone_size: str = args.phone_image_size

    conn: Optional[sqlite3.Connection] = None
    try:
        conn = get_connection(db_full_path)
        # 検索日の観測データのDataFrame取得
        exclude_to_date: str = date_add_days(find_date)
        df_find: DataFrame = get_dataframe(conn, device_name, find_date, exclude_to_date)
        # 前日の観測データのDataFrame取得
        before_date: str = date_add_days(find_date, add_days=-1)
        df_before: DataFrame = get_dataframe(conn, device_name, before_date, find_date)
        print(f"df_before:\n{df_before}")
        if df_find.shape[0] > 0 and df_before.shape[0] > 0:
            # 画像取得
            html_img_src: str = gen_plot_image(df_find, df_before, phone_size=phone_size)
            # プロット結果をPNG形式でファイル保存
            script_names: List[str] = script_name.split(".")
            save_name = f"{script_names[0]}.html"
            save_path = os.path.join("output", save_name)
            print(save_path)
            html: str = OUT_HTML.format(html_img_src)
            save_html(save_path, html)
        else:
            print("該当レコードなし")
    except sqlite3.Error as db_err:
        print(f"type({type(db_err)}): {db_err}")
        exit(1)
    except Exception as exp:
        print(exp)
        exit(1)
    finally:
        if conn is not None:
            conn.close()

5. pythonスクリプト実行

python 仮想環境に入る

$ . py_venv/py_visualize/bin/activate

5-1. SQL集計モジュールを使ったメインスクリプト実行

(py_visualize) $ python TempOutStatHtml_sqliteFunc.py \
> --db-path ~/db/weather.db --device-name esp8266_1 --find-date 2024-01-03
today_min: TempOut(appear_time='2024-01-03 23:15:03', temper=-1.4), today_max: TempOut(appear_time='2024-01-03 09:56:45', temper=1.4)
before_min: TempOut(appear_time='2024-01-02 03:35:55', temper=-6.7), before_max: TempOut(appear_time='2024-01-02 14:28:25', temper=3.5)
output/TempOutStatHtml_sqliteFunc.html

出力結果はそのままブラウザで開くことができます ※2-1 のHTML

5-2. pandas集計モジュールを使ったメインスクリプト実行

(py_visualize) $ python TempOutStatHtml_pandas.py \
> --db-path ~/db/weather.db --device-name esp8266_1 --find-date 2024-01-03
today_min: TempOut(appear_time='2024-01-03 23:15', temper=-1.4), today_max: TempOut(appear_time='2024-01-03 09:56', temper=1.4)
before_min: TempOut(appear_time='2024-01-02 03:35', temper=-6.7), before_max: TempOut(appear_time='2024-01-02 14:28', temper=3.5)
output/TempOutStatHtml_pandas.html

出力結果は 5-1 と同じになります。

5-3. 可視化画像生成モジュールを使ったメインスクリプト実行

(py_visualize) $ python PlotTodayImageForPhone_sqlite.py \
> --db-path ~/db/weather.db --find-date 2024-01-03 --device-name esp8266_1 \
> --phone-image-size '1064x1704x2.75'
df_find:
       measurement_time  temp_out
0   2024-01-03 23:54:00      -1.3
1   2024-01-03 23:44:14      -1.3
2   2024-01-03 23:34:32      -1.3
3   2024-01-03 23:24:48      -1.3
4   2024-01-03 23:15:03      -1.4
..                  ...       ...
143 2024-01-03 00:41:45       1.1
144 2024-01-03 00:32:01       1.1
145 2024-01-03 00:22:17       1.1
146 2024-01-03 00:12:34       1.1
147 2024-01-03 00:02:49       1.1

[148 rows x 2 columns]
df_before:
       measurement_time  temp_out
0   2024-01-02 23:53:05       1.1
1   2024-01-02 23:43:21       1.1
2   2024-01-02 23:33:37       1.3
3   2024-01-02 23:23:55       1.4
4   2024-01-02 23:14:10       1.5
..                  ...       ...
143 2024-01-02 00:40:44      -5.7
144 2024-01-02 00:31:01      -5.8
145 2024-01-02 00:21:17      -5.9
146 2024-01-02 00:11:34      -5.9
147 2024-01-02 00:01:50      -5.9

[148 rows x 2 columns]
output/PlotTodayImageForPhone_sqlite.html

出力結果はそのままブラウザで開くことができます ※2-2 可視化画像

この記事で紹介したソースとサンプルは下記 GitHubで公開しています

GitHub(pipito-yukio) qiita-post: python/pandas/weather_stat_sql_vs_pandas

weather_stat_sql_vs_pandas/
├── db
│   ├── get_weather_window_func.sh       # 集計処理SQL実行シェルスクリプト
│   └── weather.db                       # サンプルの気象データDB
└── src
    ├── PlotTodayImageForPhone_sqlite.py #(b) 集計結果を可視化画像に埋め込むスクリプト
    ├── TempOutStatHtml_pandas.py        #(a-1) pandas集計モジュールを使用したスクリプト 
    ├── TempOutStatHtml_sqliteFunc.py    #(a-2) SQL集計モジュールを使用したスクリプト
    ├── batch_common.py                  #(a-1)と(a-2)で使用する共通モジュール 
    └── plot_weather
        ├── __init__.py
        ├── dao
        │   ├── __init__.py
        │   └── windowfunc_sqlite.py      # SQL集計モジュール
        ├── dataloader
        │   ├── __init__.py
        │   ├── tempout_loader_sqlite.py  # 検索SQLからpandas.DataFrame生成モジュール
        │   └── tempout_stat.py           # pandas集計モジュール
        └── plotter
            ├── __init__.py
            └── plotterweather_sqlite.py  # (b)の可視化画像生成モジュール