preprocessing_functions.py

import numpy as np
import tensorflow as tf
import streamlit as st
import os
import requests
import matplotlib.pyplot as plt
import joblib
from pathlib import Path

MODEL_PATH = "output_predictions_to_meteo_smape_25.keras"
MODEL_PATH_SOLARMON = "model_24h_base_on_Me_open_meteo_weather_predictions.keras"

MODEL_PATH_SOLARMON_HIST = "model_hist/model_24h_predictions_hist.keras"

TRANSFORMER_INPUT_PATH_HIST = "model_hist/input_preprocessor.pkl"
TRANSFORMER_OUTPUT_PATH_HIST = "model_hist/output_preprocessor.pkl"

@st.cache_resource
def load_model(): 
    model = tf.keras.models.load_model(MODEL_PATH)
    return model

@st.cache_resource
def load_model_solarmon(): 
   # model = tf.keras.models.load_model(MODEL_PATH_SOLARMON)
    model = tf.keras.models.load_model(MODEL_PATH_SOLARMON_HIST)
    return model

@st.cache_resource 
def load_transformers():
    input_preprocessor = joblib.load('input_preprocessor_meteo_to_smape25.pkl')
    output_scaler = joblib.load('output_scaler_meteo_to_smape25.pkl')

    return input_preprocessor, output_scaler

@st.cache_resource 
def load_transformers_solarmon():
    #input_preprocessor = joblib.load('input_preprocessor.pkl')
    #output_scaler = joblib.load('output_preprocessor.pkl')
    
    input_preprocessor = joblib.load(TRANSFORMER_INPUT_PATH_HIST)
    output_scaler = joblib.load(TRANSFORMER_OUTPUT_PATH_HIST)    
    return input_preprocessor, output_scaler

import numpy as np

def create_sequences_solarmon(data, window, horizon, past_features, future_features):
    """
    Vytvoří sekvence vstupních dat a odpovídající cílové hodnoty pro trénování LSTM modelu.

    Parametry:
    ----------
    data : pandas.DataFrame
        DataFrame obsahující časové řady.
    window : int
        Počet časových kroků v minulosti.
    horizon : int
        Počet časových kroků do budoucnosti.
    past_features : list
        Seznam sloupců, které budou použity jako vstupní vlastnosti v minulosti.
    future_features : list
        Seznam sloupců, které budou použity jako vstupní vlastnosti v budoucnosti.
    target : str
        Název sloupce, který bude použit jako cílová hodnota.

    Návratové hodnoty:
    -------------------
    X : numpy.ndarray
        Pole tvaru (vzorky, window, past_features + future_features), obsahující sekvence vstupních dat.
    y : numpy.ndarray
        Pole tvaru (vzorky, horizon), obsahující odpovídající cílové hodnoty.
    """

    # Vytvoření historických sekvencí (past_features)
    X_past = np.lib.stride_tricks.sliding_window_view(
        data[past_features].values, (window, len(past_features))
    )[:-horizon, :, :]

    X_past = np.squeeze(X_past, axis=1)  # Výstup: (vzorky, window, len(past_features))

    # Vytvoření budoucích sekvencí (future_features)
    X_future = np.lib.stride_tricks.sliding_window_view(
        data[future_features].values, (window, len(future_features))
    )[horizon-1 : len(X_past) + horizon-1, :, :]

    X_future = np.squeeze(X_future, axis=1)  # Výstup: (vzorky, window, len(future_features))

    # Spojení historických a budoucích proměnných do jednoho pole
    X = np.concatenate([X_past, X_future], axis=2)  # (vzorky, window, past_features + future_features)

    return X

def create_sequences(data, window, horizon, past_features, future_features):
    """
    Vytvoří sekvence vstupních dat a odpovídající cílové hodnoty pro trénování LSTM modelu.

    Parametry:
    ----------
    data : pandas.DataFrame
        DataFrame obsahující časové řady.
    window : int
        Počet časových kroků v minulosti.
    horizon : int
        Počet časových kroků do budoucnosti.
    past_features : list
        Seznam sloupců, které budou použity jako vstupní vlastnosti v minulosti.
    future_features : list
        Seznam sloupců, které budou použity jako vstupní vlastnosti v budoucnosti.
    target : str
        Název sloupce, který bude použit jako cílová hodnota.

    Návratové hodnoty:
    -------------------
    X : numpy.ndarray
        Pole tvaru (vzorky, window, past_features + future_features), obsahující sekvence vstupních dat.
    y : numpy.ndarray
        Pole tvaru (vzorky, horizon), obsahující odpovídající cílové hodnoty.
    """

    X_past = np.lib.stride_tricks.sliding_window_view(
        data[past_features].values, (window, len(past_features))
    )[:-horizon, :, :]

    X_past = np.squeeze(X_past, axis=1)  

    X_future = np.lib.stride_tricks.sliding_window_view(
        data[future_features].values, (window, len(future_features))
    )[horizon-1 : len(X_past) + horizon-1, :, :]

    X_future = np.squeeze(X_future, axis=1)  

    X = np.concatenate([X_past, X_future], axis=2)  

    return X

import numpy as np
import matplotlib.pyplot as plt

def plot_solar_power_prediction(y_pred_trans, df_true=None):
    """
    Vytvoří graf predikce výkonu fotovoltaické elektrárny v průběhu dne.
    
    Args:
        y_pred_trans (numpy.ndarray): Pole s predikovanými hodnotami výkonu (kW) ve tvaru (1, 24).
        df_true (pd.DataFrame, optional): DataFrame se skutečnými hodnotami výkonu. 
                                          Musí obsahovat sloupec 'output' s 24 hodnotami.

    Returns:
        plt.Figure: Graf pro zobrazení.
    """
    hours = np.arange(24)
    
    plt.figure(figsize=(10, 6))
    plt.plot(hours, y_pred_trans.flatten(), marker='o', label='Predikce (kW)', color='tab:blue')
    
    if df_true is not None and 'output' in df_true.columns and len(df_true) >= 24:
        plt.plot(hours, df_true['output'].values[23:-1], marker='x', label='Skutečný výkon (kW)', color='tab:orange')
    
    plt.xlabel('Hodiny (UTC)')
    plt.ylabel('Výkon (kW)')
    # plt.title('Predikce vs. skutečný výkon FVE v průběhu dne')
    plt.title('Predikce průběhu výkonu FVE pro zvolený den')
    plt.xticks(hours)
    plt.grid(True)
    plt.legend()
    
    return plt