AirQualityPython/app.py - main - dplvv/Air_quality_monitoring_platform

AirQualityPython/app.py
271 строка11 KB
InventorDreamer
AirQualityPython
15 дек 2024, 15:45
15 дек 2024, 15:45d97adf0
import matplotlib
matplotlib.use('Agg')
import requests
from dash import dcc, html, Dash
from dash.dependencies import Input, Output
import dash_bootstrap_components as dbc
import matplotlib.pyplot as plt
from io import BytesIO
import base64
import logging
from datetime import datetime, timedelta

# Настройка логирования
logging.basicConfig(level=logging.DEBUG)

plt.style.use('ggplot')

API_KEY = 'f671c8937a05bca463b772d6b58d6b8a'  # Ваш API-ключ
app = Dash(__name__, external_stylesheets=[dbc.themes.LUX])
server = app.server

# --- Старый функционал: Реальные данные ---
def get_current_weather(lat, lon):
    url = f'http://api.openweathermap.org/data/2.5/weather?lat={lat}&lon={lon}&appid={API_KEY}&units=metric'
    response = requests.get(url)
    logging.debug(f"Weather API response: {response.text}")
    if response.status_code != 200:
        logging.error("Failed to fetch weather data")
        return None, None, None
    data = response.json()
    temp = data.get('main', {}).get('temp', None)
    humidity = data.get('main', {}).get('humidity', None)
    pressure = data.get('main', {}).get('pressure', None)
    return temp, humidity, pressure

def get_air_quality_data(lat, lon):
    url = f'http://api.openweathermap.org/data/2.5/air_pollution?lat={lat}&lon={lon}&appid={API_KEY}'
    response = requests.get(url)
    logging.debug(f"Air Quality API response: {response.text}")
    if response.status_code != 200 or 'list' not in response.json():
        logging.error("Failed to fetch air quality data or no data available")
        return [], []
    data = response.json()
    components = data['list'][0]['components']
    pollutants = list(components.keys())
    concentrations = list(components.values())
    return pollutants, concentrations

def create_bar_plot(pollutants, concentrations):
    fig, ax = plt.subplots(figsize=(6, 4))
    ax.bar(pollutants, concentrations, color='skyblue', edgecolor='black')
    ax.set_title("Air Pollution Components (Bar)", fontsize=14, fontweight='bold')
    ax.set_xlabel("Pollutant", fontsize=12)
    ax.set_ylabel("Concentration (µg/m³)", fontsize=12)
    plt.tight_layout()
    return fig_to_base64(fig)

def create_box_plot(pollutants, concentrations):
    fig, ax = plt.subplots(figsize=(6, 4))
    data = [[c] for c in concentrations]
    ax.boxplot(data, labels=pollutants, notch=True, patch_artist=True,
               boxprops=dict(facecolor='lightblue', color='black'),
               medianprops=dict(color='black', linewidth=1.5))
    ax.set_title("Pollutant Concentration (Boxplot)", fontsize=14, fontweight='bold')
    ax.set_xlabel("Pollutant", fontsize=12)
    ax.set_ylabel("Concentration (µg/m³)", fontsize=12)
    plt.tight_layout()
    return fig_to_base64(fig)

# --- Новый функционал: Исторические данные ---
def get_historical_data(lat, lon, start, end):
    url = f"http://history.openweathermap.org/data/2.5/history/city?lat={lat}&lon={lon}&type=hour&start={start}&end={end}&appid={API_KEY}&units=metric"
    response = requests.get(url)
    logging.debug(f"Historical API response: {response.text}")
    if response.status_code != 200 or 'list' not in response.json():
        logging.error("Failed to fetch historical data or no data available")
        return []
    return response.json()['list']

def create_histogram(data, parameter, title):
    timestamps = [datetime.utcfromtimestamp(entry['dt']).strftime('%Y-%m-%d %H:%M:%S') for entry in data]
    values = [entry['main'][parameter] for entry in data]
    fig, ax = plt.subplots(figsize=(10, 6))
    ax.bar(timestamps, values, color='skyblue', edgecolor='black')
    ax.set_title(title, fontsize=14, fontweight='bold')
    ax.set_xlabel("Time", fontsize=12)
    ax.set_ylabel(f"{parameter.capitalize()}", fontsize=12)
    ax.tick_params(axis='x', rotation=45)
    plt.tight_layout()
    return fig_to_base64(fig)

# --- Общие вспомогательные функции ---
def fig_to_base64(fig):
    img_bytes = BytesIO()
    plt.savefig(img_bytes, format='png', dpi=100)
    img_bytes.seek(0)
    img_base64 = base64.b64encode(img_bytes.getvalue()).decode('utf-8')
    plt.close(fig)
    return f"data:image/png;base64,{img_base64}"

# --- Макет приложения ---
navbar = dbc.Navbar(
    dbc.Container([
        dbc.NavbarBrand("Weather & Air Quality Dashboard", className="ms-2", style={"fontWeight": "bold", "fontSize": "1.5em"}),
    ]),
    color="dark",
    dark=True,
    className="mb-4"
)

app.layout = html.Div([
    navbar,
    dbc.Container([
        html.H1("Weather Monitoring", className="mb-4", style={"textAlign": "center"}),

        # Ввод данных для реального времени
        html.H3("Current Weather and Air Quality", className="mb-3"),
        dbc.Row([
            dbc.Col([
                html.Label("Enter Latitude:"),
                dcc.Input(id='latitude', type='number', placeholder='Enter Latitude', value=37.0902, style={'margin': '10px'}),
                html.Label("Enter Longitude:"),
                dcc.Input(id='longitude', type='number', placeholder='Enter Longitude', value=-95.7129, style={'margin': '10px'}),
            ], width=6)
        ], className="mb-4"),
        dbc.Row([
            dbc.Col(
                dbc.Card(
                    [
                        dbc.CardBody([
                            html.H5("Temperature", className="card-title"),
                            html.H2(id="temp-value", className="card-text"),
                            html.P("°C", className="text-muted")
                        ])
                    ], color="primary", inverse=True
                ), width=4
            ),
            dbc.Col(
                dbc.Card(
                    [
                        dbc.CardBody([
                            html.H5("Humidity", className="card-title"),
                            html.H2(id="humidity-value", className="card-text"),
                            html.P("%", className="text-muted")
                        ])
                    ], color="info", inverse=True
                ), width=4
            ),
            dbc.Col(
                dbc.Card(
                    [
                        dbc.CardBody([
                            html.H5("Pressure", className="card-title"),
                            html.H2(id="pressure-value", className="card-text"),
                            html.P("hPa", className="text-muted")
                        ])
                    ], color="warning", inverse=True
                ), width=4
            ),
        ], className="mb-4"),
        dbc.Row([
            dbc.Col([
                html.H4("Air Pollution Levels (Bar Chart)", className="mb-3", style={"textAlign": "center"}),
                html.Div(id='pollution-graph-bar', style={"textAlign": "center"})
            ], width=6),
            dbc.Col([
                html.H4("Pollutant Distribution (Boxplot)", className="mb-3", style={"textAlign": "center"}),
                html.Div(id='pollution-graph-box', style={"textAlign": "center"})
            ], width=6),
        ], className="mb-4"),

        # Ввод данных для исторических данных
        html.H3("Historical Weather Data", className="mb-3"),
        dbc.Row([
            dbc.Col([
                html.Label("Select Parameter:"),
                dcc.Dropdown(
                    id='parameter-dropdown',
                    options=[
                        {'label': 'Temperature', 'value': 'temp'},
                        {'label': 'Humidity', 'value': 'humidity'},
                        {'label': 'Pressure', 'value': 'pressure'}
                    ],
                    value='temp',
                    style={'margin': '10px'}
                ),
                html.Label("Select Date Range (Last 5 Days Max):"),
                dcc.DatePickerRange(
                    id='date-picker-range',
                    start_date=(datetime.utcnow() - timedelta(days=5)).strftime('%Y-%m-%d'),
                    end_date=datetime.utcnow().strftime('%Y-%m-%d'),
                    display_format='YYYY-MM-DD',
                    style={'margin': '10px'}
                ),
            ], width=6)
        ], className="mb-4"),
        dbc.Row([
            dbc.Col([
                html.H4("Historical Data Graph", className="mb-3", style={"textAlign": "center"}),
                html.Div(id='historical-graph', style={"textAlign": "center"})
            ], width=12),
        ], className="mb-4"),
    ])
])

# --- Колбэк для реального времени ---
@app.callback(
    [Output('pollution-graph-bar', 'children'),
     Output('pollution-graph-box', 'children'),
     Output('temp-value', 'children'),
     Output('humidity-value', 'children'),
     Output('pressure-value', 'children')],
    [Input('latitude', 'value'), Input('longitude', 'value')]
)
def update_realtime_dashboard(lat, lon):
    try:
        if lat is None or lon is None:
            return "No data", "No data", "N/A", "N/A", "N/A"

        temp, humidity, pressure = get_current_weather(lat, lon)
        if temp is None or humidity is None or pressure is None:
            return "Weather data unavailable", "Weather data unavailable", "N/A", "N/A", "N/A"

        pollutants, concentrations = get_air_quality_data(lat, lon)
        if not pollutants or not concentrations:
            return "Pollution data unavailable", "Pollution data unavailable", temp, humidity, pressure

        bar_img = html.Img(src=create_bar_plot(pollutants, concentrations), style={'width': '90%'})
        box_img = html.Img(src=create_box_plot(pollutants, concentrations), style={'width': '90%'})

        return bar_img, box_img, temp, humidity, pressure
    except Exception as e:
        logging.error(f"Error in real-time callback: {str(e)}")
        return "Error", "Error", "Error", "Error", "Error"

# --- Колбэк для исторических данных ---
@app.callback(
    Output('historical-graph', 'children'),
    [Input('latitude', 'value'),
     Input('longitude', 'value'),
     Input('parameter-dropdown', 'value'),
     Input('date-picker-range', 'start_date'),
     Input('date-picker-range', 'end_date')]
)
def update_historical_graph(lat, lon, parameter, start_date, end_date):
    try:
        if lat is None or lon is None or parameter is None or start_date is None or end_date is None:
            return "No data available"

        start_unix = int(datetime.strptime(start_date, '%Y-%m-%d').timestamp())
        end_unix = int(datetime.strptime(end_date, '%Y-%m-%d').timestamp())

        data = get_historical_data(lat, lon, start_unix, end_unix)
        if not data:
            return "No historical data available for the selected range."

        title_map = {
            'temp': 'Temperature Over Time (°C)',
            'humidity': 'Humidity Over Time (%)',
            'pressure': 'Pressure Over Time (hPa)'
        }
        graph_title = title_map.get(parameter, 'Historical Data')

        hist_img = html.Img(src=create_histogram(data, parameter, graph_title), style={'width': '100%'})
        return hist_img
    except Exception as e:
        logging.error(f"Error in historical callback: {str(e)}")
        return "Error generating graph."

if __name__ == '__main__':
    app.run_server(debug=True)