太陽フレア情報を画像で配信するWebサーバー

app.py

""" 太陽フレア情報を画像で配信するWebサーバー """
from datetime import datetime, timedelta
from io import BytesIO
from pathlib import Path
import requests
from PIL import Image, ImageDraw, ImageFont
from flask import Flask, send_file, redirect

# NASA API キー (以下を書き換えてください) --- (*1)
NASA_API_KEY = "DEMO_KEY"
# IPAフォントのURLと保存パス
FONT_PATH = Path(__file__).parent / "ipag.ttf"
# Flaskアプリケーションの初期化 --- (*2)
app = Flask(__name__)

def get_solar_flare_data():
    """NASA DONKI APIから太陽フレア情報を取得"""  # --- (*3)
    # 昨日から今日までの日付範囲を設定
    today = datetime.now()
    yesterday = today - timedelta(days=1)
    start_date = yesterday.strftime("%Y-%m-%d")
    end_date = today.strftime("%Y-%m-%d")
    url = f"https://api.nasa.gov/DONKI/FLR?startDate={start_date}&endDate={end_date}&api_key={NASA_API_KEY}"
    try:
        response = requests.get(url, timeout=10)
        response.raise_for_status()
        data = response.json()
        return data if isinstance(data, list) else []
    except Exception as e:
        print(f"API Error: {e}")
        return []

def get_solar_image():
    """SOHOから現在の太陽画像を取得"""  # --- (*4)
    url = "https://soho.nascom.nasa.gov/data/realtime/eit_284/512/latest.jpg"
    try:
        response = requests.get(url, timeout=10)
        response.raise_for_status()
        img = Image.open(BytesIO(response.content))
        img = img.resize((400, 400), Image.Resampling.LANCZOS)  # リサイズ
        return img
    except Exception as e:
        print(f"Image Error: {e}")
        return Image.new('RGB', (400, 400), color=(0, 0, 0))  # 取得失敗時は黒い画像を返す

def analyze_flare_data(flare_data):
    """フレアデータを分析"""
    class_count = {'C': 0, 'M': 0, 'X': 0, 'Other': 0}
    for flare in flare_data:
        cls = flare.get('classType', 'Other')
        key = cls[0] if cls and cls[0] in 'CMX' else 'Other'
        class_count[key] += 1
    return class_count

def create_flare_info_image(flare_data, width=540, height=540):
    """太陽フレアの情報を表示する画像を作成"""  # --- (*5)
    # フレア情報を取得
    total_flares = len(flare_data)
    class_count = analyze_flare_data(flare_data)
    # 空の画像を作成
    img = Image.new('RGB', (width, height), color=(0, 0, 0))
    draw = ImageDraw.Draw(img)  # 描画用オブジェクトを取得
    font = ImageFont.truetype(str(FONT_PATH), 18)  # フォントサイズ
    text_color = (255, 255, 255)  # テキストの色
    # タイトル部分を描画
    draw.rectangle([(0, 0), (width, 50)], fill=(50, 50, 50))
    y_pos = 15
    draw.text((140, y_pos), "■ 太陽フレアの情報 ■", fill=text_color, font=font)
    y_pos += 70
    # テキスト情報を表示
    draw.text((20, y_pos), f"● 発生件数: {total_flares}", fill=text_color, font=font)
    y_pos += 40
    draw.text((20, y_pos), "● クラス毎の件数:", fill=text_color, font=font)
    y_pos += 40
    # 棒グラフを描画
    bar_colors = [(0, 255, 0), (255, 165, 0), (255, 0, 0)]  # C:緑, M:オレンジ, X:赤
    class_info = ["[C] 小規模", "[M] 中規模", "[X] 大規模"]  # クラスの説明
    bar_height = 20
    bar_width_unit = 500/ 10  # 最大10件まで表示可能と仮定
    # C/M/Xの順番でグラフを描画 --- (*6)
    for i, class_name in enumerate(['C', 'M', 'X']):
        graph_y = y_pos + 30
        # グラフのラベルを描画
        draw.text((20, y_pos), f"  {class_info[i]}:", fill=text_color, font=font)
        # 棒グラフの幅を計算
        v = class_count[class_name]
        bar_width = v * bar_width_unit
        draw.rectangle([(40, graph_y), (50 + bar_width, graph_y + bar_height)], fill=bar_colors[i])
        draw.text((50 + bar_width + 10, graph_y + 2), f"{v}", fill=text_color, font=font)
        y_pos += 70
    draw.text((20, y_pos), "● 詳細:", fill=text_color, font=font)
    y_pos += 40
    # 発生時刻リストを最大3件表示
    for flare in flare_data[:3]:
        begin_time = flare.get('beginTime', 'N/A').replace('T', ' ').replace('Z', '')
        classification = flare.get('classType', 'N/A')
        info_text = f"  - {begin_time} → {classification}"
        draw.text((20, y_pos), info_text, fill=text_color, font=font)
        y_pos += 40
    return img

def create_combined_image(solar_image, flare_info_image):
    """太陽画像とフレア情報を結合した960x540の画像を作成""" # --- (*7)
    # 最終画像を作成（黒い背景）
    final_img = Image.new('RGB', (960, 540), color=(0, 0, 0))
    # 左側に太陽画像を配置（上下中央揃え）
    y_offset = (540 - 400) // 2
    final_img.paste(solar_image, (20, y_offset))    
    # 右側にフレア情報を配置
    final_img.paste(flare_info_image, (460, 0))
    return final_img

@app.route('/image.jpg')
def get_image():
    """太陽フレア情報と画像を結合した画像を返す"""  # --- (*8)
    try:
        # データを取得して画像を生成
        flare_data, solar_image = [get_solar_flare_data(), get_solar_image()]
        flare_info_img = create_flare_info_image(flare_data, width=500, height=540)
        combined_image = create_combined_image(solar_image, flare_info_img)
        # メモリにJPEGとして保存して出力
        img_io = BytesIO()
        combined_image.save(img_io, 'JPEG', quality=85)
        img_io.seek(0)
        return send_file(img_io, mimetype='image/jpeg')
    except Exception as e:
        print(f"Error: {e}")
        return "Error generating image", 500

@app.route('/')
def root():
    return redirect('/image.jpg')

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5555, debug=True)  # --- (*9)

UIFlow2に書き込むMicroPythonのプログラム:

import os, sys, io
import M5
from M5 import *
from image_plus import ImagePlus

image_plus0 = None

def setup():  # 初期化処理 --- (*1)
  global image_plus0
  Widgets.setRotation(1)
  M5.begin()
  Widgets.fillScreen(0xeeeeee)
  # 定期的にWebサーバーから画像を取得する「画像+」を利用 --- (*2)
  image_plus0 = ImagePlus("http://192.168.1.14:5555/image.jpg", 0, 0, True, 60000, default_img="/flash/res/img/default.png")
  Power.setLed(255)

def loop():  # メインループ
  global image_plus0
  M5.update()

if __name__ == '__main__':
  try:
    setup()  # 初期化
    while True:  # メインループ
      loop()
  except (Exception, KeyboardInterrupt) as e:
    try:
      image_plus0.deinit()
      from utility import print_error_msg
      print_error_msg(e)
    except ImportError:
      print("please update to latest firmware")

ローカルPCのサーバーに接続できない場合は以下のテストサーバーのものをご利用ください

UIFlow2に書き込むMicroPythonのプログラム(テストバージョン):

import os, sys, io
import M5
from M5 import *
from image_plus import ImagePlus

image_plus0 = None

def setup():  # 初期化処理 --- (*1)
  global image_plus0
  Widgets.setRotation(1)
  M5.begin()
  Widgets.fillScreen(0xeeeeee)
  # 定期的にWebサーバーから画像を取得する「画像+」を利用 --- (*2)
  image_plus0 = ImagePlus("https://api.aoikujira.com/test/mynavi-py-flare/image.jpg", 0, 0, True, 60000, default_img="/flash/res/img/default.png")
  Power.setLed(255)

def loop():  # メインループ
  global image_plus0
  M5.update()

if __name__ == '__main__':
  try:
    setup()  # 初期化
    while True:  # メインループ
      loop()
  except (Exception, KeyboardInterrupt) as e:
    try:
      image_plus0.deinit()
      from utility import print_error_msg
      print_error_msg(e)
    except ImportError:
      print("please update to latest firmware")