gitea/th-air-quality-etl-ml
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ปรับตารางให้แสดงข้อมูลเวลาทำนายได้ สำหรับแสดงผลใน Dashboard

Browse Source
This commit is contained in:
Flook 2025-05-12 09:46:18 +07:00
parent 0af387adcc
commit d90b233bbc
2 changed files with 31 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
clickhouse/air_quality_forecast.py
View File

@ -4,7 +4,7 @@ import os
load_dotenv() load_dotenv()
# ----- สร้าง Table ถ้ายังไม่มี ----- # ----- สร้าง Table ใหม่ โดยลบของเดิมถ้ามี -----
def air_quality_forecast(): def air_quality_forecast():
try: try:
# อ่านค่าจาก .env # อ่านค่าจาก .env
@ -14,34 +14,40 @@ def air_quality_forecast():
# สร้าง client ด้วย clickhouse-connect # สร้าง client ด้วย clickhouse-connect
client = clickhouse_connect.get_client( client = clickhouse_connect.get_client(
host=ch_host, # ระบุ Host โดยไม่ต้องใช้ https:// อีก host=ch_host,
port=443, # ใช้พอร์ต HTTPS 443 port=443,
username=ch_user, # ชื่อผู้ใช้ username=ch_user,
password=ch_password, # รหัสผ่าน password=ch_password,
secure=True # ใช้การเชื่อมต่อที่ปลอดภัย (HTTPS) secure=True
) )
# สร้าง Table ด้วยคำสั่ง SQL # 🔥 ลบตารางเดิม (ถ้ามี) ก่อนสร้างใหม่
drop_table_sql = "DROP TABLE IF EXISTS air_quality_forecast"
client.command(drop_table_sql)
print("🗑️ Table 'air_quality_forecast' dropped (if existed).")
# ✅ สร้าง Table ใหม่ที่ใช้ DateTime
create_table_sql = """ create_table_sql = """
CREATE TABLE IF NOT EXISTS air_quality_forecast ( CREATE TABLE air_quality_forecast (
predicted_for_date Date, -- วันที่ที่ทำการทำนาย predicted_for_date DateTime, -- วันที่-เวลาที่ทำการทำนาย
predicted_aqi Int32, -- ค่า AQI ที่ทำนายได้ predicted_aqi Int32, -- ค่า AQI ที่ทำนายได้
aqi_explain String, -- คำอธิบายหรือเหตุผลจากโมเดล aqi_explain String, -- คำอธิบายหรือเหตุผลจากโมเดล
predicted_at DateTime DEFAULT now() -- วันที่-เวลาที่สร้างบันทึกนี้ predicted_at DateTime DEFAULT now() -- วันที่-เวลาที่สร้างบันทึกนี้
) ENGINE = MergeTree() ) ENGINE = MergeTree()
ORDER BY predicted_for_date; ORDER BY predicted_for_date; \
""" """
client.command(create_table_sql) client.command(create_table_sql)
print("✅ Table 'air_quality_forecast' created successfully.")
verify_air_quality_forecast(client) verify_air_quality_forecast(client)
except Exception as e: except Exception as e:
print("❌ ClickHouse Error:", e) print("❌ ClickHouse Error:", e)
# ----- ตรวจสอบการสร้าง Table -----
def verify_air_quality_forecast(client): def verify_air_quality_forecast(client):
try: try:
verify_table_sql = """ verify_table_sql = "SHOW TABLES LIKE 'air_quality_forecast'"
SHOW TABLES LIKE 'air_quality_forecast'
"""
result = client.command(verify_table_sql) result = client.command(verify_table_sql)
if result.strip(): if result.strip():

15
pipelines/predict_and_store.py
View File

@ -4,7 +4,6 @@ import json
from datetime import datetime, timedelta from datetime import datetime, timedelta
from clickhouse_connect import get_client from clickhouse_connect import get_client
from dotenv import load_dotenv from dotenv import load_dotenv
from requests.auth import HTTPBasicAuth
load_dotenv() load_dotenv()
@ -90,23 +89,23 @@ def predict_aqi(latest_record):
return None return None
# บันทึกผลทำนายลง ClickHouse # บันทึกผลทำนายลง ClickHouse
def store_prediction(client, predicted_data, target_date): def store_prediction(client, predicted_data, target_datetime):
aqi = predicted_data.get('aqi') aqi = predicted_data.get('aqi')
aqi_explain = predicted_data.get('aqi_explain') aqi_explain = predicted_data.get('aqi_explain')
if isinstance(aqi_explain, dict): if isinstance(aqi_explain, dict):
aqi_explain_json = json.dumps(aqi_explain) aqi_explain_json = json.dumps(aqi_explain)
else: else:
aqi_explain_json = aqi_explain # assume already JSON string aqi_explain_json = aqi_explain
insert_query = """ insert_query = """
INSERT INTO air_quality_forecast (predicted_for_date, predicted_aqi, aqi_explain) INSERT INTO air_quality_forecast (predicted_for_date, predicted_aqi, aqi_explain)
VALUES \ VALUES
""" """
client.command( client.command(
insert_query + f"('{target_date}', {aqi}, '{aqi_explain_json}')" insert_query + f"('{target_datetime.isoformat()}', {aqi}, '{aqi_explain_json}')"
) )
print(f"✅ Prediction for {target_date} saved to ClickHouse.") print(f"✅ Prediction for {target_datetime} saved to ClickHouse.")
def run_prediction_pipeline(): def run_prediction_pipeline():
client = get_clickhouse_client() client = get_clickhouse_client()
@ -120,8 +119,8 @@ def run_prediction_pipeline():
record_time = latest_record['record_time'] record_time = latest_record['record_time']
if isinstance(record_time, str): if isinstance(record_time, str):
record_time = datetime.fromisoformat(record_time) record_time = datetime.fromisoformat(record_time)
target_date = (record_time + timedelta(days=1)).date() target_datetime = record_time + timedelta(days=1)
store_prediction(client, prediction, target_date) store_prediction(client, prediction, target_datetime)
if __name__ == "__main__": if __name__ == "__main__":
run_prediction_pipeline() run_prediction_pipeline()