"""Create netCDF files controlling time evolution."""
import os
import matplotlib.pyplot as plt
import numpy as np
import xarray as xr
# GENERAL CONSTANTS
OUT_PATH = "."
# SCALING CONSTANTS
SCALING_TIME = np.arange(1990, 2200, 1)
SCALING_ARR = np.sin(SCALING_TIME * 0.2) * 0.6 + 1.0
SCALING_ARR = SCALING_ARR.astype("float32")
# NORMALIZATION CONSTANTS
NORM_TIME = np.array(
[
2020,
2025,
2030,
2035,
2040,
2045,
2050,
2055,
2060,
2065,
2070,
2075,
2080,
2085,
2090,
2095,
2100,
2105,
2110,
2115,
2120,
]
)
# Reference for fuel consumption until year 2050:
# Energy Insights Global Energy Perspective, Reference Case A3 October 2020;
# IATA; ICAO (fuel consumption values beyond 2050 are customized)
FUEL_ARR = np.array(
[
215,
364,
407,
446,
479,
503,
520,
536,
552,
568,
585,
603,
621,
639,
658,
678,
699,
720,
741,
763,
786,
]
).astype("float32")
EI_CO2_ARR = 3.115 * np.ones(len(NORM_TIME), dtype="float32")
EI_H2O_ARR = 1.25 * np.ones(len(NORM_TIME), dtype="float32")
DIS_PER_FUEL_ARR = 0.3 * np.ones(len(NORM_TIME), dtype="float32")
# TIME SCALING
[docs]
def plot_time_scaling(
scaling_time: np.ndarray, scaling_arr: np.ndarray, out_path: str = OUT_PATH
) -> None:
"""Plots the time scaling factors.
Args:
scaling_time (numpy.ndarray): The time values for the scaling factors.
scaling_arr (numpy.ndarray): The scaling factors to plot.
out_path (str, optional): The path to the output directory.
Defaults to OUT_PATH.
Returns:
None
"""
fig, ax = plt.subplots()
ax.plot(scaling_time, scaling_arr)
ax.set_xlabel("year")
ax.set_ylabel("scaling factor")
fig.savefig(os.path.join(out_path, "time_scaling.png"))
[docs]
def create_time_scaling_xr(
scaling_time: np.ndarray, scaling_arr: np.ndarray
) -> xr.Dataset:
"""Create an :class:`xarray.Dataset` containing time scaling factors.
Args:
scaling_time (numpy.ndarray): The time values for the scaling factors.
scaling_arr (numpy.ndarray): The scaling factors to plot.
Returns:
xarray.Dataset: The dataset containing the time scaling factors.
"""
evolution = xr.Dataset(
data_vars={"scaling": (["time"], scaling_arr)},
coords={"time": scaling_time},
)
evolution.time.attrs = {"units": "years"}
evolution.scaling.attrs = {"species": "all"}
evolution.attrs = {
"Title": "Time scaling example",
"Convention": "CF-XXX",
"Type": "scaling",
"Author": "Stefan Völk",
"Contact": "stefan.voelk@dlr.de",
}
return evolution
# TIME NORMALIZATION
[docs]
def create_time_normalization_xr(
time_arr: np.ndarray,
fuel_arr: np.ndarray,
ei_co2_arr: np.ndarray,
ei_h2o_arr: np.ndarray,
dis_per_fuel_arr: np.ndarray,
) -> xr.Dataset:
"""Create an :class:`xarray.Dataset` containing normalization factors.
Args:
time_arr (numpy.ndarray): Time values (years)
fuel_arr (numpy.ndarray): Fuel consumption
ei_co2_arr (numpy.ndarray): Emission indices for CO2
ei_h2o_arr (numpy.ndarray): Emission indices for H2O
dis_per_fuel_arr (numpy.ndarray): Distance per fuel
Returns:
xarray.Dataset: The dataset containing the normalization factors
"""
evolution = xr.Dataset(
data_vars={
"fuel": (["time"], fuel_arr),
"EI_CO2": (["time"], ei_co2_arr),
"EI_H2O": (["time"], ei_h2o_arr),
"dis_per_fuel": (["time"], dis_per_fuel_arr),
},
coords={"time": time_arr},
)
evolution.time.attrs = {"units": "years"}
evolution.fuel.attrs = {
"long_name": "fuel consumption",
"units": "Tg yr-1",
}
evolution.EI_CO2.attrs = {"long_name": "CO2 emission index", "units": ""}
evolution.EI_H2O.attrs = {"long_name": "H2O emission index", "units": ""}
evolution.dis_per_fuel.attrs = {
"long_name": "distance per fuel",
"units": "km kg-1",
}
evolution.attrs = {
"Title": "Time normalization example",
"Convention": "CF-XXX",
"Type": "norm",
"Author": "Stefan Völk",
"Contact": "stefan.voelk@dlr.de",
}
return evolution
[docs]
def plot_time_norm(evolution: xr.Dataset, out_path: str = OUT_PATH) -> None:
"""Plot normalized values.
Args:
evolution (xarray.Dataset): The xarray Dataset containing the
normalization factors.
out_path (str, optional): The path to the output directory.
Defaults to OUT_PATH.
Returns:
None
"""
co2_emi_arr = np.multiply(evolution.fuel.values, evolution.EI_CO2.values)
fig, axs = plt.subplots(nrows=2)
axs[0].grid(True)
axs[1].grid(True)
evolution.fuel.plot.line("-o", ax=axs[0])
axs[1].plot(evolution.time.values, co2_emi_arr, "-o")
axs[1].set_xlabel("time [years]")
axs[1].set_ylabel("CO2 emissions [Tg]")
fig.savefig(os.path.join(out_path, "time_norm.png"))
# WRITE OUTPUT netCDF
[docs]
def convert_xr_to_nc(ds: xr.Dataset, file_name: str, out_path: str = OUT_PATH) -> None:
"""Convert a xarray dataset to a netCDF file and write to out_path.
Create out_path if not existing.
Args:
ds (xarray.Dataset): The xarray dataset to write to netCDF.
file_name (str): The name of the output file, including the extension.
out_path (str, optional): The path to the output directory.
Defaults to OUT_PATH.
Returns:
None
"""
os.makedirs(out_path, exist_ok=True)
out_file = os.path.join(out_path, f"{file_name}.nc")
ds.to_netcdf(out_file)
[docs]
def main():
"""Parse command-line arguments and create time evolution files."""
import argparse
parser = argparse.ArgumentParser(
description="Create netCDF files controlling time evolution: "
"time scaling and time normalization.",
)
parser.add_argument(
"-o",
"--output-dir",
type=str,
default=OUT_PATH,
help="Directory to write the generated files into "
"(default: current directory).",
)
parser.add_argument(
"-p",
"--plot",
action="store_true",
default=False,
help="Save plots of the generated time evolution files to output-dir "
"(default: False).",
)
args = parser.parse_args()
scaling_ds = create_time_scaling_xr(SCALING_TIME, SCALING_ARR)
convert_xr_to_nc(scaling_ds, "time_scaling_example", out_path=args.output_dir)
if args.plot:
plot_time_scaling(SCALING_TIME, SCALING_ARR, out_path=args.output_dir)
norm_ds = create_time_normalization_xr(
NORM_TIME, FUEL_ARR, EI_CO2_ARR, EI_H2O_ARR, DIS_PER_FUEL_ARR
)
convert_xr_to_nc(norm_ds, "time_norm_example", out_path=args.output_dir)
if args.plot:
plot_time_norm(norm_ds, out_path=args.output_dir)
if __name__ == "__main__":
main()