気圧と健康の気象病予報士@東京

OpenWeatherMapのデータと生成AIを用いて記事を作成しています

現在運用中のスクリプト -ver.0.03-20240606

Python スクリプトのみ。件名(カテゴリー情報を付加)、ログファイルの圧縮処理などの、いくつかの変更があります。まずは、Mermaid のフローチャートから紹介。

フローチャート作成

import os

def create_mermaid_chart(mermaid_code, output_file, width=800, height=800):
    # 保存するMermaidコードをファイルに書き込む
    with open("chart.mmd", "w") as f:
        f.write(mermaid_code)
    # mermaid-cliを使用して画像ファイルに変換する
    result = os.system(f"mmdc -i chart.mmd -o {output_file} -w {width} -H {height}")
    if result == 0:
        print(f"Chart created successfully: {output_file}")
    else:
        print("Failed to create chart")

mermaid_code = """
graph TD
    A[Start] --> B[Load Environment Variables]
    B --> C[Setup Logging]
    C --> D[Decompress Log File]
    D --> E[Configure Logging]
    E --> F[Fetch Pressure Data]
    F -->|Retry if fails| F
    F --> G{If Data Fetched}
    G -->|Yes| H[Extract Timestamps and Pressures]
    H --> I[Highlight Extreme Downward Trend]
    I --> J[Save Graph as file_name]
    J --> K[Analyze Pressure Data Using OpenAI API]
    K --> L[Generate Comment]
    L --> M[Send Email with Analysis and Graph]
    M --> N[Compress Log File]
    N --> O[End]
    G -->|No| O

    style A fill:#f9f,stroke:#333,stroke-width:2px
    style B fill:#bbf,stroke:#333,stroke-width:2px
    style C fill:#bbf,stroke:#333,stroke-width:2px
    style D fill:#bbf,stroke:#333,stroke-width:2px
    style E fill:#bbf,stroke:#333,stroke-width:2px
    style F fill:#bbf,stroke:#333,stroke-width:2px
    style G fill:#ff9,stroke:#333,stroke-width:2px
    style H fill:#bbf,stroke:#333,stroke-width:2px
    style I fill:#bbf,stroke:#333,stroke-width:2px
    style J fill:#bbf,stroke:#333,stroke-width:2px
    style K fill:#bbf,stroke:#333,stroke-width:2px
    style L fill:#bbf,stroke:#333,stroke-width:2px
    style M fill:#bbf,stroke:#333,stroke-width:2px
    style N fill:#bbf,stroke:#333,stroke-width:2px
    style O fill:#f9f,stroke:#333,stroke-width:2px

    linkStyle 0 stroke:#f66,stroke-width:2px
    linkStyle 1 stroke:#f66,stroke-width:2px
    linkStyle 2 stroke:#f66,stroke-width:2px
    linkStyle 3 stroke:#f66,stroke-width:2px
    linkStyle 4 stroke:#f66,stroke-width:2px
    linkStyle 5 stroke:#f66,stroke-width:2px
    linkStyle 6 stroke:#f66,stroke-width:2px
    linkStyle 7 stroke:#f66,stroke-width:2px
    linkStyle 8 stroke:#f66,stroke-width:2px
    linkStyle 9 stroke:#f66,stroke-width:2px
    linkStyle 10 stroke:#f66,stroke-width:2px
    linkStyle 11 stroke:#f66,stroke-width:2px
    linkStyle 12 stroke:#f66,stroke-width:2px
    linkStyle 13 stroke:#f66,stroke-width:2px
    linkStyle 14 stroke:#f66,stroke-width:2px
"""

output_file = "chart.png"
create_mermaid_chart(mermaid_code, output_file, width=1000, height=1000)

最近、GPT4o にフローチャートの作成を依頼すると、わけのわからない図が出力されます。なんななんだろう。webp 形式のファイルなので、png に変換する ffmpeg コマンドも。

$ ffmpeg -i input.webp output.png

どこの言語?

ソースコード

import os
import time
import gzip
import shutil
import logging
from datetime import datetime, timedelta, timezone
import smtplib
from email.mime.image import MIMEImage
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText

import requests
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
from dotenv import load_dotenv
import openai


# 環境変数の読み込み
load_dotenv()

# 環境変数から設定を取得
log_dir = os.getenv('LOG_DIR')
api_key = os.getenv('OPENWEATHERMAP_API_KEY')
openai_api_key = os.getenv('OPENAI_API_KEY')
smtp_server = os.getenv('SMTP_SERVER')
smtp_port = int(os.getenv('SMTP_PORT'))
from_email = os.getenv('FROM_EMAIL')
to_addr = os.getenv('TO_ADDR')
email_password = os.getenv('EMAIL_PASSWORD')

# 日付ごとにログファイルを生成
current_date = datetime.now().strftime("%Y-%m-%d")
log_file_path = os.path.join(log_dir, f"{current_date}.log")
compressed_log_file_path = f"{log_file_path}.gz"

def decompress_log_file():
    if os.path.exists(compressed_log_file_path):
        try:
            with gzip.open(compressed_log_file_path, 'rb') as f_in:
                with open(log_file_path, 'wb') as f_out:
                    shutil.copyfileobj(f_in, f_out)
        except gzip.BadGzipFile:
            shutil.copy(compressed_log_file_path, log_file_path)
    else:
        open(log_file_path, 'w').close()

def compress_log_file():
    try:
        with open(log_file_path, 'rb') as f_in:
            with gzip.open(compressed_log_file_path, 'wb') as f_out:  # 修正箇所
                shutil.copyfileobj(f_in, f_out)
        os.remove(log_file_path)
        logging.debug("Log file compressed successfully")
    except Exception as e:
        logging.error(f"Error compressing log file: {e}")

# gzファイルの展開または新規作成
decompress_log_file()

# ログの設定
logging.basicConfig(filename=log_file_path, level=logging.DEBUG,
                    format='%(asctime)s %(levelname)s %(message)s')

logging.debug("Starting script")

if log_file_path:
    # ログの記録
    with open(log_file_path, "a") as log:
        log.write(f"Script executed at {datetime.now()}\n")
else:
    logging.error("LOG_FILE_PATH environment variable not set.")
    print("LOG_FILE_PATH environment variable not set.")

# 定数
CITY_NAME = 'Shibuya,JP'
API_URL = f'http://api.openweathermap.org/data/2.5/forecast?q={CITY_NAME}&appid={api_key}&units=metric'
THRESHOLD = -0.1  # 極端な下降トレンドとみなす閾値(調整可能)
MIN_INTERVAL = 2 * 3600  # 強調表示する最小間隔(2時間)

# OpenAI APIキーの設定
openai.api_key = openai_api_key

logging.debug("Environment variables loaded")

# 気圧データ取得関数
def get_pressure_data():
    logging.debug("Starting to fetch pressure data")
    max_retries = 3
    retries = 0
    while retries < max_retries:
        try:
            response = requests.get(API_URL)
            response.raise_for_status()
            data = response.json()
            logging.debug("Pressure data fetched successfully")
            return data
        except requests.RequestException as e:
            retries += 1
            logging.error(f"Error fetching data: {e}, retrying ({retries}/{max_retries})...")
            time.sleep(5)  # 5秒待ってからリトライ
    return None

# 気圧データの解析関数 (OpenAI API)
def analyze_pressure_data(timestamps, pressures):
    logging.debug("Starting to analyze pressure data")
    try:
        # 気圧データを簡略化してテキスト形式で整形
        data_text = "\n".join([f"{t.strftime('%Y-%m-%d %H:%M')}: {p}" for t, p in zip(timestamps, pressures)])

        # プロンプトに気圧データを説明する
        prompt = (
            "以下は東京渋谷の気圧データです。このデータに基づいて、気圧と気象病についてコメントを生成してください。コメントの最初に結論をやや長めに述べてください。その後、コメントは短いパラグラフを用い、見出しや箇条書き、番号付きリストなどを活用して読みやすくしてください。\n\n"
            + data_text
        )

        # テキストプロンプトに基づいた解析を行う
        response = openai.ChatCompletion.create(
            model="gpt-4o",  # 使用するモデル
            messages=[
                {"role": "system", "content": "You are an assistant that analyzes pressure data trends."},
                {"role": "user", "content": prompt}
            ],
            max_tokens=1000  # 生成されるコメントの最大トークン数
        )

        # 分析結果のコメントを取得
        comment = response['choices'][0]['message']['content'].strip()
        logging.debug("Comment from OpenAI: %s", comment)
        return format_comment(comment)

    except Exception as e:
        logging.error(f"Error analyzing pressure data: {e}")
        return None

# コメントの形式を整える関数
def format_comment(comment):
    return f"{comment}"

# 気圧グラフ作成関数 (極端な下降トレンドをハイライト)
def highlight_extreme_downward_trend(timestamps, pressures, threshold, min_interval):
    logging.debug("Starting to create pressure graph")
    plt.figure(figsize=(12, 6))
    plt.plot(timestamps, pressures, marker='o', linestyle='-')

    highlight_start = None
    for i in range(len(pressures) - 1):
        rate_of_change = (pressures[i + 1] - pressures[i]) / ((timestamps[i + 1] - timestamps[i]).total_seconds() / 3600)
        if rate_of_change < threshold:
            if highlight_start is None:
                highlight_start = i
        else:
            if highlight_start is not None:
                interval = (timestamps[i] - timestamps[highlight_start]).total_seconds()
                if interval >= min_interval:
                    alpha = max(0.1, min(1.0, abs(rate_of_change)/2))
                    color = (1.0, 0.0, 0.0, alpha)

                    plt.axvspan(timestamps[highlight_start], timestamps[i], color=color)
                highlight_start = None

    plt.title('Pressure Data in Shibuya, Tokyo')
    plt.xlabel('Date')
    plt.ylabel('Pressure (hPa)')
    plt.xticks(rotation=45)

    # 横軸の日付フォーマットを変更
    ax = plt.gca()
    ax.xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m-%d'))
    ax.xaxis.set_major_locator(mdates.DayLocator())
    ax.xaxis.set_minor_locator(mdates.HourLocator(interval=6))

    plt.grid(True)
    plt.tight_layout()

    # ファイル名に日付と時刻を含めて保存
    output_dir = os.getenv('OUTPUT_DIR', os.path.join(os.path.dirname(__file__), "png"))
    os.makedirs(output_dir, exist_ok=True)

    current_datetime = datetime.now().strftime("%Y%m%d_%H%M%S")
    file_name = os.path.join(output_dir, f"{current_datetime}_fig.png")
    plt.savefig(file_name)
    logging.debug("Pressure graph saved as %s", file_name)
    return file_name

# メール送信関数
def send_email(file_path, comment):
    logging.debug("Starting to send email")
    current_time = datetime.now().strftime("%I %p %d %b")

    msg = MIMEMultipart()
    msg['Subject'] = f'[meteoropathy]Tokyo Pressure - {current_time}'
    msg['From'] = from_email
    msg['To'] = to_addr

    # メール本文をMarkdown形式で作成
    if comment:
        email_body = f"""
**Current time:** {current_time}

{comment}

        """
        msg.attach(MIMEText(email_body, 'plain'))  # Markdown形式で送信
    else:
        email_body = f"""
**Current time:** {current_time}

No analysis comment available.

        """
        msg.attach(MIMEText(email_body, 'plain'))  # Markdown形式で送信

    with open(file_path, 'rb') as f:
        img_data = f.read()
        image = MIMEImage(img_data, name=os.path.basename(file_path))
        msg.attach(image)

    try:
        with smtplib.SMTP(smtp_server, smtp_port) as server:
            server.starttls()
            server.login(from_email, email_password)
            server.send_message(msg)
            logging.debug("Email sent successfully")
    except smtplib.SMTPException as e:
        logging.error(f"Error sending email: {e}")

# メイン関数
def main():
    logging.debug("Main function started")
    try:
        data = get_pressure_data()
        if data:
            timestamps, pressures = [], []
            jst = timezone(timedelta(hours=9), 'JST')
            current_time = datetime.now(jst)
            start_time = current_time - timedelta(hours=24)
            for item in data.get('list', []):
                timestamp = datetime.utcfromtimestamp(item.get('dt', 0)).replace(tzinfo=timezone.utc).astimezone(jst)
                if timestamp >= start_time:
                    timestamps.append(timestamp)
                    pressures.append(item.get('main', {}).get('pressure', 0))

            # 画像データのBase64エンコード
            file_name = highlight_extreme_downward_trend(timestamps, pressures, THRESHOLD, MIN_INTERVAL)

            # 気圧データの解析コメント取得
            comment = analyze_pressure_data(timestamps, pressures)

            # メール送信
            send_email(file_name, comment)
        else:
            logging.error("Failed to fetch pressure data after retries.")
    except Exception as e:
        logging.error(f"Error in main function: {e}")

    # ログファイルをgzipで再圧縮
    compress_log_file()

if __name__ == "__main__":
    main()

現在の残高