Get mean absolute error#
Show 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
Show 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)
Show 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()
)
Show 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()
)
Show 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")
)
Show 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())
Show 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())
Show 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