Extending Gridded DSS Forcing Through the Simulation Window¶
# True uses this checkout; False uses the installed package.
import sys
from hashlib import sha256
import os
from pathlib import Path
import subprocess
from tempfile import mkdtemp
import numpy as np
import pandas as pd
USE_LOCAL_SOURCE = False
REPO_ROOT = Path.cwd().resolve()
if REPO_ROOT.name.lower() == "examples":
REPO_ROOT = REPO_ROOT.parent
if USE_LOCAL_SOURCE:
local_path = str(REPO_ROOT)
if local_path not in sys.path:
sys.path.insert(0, local_path)
print("LOCAL SOURCE MODE")
else:
print("PIP PACKAGE MODE")
from ras_commander import RasDss # noqa: E402
def sha256_file(path, chunk_size=1024 * 1024):
digest = sha256()
with Path(path).open("rb") as stream:
for chunk in iter(lambda: stream.read(chunk_size), b""):
digest.update(chunk)
return digest.hexdigest()
Why this derivative exists¶
A gridded precipitation or rainfall-excess DSS family often ends when the observed or forecast rainfall ends, while a hydraulic simulation continues so runoff can route and water can recede. RasDss.copy_grid_with_zero_tail() creates a run-local DSS derivative that keeps the selected source grids and appends explicit zero-precipitation intervals through the required forcing window. The source PR used this derivative as the gridded DSS forcing later supplied to scenario preparation.
The same operation can mechanically align a reviewed source with a model workflow: shift every record window by a specified number of minutes, rename the A/B/C/F pathname family, or translate the grid origin by exact whole-cell increments. The source file remains byte-for-byte unchanged. The derivative is a newly encoded DSS file, so its bytes are not expected to match the source.
This is preparation of an already accepted gridded forcing product. It is not a rainfall or rainfall-excess calculation.
Scope and engineering boundary¶
This notebook demonstrates only deterministic file preparation and QA. It does not:
- perform scientific storm transposition, even if a pathname label contains words such as
TRANSPOSED; - reproject, resample, interpolate, rotate, or redistribute grid values;
- infer a time zone or daylight-saving offset;
- infer how many dry intervals a hydraulic model needs;
- determine whether rainfall, loss, infiltration, runoff, or interpolation assumptions are suitable; or
- modify or execute a HEC-RAS project.
An engineer must approve the source product, time basis, simulation window, pathname convention, spatial relationship, and HEC-RAS meteorological configuration before the derivative is attached to a model.
1. Configure the source and mechanical transforms¶
Set RAS_COMMANDER_GRID_SOURCE_DSS and RAS_COMMANDER_GRID_SOURCE_FAMILY to exercise a reviewed real-world grid family. Optional RAS_COMMANDER_GRID_OUTPUT_FAMILY, RAS_COMMANDER_GRID_TIME_SHIFT_MINUTES, RAS_COMMANDER_GRID_X_SHIFT, and RAS_COMMANDER_GRID_Y_SHIFT settings declare reviewed mechanical transforms. External sources default to no pathname, time, or origin translation. With no source environment variable, the notebook creates a small deterministic native DSS7 demonstration under the ignored working/ folder. No generated DSS is written under examples/data/.
WORK_ROOT = REPO_ROOT / "working" / "example_728_extended_grid_forcing"
WORK_ROOT.mkdir(parents=True, exist_ok=True)
RUN_ROOT = Path(mkdtemp(prefix="run_", dir=WORK_ROOT))
source_dss_setting = os.environ.get("RAS_COMMANDER_GRID_SOURCE_DSS")
source_family_setting = os.environ.get("RAS_COMMANDER_GRID_SOURCE_FAMILY")
if source_dss_setting and not source_family_setting:
raise RuntimeError("Set RAS_COMMANDER_GRID_SOURCE_FAMILY with an external source DSS")
SOURCE_DSS = (
Path(source_dss_setting).expanduser()
if source_dss_setting
else RUN_ROOT / "demo_aorc_source_v7.dss"
)
SOURCE_FAMILY = source_family_setting or "/SHG/DEMO/PRECIPITATION///AORC/"
USING_DEMO_SOURCE = source_dss_setting is None
OUTPUT_DSS = RUN_ROOT / "extended_grid_forcing_derivative.dss"
OUTPUT_FAMILY = os.environ.get(
"RAS_COMMANDER_GRID_OUTPUT_FAMILY",
"/SHG/DEMO/PRECIPITATION///EXTENDED-WINDOW-DEMO/" if USING_DEMO_SOURCE else SOURCE_FAMILY,
)
TAIL_INTERVALS = int(os.environ.get("RAS_COMMANDER_GRID_TAIL_INTERVALS", "3"))
TIME_SHIFT_MINUTES = int(
os.environ.get("RAS_COMMANDER_GRID_TIME_SHIFT_MINUTES", "0")
) # reviewed fixed offset; no time-zone inference
X_SHIFT = float(
os.environ.get("RAS_COMMANDER_GRID_X_SHIFT", "0")
) # optional reviewed translation in source CRS units
Y_SHIFT = float(
os.environ.get("RAS_COMMANDER_GRID_Y_SHIFT", "0")
) # optional reviewed translation in source CRS units
configuration = pd.Series(
{
"source mode": "deterministic native DSS7 demo" if USING_DEMO_SOURCE else "reviewed external DSS",
"source DSS": str(SOURCE_DSS),
"source family": SOURCE_FAMILY,
"output DSS": str(OUTPUT_DSS),
"output family": OUTPUT_FAMILY,
"zero-tail intervals": TAIL_INTERVALS,
"fixed time shift (minutes)": TIME_SHIFT_MINUTES,
"x/y origin shift": f"{X_SHIFT}, {Y_SHIFT}",
},
name="reviewed configuration",
)
display(configuration.to_frame())
2. Prepare or locate the source DSS¶
The fallback fixture is written through public RasDss APIs and includes an unrelated time-series record. That extra record lets us verify that the derivative contains only the selected grid family and its zero tail. Replace the fallback with an independently sourced DSS for project-specific review.
if USING_DEMO_SOURCE:
# Build the fixture in a short-lived process so native DSS handles are
# released before this notebook takes its independent pre-copy hash.
demo_script = r'''
import sys
from datetime import datetime
from pathlib import Path
import numpy as np
import pandas as pd
from ras_commander import RasDss
source = Path(sys.argv[1])
RasDss.write_timeseries(
source,
"/UNRELATED/GAGE/FLOW//1HOUR/DEMO/",
pd.date_range("2020-01-01", periods=2, freq="h"),
[10.0, 20.0],
units="CFS",
data_type="INST-VAL",
dss_version=7,
)
RasDss.write_grid_timeseries(
source,
"/SHG/DEMO/PRECIPITATION///AORC/",
np.array(
[
[[1.0, np.nan, 2.0], [3.0, 4.0, 5.0]],
[[6.0, np.nan, 7.0], [8.0, 9.0, 10.0]],
],
dtype=np.float32,
),
[
datetime(2020, 1, 1, 0),
datetime(2020, 1, 1, 1),
datetime(2020, 1, 1, 2),
],
{
"cell_size": 1000.0,
"origin": (259000.0, 1024000.0),
"crs": "SHG",
"units": "MM",
"data_type": "PER-CUM",
"compression": "PRECIP_2_BYTE",
},
create_if_missing=False,
)
'''
demo_env = os.environ.copy()
if USE_LOCAL_SOURCE:
demo_env["PYTHONPATH"] = os.pathsep.join(
value for value in (str(REPO_ROOT), demo_env.get("PYTHONPATH")) if value
)
completed = subprocess.run(
[sys.executable, "-c", demo_script, str(SOURCE_DSS)],
cwd=REPO_ROOT,
env=demo_env,
capture_output=True,
text=True,
check=False,
)
if completed.returncode != 0:
raise RuntimeError("Native DSS demo creation failed:\n" + completed.stdout + completed.stderr)
if "access violation" in (completed.stdout + completed.stderr).casefold():
raise RuntimeError("Native DSS demo creation emitted an access-violation diagnostic")
elif not SOURCE_DSS.is_file():
raise FileNotFoundError(f"Configured source DSS does not exist: {SOURCE_DSS}")
source_sha_before = sha256_file(SOURCE_DSS)
if USING_DEMO_SOURCE:
print(f"Generated deterministic DSS7 demonstration dataset: {SOURCE_DSS}")
else:
print(f"Reading existing reviewed source DSS without modifying it: {SOURCE_DSS}")
3. Inspect the selected grid family¶
A family selector has six A-F parts with blank D and E time-window parts. Selection is case-insensitive and must resolve to one regular, contiguous grid family. Here we display the matching native records and inspect one frame before authoring the derivative.
selector_parts = SOURCE_FAMILY.split("/")[1:-1]
if len(selector_parts) != 6 or selector_parts[3] or selector_parts[4]:
raise ValueError("SOURCE_FAMILY must be a six-part selector with blank D/E parts")
source_catalog = RasDss.get_catalog(SOURCE_DSS)
catalog_parts = source_catalog["pathname"].map(lambda value: value.split("/")[1:-1])
selected_mask = catalog_parts.map(
lambda parts: len(parts) == 6
and all(parts[index].casefold() == selector_parts[index].casefold() for index in (0, 1, 2, 5))
)
selected_paths = source_catalog.loc[selected_mask, "pathname"].sort_values().tolist()
if not selected_paths:
raise ValueError(f"No grid records matched {SOURCE_FAMILY}")
source_first = RasDss.read_grid(SOURCE_DSS, selected_paths[0])
source_summary = pd.Series(
{
"DSS version": RasDss.get_file_version(SOURCE_DSS),
"catalog records (all types)": len(source_catalog),
"selected grid records": len(selected_paths),
"first window": f"{source_first['start_time']} to {source_first['end_time']}",
"shape": source_first["shape"],
"cell size": source_first["cell_size"],
"origin": source_first["metadata"]["origin"],
"NoData cells": int(np.isnan(source_first["data"]).sum()),
"independent pre-copy SHA-256": source_sha_before,
},
name="selected source family",
)
display(source_summary.to_frame())
display(pd.DataFrame({"source pathname": selected_paths}))
4. Create the run-local derivative¶
The call streams one frame at a time, rewrites only the selected family, appends zero grids with the original NoData footprint, completely reads back the temporary DSS, and then publishes it. Because RUN_ROOT is unique, overwrite=False preserves fail-closed create-if-absent behavior.
result = RasDss.copy_grid_with_zero_tail(
source_dss=SOURCE_DSS,
output_dss=OUTPUT_DSS,
pathname=SOURCE_FAMILY,
tail_intervals=TAIL_INTERVALS,
time_shift_minutes=TIME_SHIFT_MINUTES,
output_pathname=OUTPUT_FAMILY,
x_shift=X_SHIFT,
y_shift=Y_SHIFT,
overwrite=False,
)
display(pd.Series(result, name="derivative result").to_frame())
5. Verify source immutability, version, and time coverage¶
The notebook hashes the source independently before the copy. The API also hashes it before native access, checks it again after validation and writing, and returns the accepted SHA-256. Equality of the independent pre-copy hash, API-accepted hash, and independent post-operation hash is the byte-preservation assertion. The derivative has its own encoding and contains only the renamed grid family: copied forcing frames followed by the requested dry tail.
source_sha_after = sha256_file(SOURCE_DSS)
source_sha_accepted = result["source_sha256"]
output_catalog = RasDss.get_catalog(OUTPUT_DSS)["pathname"].sort_values().tolist()
expected_output_count = result["source_record_count"] + result["appended_record_count"]
assert source_sha_before == source_sha_after == source_sha_accepted
assert RasDss.get_file_version(OUTPUT_DSS) == RasDss.get_file_version(SOURCE_DSS)
assert len(output_catalog) == expected_output_count
assert set(output_catalog) == set(result["written_source_pathnames"] + result["appended_pathnames"])
assert not any("/UNRELATED/" in pathname.upper() for pathname in output_catalog)
coverage = pd.DataFrame({"pathname": output_catalog})
coverage["record role"] = coverage["pathname"].map(
lambda pathname: "zero tail" if pathname in result["appended_pathnames"] else "copied source"
)
verification = pd.Series(
{
"source SHA unchanged": source_sha_before == source_sha_after == source_sha_accepted,
"independent pre-copy SHA-256": source_sha_before,
"accepted source SHA-256": source_sha_accepted,
"independent post-copy SHA-256": source_sha_after,
"source/output DSS version": result["dss_version"],
"output start": result["output_start"],
"copied forcing ends": result["output_source_end"],
"zero tail ends": result["padded_end"],
"output records": len(output_catalog),
},
name="coverage QA",
)
display(verification.to_frame())
display(coverage)
6. Verify values, NoData, and mechanical translations¶
A whole-cell origin translation changes spatial metadata only. The copied grid array and shape remain unchanged. Valid tail cells are exactly zero, while every source NoData cell remains NoData. The timestamp check verifies the configured fixed offset; it does not claim that the offset is the correct civil-time conversion for another event.
output_first = RasDss.read_grid(OUTPUT_DSS, result["written_source_pathnames"][0])
np.testing.assert_allclose(
output_first["data"],
source_first["data"],
rtol=0.0,
atol=0.0,
equal_nan=True,
)
assert output_first["shape"] == source_first["shape"]
source_origin = np.asarray(source_first["metadata"]["origin"], dtype=float)
output_origin = np.asarray(output_first["metadata"]["origin"], dtype=float)
np.testing.assert_allclose(output_origin - source_origin, [X_SHIFT, Y_SHIFT])
actual_time_shift = int(
(pd.Timestamp(output_first["start_time"]) - pd.Timestamp(source_first["start_time"])).total_seconds()
// 60
)
assert actual_time_shift == TIME_SHIFT_MINUTES
source_nodata = np.isnan(source_first["data"])
tail_qa = []
for pathname in result["appended_pathnames"]:
tail_grid = RasDss.read_grid(OUTPUT_DSS, pathname)
tail_nodata = np.isnan(tail_grid["data"])
assert np.array_equal(tail_nodata, source_nodata)
assert np.all(tail_grid["data"][~tail_nodata] == 0.0)
tail_qa.append(
{
"pathname": pathname,
"valid-cell maximum": float(tail_grid["data"][~tail_nodata].max()),
"NoData cells": int(tail_nodata.sum()),
}
)
transform_qa = pd.Series(
{
"copied values unchanged": True,
"shape unchanged": output_first["shape"] == source_first["shape"],
"fixed time shift observed (minutes)": actual_time_shift,
"source origin": tuple(source_origin),
"output origin": tuple(output_origin),
"origin delta": tuple(output_origin - source_origin),
"tail NoData mask preserved": True,
},
name="mechanical transform QA",
)
display(transform_qa.to_frame())
display(pd.DataFrame(tail_qa))
7. Record the real ras-commander handoff without modifying a project¶
The generated derivative can be attached to a copied HEC-RAS project with the existing public RasUnsteady.configure_gridded_dss_precipitation() API. That call modifies the copied .u## file and its HDF sidecar; it does not write the DSS dataset. This notebook does not invoke the mutating handoff.
The real public call, shown here but deliberately not executed, has this shape:
from ras_commander import RasUnsteady
RasUnsteady.configure_gridded_dss_precipitation(
unsteady_file=COPIED_UNSTEADY_FILE,
dss_filename=OUTPUT_DSS,
dss_pathname=OUTPUT_FAMILY,
interpolation=REVIEWED_INTERPOLATION,
)
Run that call only against an isolated project copy after approving the derivative, pathname, interpolation setting, and simulation window.
REVIEWED_INTERPOLATION = None # e.g., a project-approved HEC-RAS setting
scenario_handoff = pd.Series(
{
"public handoff API": "RasUnsteady.configure_gridded_dss_precipitation",
"dss_filename": str(OUTPUT_DSS),
"dss_pathname": OUTPUT_FAMILY,
"interpolation": REVIEWED_INTERPOLATION,
"handoff invoked": False,
"HEC-RAS project modified": False,
"HEC-RAS executed": False,
},
name="conceptual downstream handoff",
)
display(scenario_handoff.to_frame())
Review record¶
Before attaching this derivative to a model, record approval of: source provenance and completeness; precipitation versus rainfall-excess meaning; units and accumulation convention; source and model clocks; tail duration; grid CRS, resolution, origin, and extent; pathname family; interpolation behavior; loss/infiltration assumptions; simulation window; and independent HEC-RAS execution authorization.
The completed artifact from this notebook is a reviewable DSS forcing derivative, not an approved hydraulic model run.