Get mean absolute error

Get mean absolute error#

Hide code cell source
from collections.abc import Iterable
from inspect import Signature

from devtools import pprint
from matplotlib.pyplot import subplots
from more_itertools import one
from numpy import inf, nan, pi
from pandas import DataFrame, Series, merge_ordered, read_hdf
from seaborn import scatterplot

from boilercv.correlations import GROUPS, beta
from boilercv.correlations import nusselt as correlations_nusselt
from boilercv.dimensionless_params import jakob, prandtl, thermal_diffusivity
from boilercv_docs.nbs import init
from boilercv_pipeline.models.column import Col
from boilercv_pipeline.models.df import GBC
from boilercv_pipeline.models.subcool import const
from boilercv_pipeline.palettes import cat10
from boilercv_pipeline.stages import find_tracks, get_thermal_data
from boilercv_pipeline.stages.get_mae import GetMae as Params

PARAMS = None
Hide code cell source
from datetime import datetime

from boilercore.paths import ISOLIKE, dt_fromisolike

if isinstance(PARAMS, str):
    params = Params.model_validate_json(PARAMS)
else:
    params = Params(context=init(), include_patterns=const.nb_include_patterns)
params.format.set_display_options()
data = params.data
C = params.cols
context = params.context

# Physical parameters
LATENT_HEAT_OF_VAPORIZATION = 2.23e6  # J/kg
LIQUID_DENSITY = 960  # kg/m^3
LIQUID_DYNAMIC_VISCOSITY = 2.88e-4  # Pa-s
LIQUID_ISOBARIC_SPECIFIC_HEAT = 4213  # J/kg-K
LIQUID_THERMAL_CONDUCTIVITY = 0.676  # W/m-K
VAPOR_DENSITY = 0.804  # kg/m^3


# Plotting
CORRELATIONS_PALETTE = cat10
"""For plotting one approach."""

# Plotting
MAX_BETA_MAE = 0.3
"""Maximum mean absolute error of beta to plot."""
MAX_NUSSELT_MAE = 4000
"""Maximum mean absolute error of nusselt to plot."""
WIDTH_SCALE = 2.00
"""Width scale for plotting."""
HEIGHT_SCALE = 1.30
"""Height scale for plotting."""

tracks_path = one(params.tracks)
BC = find_tracks.Cols()
tracks = (
    DataFrame(read_hdf(tracks_path))
    .pipe(lambda df: df[df[BC.bub_fourier()] > 0])[
        [
            BC.bub(),
            BC.bub_reynolds(),
            BC.bub_reynolds0(),
            BC.bub_fourier(),
            BC.bub_beta(),
            BC.bub_nusselt(),
        ]
    ]
    .set_index(BC.bub())
    .pipe(
        lambda df: df[
            (df[BC.bub_beta()] > 0.05 * df[BC.bub_beta()].max())
            & (df[BC.bub_nusselt()] > 0.05 * df[BC.bub_nusselt()].max())
        ]
    )
)


def get_time() -> datetime:
    """Get time from path."""
    if match := ISOLIKE.search(tracks_path.stem):
        return dt_fromisolike(match)
    else:
        raise ValueError("No time found in path.")


time = get_time()
thermal = read_hdf(params.deps.thermal)
TC = get_thermal_data.Cols()
subcooling = thermal.set_index(TC.time())[TC.subcool()][time]

beta_correlations = beta.get_correlations()
nusselt_correlations = correlations_nusselt.get_correlations()
object_averages = tracks.groupby(BC.bub(), **GBC).mean().mean()
constants = {
    "Ja": jakob(
        liquid_density=LIQUID_DENSITY,
        vapor_density=VAPOR_DENSITY,
        liquid_isobaric_specific_heat=LIQUID_ISOBARIC_SPECIFIC_HEAT,
        subcooling=subcooling,
        latent_heat_of_vaporization=LATENT_HEAT_OF_VAPORIZATION,
    ),
    "Pr": prandtl(
        dynamic_viscosity=LIQUID_DYNAMIC_VISCOSITY,
        isobaric_specific_heat=LIQUID_ISOBARIC_SPECIFIC_HEAT,
        thermal_conductivity=LIQUID_THERMAL_CONDUCTIVITY,
    ),
    "alpha": thermal_diffusivity(
        thermal_conductivity=LIQUID_THERMAL_CONDUCTIVITY,
        density=LIQUID_DENSITY,
        isobaric_specific_heat=LIQUID_ISOBARIC_SPECIFIC_HEAT,
    ),
    "pi": pi,
}


def get_mae(df: DataFrame, kind: str, rel: bool = False) -> "Series[float]":
    """Get mean absolute error for dimensionless bubble diameter.

    Get mean absolute percentage error if `rel`.
    """
    df = (
        df.assign(**{
            label: (
                abs(
                    (
                        (
                            corr.expr(**{
                                kwd: value
                                for kwd, value in {
                                    **constants,
                                    "Re_b": df[BC.bub_reynolds()],
                                    "Re_b0": df[BC.bub_reynolds0()],
                                    "Fo_0": df[BC.bub_fourier()],
                                    **(
                                        {}
                                        if kind == "beta"
                                        else {"beta": df[BC.bub_beta()]}
                                    ),
                                }.items()
                                if kwd in Signature.from_callable(corr.expr).parameters
                            })
                        )
                        - df["experimental"]
                    )
                    / (df["experimental"] if rel else 1)
                )
            )
            for label, corr in (
                beta_correlations if kind == "beta" else nusselt_correlations
            ).items()
        })
        .drop(
            columns=[
                "experimental",
                BC.bub_reynolds(),
                BC.bub_reynolds0(),
                BC.bub_fourier(),
                BC.bub_nusselt() if kind == "beta" else BC.bub_beta(),
            ]
        )
        .replace(inf, nan)
        .dropna()
    )
    return df.sum() / len(df)


def preview(
    df: DataFrame, cols: Iterable[Col] | None = None, index: Col | None = None
) -> DataFrame:
    """Preview a dataframe in the notebook."""
    df = params.format.preview(cols=cols, df=df, index=index, f=lambda df: df.head(16))
    return df


pprint(params)
Hide code cell source
beta_mae = preview(
    tracks.rename(columns={BC.bub_beta(): "experimental"})
    .groupby(BC.bub(), **GBC)
    .apply(lambda df: get_mae(df, kind="beta"))
    .rename_axis(index=BC.bub())
    .reset_index()
)
Hide code cell source
nusselt_mae = preview(
    tracks.rename(columns={BC.bub_nusselt(): "experimental"})
    .groupby(BC.bub(), **GBC)
    .apply(lambda df: get_mae(df, kind="nusselt", rel=False))
    .rename_axis(index=BC.bub())
    .reset_index()
)
Hide code cell source
maes = preview(
    merge_ordered(
        beta_mae.set_index(BC.bub()).melt(
            var_name="Correlation", value_name="Beta (MAE)", ignore_index=False
        ),
        nusselt_mae.set_index(BC.bub()).melt(
            var_name="Correlation", value_name="Nusselt (MAE)", ignore_index=False
        ),
        on=[BC.bub(), "Correlation"],
    )
    .assign(**{"Group": lambda df: df["Correlation"].map(GROUPS)})
    .sort_values("Group")
)
Hide code cell source
figure, ax = subplots()
ax.set_xlim(0, MAX_BETA_MAE)
ax.set_ylim(0, MAX_NUSSELT_MAE)
scatterplot(
    ax=ax,
    x="Beta (MAE)",
    y="Nusselt (MAE)",
    hue="Correlation",
    palette=CORRELATIONS_PALETTE,
    s=params.format.size,
    data=maes,
)
figure.set_figwidth(WIDTH_SCALE * figure.get_figwidth())
figure.set_figheight(HEIGHT_SCALE * figure.get_figheight())
Hide code cell source
figure, ax = subplots()
ax.set_xlim(0, MAX_BETA_MAE)
ax.set_ylim(0, MAX_NUSSELT_MAE)
scatterplot(
    ax=ax,
    x="Beta (MAE)",
    y="Nusselt (MAE)",
    hue="Group",
    palette=CORRELATIONS_PALETTE,
    s=params.format.size,
    data=maes,
)
figure.set_figwidth(WIDTH_SCALE * figure.get_figwidth())
figure.set_figheight(HEIGHT_SCALE * figure.get_figheight())
Hide code cell source
mae = (
    DataFrame([
        (
            tracks.rename(columns={BC.bub_beta(): "experimental"})
            .pipe(get_mae, "beta")
            .rename("Beta (MAE)")
        ),
        (
            tracks.rename(columns={BC.bub_nusselt(): "experimental"})
            .pipe(get_mae, "nusselt", rel=False)
            .rename("Nusselt (MAE)")
        ),
    ])
    .T.rename_axis(index="Correlation")
    .assign(**{"subcooling": subcooling})
)

mae