transformadores-backend/funciones_analisis.py

import numpy as np
import pandas as pd
import os, shutil
from scipy.fft import fft, fftfreq


def crear_cache():
   try: 
       os.makedirs('temp/T1')
       os.makedirs('temp/T2')
       os.makedirs('temp/Tiempo')


   except FileExistsError: 
       shutil.rmtree('temp')
       os.makedirs('temp/T1')
       os.makedirs('temp/T2')
       os.makedirs('temp/Tiempo')



def plot_VI(ax, t, V, I, analisis = False, Titulo = False): #parametro reset plot sirve para dar la opcion a superponer graficas 

    linea_V, = ax.plot(t, V, label="Tension")  # Graficar la Tension
    linea_I, = ax.plot(t, I, label="Corriente")  # Graficar la Corriente
    ax.set_ylabel('Amplitud')
    ax.set_xlabel('Tiempo [ms]')
    ax.grid(True)

    if Titulo: ax.set_title(Titulo) #si hay un titulo que lo ponga

    ax.legend(loc='upper right')

    if analisis:
        V_rms, I_rms, FP, tipo_FP = analizar_ondas(V,I)
        data = [f"Tension RMS: {V_rms}[V]", f"Corriente RMS: {I_rms}[A]", f"FP: {FP} {tipo_FP}"]

        for i, text in enumerate(data):
            ax.text(1.05, 0.8 - i*0.1, text, fontsize=12, ha='left', va='center', transform=ax.transAxes)

    return linea_V, linea_I



def plot_termico(ax, t, T, reset_plot = True, tags = False, Titulo = False):
    
    if reset_plot: ax.clear() # que limpie la  grafica
    if Titulo: ax.set_title(Titulo) #que ponga un titulo si lo hay

    linea, = ax.plot(t, T, label = tags)  # Graficar temperatura
    ax.set_ylabel('Temperatura [ºC]')
    ax.set_xlabel('Tiempo [s]')
    ax.grid(True)

    if tags: ax.legend(loc='upper right')

    return linea



def plot_fft(ax, t, onda, fs, threshold_picos = False, reset_plot = False, Titulo = False, tags = False):

    if reset_plot: ax.clear() # que limpie la  grafica
    if Titulo: ax.set_title(Titulo) #que ponga un titulo si lo hay

    x = fftfreq(len(t), 1/fs)[:len(t)//2]
    y = 1/len(x) * np.abs(fft(onda)[0:len(t)//2])

    linea, = ax.plot(x,y)
    ax.set_ylabel('Amplitud')
    ax.set_xlabel('Frecuencias')

    if threshold_picos: 
        peaks = [(freq, amp) for freq, amp in zip(x, y) if amp > threshold_picos]

        for freq, amp in peaks:
            ax.annotate(f"{freq:.2f} Hz\n Amplitud: {amp:.2f}", 
                        xy =(freq, amp), 
                        xytext=(freq, amp +.05), 
                        fontsize=10, 
                        ha="left",
                        va = 'center')

            ax.set_xlim(left=min(x), right=max(x) + 10)
            ax.set_ylim(bottom=0, top=max(y) + 0.2)          

    if tags: ax.legend(loc='upper right')

    return linea



def analizar_ondas(V, I):

    V_rms = round(np.sqrt(np.mean(np.array(V)**2)))
    I_rms = round(np.sqrt(np.mean(np.array(I)**2)))

    V = fft(V)
    I = fft(I)

    idx = np.argmax(np.abs(V))

    fase_1 = np.angle(V[idx])
    fase_2 = np.angle(I[idx])

    desfase = fase_2 - fase_1

    tipo_desfase = 'Inductivo'

    if desfase > 0:
        tipo_desfase = 'Capacitivo'

    factor_potencia = round(np.cos(abs(desfase)),2)

    return V_rms, I_rms, factor_potencia, tipo_desfase



def guardar_todo(datos_temp, datos_electricos):

    print('Guardando Archivos')

    guardar_temperatura_npy(datos_temp['Tiempo'], datos_temp['T1'], datos_temp['T2'])
    guardar_temperatura()
    guardar_VI(datos_electricos['V1'], datos_electricos['I1'], datos_electricos['V2'], datos_electricos['I2'])

    shutil.rmtree('temp') #borro chache

    print('Archivos Guardados')



def guardar_temperatura_npy(Tiempo, T1, T2):

    n = len(os.listdir('temp/Tiempo'))

    np.save(f"temp/Tiempo/{n}.npy", Tiempo)
    np.save(f"temp/T1/{n}.npy", T1)
    np.save(f"temp/T2/{n}.npy", T2)



def guardar_temperatura(): # Mira los datos del directorio temp y guarda los datos de temperatura en un csv
    
    try: os.makedirs('Resultados/Termicos')
    except:pass

    elementos = os.listdir('temp')
    data = []

    for i in elementos:
        temp = [0]

        for j in os.listdir(f"temp/{i}"):
            temp = np.concatenate((temp, np.load(f"temp/{i}/{j}")))

        data.append(temp[-(len(temp)-1):])


    data = np.array(data)
    df = pd.DataFrame(np.transpose(data), columns= elementos)
    n = len(os.listdir('Resultados/Termicos')) + 1

    df.to_csv(f"Resultados/Termicos/Datos_Termicos_{n}.csv", index=False)        



def guardar_VI(V1, I1, V2, I2): # guarda los utimos datos electricos en un csv
    
    try:os.makedirs('Resultados/Electricos')
    except:pass

    data = np.array([V1, I1, V2, I2])

    columnas = ['Tension Primario [V]', 'Corriente Primario [A]', 'Tension Secundario [V]', 'Corriente Secundario [A]']
    df = pd.DataFrame(np.transpose(data), columns = columnas)

    n = len(os.listdir('Resultados/Termicos')) + 1

    df.to_csv(f'Resultados/Electricos/Datos_Electricos_{n}.csv')