Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 0 additions & 2 deletions src/ebfm/core/INIT.py
Original file line number Diff line number Diff line change
Expand Up @@ -106,9 +106,7 @@ def init_config(time_config: TimeConfig, grid_config, restartdir: Path, initiali
write_restart_file = False

io["writebootfile"] = write_restart_file

io["freqout"] = 8 # OUTPUT: frequency of storing output (every n-th time-step)
io["output_type"] = 2 # Set output file type: 1 = binary files, 2 = netCDF file

# Ensure output and reboot directories exist
os.makedirs(io["outdir"], exist_ok=True)
Expand Down
223 changes: 77 additions & 146 deletions src/ebfm/core/LOOP_write_to_file.py
Original file line number Diff line number Diff line change
Expand Up @@ -164,7 +164,10 @@ def _write_output_variable(nc_file, varname, var_1D, time_index, grid, column):

def main(OUTFILE, io, OUT, grid, t, time, column, gather: ColumnGather):
"""
Accumulate the output variables of this time step and write them to file every io["freqout"] steps.
Accumulate the output variables of this time step and write them to a NetCDF file every io["freqout"] steps.

Structured grids are written on dimensions (time, y, x). Unstructured grids are written on dimensions
(time, cell), with optional mesh topology if grid["mesh"] is available.

Collective over gather.comm: each rank accumulates its own columns, the root rank gathers the columns of
all ranks and writes them to a single file.
Expand Down Expand Up @@ -214,10 +217,6 @@ def main(OUTFILE, io, OUT, grid, t, time, column, gather: ColumnGather):
["ebm_Emelt", "W m^-2", "mean", "Melt energy of EBFM's own energy balance"],
]

io["varsout"] = [
{"varname": v[0], "units": v[1], "type": v[2], "description": v[3]} for v in OUTFILE["varsout"]
]

# Update OUTFILE.TEMP with variables to be stored
for entry in OUTFILE["varsout"]:
varname, var_type = entry[0], entry[2]
Expand Down Expand Up @@ -245,147 +244,79 @@ def main(OUTFILE, io, OUT, grid, t, time, column, gather: ColumnGather):
elif var_type == "sum":
OUTFILE["TEMP"][varname] += temp_long

def save_binary_files():
"""
Write model output to binary files and save run information.

Parameters:
- OUTFILE: Dictionary storing output details and temporary data.
- io: Dictionary holding I/O parameters, e.g., freqout, outdir.
- OUT: Dictionary with variables to be output.
- grid: Grid information.
- t: Current time step.
- time: Dictionary containing time-related variables.
- C: Dictionary of constants.

Returns:
- Updated OUTFILE and io dictionaries.
"""

# Save output to binary files at the first time step
if is_first_time_step(t) and gather.is_root:
if not os.path.exists(io["outdir"]):
os.makedirs(io["outdir"])

io["fid"] = {}
for entry in OUTFILE["varsout"]:
varname = entry[0]
filepath = os.path.join(io["outdir"], f"OUT_{varname}.bin")
io["fid"][varname] = open(filepath, "wb")

# Write variables to binary files when `freqout` is met
if is_output_time_step(t, io): # TODO: correct?
for entry in OUTFILE["varsout"]:
varname = entry[0]
OUTFILE[varname] = gather.gather(OUTFILE["TEMP"][varname])

if is_output_time_step(t, io) and gather.is_root:
for entry in OUTFILE["varsout"]:
varname = entry[0]
io["fid"][varname].write(OUTFILE[varname].astype("float32").tobytes())

# Close all file streams and save run metadata at the final time step
if is_final_time_step(t, time) and gather.is_root:
for file in io["fid"].values():
file.close()

# TODO: WORK IN PROGRESS
# Prepare the runinfo dictionary
# runinfo = {"grid": grid, "time": time, "IOout": io, "Cout": C}

return True

def save_netCDF_file():
"""
Save model output stored in OUTFILE to a NetCDF file.

Structured grids are written on dimensions (time, y, x).
Unstructured grids are written on dimensions (time, cell), with
optional mesh topology if grid["mesh"] is available.
"""
# Epoch for time variable
time_units = "days since 1970-01-01 00:00:00"
time_calendar = "gregorian"

# The root rank defines the file from the grid of all ranks
if is_first_time_step(t):
io["output_grid"] = _gather_grid(gather, grid)

# Initialize NetCDF file at the first time step
if is_first_time_step(t) and gather.is_root:
if not os.path.exists(io["outdir"]):
os.makedirs(io["outdir"])

# Create NetCDF file
nc_filepath = os.path.join(io["outdir"], "model_output.nc")
io["nc_file"] = Dataset(nc_filepath, "w", format="NETCDF4")
io["nc_file"].createDimension("time", None) # Unlimited time dimension

if io["output_grid"]["is_unstructured"]:
_write_unstructured_grid(io["nc_file"], io["output_grid"])
else:
_write_structured_grid(io["nc_file"], io["output_grid"])

io["nc_file"].createDimension("nl", column.nl) # Vertical layers for `sub` variables

# Define standard output variables
for entry in OUTFILE["varsout"]:
varname = entry[0]
var_units = entry[1]
var_desc = entry[3]

dimensions, chunksizes = _output_dimensions(varname, io["output_grid"], column)

nc_var = io["nc_file"].createVariable(
varname=varname,
datatype=np.float32,
dimensions=dimensions,
zlib=True,
complevel=4,
fill_value=_FILL_VALUE,
chunksizes=chunksizes,
)

# Assign metadata
nc_var.units = var_units
nc_var.description = var_desc

# Define a time variable to track simulation steps
nc_time = io["nc_file"].createVariable("time", np.float64, ("time",), zlib=True, fill_value=_FILL_VALUE)
nc_time.units = time_units
nc_time.calendar = time_calendar
nc_time.description = "Time at which data is recorded, in days since 1970-01-01 00:00:00"

# Write data to NetCDF at the specified frequency (e.g., for average/summed values)
if is_output_time_step(t, io): # TODO: correct?
time_index = t // io["freqout"] # Determine the time slice index

# Calculate time in "days since 1970-01-01"
time_days_since_1970 = date2num(time["TCUR"], units=time_units, calendar=time_calendar)

# Write time variable
# Epoch for time variable
time_units = "days since 1970-01-01 00:00:00"
time_calendar = "gregorian"

# The root rank defines the file from the grid of all ranks
if is_first_time_step(t):
io["output_grid"] = _gather_grid(gather, grid)

# Initialize NetCDF file at the first time step
if is_first_time_step(t) and gather.is_root:
if not os.path.exists(io["outdir"]):
os.makedirs(io["outdir"])

# Create NetCDF file
nc_filepath = os.path.join(io["outdir"], "model_output.nc")
io["nc_file"] = Dataset(nc_filepath, "w", format="NETCDF4")
io["nc_file"].createDimension("time", None) # Unlimited time dimension

if io["output_grid"]["is_unstructured"]:
_write_unstructured_grid(io["nc_file"], io["output_grid"])
else:
_write_structured_grid(io["nc_file"], io["output_grid"])

io["nc_file"].createDimension("nl", column.nl) # Vertical layers for `sub` variables

# Define standard output variables
for entry in OUTFILE["varsout"]:
varname = entry[0]
var_units = entry[1]
var_desc = entry[3]

dimensions, chunksizes = _output_dimensions(varname, io["output_grid"], column)

nc_var = io["nc_file"].createVariable(
varname=varname,
datatype=np.float32,
dimensions=dimensions,
zlib=True,
complevel=4,
fill_value=_FILL_VALUE,
chunksizes=chunksizes,
)

# Assign metadata
nc_var.units = var_units
nc_var.description = var_desc

# Define a time variable to track simulation steps
nc_time = io["nc_file"].createVariable("time", np.float64, ("time",), zlib=True, fill_value=_FILL_VALUE)
nc_time.units = time_units
nc_time.calendar = time_calendar
nc_time.description = "Time at which data is recorded, in days since 1970-01-01 00:00:00"

# Write data to NetCDF at the specified frequency (e.g., for average/summed values)
if is_output_time_step(t, io): # TODO: correct?
time_index = t // io["freqout"] # Determine the time slice index

# Calculate time in "days since 1970-01-01"
time_days_since_1970 = date2num(time["TCUR"], units=time_units, calendar=time_calendar)

# Write time variable
if gather.is_root:
io["nc_file"]["time"][time_index] = time_days_since_1970

# Write variables to NetCDF, gathering one variable at a time to limit the memory needed on the root rank
for entry in OUTFILE["varsout"]:
varname = entry[0]
var_1D = gather.gather(OUTFILE["TEMP"][varname])
if gather.is_root:
io["nc_file"]["time"][time_index] = time_days_since_1970

# Write variables to NetCDF, gathering one variable at a time to limit the memory needed on the root rank
for entry in OUTFILE["varsout"]:
varname = entry[0]
var_1D = gather.gather(OUTFILE["TEMP"][varname])
if gather.is_root:
_write_output_variable(io["nc_file"], varname, var_1D, time_index, io["output_grid"], column)

# Close the NetCDF file at the final time step
if is_final_time_step(t, time) and gather.is_root:
io["nc_file"].close()

return True

# Save output as binary or netCDF files
if io["output_type"] == 1:
save_binary_files()
elif io["output_type"] == 2:
save_netCDF_file()
else:
print("Invalid output type. Please choose 1 for binary files or 2 for NetCDF.")
_write_output_variable(io["nc_file"], varname, var_1D, time_index, io["output_grid"], column)

# Close the NetCDF file at the final time step
if is_final_time_step(t, time) and gather.is_root:
io["nc_file"].close()

return io, OUTFILE
15 changes: 6 additions & 9 deletions tests/core/test_loop_write_to_file.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,8 +27,6 @@
# Element ids of the unstructured grid. Like the ids of an Elmer mesh, they neither start at 0 nor are contiguous.
ELEMENT_IDS = (3, 5, 12, 27, 40)

_NETCDF_OUTPUT = 2


def _make_column():
return ColumnDiscretizationConfig(nl=NL, split=(2,))
Expand Down Expand Up @@ -135,22 +133,21 @@ def _write_single_step(outdir, grid, extra_out=None):
out = _fake_out(GPSUM, NL)
if extra_out:
out.update(extra_out)
io = {"outdir": str(outdir), "freqout": 1, "output_type": _NETCDF_OUTPUT}
io = {"outdir": str(outdir), "freqout": 1}
time = {"TCUR": datetime(2020, 1, 1), "tn": 1}

io, outfile = write_output({}, io, out, grid, 0, time, column, ColumnGather(defaultComm, grid["column_id"]))
_, outfile = write_output({}, io, out, grid, 0, time, column, ColumnGather(defaultComm, grid["column_id"]))

return Path(outdir) / "model_output.nc", outfile, out


class TestNetCDFOutputSmoke(unittest.TestCase):
"""
End-to-end checks for the NetCDF output path (io["output_type"] == 2).
End-to-end checks for the NetCDF output.

This is the default output type set in INIT, and the writer takes the column
discretization as a separate argument from the grid. These tests drive the
real entry point so that a helper whose signature or arguments drift out of
sync fails here instead of at runtime.
The writer takes the column discretization as a separate argument from the
grid. These tests drive the real entry point so that a helper whose
signature or arguments drift out of sync fails here instead of at runtime.
"""

def test_structured_grid_writes_all_declared_variables(self):
Expand Down
Loading