SCHISM Boundary Conditions
Overview
The SCHISM boundary conditions system provides a unified interface for configuring all types of boundary conditions in SCHISM simulations. This system replaces the previous separate tidal and ocean configurations with a single, flexible approach that supports:
- Harmonic boundaries - Pure harmonic tidal forcing using tidal constituents
- Hybrid boundaries - Combined harmonic and external data forcing
- River boundaries - Constant or time-varying river inputs
- Nested boundaries - Coupling with parent model outputs
- Custom configurations - Flexible mixing of different boundary types
Key Classes
SCHISMDataBoundaryConditions
The main class for configuring boundary conditions. This unified interface handles all boundary types and their associated data sources.
SCHISMDataBoundaryConditions
Bases: RompyBaseModel
This class configures all boundary conditions for SCHISM including tidal,
ocean, river, and nested model boundaries.
It provides a unified interface for specifying boundary conditions and their
data sources, replacing the separate tides and ocean configurations.
Source code in rompy_schism/data.py
| class SCHISMDataBoundaryConditions(RompyBaseModel):
"""
This class configures all boundary conditions for SCHISM including tidal,
ocean, river, and nested model boundaries.
It provides a unified interface for specifying boundary conditions and their
data sources, replacing the separate tides and ocean configurations.
"""
# Allow arbitrary types for schema generation
model_config = ConfigDict(arbitrary_types_allowed=True)
data_type: Literal["boundary_conditions"] = Field(
default="boundary_conditions",
description="Model type discriminator",
)
# Tidal dataset specification
tidal_data: Optional[TidalDataset] = Field(
None,
description="Tidal forcing dataset",
)
# Boundary configurations with integrated data sources
boundaries: Dict[int, BoundarySetupWithSource] = Field(
default_factory=dict,
description="Boundary configuration by boundary index",
)
# Predefined configuration types
setup_type: Optional[Literal["tidal", "hybrid", "river", "nested"]] = Field(
None, description="Predefined boundary setup type"
)
# Hotstart configuration
hotstart_config: Optional[HotstartConfig] = Field(
None, description="Configuration for hotstart file generation"
)
@model_validator(mode="before")
@classmethod
def convert_numpy_types(cls, data):
"""Convert any numpy values to Python native types"""
if not isinstance(data, dict):
return data
for key, value in list(data.items()):
if isinstance(value, (np.bool_, np.integer, np.floating, np.ndarray)):
data[key] = to_python_type(value)
return data
@model_validator(mode="after")
def validate_tidal_data(self):
"""Ensure tidal data is provided when needed for TIDAL or TIDALSPACETIME boundaries."""
boundaries = self.boundaries or {}
needs_tidal_data = False
# Check setup_type first
if self.setup_type in ["tidal", "hybrid"]:
needs_tidal_data = True
# Then check individual boundaries
for setup in boundaries.values():
if (
hasattr(setup, "elev_type")
and setup.elev_type
in [ElevationType.HARMONIC, ElevationType.HARMONICEXTERNAL]
) or (
hasattr(setup, "vel_type")
and setup.vel_type
in [VelocityType.HARMONIC, VelocityType.HARMONICEXTERNAL]
):
needs_tidal_data = True
break
if needs_tidal_data and not self.tidal_data:
raise ValueError(
"Tidal data is required for HARMONIC or HARMONICEXTERNAL boundary types but was not provided"
)
return self
@model_validator(mode="after")
def validate_setup_type(self):
"""Validate setup type specific requirements."""
# Skip validation if setup_type is not set
if not self.setup_type:
return self
if self.setup_type in ["tidal", "hybrid"]:
if not self.tidal_data:
raise ValueError(
"tidal_data is required for tidal or hybrid setup_type"
)
elif self.setup_type == "river":
if self.boundaries:
has_flow = any(
hasattr(s, "const_flow") and s.const_flow is not None
for s in self.boundaries.values()
)
if not has_flow:
raise ValueError(
"At least one boundary should have const_flow for river setup_type"
)
elif self.setup_type == "nested":
if self.boundaries:
for idx, setup in self.boundaries.items():
if (
hasattr(setup, "vel_type")
and setup.vel_type == VelocityType.RELAXED
):
if not hasattr(setup, "inflow_relax") or not hasattr(
setup, "outflow_relax"
):
logger.warning(
f"inflow_relax and outflow_relax are recommended for nested setup_type in boundary {idx}"
)
else:
raise ValueError(
f"Unknown setup_type: {self.setup_type}. Expected one of: tidal, hybrid, river, nested"
)
return self
def _create_boundary_config(self, grid):
"""Create a TidalBoundary object based on the configuration."""
# Get tidal data paths
if self.tidal_data:
if (
hasattr(self.tidal_data, "tidal_database")
and self.tidal_data.tidal_database
):
str(self.tidal_data.tidal_database)
# Ensure boundary information is computed
if hasattr(grid.pylibs_hgrid, "compute_bnd"):
grid.pylibs_hgrid.compute_bnd()
else:
logger.warning(
"Grid object doesn't have compute_bnd method. Boundary information may be missing."
)
# Create a new TidalBoundary with all the configuration
# Ensure boundary information is computed before creating the boundary
if not hasattr(grid.pylibs_hgrid, "nob") or not hasattr(
grid.pylibs_hgrid, "nobn"
):
logger.info("Computing boundary information before creating TidalBoundary")
# First try compute_bnd if available
if hasattr(grid.pylibs_hgrid, "compute_bnd"):
grid.pylibs_hgrid.compute_bnd()
# Then try compute_all if nob is still missing
if not hasattr(grid.pylibs_hgrid, "nob") and hasattr(
grid.pylibs_hgrid, "compute_all"
):
if hasattr(grid.pylibs_hgrid, "compute_all"):
grid.pylibs_hgrid.compute_all()
# Verify boundary attributes are available
if not hasattr(grid.pylibs_hgrid, "nob"):
logger.error("Failed to set 'nob' attribute on grid.pylibs_hgrid")
raise AttributeError(
"Missing required 'nob' attribute on grid.pylibs_hgrid"
)
# Create TidalBoundary with pre-computed grid to avoid losing boundary info
# Get the grid path for TidalBoundary
grid_path = (
str(grid.hgrid.path)
if hasattr(grid, "hgrid") and hasattr(grid.hgrid, "path")
else None
)
if grid_path is None:
# Create a temporary file with the grid if needed
import tempfile
temp_file = tempfile.NamedTemporaryFile(suffix=".gr3", delete=False)
temp_path = temp_file.name
temp_file.close()
grid.pylibs_hgrid.write_hgrid(temp_path)
grid_path = temp_path
boundary = BoundaryHandler(grid_path=grid_path, tidal_data=self.tidal_data)
# Replace the TidalBoundary's grid with our pre-computed one to preserve boundary info
boundary.grid = grid.pylibs_hgrid
# Configure each boundary segment
for idx, setup in self.boundaries.items():
boundary_config = setup.to_boundary_config()
boundary.set_boundary_config(idx, boundary_config)
return boundary
def get(
self,
destdir: str | Path,
grid: SCHISMGrid,
time: TimeRange,
) -> Dict[str, str]:
"""
Process all boundary data and generate necessary input files.
Parameters
----------
destdir : str | Path
Destination directory
grid : SCHISMGrid
SCHISM grid instance
time : TimeRange
Time range for the simulation
Returns
-------
Dict[str, str]
Paths to generated files
"""
# Processing boundary conditions
# Convert destdir to Path object
destdir = Path(destdir)
# Create destdir if it doesn't exist
if not destdir.exists():
logger.info(f"Creating destination directory: {destdir}")
destdir.mkdir(parents=True, exist_ok=True)
# # 1. Process tidal data if needed
if self.tidal_data:
logger.info(
f"{ARROW} Processing tidal constituents: {', '.join(self.tidal_data.constituents) if hasattr(self.tidal_data, 'constituents') else 'default'}"
)
self.tidal_data.get(grid)
# 2. Create boundary condition file (bctides.in)
boundary = self._create_boundary_config(grid)
# Set start time and run duration
start_time = time.start
if time.end is not None and time.start is not None:
run_days = (
time.end - time.start
).total_seconds() / 86400.0 # Convert to days
else:
run_days = 1.0 # Default to 1 day if time is not properly specified
boundary.set_run_parameters(start_time, run_days)
# Generate bctides.in file
bctides_path = destdir / "bctides.in"
logger.info(f"{ARROW} Generating boundary condition file: bctides.in")
# Ensure grid object has complete boundary information before writing
if hasattr(grid.pylibs_hgrid, "compute_all"):
grid.pylibs_hgrid.compute_all()
# Double-check all required attributes are present
required_attrs = ["nob", "nobn", "iobn"]
missing_attrs = [
attr
for attr in required_attrs
if not (grid.pylibs_hgrid and hasattr(grid.pylibs_hgrid, attr))
]
if missing_attrs:
error_msg = (
f"Grid is missing required attributes: {', '.join(missing_attrs)}"
)
logger.error(error_msg)
raise AttributeError(error_msg)
# Write the boundary file - no fallbacks
boundary.write_boundary_file(bctides_path)
logger.info(f"{ARROW} Boundary conditions written successfully")
# 3. Process ocean data based on boundary configurations
processed_files = {"bctides": str(bctides_path)}
# Collect variables to process and source information for logging
variables_to_process = []
source_files = set()
for idx, setup in self.boundaries.items():
if (
setup.elev_type
in [ElevationType.EXTERNAL, ElevationType.HARMONICEXTERNAL]
and setup.elev_source
):
variables_to_process.append("elevation")
if hasattr(setup.elev_source, "source") and hasattr(
setup.elev_source.source, "uri"
):
source_files.add(str(setup.elev_source.source.uri))
if (
setup.vel_type
in [
VelocityType.EXTERNAL,
VelocityType.HARMONICEXTERNAL,
VelocityType.RELAXED,
]
and setup.vel_source
):
variables_to_process.append("velocity")
if hasattr(setup.vel_source, "source") and hasattr(
setup.vel_source.source, "uri"
):
source_files.add(str(setup.vel_source.source.uri))
if setup.temp_type == TracerType.EXTERNAL and setup.temp_source:
variables_to_process.append("temperature")
if hasattr(setup.temp_source, "source") and hasattr(
setup.temp_source.source, "uri"
):
source_files.add(str(setup.temp_source.source.uri))
if setup.salt_type == TracerType.EXTERNAL and setup.salt_source:
variables_to_process.append("salinity")
if hasattr(setup.salt_source, "source") and hasattr(
setup.salt_source.source, "uri"
):
source_files.add(str(setup.salt_source.source.uri))
if variables_to_process:
unique_vars = list(
dict.fromkeys(variables_to_process)
) # Remove duplicates while preserving order
logger.info(f"{ARROW} Processing boundary data: {', '.join(unique_vars)}")
if source_files:
if len(source_files) == 1:
logger.info(f" • Source: {list(source_files)[0]}")
else:
logger.info(f" • Sources: {len(source_files)} files")
# Process each data source based on the boundary type
for idx, setup in self.boundaries.items():
# Process elevation data if needed
if setup.elev_type in [
ElevationType.EXTERNAL,
ElevationType.HARMONICEXTERNAL,
]:
if setup.elev_source:
if (
hasattr(setup.elev_source, "data_type")
and setup.elev_source.data_type == "boundary"
):
# Process using SCHISMDataBoundary interface
setup.elev_source.id = "elev2D" # Set the ID for the boundary
file_path = setup.elev_source.get(destdir, grid, time)
else:
# Process using DataBlob interface
file_path = setup.elev_source.get(str(destdir))
processed_files[f"elev_boundary_{idx}"] = file_path
# Process velocity data if needed
if setup.vel_type in [
VelocityType.EXTERNAL,
VelocityType.HARMONICEXTERNAL,
VelocityType.RELAXED,
]:
if setup.vel_source:
if (
hasattr(setup.vel_source, "data_type")
and setup.vel_source.data_type == "boundary"
):
# Process using SCHISMDataBoundary interface
setup.vel_source.id = "uv3D" # Set the ID for the boundary
file_path = setup.vel_source.get(destdir, grid, time)
else:
# Process using DataBlob interface
file_path = setup.vel_source.get(str(destdir))
processed_files[f"vel_boundary_{idx}"] = file_path
# Process temperature data if needed
if setup.temp_type == TracerType.EXTERNAL:
if setup.temp_source:
if (
hasattr(setup.temp_source, "data_type")
and setup.temp_source.data_type == "boundary"
):
# Process using SCHISMDataBoundary interface
setup.temp_source.id = "TEM_3D" # Set the ID for the boundary
file_path = setup.temp_source.get(destdir, grid, time)
else:
# Process using DataBlob interface
file_path = setup.temp_source.get(str(destdir))
processed_files[f"temp_boundary_{idx}"] = file_path
# Process salinity data if needed
if setup.salt_type == TracerType.EXTERNAL:
if setup.salt_source:
if (
hasattr(setup.salt_source, "data_type")
and setup.salt_source.data_type == "boundary"
):
# Process using SCHISMDataBoundary interface
setup.salt_source.id = "SAL_3D" # Set the ID for the boundary
file_path = setup.salt_source.get(destdir, grid, time)
else:
# Process using DataBlob interface
file_path = setup.salt_source.get(str(destdir))
processed_files[f"salt_boundary_{idx}"] = file_path
# Generate hotstart file if configured
if self.hotstart_config and self.hotstart_config.enabled:
logger.info(f"{ARROW} Generating hotstart file")
hotstart_path = self._generate_hotstart(destdir, grid, time)
processed_files["hotstart"] = hotstart_path
logger.info(f" • Output: {hotstart_path}")
# Log summary of processed files with more details
boundary_data_files = [f for k, f in processed_files.items() if "boundary" in k]
if boundary_data_files:
logger.info(
f" • Files: {', '.join([Path(f).name for f in boundary_data_files])}"
)
return processed_files
def _generate_hotstart(
self,
destdir: Union[str, Path],
grid: SCHISMGrid,
time: Optional[TimeRange] = None,
) -> str:
"""
Generate hotstart file using boundary condition data sources.
Args:
destdir: Destination directory for the hotstart file
grid: SCHISM grid object
time: Time range for the data
Returns:
Path to the generated hotstart file
"""
from rompy_schism.hotstart import SCHISMDataHotstart
# Find a boundary that has both temperature and salinity sources
temp_source = None
salt_source = None
for boundary_config in self.boundaries.values():
if boundary_config.temp_source is not None:
temp_source = boundary_config.temp_source
if boundary_config.salt_source is not None:
salt_source = boundary_config.salt_source
# If we found both, we can proceed
if temp_source is not None and salt_source is not None:
break
if temp_source is None or salt_source is None:
raise ValueError(
"Hotstart generation requires both temperature and salinity sources "
"to be configured in boundary conditions"
)
# Create hotstart instance using the first available source
# (assuming temp and salt sources point to the same dataset)
# Include both temperature and salinity variables for hotstart generation
temp_var_name = (
self.hotstart_config.temp_var if self.hotstart_config else "temperature"
)
salt_var_name = (
self.hotstart_config.salt_var if self.hotstart_config else "salinity"
)
# Log hotstart generation details
logger.info(f" • Variables: {temp_var_name}, {salt_var_name}")
if hasattr(temp_source, "source") and hasattr(temp_source.source, "uri"):
logger.info(f" • Source: {temp_source.source.uri}")
hotstart_data = SCHISMDataHotstart(
source=temp_source.source,
variables=[temp_var_name, salt_var_name],
coords=getattr(temp_source, "coords", None),
temp_var=temp_var_name,
salt_var=salt_var_name,
time_offset=(
self.hotstart_config.time_offset if self.hotstart_config else 0.0
),
time_base=(
self.hotstart_config.time_base
if self.hotstart_config
else datetime(2000, 1, 1)
),
output_filename=(
self.hotstart_config.output_filename
if self.hotstart_config
else "hotstart.nc"
),
)
return hotstart_data.get(str(destdir), grid=grid, time=time)
|
Attributes
model_config
class-attribute
instance-attribute
model_config = ConfigDict(arbitrary_types_allowed=True)
data_type
class-attribute
instance-attribute
data_type: Literal['boundary_conditions'] = Field(default='boundary_conditions', description='Model type discriminator')
tidal_data
class-attribute
instance-attribute
tidal_data: Optional[TidalDataset] = Field(None, description='Tidal forcing dataset')
boundaries
class-attribute
instance-attribute
boundaries: Dict[int, BoundarySetupWithSource] = Field(default_factory=dict, description='Boundary configuration by boundary index')
setup_type
class-attribute
instance-attribute
setup_type: Optional[Literal['tidal', 'hybrid', 'river', 'nested']] = Field(None, description='Predefined boundary setup type')
hotstart_config
class-attribute
instance-attribute
hotstart_config: Optional[HotstartConfig] = Field(None, description='Configuration for hotstart file generation')
Functions
convert_numpy_types
classmethod
convert_numpy_types(data)
Convert any numpy values to Python native types
Source code in rompy_schism/data.py
| @model_validator(mode="before")
@classmethod
def convert_numpy_types(cls, data):
"""Convert any numpy values to Python native types"""
if not isinstance(data, dict):
return data
for key, value in list(data.items()):
if isinstance(value, (np.bool_, np.integer, np.floating, np.ndarray)):
data[key] = to_python_type(value)
return data
|
validate_tidal_data
Ensure tidal data is provided when needed for TIDAL or TIDALSPACETIME boundaries.
Source code in rompy_schism/data.py
| @model_validator(mode="after")
def validate_tidal_data(self):
"""Ensure tidal data is provided when needed for TIDAL or TIDALSPACETIME boundaries."""
boundaries = self.boundaries or {}
needs_tidal_data = False
# Check setup_type first
if self.setup_type in ["tidal", "hybrid"]:
needs_tidal_data = True
# Then check individual boundaries
for setup in boundaries.values():
if (
hasattr(setup, "elev_type")
and setup.elev_type
in [ElevationType.HARMONIC, ElevationType.HARMONICEXTERNAL]
) or (
hasattr(setup, "vel_type")
and setup.vel_type
in [VelocityType.HARMONIC, VelocityType.HARMONICEXTERNAL]
):
needs_tidal_data = True
break
if needs_tidal_data and not self.tidal_data:
raise ValueError(
"Tidal data is required for HARMONIC or HARMONICEXTERNAL boundary types but was not provided"
)
return self
|
validate_setup_type
Validate setup type specific requirements.
Source code in rompy_schism/data.py
| @model_validator(mode="after")
def validate_setup_type(self):
"""Validate setup type specific requirements."""
# Skip validation if setup_type is not set
if not self.setup_type:
return self
if self.setup_type in ["tidal", "hybrid"]:
if not self.tidal_data:
raise ValueError(
"tidal_data is required for tidal or hybrid setup_type"
)
elif self.setup_type == "river":
if self.boundaries:
has_flow = any(
hasattr(s, "const_flow") and s.const_flow is not None
for s in self.boundaries.values()
)
if not has_flow:
raise ValueError(
"At least one boundary should have const_flow for river setup_type"
)
elif self.setup_type == "nested":
if self.boundaries:
for idx, setup in self.boundaries.items():
if (
hasattr(setup, "vel_type")
and setup.vel_type == VelocityType.RELAXED
):
if not hasattr(setup, "inflow_relax") or not hasattr(
setup, "outflow_relax"
):
logger.warning(
f"inflow_relax and outflow_relax are recommended for nested setup_type in boundary {idx}"
)
else:
raise ValueError(
f"Unknown setup_type: {self.setup_type}. Expected one of: tidal, hybrid, river, nested"
)
return self
|
get
get(destdir: str | Path, grid: SCHISMGrid, time: TimeRange) -> Dict[str, str]
Process all boundary data and generate necessary input files.
Parameters
destdir : str | Path
Destination directory
grid : SCHISMGrid
SCHISM grid instance
time : TimeRange
Time range for the simulation
Returns
Dict[str, str]
Paths to generated files
Source code in rompy_schism/data.py
| def get(
self,
destdir: str | Path,
grid: SCHISMGrid,
time: TimeRange,
) -> Dict[str, str]:
"""
Process all boundary data and generate necessary input files.
Parameters
----------
destdir : str | Path
Destination directory
grid : SCHISMGrid
SCHISM grid instance
time : TimeRange
Time range for the simulation
Returns
-------
Dict[str, str]
Paths to generated files
"""
# Processing boundary conditions
# Convert destdir to Path object
destdir = Path(destdir)
# Create destdir if it doesn't exist
if not destdir.exists():
logger.info(f"Creating destination directory: {destdir}")
destdir.mkdir(parents=True, exist_ok=True)
# # 1. Process tidal data if needed
if self.tidal_data:
logger.info(
f"{ARROW} Processing tidal constituents: {', '.join(self.tidal_data.constituents) if hasattr(self.tidal_data, 'constituents') else 'default'}"
)
self.tidal_data.get(grid)
# 2. Create boundary condition file (bctides.in)
boundary = self._create_boundary_config(grid)
# Set start time and run duration
start_time = time.start
if time.end is not None and time.start is not None:
run_days = (
time.end - time.start
).total_seconds() / 86400.0 # Convert to days
else:
run_days = 1.0 # Default to 1 day if time is not properly specified
boundary.set_run_parameters(start_time, run_days)
# Generate bctides.in file
bctides_path = destdir / "bctides.in"
logger.info(f"{ARROW} Generating boundary condition file: bctides.in")
# Ensure grid object has complete boundary information before writing
if hasattr(grid.pylibs_hgrid, "compute_all"):
grid.pylibs_hgrid.compute_all()
# Double-check all required attributes are present
required_attrs = ["nob", "nobn", "iobn"]
missing_attrs = [
attr
for attr in required_attrs
if not (grid.pylibs_hgrid and hasattr(grid.pylibs_hgrid, attr))
]
if missing_attrs:
error_msg = (
f"Grid is missing required attributes: {', '.join(missing_attrs)}"
)
logger.error(error_msg)
raise AttributeError(error_msg)
# Write the boundary file - no fallbacks
boundary.write_boundary_file(bctides_path)
logger.info(f"{ARROW} Boundary conditions written successfully")
# 3. Process ocean data based on boundary configurations
processed_files = {"bctides": str(bctides_path)}
# Collect variables to process and source information for logging
variables_to_process = []
source_files = set()
for idx, setup in self.boundaries.items():
if (
setup.elev_type
in [ElevationType.EXTERNAL, ElevationType.HARMONICEXTERNAL]
and setup.elev_source
):
variables_to_process.append("elevation")
if hasattr(setup.elev_source, "source") and hasattr(
setup.elev_source.source, "uri"
):
source_files.add(str(setup.elev_source.source.uri))
if (
setup.vel_type
in [
VelocityType.EXTERNAL,
VelocityType.HARMONICEXTERNAL,
VelocityType.RELAXED,
]
and setup.vel_source
):
variables_to_process.append("velocity")
if hasattr(setup.vel_source, "source") and hasattr(
setup.vel_source.source, "uri"
):
source_files.add(str(setup.vel_source.source.uri))
if setup.temp_type == TracerType.EXTERNAL and setup.temp_source:
variables_to_process.append("temperature")
if hasattr(setup.temp_source, "source") and hasattr(
setup.temp_source.source, "uri"
):
source_files.add(str(setup.temp_source.source.uri))
if setup.salt_type == TracerType.EXTERNAL and setup.salt_source:
variables_to_process.append("salinity")
if hasattr(setup.salt_source, "source") and hasattr(
setup.salt_source.source, "uri"
):
source_files.add(str(setup.salt_source.source.uri))
if variables_to_process:
unique_vars = list(
dict.fromkeys(variables_to_process)
) # Remove duplicates while preserving order
logger.info(f"{ARROW} Processing boundary data: {', '.join(unique_vars)}")
if source_files:
if len(source_files) == 1:
logger.info(f" • Source: {list(source_files)[0]}")
else:
logger.info(f" • Sources: {len(source_files)} files")
# Process each data source based on the boundary type
for idx, setup in self.boundaries.items():
# Process elevation data if needed
if setup.elev_type in [
ElevationType.EXTERNAL,
ElevationType.HARMONICEXTERNAL,
]:
if setup.elev_source:
if (
hasattr(setup.elev_source, "data_type")
and setup.elev_source.data_type == "boundary"
):
# Process using SCHISMDataBoundary interface
setup.elev_source.id = "elev2D" # Set the ID for the boundary
file_path = setup.elev_source.get(destdir, grid, time)
else:
# Process using DataBlob interface
file_path = setup.elev_source.get(str(destdir))
processed_files[f"elev_boundary_{idx}"] = file_path
# Process velocity data if needed
if setup.vel_type in [
VelocityType.EXTERNAL,
VelocityType.HARMONICEXTERNAL,
VelocityType.RELAXED,
]:
if setup.vel_source:
if (
hasattr(setup.vel_source, "data_type")
and setup.vel_source.data_type == "boundary"
):
# Process using SCHISMDataBoundary interface
setup.vel_source.id = "uv3D" # Set the ID for the boundary
file_path = setup.vel_source.get(destdir, grid, time)
else:
# Process using DataBlob interface
file_path = setup.vel_source.get(str(destdir))
processed_files[f"vel_boundary_{idx}"] = file_path
# Process temperature data if needed
if setup.temp_type == TracerType.EXTERNAL:
if setup.temp_source:
if (
hasattr(setup.temp_source, "data_type")
and setup.temp_source.data_type == "boundary"
):
# Process using SCHISMDataBoundary interface
setup.temp_source.id = "TEM_3D" # Set the ID for the boundary
file_path = setup.temp_source.get(destdir, grid, time)
else:
# Process using DataBlob interface
file_path = setup.temp_source.get(str(destdir))
processed_files[f"temp_boundary_{idx}"] = file_path
# Process salinity data if needed
if setup.salt_type == TracerType.EXTERNAL:
if setup.salt_source:
if (
hasattr(setup.salt_source, "data_type")
and setup.salt_source.data_type == "boundary"
):
# Process using SCHISMDataBoundary interface
setup.salt_source.id = "SAL_3D" # Set the ID for the boundary
file_path = setup.salt_source.get(destdir, grid, time)
else:
# Process using DataBlob interface
file_path = setup.salt_source.get(str(destdir))
processed_files[f"salt_boundary_{idx}"] = file_path
# Generate hotstart file if configured
if self.hotstart_config and self.hotstart_config.enabled:
logger.info(f"{ARROW} Generating hotstart file")
hotstart_path = self._generate_hotstart(destdir, grid, time)
processed_files["hotstart"] = hotstart_path
logger.info(f" • Output: {hotstart_path}")
# Log summary of processed files with more details
boundary_data_files = [f for k, f in processed_files.items() if "boundary" in k]
if boundary_data_files:
logger.info(
f" • Files: {', '.join([Path(f).name for f in boundary_data_files])}"
)
return processed_files
|
BoundarySetupWithSource
Configures individual boundary segments with their data sources and boundary condition types.
BoundarySetupWithSource
Bases: BoundarySetup
Enhanced boundary setup that includes data sources.
This class extends BoundarySetup to provide a unified configuration
for both boundary conditions and their data sources.
Source code in rompy_schism/data.py
| class BoundarySetupWithSource(BoundarySetup):
"""
Enhanced boundary setup that includes data sources.
This class extends BoundarySetup to provide a unified configuration
for both boundary conditions and their data sources.
"""
elev_source: Optional[Union[DataBlob, DataGrid, SCHISMDataBoundary]] = Field(
None, description="Data source for elevation boundary condition"
)
vel_source: Optional[Union[DataBlob, DataGrid, SCHISMDataBoundary]] = Field(
None, description="Data source for velocity boundary condition"
)
temp_source: Optional[Union[DataBlob, DataGrid, SCHISMDataBoundary]] = Field(
None, description="Data source for temperature boundary condition"
)
salt_source: Optional[Union[DataBlob, DataGrid, SCHISMDataBoundary]] = Field(
None, description="Data source for salinity boundary condition"
)
@model_validator(mode="after")
def validate_data_sources(self):
"""Ensure data sources are provided when needed for space-time boundary types."""
# Check elevation data source
if (
self.elev_type in [ElevationType.EXTERNAL, ElevationType.HARMONICEXTERNAL]
and self.elev_source is None
):
logger.warning(
"elev_source should be provided for EXTERNAL or HARMONICEXTERNAL elevation type"
)
# Check velocity data source
if (
self.vel_type
in [
VelocityType.EXTERNAL,
VelocityType.HARMONICEXTERNAL,
VelocityType.RELAXED,
]
and self.vel_source is None
):
logger.warning(
"vel_source should be provided for EXTERNAL, HARMONICEXTERNAL, or RELAXED velocity type"
)
# Check temperature data source
if self.temp_type == TracerType.EXTERNAL and self.temp_source is None:
logger.warning(
"temp_source should be provided for EXTERNAL temperature type"
)
# Check salinity data source
if self.salt_type == TracerType.EXTERNAL and self.salt_source is None:
logger.warning("salt_source should be provided for EXTERNAL salinity type")
return self
|
Attributes
elev_source
class-attribute
instance-attribute
elev_source: Optional[Union[DataBlob, DataGrid, SCHISMDataBoundary]] = Field(None, description='Data source for elevation boundary condition')
vel_source
class-attribute
instance-attribute
vel_source: Optional[Union[DataBlob, DataGrid, SCHISMDataBoundary]] = Field(None, description='Data source for velocity boundary condition')
temp_source
class-attribute
instance-attribute
temp_source: Optional[Union[DataBlob, DataGrid, SCHISMDataBoundary]] = Field(None, description='Data source for temperature boundary condition')
salt_source
class-attribute
instance-attribute
salt_source: Optional[Union[DataBlob, DataGrid, SCHISMDataBoundary]] = Field(None, description='Data source for salinity boundary condition')
Functions
validate_data_sources
Ensure data sources are provided when needed for space-time boundary types.
Source code in rompy_schism/data.py
| @model_validator(mode="after")
def validate_data_sources(self):
"""Ensure data sources are provided when needed for space-time boundary types."""
# Check elevation data source
if (
self.elev_type in [ElevationType.EXTERNAL, ElevationType.HARMONICEXTERNAL]
and self.elev_source is None
):
logger.warning(
"elev_source should be provided for EXTERNAL or HARMONICEXTERNAL elevation type"
)
# Check velocity data source
if (
self.vel_type
in [
VelocityType.EXTERNAL,
VelocityType.HARMONICEXTERNAL,
VelocityType.RELAXED,
]
and self.vel_source is None
):
logger.warning(
"vel_source should be provided for EXTERNAL, HARMONICEXTERNAL, or RELAXED velocity type"
)
# Check temperature data source
if self.temp_type == TracerType.EXTERNAL and self.temp_source is None:
logger.warning(
"temp_source should be provided for EXTERNAL temperature type"
)
# Check salinity data source
if self.salt_type == TracerType.EXTERNAL and self.salt_source is None:
logger.warning("salt_source should be provided for EXTERNAL salinity type")
return self
|
BoundaryHandler
Core boundary handler that extends BoundaryData and supports all SCHISM boundary types.
BoundaryHandler
Bases: BoundaryData
Handler for SCHISM boundary conditions.
This class extends BoundaryData to handle all SCHISM boundary condition types
including tidal, river, nested, and hybrid configurations.
Source code in rompy_schism/boundary_core.py
| class BoundaryHandler(BoundaryData):
"""Handler for SCHISM boundary conditions.
This class extends BoundaryData to handle all SCHISM boundary condition types
including tidal, river, nested, and hybrid configurations.
"""
def __init__(
self,
grid_path: Union[str, Path],
tidal_data: Optional[TidalDataset] = None,
boundary_configs: Optional[Dict[int, BoundaryConfig]] = None,
*args,
**kwargs,
):
"""Initialize the boundary handler.
Parameters
----------
grid_path : str or Path
Path to the SCHISM grid file
tidal_data : TidalDataset, optional
Tidal dataset containing specification of tidal forcing
boundary_configs : dict, optional
Configuration for each boundary, keyed by boundary index
"""
super().__init__(grid_path, *args, **kwargs)
self.tidal_data = tidal_data
self.boundary_configs = boundary_configs if boundary_configs is not None else {}
# For storing start time and run duration
self._start_time = None
self._rnday = None
# Additional file paths for various boundary types
self.temp_th_path = None # Temperature time history
self.temp_3d_path = None # 3D temperature
self.salt_th_path = None # Salinity time history
self.salt_3d_path = None # 3D salinity
self.flow_th_path = None # Flow time history
self.elev_st_path = None # Space-time elevation
self.vel_st_path = None # Space-time velocity
def set_boundary_config(self, boundary_index: int, config: BoundaryConfig):
"""Set configuration for a specific boundary.
Parameters
----------
boundary_index : int
Index of the boundary
config : BoundaryConfig
Configuration for the boundary
"""
self.boundary_configs[boundary_index] = config
def set_boundary_type(
self,
boundary_index: int,
elev_type: ElevationType,
vel_type: VelocityType,
temp_type: TracerType = TracerType.NONE,
salt_type: TracerType = TracerType.NONE,
**kwargs,
):
"""Set boundary types for a specific boundary.
Parameters
----------
boundary_index : int
Index of the boundary
elev_type : ElevationType
Elevation boundary condition type
vel_type : VelocityType
Velocity boundary condition type
temp_type : TracerType, optional
Temperature boundary condition type
salt_type : TracerType, optional
Salinity boundary condition type
**kwargs
Additional parameters for the boundary configuration
"""
config = BoundaryConfig(
elev_type=elev_type,
vel_type=vel_type,
temp_type=temp_type,
salt_type=salt_type,
**kwargs,
)
self.set_boundary_config(boundary_index, config)
def set_run_parameters(self, start_time, run_days):
"""Set start time and run duration.
Parameters
----------
start_time : datetime or list
Start time for the simulation
run_days : float
Duration of the simulation in days
"""
self._start_time = start_time
self._rnday = run_days
def get_flags_list(self) -> List[List[int]]:
"""Get list of boundary flags for Bctides.
Returns
-------
list of list of int
Boundary flags for each boundary
"""
if not self.boundary_configs:
return [[5, 5, 0, 0]] # Default to tidal
# Find max boundary without using default parameter
if self.boundary_configs:
# Convert keys to list and find max
boundary_keys = list(self.boundary_configs.keys())
max_boundary = max(boundary_keys) if boundary_keys else -1
else:
max_boundary = -1
flags = []
for i in range(int(max_boundary) + 1):
if i in self.boundary_configs:
config = self.boundary_configs[i]
flags.append(
[
int(config.elev_type),
int(config.vel_type),
int(config.temp_type),
int(config.salt_type),
]
)
else:
flags.append([0, 0, 0, 0])
return flags
def get_constant_values(self) -> Dict[str, List[float]]:
"""Get constant values for boundaries.
Returns
-------
dict
Dictionary of constant values for each boundary type
"""
result = {
"ethconst": [],
"vthconst": [],
"tthconst": [],
"sthconst": [],
"tobc": [],
"sobc": [],
"inflow_relax": [],
"outflow_relax": [],
"eta_mean": [],
"vn_mean": [],
"temp_th_path": [],
"temp_3d_path": [],
"salt_th_path": [],
"salt_3d_path": [],
"flow_th_path": [],
"elev_st_path": [],
"vel_st_path": [],
}
if not self.boundary_configs:
return result
# Find max boundary without using default parameter
if self.boundary_configs:
# Convert keys to list and find max
boundary_keys = list(self.boundary_configs.keys())
max_boundary = max(boundary_keys) if boundary_keys else -1
else:
max_boundary = -1
for i in range(int(max_boundary) + 1):
if i in self.boundary_configs:
config = self.boundary_configs[i]
# Handle type 2 (constant) boundaries
if config.elev_type == ElevationType.CONSTANT:
result["ethconst"].append(
config.ethconst if config.ethconst is not None else 0.0
)
else:
result["ethconst"].append(0.0)
if config.vel_type == VelocityType.CONSTANT:
result["vthconst"].append(
config.vthconst if config.vthconst is not None else 0.0
)
else:
result["vthconst"].append(0.0)
if config.temp_type == TracerType.CONSTANT:
result["tthconst"].append(
config.tthconst if config.tthconst is not None else 0.0
)
else:
result["tthconst"].append(0.0)
if config.salt_type == TracerType.CONSTANT:
result["sthconst"].append(
config.sthconst if config.sthconst is not None else 0.0
)
else:
result["sthconst"].append(0.0)
# Nudging factors for temperature and salinity
result["tobc"].append(config.tobc if config.tobc is not None else 1.0)
result["sobc"].append(config.sobc if config.sobc is not None else 1.0)
# Temperature and salinity file paths
result["temp_th_path"].append(config.temp_th_path)
result["temp_3d_path"].append(config.temp_3d_path)
result["salt_th_path"].append(config.salt_th_path)
result["salt_3d_path"].append(config.salt_3d_path)
# Flow time history path
result["flow_th_path"].append(config.flow_th_path)
# Space-time file paths
result["elev_st_path"].append(config.elev_st_path)
result["vel_st_path"].append(config.vel_st_path)
# Relaxation factors for velocity
if config.vel_type == VelocityType.RELAXED:
result["inflow_relax"].append(
config.inflow_relax if config.inflow_relax is not None else 0.5
)
result["outflow_relax"].append(
config.outflow_relax
if config.outflow_relax is not None
else 0.1
)
else:
result["inflow_relax"].append(0.5) # Default values
result["outflow_relax"].append(0.1)
# Handle Flather boundaries
if config.vel_type == VelocityType.FLATHER:
# Create default values if none provided
if config.eta_mean is None:
# For testing, create a simple array of zeros with size = num nodes on this boundary
# In practice, this should be filled with actual mean elevation values
num_nodes = (
self.grid.nobn[i]
if hasattr(self.grid, "nobn") and i < len(self.grid.nobn)
else 1
)
eta_mean = [0.0] * num_nodes
else:
eta_mean = config.eta_mean
if config.vn_mean is None:
# For testing, create a simple array of arrays with zeros
num_nodes = (
self.grid.nobn[i]
if hasattr(self.grid, "nobn") and i < len(self.grid.nobn)
else 1
)
# Assume 5 vertical levels for testing
vn_mean = [[0.0] * 5 for _ in range(num_nodes)]
else:
vn_mean = config.vn_mean
result["eta_mean"].append(eta_mean)
result["vn_mean"].append(vn_mean)
else:
result["eta_mean"].append(None)
result["vn_mean"].append(None)
else:
# Default values for missing boundaries
result["ethconst"].append(0.0)
result["vthconst"].append(0.0)
result["tthconst"].append(0.0)
result["sthconst"].append(0.0)
result["tobc"].append(1.0)
result["sobc"].append(1.0)
result["inflow_relax"].append(0.5)
result["outflow_relax"].append(0.1)
result["eta_mean"].append(None)
result["vn_mean"].append(None)
result["temp_th_path"].append(None)
result["temp_3d_path"].append(None)
result["salt_th_path"].append(None)
result["salt_3d_path"].append(None)
result["flow_th_path"].append(None)
result["elev_st_path"].append(None)
result["vel_st_path"].append(None)
return result
def create_bctides(self) -> Bctides:
"""Create a Bctides instance from this configuration.
Returns
-------
Bctides
Configured Bctides instance
"""
flags = self.get_flags_list()
constants = self.get_constant_values()
# Clean up lists to avoid None values
ethconst = constants["ethconst"] if constants["ethconst"] else None
vthconst = constants["vthconst"] if constants["vthconst"] else None
tthconst = constants["tthconst"] if constants["tthconst"] else None
sthconst = constants["sthconst"] if constants["sthconst"] else None
tobc = constants["tobc"] if constants["tobc"] else None
sobc = constants["sobc"] if constants["sobc"] else None
inflow_relax = constants["inflow_relax"] if constants["inflow_relax"] else None
outflow_relax = (
constants["outflow_relax"] if constants["outflow_relax"] else None
)
# Add flow and flux boundary information
ncbn = 0
nfluxf = 0
# Count the number of flow and flux boundaries
for i, config in self.boundary_configs.items():
# Count flow boundaries - both CONSTANT type with non-zero flow value
# and type 1 (time history) are considered flow boundaries
if (
config.vel_type == VelocityType.CONSTANT and config.vthconst is not None
) or (config.vel_type == VelocityType.TIMEHIST):
ncbn += 1
# Count flux boundaries - type 3 with flux specified
if config.vel_type == VelocityType.HARMONIC:
nfluxf += 1
# Extract file paths
temp_th_path = (
constants.get("temp_th_path", [None])[0]
if constants.get("temp_th_path")
else None
)
temp_3d_path = (
constants.get("temp_3d_path", [None])[0]
if constants.get("temp_3d_path")
else None
)
salt_th_path = (
constants.get("salt_th_path", [None])[0]
if constants.get("salt_th_path")
else None
)
salt_3d_path = (
constants.get("salt_3d_path", [None])[0]
if constants.get("salt_3d_path")
else None
)
flow_th_path = (
constants.get("flow_th_path", [None])[0]
if constants.get("flow_th_path")
else None
)
elev_st_path = (
constants.get("elev_st_path", [None])[0]
if constants.get("elev_st_path")
else None
)
vel_st_path = (
constants.get("vel_st_path", [None])[0]
if constants.get("vel_st_path")
else None
)
# Extract Flather boundary data if available
eta_mean = (
constants.get("eta_mean", [None]) if constants.get("eta_mean") else None
)
vn_mean = constants.get("vn_mean", [None]) if constants.get("vn_mean") else None
# Ensure grid boundaries are computed before creating Bctides
if self.grid is not None:
if hasattr(self.grid, "compute_bnd") and not hasattr(self.grid, "nob"):
logger.info("Computing grid boundaries for Bctides")
self.grid.compute_bnd()
elif not hasattr(self.grid, "nob") and hasattr(self.grid, "compute_all"):
logger.info(
"Running compute_all to ensure grid boundaries are available"
)
self.grid.compute_all()
# Verify boundaries were computed
if not hasattr(self.grid, "nob"):
logger.error(
"Failed to compute grid boundaries - grid has no 'nob' attribute"
)
raise AttributeError("Grid boundaries could not be computed")
# Create Bctides object with all the enhanced parameters
bctides = Bctides(
hgrid=self.grid,
flags=flags,
constituents=self.tidal_data.constituents,
tidal_database=self.tidal_data.tidal_database,
tidal_model=self.tidal_data.tidal_model,
tidal_potential=self.tidal_data.tidal_potential,
cutoff_depth=self.tidal_data.cutoff_depth,
nodal_corrections=self.tidal_data.nodal_corrections,
tide_interpolation_method=self.tidal_data.tide_interpolation_method,
extrapolate_tides=self.tidal_data.extrapolate_tides,
extrapolation_distance=self.tidal_data.extrapolation_distance,
extra_databases=self.tidal_data.extra_databases,
mdt=getattr(
self.tidal_data, "_mdt", self.tidal_data.mean_dynamic_topography
),
ethconst=ethconst,
vthconst=vthconst,
tthconst=tthconst,
sthconst=sthconst,
tobc=tobc,
sobc=sobc,
relax=constants.get("inflow_relax", []), # For backward compatibility
inflow_relax=inflow_relax,
outflow_relax=outflow_relax,
ncbn=ncbn,
nfluxf=nfluxf,
elev_th_path=None, # Time history of elevation is not handled by this path yet
elev_st_path=elev_st_path,
flow_th_path=flow_th_path,
vel_st_path=vel_st_path,
temp_th_path=temp_th_path,
temp_3d_path=temp_3d_path,
salt_th_path=salt_th_path,
salt_3d_path=salt_3d_path,
)
# Set additional properties for Flather boundaries
if eta_mean and any(x is not None for x in eta_mean):
bctides.eta_mean = eta_mean
if vn_mean and any(x is not None for x in vn_mean):
bctides.vn_mean = vn_mean
# Set start time and run duration
if self._start_time and self._rnday is not None:
bctides._start_time = self._start_time
bctides._rnday = self._rnday
return bctides
def write_boundary_file(self, output_path: Union[str, Path]) -> Path:
"""Write the bctides.in file.
Parameters
----------
output_path : str or Path
Path to write the file
Returns
-------
Path
Path to the written file
Raises
------
ValueError
If start_time and rnday are not set
"""
if not self._start_time or self._rnday is None:
raise ValueError(
"start_time and rnday must be set before writing boundary file"
)
# Create Bctides object
bctides = self.create_bctides()
# Write file
output_path = Path(output_path)
bctides.write_bctides(output_path)
return output_path
|
Attributes
tidal_data
instance-attribute
boundary_configs
instance-attribute
boundary_configs = boundary_configs if boundary_configs is not None else {}
temp_th_path
instance-attribute
temp_3d_path
instance-attribute
salt_th_path
instance-attribute
salt_3d_path
instance-attribute
flow_th_path
instance-attribute
elev_st_path
instance-attribute
vel_st_path
instance-attribute
Functions
set_boundary_config
Set configuration for a specific boundary.
Parameters
boundary_index : int
Index of the boundary
config : BoundaryConfig
Configuration for the boundary
Source code in rompy_schism/boundary_core.py
| def set_boundary_config(self, boundary_index: int, config: BoundaryConfig):
"""Set configuration for a specific boundary.
Parameters
----------
boundary_index : int
Index of the boundary
config : BoundaryConfig
Configuration for the boundary
"""
self.boundary_configs[boundary_index] = config
|
set_boundary_type
Set boundary types for a specific boundary.
Parameters
boundary_index : int
Index of the boundary
elev_type : ElevationType
Elevation boundary condition type
vel_type : VelocityType
Velocity boundary condition type
temp_type : TracerType, optional
Temperature boundary condition type
salt_type : TracerType, optional
Salinity boundary condition type
**kwargs
Additional parameters for the boundary configuration
Source code in rompy_schism/boundary_core.py
| def set_boundary_type(
self,
boundary_index: int,
elev_type: ElevationType,
vel_type: VelocityType,
temp_type: TracerType = TracerType.NONE,
salt_type: TracerType = TracerType.NONE,
**kwargs,
):
"""Set boundary types for a specific boundary.
Parameters
----------
boundary_index : int
Index of the boundary
elev_type : ElevationType
Elevation boundary condition type
vel_type : VelocityType
Velocity boundary condition type
temp_type : TracerType, optional
Temperature boundary condition type
salt_type : TracerType, optional
Salinity boundary condition type
**kwargs
Additional parameters for the boundary configuration
"""
config = BoundaryConfig(
elev_type=elev_type,
vel_type=vel_type,
temp_type=temp_type,
salt_type=salt_type,
**kwargs,
)
self.set_boundary_config(boundary_index, config)
|
set_run_parameters
set_run_parameters(start_time, run_days)
Set start time and run duration.
Parameters
start_time : datetime or list
Start time for the simulation
run_days : float
Duration of the simulation in days
Source code in rompy_schism/boundary_core.py
| def set_run_parameters(self, start_time, run_days):
"""Set start time and run duration.
Parameters
----------
start_time : datetime or list
Start time for the simulation
run_days : float
Duration of the simulation in days
"""
self._start_time = start_time
self._rnday = run_days
|
get_flags_list
get_flags_list() -> List[List[int]]
Get list of boundary flags for Bctides.
Returns
list of list of int
Boundary flags for each boundary
Source code in rompy_schism/boundary_core.py
| def get_flags_list(self) -> List[List[int]]:
"""Get list of boundary flags for Bctides.
Returns
-------
list of list of int
Boundary flags for each boundary
"""
if not self.boundary_configs:
return [[5, 5, 0, 0]] # Default to tidal
# Find max boundary without using default parameter
if self.boundary_configs:
# Convert keys to list and find max
boundary_keys = list(self.boundary_configs.keys())
max_boundary = max(boundary_keys) if boundary_keys else -1
else:
max_boundary = -1
flags = []
for i in range(int(max_boundary) + 1):
if i in self.boundary_configs:
config = self.boundary_configs[i]
flags.append(
[
int(config.elev_type),
int(config.vel_type),
int(config.temp_type),
int(config.salt_type),
]
)
else:
flags.append([0, 0, 0, 0])
return flags
|
get_constant_values
get_constant_values() -> Dict[str, List[float]]
Get constant values for boundaries.
Returns
dict
Dictionary of constant values for each boundary type
Source code in rompy_schism/boundary_core.py
| def get_constant_values(self) -> Dict[str, List[float]]:
"""Get constant values for boundaries.
Returns
-------
dict
Dictionary of constant values for each boundary type
"""
result = {
"ethconst": [],
"vthconst": [],
"tthconst": [],
"sthconst": [],
"tobc": [],
"sobc": [],
"inflow_relax": [],
"outflow_relax": [],
"eta_mean": [],
"vn_mean": [],
"temp_th_path": [],
"temp_3d_path": [],
"salt_th_path": [],
"salt_3d_path": [],
"flow_th_path": [],
"elev_st_path": [],
"vel_st_path": [],
}
if not self.boundary_configs:
return result
# Find max boundary without using default parameter
if self.boundary_configs:
# Convert keys to list and find max
boundary_keys = list(self.boundary_configs.keys())
max_boundary = max(boundary_keys) if boundary_keys else -1
else:
max_boundary = -1
for i in range(int(max_boundary) + 1):
if i in self.boundary_configs:
config = self.boundary_configs[i]
# Handle type 2 (constant) boundaries
if config.elev_type == ElevationType.CONSTANT:
result["ethconst"].append(
config.ethconst if config.ethconst is not None else 0.0
)
else:
result["ethconst"].append(0.0)
if config.vel_type == VelocityType.CONSTANT:
result["vthconst"].append(
config.vthconst if config.vthconst is not None else 0.0
)
else:
result["vthconst"].append(0.0)
if config.temp_type == TracerType.CONSTANT:
result["tthconst"].append(
config.tthconst if config.tthconst is not None else 0.0
)
else:
result["tthconst"].append(0.0)
if config.salt_type == TracerType.CONSTANT:
result["sthconst"].append(
config.sthconst if config.sthconst is not None else 0.0
)
else:
result["sthconst"].append(0.0)
# Nudging factors for temperature and salinity
result["tobc"].append(config.tobc if config.tobc is not None else 1.0)
result["sobc"].append(config.sobc if config.sobc is not None else 1.0)
# Temperature and salinity file paths
result["temp_th_path"].append(config.temp_th_path)
result["temp_3d_path"].append(config.temp_3d_path)
result["salt_th_path"].append(config.salt_th_path)
result["salt_3d_path"].append(config.salt_3d_path)
# Flow time history path
result["flow_th_path"].append(config.flow_th_path)
# Space-time file paths
result["elev_st_path"].append(config.elev_st_path)
result["vel_st_path"].append(config.vel_st_path)
# Relaxation factors for velocity
if config.vel_type == VelocityType.RELAXED:
result["inflow_relax"].append(
config.inflow_relax if config.inflow_relax is not None else 0.5
)
result["outflow_relax"].append(
config.outflow_relax
if config.outflow_relax is not None
else 0.1
)
else:
result["inflow_relax"].append(0.5) # Default values
result["outflow_relax"].append(0.1)
# Handle Flather boundaries
if config.vel_type == VelocityType.FLATHER:
# Create default values if none provided
if config.eta_mean is None:
# For testing, create a simple array of zeros with size = num nodes on this boundary
# In practice, this should be filled with actual mean elevation values
num_nodes = (
self.grid.nobn[i]
if hasattr(self.grid, "nobn") and i < len(self.grid.nobn)
else 1
)
eta_mean = [0.0] * num_nodes
else:
eta_mean = config.eta_mean
if config.vn_mean is None:
# For testing, create a simple array of arrays with zeros
num_nodes = (
self.grid.nobn[i]
if hasattr(self.grid, "nobn") and i < len(self.grid.nobn)
else 1
)
# Assume 5 vertical levels for testing
vn_mean = [[0.0] * 5 for _ in range(num_nodes)]
else:
vn_mean = config.vn_mean
result["eta_mean"].append(eta_mean)
result["vn_mean"].append(vn_mean)
else:
result["eta_mean"].append(None)
result["vn_mean"].append(None)
else:
# Default values for missing boundaries
result["ethconst"].append(0.0)
result["vthconst"].append(0.0)
result["tthconst"].append(0.0)
result["sthconst"].append(0.0)
result["tobc"].append(1.0)
result["sobc"].append(1.0)
result["inflow_relax"].append(0.5)
result["outflow_relax"].append(0.1)
result["eta_mean"].append(None)
result["vn_mean"].append(None)
result["temp_th_path"].append(None)
result["temp_3d_path"].append(None)
result["salt_th_path"].append(None)
result["salt_3d_path"].append(None)
result["flow_th_path"].append(None)
result["elev_st_path"].append(None)
result["vel_st_path"].append(None)
return result
|
create_bctides
create_bctides() -> Bctides
Create a Bctides instance from this configuration.
Returns
Bctides
Configured Bctides instance
Source code in rompy_schism/boundary_core.py
| def create_bctides(self) -> Bctides:
"""Create a Bctides instance from this configuration.
Returns
-------
Bctides
Configured Bctides instance
"""
flags = self.get_flags_list()
constants = self.get_constant_values()
# Clean up lists to avoid None values
ethconst = constants["ethconst"] if constants["ethconst"] else None
vthconst = constants["vthconst"] if constants["vthconst"] else None
tthconst = constants["tthconst"] if constants["tthconst"] else None
sthconst = constants["sthconst"] if constants["sthconst"] else None
tobc = constants["tobc"] if constants["tobc"] else None
sobc = constants["sobc"] if constants["sobc"] else None
inflow_relax = constants["inflow_relax"] if constants["inflow_relax"] else None
outflow_relax = (
constants["outflow_relax"] if constants["outflow_relax"] else None
)
# Add flow and flux boundary information
ncbn = 0
nfluxf = 0
# Count the number of flow and flux boundaries
for i, config in self.boundary_configs.items():
# Count flow boundaries - both CONSTANT type with non-zero flow value
# and type 1 (time history) are considered flow boundaries
if (
config.vel_type == VelocityType.CONSTANT and config.vthconst is not None
) or (config.vel_type == VelocityType.TIMEHIST):
ncbn += 1
# Count flux boundaries - type 3 with flux specified
if config.vel_type == VelocityType.HARMONIC:
nfluxf += 1
# Extract file paths
temp_th_path = (
constants.get("temp_th_path", [None])[0]
if constants.get("temp_th_path")
else None
)
temp_3d_path = (
constants.get("temp_3d_path", [None])[0]
if constants.get("temp_3d_path")
else None
)
salt_th_path = (
constants.get("salt_th_path", [None])[0]
if constants.get("salt_th_path")
else None
)
salt_3d_path = (
constants.get("salt_3d_path", [None])[0]
if constants.get("salt_3d_path")
else None
)
flow_th_path = (
constants.get("flow_th_path", [None])[0]
if constants.get("flow_th_path")
else None
)
elev_st_path = (
constants.get("elev_st_path", [None])[0]
if constants.get("elev_st_path")
else None
)
vel_st_path = (
constants.get("vel_st_path", [None])[0]
if constants.get("vel_st_path")
else None
)
# Extract Flather boundary data if available
eta_mean = (
constants.get("eta_mean", [None]) if constants.get("eta_mean") else None
)
vn_mean = constants.get("vn_mean", [None]) if constants.get("vn_mean") else None
# Ensure grid boundaries are computed before creating Bctides
if self.grid is not None:
if hasattr(self.grid, "compute_bnd") and not hasattr(self.grid, "nob"):
logger.info("Computing grid boundaries for Bctides")
self.grid.compute_bnd()
elif not hasattr(self.grid, "nob") and hasattr(self.grid, "compute_all"):
logger.info(
"Running compute_all to ensure grid boundaries are available"
)
self.grid.compute_all()
# Verify boundaries were computed
if not hasattr(self.grid, "nob"):
logger.error(
"Failed to compute grid boundaries - grid has no 'nob' attribute"
)
raise AttributeError("Grid boundaries could not be computed")
# Create Bctides object with all the enhanced parameters
bctides = Bctides(
hgrid=self.grid,
flags=flags,
constituents=self.tidal_data.constituents,
tidal_database=self.tidal_data.tidal_database,
tidal_model=self.tidal_data.tidal_model,
tidal_potential=self.tidal_data.tidal_potential,
cutoff_depth=self.tidal_data.cutoff_depth,
nodal_corrections=self.tidal_data.nodal_corrections,
tide_interpolation_method=self.tidal_data.tide_interpolation_method,
extrapolate_tides=self.tidal_data.extrapolate_tides,
extrapolation_distance=self.tidal_data.extrapolation_distance,
extra_databases=self.tidal_data.extra_databases,
mdt=getattr(
self.tidal_data, "_mdt", self.tidal_data.mean_dynamic_topography
),
ethconst=ethconst,
vthconst=vthconst,
tthconst=tthconst,
sthconst=sthconst,
tobc=tobc,
sobc=sobc,
relax=constants.get("inflow_relax", []), # For backward compatibility
inflow_relax=inflow_relax,
outflow_relax=outflow_relax,
ncbn=ncbn,
nfluxf=nfluxf,
elev_th_path=None, # Time history of elevation is not handled by this path yet
elev_st_path=elev_st_path,
flow_th_path=flow_th_path,
vel_st_path=vel_st_path,
temp_th_path=temp_th_path,
temp_3d_path=temp_3d_path,
salt_th_path=salt_th_path,
salt_3d_path=salt_3d_path,
)
# Set additional properties for Flather boundaries
if eta_mean and any(x is not None for x in eta_mean):
bctides.eta_mean = eta_mean
if vn_mean and any(x is not None for x in vn_mean):
bctides.vn_mean = vn_mean
# Set start time and run duration
if self._start_time and self._rnday is not None:
bctides._start_time = self._start_time
bctides._rnday = self._rnday
return bctides
|
write_boundary_file
write_boundary_file(output_path: Union[str, Path]) -> Path
Write the bctides.in file.
Parameters
output_path : str or Path
Path to write the file
Returns
Path
Path to the written file
Raises
ValueError
If start_time and rnday are not set
Source code in rompy_schism/boundary_core.py
| def write_boundary_file(self, output_path: Union[str, Path]) -> Path:
"""Write the bctides.in file.
Parameters
----------
output_path : str or Path
Path to write the file
Returns
-------
Path
Path to the written file
Raises
------
ValueError
If start_time and rnday are not set
"""
if not self._start_time or self._rnday is None:
raise ValueError(
"start_time and rnday must be set before writing boundary file"
)
# Create Bctides object
bctides = self.create_bctides()
# Write file
output_path = Path(output_path)
bctides.write_bctides(output_path)
return output_path
|
BoundaryConfig
Configuration for individual boundary segments.
BoundaryConfig
Bases: BaseModel
Configuration for a single SCHISM boundary segment.
Source code in rompy_schism/boundary_core.py
| class BoundaryConfig(BaseModel):
"""Configuration for a single SCHISM boundary segment."""
# Required fields with default values
elev_type: ElevationType = Field(
default=ElevationType.NONE, description="Elevation boundary condition type"
)
vel_type: VelocityType = Field(
default=VelocityType.NONE, description="Velocity boundary condition type"
)
temp_type: TracerType = Field(
default=TracerType.NONE, description="Temperature boundary condition type"
)
salt_type: TracerType = Field(
default=TracerType.NONE, description="Salinity boundary condition type"
)
# Optional fields for specific boundary types
# Elevation constants (for ElevationType.CONSTANT)
ethconst: Optional[float] = Field(
default=None, description="Constant elevation value (for CONSTANT type)"
)
# Velocity/flow constants (for VelocityType.CONSTANT)
vthconst: Optional[float] = Field(
default=None, description="Constant velocity/flow value (for CONSTANT type)"
)
# Temperature constants and parameters
tthconst: Optional[float] = Field(
default=None, description="Constant temperature value (for CONSTANT type)"
)
tobc: Optional[float] = Field(
default=1.0,
description="Temperature nudging factor (0-1, 1 is strongest nudging)",
)
temp_th_path: Optional[str] = Field(
default=None, description="Path to temperature time history file (for type 1)"
)
temp_3d_path: Optional[str] = Field(
default=None, description="Path to 3D temperature file (for type 4)"
)
# Salinity constants and parameters
sthconst: Optional[float] = Field(
default=None, description="Constant salinity value (for CONSTANT type)"
)
sobc: Optional[float] = Field(
default=1.0, description="Salinity nudging factor (0-1, 1 is strongest nudging)"
)
salt_th_path: Optional[str] = Field(
default=None, description="Path to salinity time history file (for type 1)"
)
salt_3d_path: Optional[str] = Field(
default=None, description="Path to 3D salinity file (for type 4)"
)
# Velocity/flow time history parameters (for VelocityType.TIMEHIST)
flow_th_path: Optional[str] = Field(
default=None, description="Path to flow time history file (for type 1)"
)
# Relaxation parameters for velocity (for VelocityType.RELAXED)
inflow_relax: Optional[float] = Field(
default=0.5,
description="Relaxation factor for inflow (0-1, 1 is strongest nudging)",
)
outflow_relax: Optional[float] = Field(
default=0.1,
description="Relaxation factor for outflow (0-1, 1 is strongest nudging)",
)
# Flather boundary values (for VelocityType.FLATHER)
eta_mean: Optional[List[float]] = Field(
default=None, description="Mean elevation profile for Flather boundary"
)
vn_mean: Optional[List[List[float]]] = Field(
default=None, description="Mean velocity profile for Flather boundary"
)
# Space-time parameters
elev_st_path: Optional[str] = Field(
default=None,
description="Path to space-time elevation file (for SPACETIME type)",
)
vel_st_path: Optional[str] = Field(
default=None,
description="Path to space-time velocity file (for SPACETIME type)",
)
model_config = ConfigDict(arbitrary_types_allowed=True)
def __str__(self):
"""String representation of the boundary configuration."""
return (
f"BoundaryConfig(elev_type={self.elev_type}, vel_type={self.vel_type}, "
f"temp_type={self.temp_type}, salt_type={self.salt_type})"
)
|
Attributes
elev_type
class-attribute
instance-attribute
elev_type: ElevationType = Field(default=NONE, description='Elevation boundary condition type')
vel_type
class-attribute
instance-attribute
vel_type: VelocityType = Field(default=NONE, description='Velocity boundary condition type')
temp_type
class-attribute
instance-attribute
temp_type: TracerType = Field(default=NONE, description='Temperature boundary condition type')
salt_type
class-attribute
instance-attribute
salt_type: TracerType = Field(default=NONE, description='Salinity boundary condition type')
ethconst
class-attribute
instance-attribute
ethconst: Optional[float] = Field(default=None, description='Constant elevation value (for CONSTANT type)')
vthconst
class-attribute
instance-attribute
vthconst: Optional[float] = Field(default=None, description='Constant velocity/flow value (for CONSTANT type)')
tthconst
class-attribute
instance-attribute
tthconst: Optional[float] = Field(default=None, description='Constant temperature value (for CONSTANT type)')
tobc
class-attribute
instance-attribute
tobc: Optional[float] = Field(default=1.0, description='Temperature nudging factor (0-1, 1 is strongest nudging)')
temp_th_path
class-attribute
instance-attribute
temp_th_path: Optional[str] = Field(default=None, description='Path to temperature time history file (for type 1)')
temp_3d_path
class-attribute
instance-attribute
temp_3d_path: Optional[str] = Field(default=None, description='Path to 3D temperature file (for type 4)')
sthconst
class-attribute
instance-attribute
sthconst: Optional[float] = Field(default=None, description='Constant salinity value (for CONSTANT type)')
sobc
class-attribute
instance-attribute
sobc: Optional[float] = Field(default=1.0, description='Salinity nudging factor (0-1, 1 is strongest nudging)')
salt_th_path
class-attribute
instance-attribute
salt_th_path: Optional[str] = Field(default=None, description='Path to salinity time history file (for type 1)')
salt_3d_path
class-attribute
instance-attribute
salt_3d_path: Optional[str] = Field(default=None, description='Path to 3D salinity file (for type 4)')
flow_th_path
class-attribute
instance-attribute
flow_th_path: Optional[str] = Field(default=None, description='Path to flow time history file (for type 1)')
inflow_relax
class-attribute
instance-attribute
inflow_relax: Optional[float] = Field(default=0.5, description='Relaxation factor for inflow (0-1, 1 is strongest nudging)')
outflow_relax
class-attribute
instance-attribute
outflow_relax: Optional[float] = Field(default=0.1, description='Relaxation factor for outflow (0-1, 1 is strongest nudging)')
eta_mean
class-attribute
instance-attribute
eta_mean: Optional[List[float]] = Field(default=None, description='Mean elevation profile for Flather boundary')
vn_mean
class-attribute
instance-attribute
vn_mean: Optional[List[List[float]]] = Field(default=None, description='Mean velocity profile for Flather boundary')
elev_st_path
class-attribute
instance-attribute
elev_st_path: Optional[str] = Field(default=None, description='Path to space-time elevation file (for SPACETIME type)')
vel_st_path
class-attribute
instance-attribute
vel_st_path: Optional[str] = Field(default=None, description='Path to space-time velocity file (for SPACETIME type)')
model_config
class-attribute
instance-attribute
model_config = ConfigDict(arbitrary_types_allowed=True)
Boundary Type Enums
ElevationType
ElevationType
Bases: IntEnum
Elevation boundary condition types.
Source code in rompy_schism/boundary_core.py
| class ElevationType(IntEnum):
"""Elevation boundary condition types."""
NONE = 0 # Not specified
TIMEHIST = 1 # Time history from elev.th
CONSTANT = 2 # Constant elevation
HARMONIC = 3 # Harmonic tidal constituents
EXTERNAL = 4 # External model data from elev2D.th.nc
HARMONICEXTERNAL = 5 # Combination of harmonic and external data
|
Attributes
NONE
class-attribute
instance-attribute
TIMEHIST
class-attribute
instance-attribute
CONSTANT
class-attribute
instance-attribute
HARMONIC
class-attribute
instance-attribute
EXTERNAL
class-attribute
instance-attribute
HARMONICEXTERNAL
class-attribute
instance-attribute
VelocityType
VelocityType
Bases: IntEnum
Velocity boundary condition types.
Source code in rompy_schism/boundary_core.py
| class VelocityType(IntEnum):
"""Velocity boundary condition types."""
NONE = 0 # Not specified
TIMEHIST = 1 # Time history from flux.th
CONSTANT = 2 # Constant discharge
HARMONIC = 3 # Harmonic tidal constituents
EXTERNAL = 4 # External model data from uv3D.th.nc
HARMONICEXTERNAL = 5 # Combination of harmonic and external data
FLATHER = -1 # Flather type radiation boundary
RELAXED = -4 # 3D input with relaxation
|
Attributes
NONE
class-attribute
instance-attribute
TIMEHIST
class-attribute
instance-attribute
CONSTANT
class-attribute
instance-attribute
HARMONIC
class-attribute
instance-attribute
EXTERNAL
class-attribute
instance-attribute
HARMONICEXTERNAL
class-attribute
instance-attribute
FLATHER
class-attribute
instance-attribute
RELAXED
class-attribute
instance-attribute
TracerType
TracerType
Bases: IntEnum
Temperature/salinity boundary condition types.
Source code in rompy_schism/boundary_core.py
| class TracerType(IntEnum):
"""Temperature/salinity boundary condition types."""
NONE = 0 # Not specified
TIMEHIST = 1 # Time history from temp/salt.th
CONSTANT = 2 # Constant temperature/salinity
INITIAL = 3 # Initial profile for inflow
EXTERNAL = 4 # External model 3D input
|
Attributes
NONE
class-attribute
instance-attribute
TIMEHIST
class-attribute
instance-attribute
CONSTANT
class-attribute
instance-attribute
INITIAL
class-attribute
instance-attribute
EXTERNAL
class-attribute
instance-attribute
Factory Functions
The boundary conditions module provides convenient factory functions for creating common boundary configurations. These functions return SCHISMDataBoundaryConditions objects that can be directly used in SCHISM simulations.
High-Level Configuration Functions
create_tidal_only_boundary_config
create_tidal_only_boundary_config
create_tidal_only_boundary_config(constituents: Union[str, List[str]] = 'major', tidal_database: Union[str, Path] = None, tidal_model: Optional[str] = 'FES2014', nodal_corrections: bool = True, tidal_potential: bool = True, cutoff_depth: float = 50.0, tide_interpolation_method: str = 'bilinear')
Create a configuration where all open boundaries are treated as tidal boundaries.
Parameters
constituents : str or list, optional
Tidal constituents to include, by default "major"
tidal_database : str or Path, optional
Path to tidal database for pyTMD, by default None
tidal_model : str, optional
Tidal model to use, by default 'FES2014'
nodal_corrections : bool, optional
Whether to apply nodal corrections, by default True
tidal_potential : bool, optional
Whether to include tidal potential, by default True
cutoff_depth : float, optional
Depth threshold for tidal potential, by default 50.0
tide_interpolation_method : str, optional
Method for tide interpolation, by default "bilinear"
Returns
SCHISMDataBoundaryConditions
Configured boundary conditions
Source code in rompy_schism/boundary_core.py
| def create_tidal_only_boundary_config(
constituents: Union[str, List[str]] = "major",
tidal_database: Union[str, Path] = None,
tidal_model: Optional[str] = "FES2014",
nodal_corrections: bool = True,
tidal_potential: bool = True,
cutoff_depth: float = 50.0,
tide_interpolation_method: str = "bilinear",
):
"""
Create a configuration where all open boundaries are treated as tidal boundaries.
Parameters
----------
constituents : str or list, optional
Tidal constituents to include, by default "major"
tidal_database : str or Path, optional
Path to tidal database for pyTMD, by default None
tidal_model : str, optional
Tidal model to use, by default 'FES2014'
nodal_corrections : bool, optional
Whether to apply nodal corrections, by default True
tidal_potential : bool, optional
Whether to include tidal potential, by default True
cutoff_depth : float, optional
Depth threshold for tidal potential, by default 50.0
tide_interpolation_method : str, optional
Method for tide interpolation, by default "bilinear"
Returns
-------
SCHISMDataBoundaryConditions
Configured boundary conditions
"""
from rompy_schism.data import SCHISMDataBoundaryConditions
# Create tidal dataset
tidal_data = TidalDataset(
constituents=constituents,
tidal_database=tidal_database,
tidal_model=tidal_model,
nodal_corrections=nodal_corrections,
tidal_potential=tidal_potential,
cutoff_depth=cutoff_depth,
tide_interpolation_method=tide_interpolation_method,
)
# Create the config with tidal setup
config = SCHISMDataBoundaryConditions(
tidal_data=tidal_data,
setup_type="tidal",
boundaries={},
hotstart_config=None,
)
return config
|
Example Usage:
from rompy_schism.boundary_conditions import create_tidal_only_boundary_config
# Basic tidal configuration
bc = create_tidal_only_boundary_config(
constituents=["M2", "S2", "N2", "K1", "O1"],
tidal_elevations="/path/to/h_tpxo9.nc",
tidal_velocities="/path/to/u_tpxo9.nc"
)
# With earth tidal potential
bc = create_tidal_only_boundary_config(
constituents=["M2", "S2", "K1", "O1"],
ntip=1 # Enable earth tidal potential
)
create_hybrid_boundary_config
create_hybrid_boundary_config
create_hybrid_boundary_config(constituents: Union[str, List[str]] = 'major', tidal_database: Union[str, Path] = None, tidal_model: Optional[str] = 'FES2014', nodal_corrections: bool = True, tidal_potential: bool = True, cutoff_depth: float = 50.0, tide_interpolation_method: str = 'bilinear', elev_source: Optional[Union[Any, Any]] = None, vel_source: Optional[Union[Any, Any]] = None, temp_source: Optional[Union[Any, Any]] = None, salt_source: Optional[Union[Any, Any]] = None)
Create a configuration for hybrid harmonic + external data boundaries.
Parameters
constituents : str or list, optional
Tidal constituents to include, by default "major"
tidal_database : str or Path, optional
Path to tidal database for pyTMD, by default None
tidal_model : str, optional
Tidal model to use, by default 'FES2014'
nodal_corrections : bool, optional
Whether to apply nodal corrections, by default True
tidal_potential : bool, optional
Whether to include tidal potential, by default True
cutoff_depth : float, optional
Depth threshold for tidal potential, by default 50.0
tide_interpolation_method : str, optional
Method for tide interpolation, by default "bilinear"
elev_source : Union[DataBlob, SCHISMDataBoundary], optional
Data source for elevation
vel_source : Union[DataBlob, SCHISMDataBoundary], optional
Data source for velocity
temp_source : Union[DataBlob, SCHISMDataBoundary], optional
Data source for temperature
salt_source : Union[DataBlob, SCHISMDataBoundary], optional
Data source for salinity
Returns
SCHISMDataBoundaryConditions
Configured boundary conditions
Source code in rompy_schism/boundary_core.py
| def create_hybrid_boundary_config(
constituents: Union[str, List[str]] = "major",
tidal_database: Union[str, Path] = None,
tidal_model: Optional[str] = "FES2014",
nodal_corrections: bool = True,
tidal_potential: bool = True,
cutoff_depth: float = 50.0,
tide_interpolation_method: str = "bilinear",
elev_source: Optional[Union[Any, Any]] = None,
vel_source: Optional[Union[Any, Any]] = None,
temp_source: Optional[Union[Any, Any]] = None,
salt_source: Optional[Union[Any, Any]] = None,
):
"""
Create a configuration for hybrid harmonic + external data boundaries.
Parameters
----------
constituents : str or list, optional
Tidal constituents to include, by default "major"
tidal_database : str or Path, optional
Path to tidal database for pyTMD, by default None
tidal_model : str, optional
Tidal model to use, by default 'FES2014'
nodal_corrections : bool, optional
Whether to apply nodal corrections, by default True
tidal_potential : bool, optional
Whether to include tidal potential, by default True
cutoff_depth : float, optional
Depth threshold for tidal potential, by default 50.0
tide_interpolation_method : str, optional
Method for tide interpolation, by default "bilinear"
elev_source : Union[DataBlob, SCHISMDataBoundary], optional
Data source for elevation
vel_source : Union[DataBlob, SCHISMDataBoundary], optional
Data source for velocity
temp_source : Union[DataBlob, SCHISMDataBoundary], optional
Data source for temperature
salt_source : Union[DataBlob, SCHISMDataBoundary], optional
Data source for salinity
Returns
-------
SCHISMDataBoundaryConditions
Configured boundary conditions
"""
from rompy_schism.data import BoundarySetupWithSource, SCHISMDataBoundaryConditions
from rompy_schism.tides_enhanced import TidalDataset
# Create tidal dataset
tidal_data = TidalDataset(
constituents=constituents,
tidal_database=tidal_database,
tidal_model=tidal_model,
nodal_corrections=nodal_corrections,
tidal_potential=tidal_potential,
cutoff_depth=cutoff_depth,
tide_interpolation_method=tide_interpolation_method,
)
# Create the config with hybrid setup
config = SCHISMDataBoundaryConditions(
tidal_data=tidal_data,
setup_type="hybrid",
boundaries={
0: BoundarySetupWithSource(
elev_type=ElevationType.HARMONICEXTERNAL,
vel_type=VelocityType.HARMONICEXTERNAL
if vel_source
else VelocityType.NONE,
temp_type=TracerType.EXTERNAL if temp_source else TracerType.INITIAL,
salt_type=TracerType.EXTERNAL if salt_source else TracerType.INITIAL,
elev_source=elev_source,
vel_source=vel_source,
temp_source=temp_source,
salt_source=salt_source,
)
},
hotstart_config=None,
)
return config
|
Example Usage:
from rompy_schism.boundary_conditions import create_hybrid_boundary_config
from rompy.core.data import DataBlob
# Hybrid configuration with external data
bc = create_hybrid_boundary_config(
constituents=["M2", "S2"],
tidal_elevations="/path/to/h_tpxo9.nc",
tidal_velocities="/path/to/u_tpxo9.nc",
elev_source=DataBlob(source="/path/to/elev2D.th.nc"),
vel_source=DataBlob(source="/path/to/uv3D.th.nc"),
temp_source=DataBlob(source="/path/to/TEM_3D.th.nc"),
salt_source=DataBlob(source="/path/to/SAL_3D.th.nc")
)
create_river_boundary_config
create_river_boundary_config
create_river_boundary_config(river_boundary_index: int = 0, river_flow: float = -100.0, other_boundaries: Literal['tidal', 'hybrid', 'none'] = 'tidal', constituents: Union[str, List[str]] = 'major', tidal_database: Union[str, Path] = None, tidal_model: Optional[str] = 'FES2014', nodal_corrections: bool = True, tidal_potential: bool = True, cutoff_depth: float = 50.0, tide_interpolation_method: str = 'bilinear')
Create a configuration with a designated river boundary and optional tidal boundaries.
Parameters
river_boundary_index : int
Index of the river boundary
river_flow : float
Flow rate (negative for inflow)
other_boundaries : str
How to treat other boundaries ("tidal", "hybrid", or "none")
constituents : str or list, optional
Tidal constituents to include, by default "major"
tidal_database : str or Path, optional
Path to tidal database for pyTMD, by default None
tidal_model : str, optional
Tidal model to use, by default 'FES2014'
nodal_corrections : bool, optional
Whether to apply nodal corrections, by default True
tidal_potential : bool, optional
Whether to include tidal potential, by default True
cutoff_depth : float, optional
Depth threshold for tidal potential, by default 50.0
tide_interpolation_method : str, optional
Method for tide interpolation, by default "bilinear"
Returns
SCHISMDataBoundaryConditions
Configured boundary conditions
Source code in rompy_schism/boundary_core.py
| def create_river_boundary_config(
river_boundary_index: int = 0,
river_flow: float = -100.0, # Negative for inflow
other_boundaries: Literal["tidal", "hybrid", "none"] = "tidal",
constituents: Union[str, List[str]] = "major",
tidal_database: Union[str, Path] = None,
tidal_model: Optional[str] = "FES2014",
nodal_corrections: bool = True,
tidal_potential: bool = True,
cutoff_depth: float = 50.0,
tide_interpolation_method: str = "bilinear",
):
"""
Create a configuration with a designated river boundary and optional tidal boundaries.
Parameters
----------
river_boundary_index : int
Index of the river boundary
river_flow : float
Flow rate (negative for inflow)
other_boundaries : str
How to treat other boundaries ("tidal", "hybrid", or "none")
constituents : str or list, optional
Tidal constituents to include, by default "major"
tidal_database : str or Path, optional
Path to tidal database for pyTMD, by default None
tidal_model : str, optional
Tidal model to use, by default 'FES2014'
nodal_corrections : bool, optional
Whether to apply nodal corrections, by default True
tidal_potential : bool, optional
Whether to include tidal potential, by default True
cutoff_depth : float, optional
Depth threshold for tidal potential, by default 50.0
tide_interpolation_method : str, optional
Method for tide interpolation, by default "bilinear"
Returns
-------
SCHISMDataBoundaryConditions
Configured boundary conditions
"""
from rompy_schism.data import BoundarySetupWithSource, SCHISMDataBoundaryConditions
from rompy_schism.tides_enhanced import TidalDataset
# Create tidal dataset if both paths are provided and needed
tidal_data = None
if other_boundaries in ["tidal", "hybrid"]:
tidal_data = TidalDataset(
constituents=constituents,
tidal_database=tidal_database,
tidal_model=tidal_model,
nodal_corrections=nodal_corrections,
tidal_potential=tidal_potential,
cutoff_depth=cutoff_depth,
tide_interpolation_method=tide_interpolation_method,
)
# Create the basic config
config = SCHISMDataBoundaryConditions(
tidal_data=tidal_data,
setup_type="river",
hotstart_config=None,
)
# Add the river boundary
config.boundaries[river_boundary_index] = BoundarySetupWithSource(
elev_type=ElevationType.NONE,
vel_type=VelocityType.CONSTANT,
temp_type=TracerType.NONE,
salt_type=TracerType.NONE,
const_flow=river_flow,
)
return config
|
Example Usage:
from rompy_schism.boundary_conditions import create_river_boundary_config
# River boundary with tidal forcing on other boundaries
bc = create_river_boundary_config(
river_boundary_index=1,
river_flow=-500.0, # 500 m³/s inflow
river_temp=15.0, # 15°C
river_salt=0.1, # 0.1 PSU (fresh water)
other_boundaries="tidal",
constituents=["M2", "S2", "N2"]
)
# River-only configuration
bc = create_river_boundary_config(
river_boundary_index=0,
river_flow=-200.0,
other_boundaries="none"
)
create_nested_boundary_config
create_nested_boundary_config
create_nested_boundary_config(with_tides: bool = True, inflow_relax: float = 0.8, outflow_relax: float = 0.2, elev_source: Optional[Union[Any, Any]] = None, vel_source: Optional[Union[Any, Any]] = None, temp_source: Optional[Union[Any, Any]] = None, salt_source: Optional[Union[Any, Any]] = None, constituents: Union[str, List[str]] = 'major', tidal_database: Union[str, Path] = None, tidal_model: Optional[str] = 'FES2014', nodal_corrections: bool = True, tidal_potential: bool = True, cutoff_depth: float = 50.0, tide_interpolation_method: str = 'bilinear')
Create a configuration for nested model boundaries with external data.
Parameters
with_tides : bool
Include tidal components
inflow_relax : float
Relaxation parameter for inflow (0-1)
outflow_relax : float
Relaxation parameter for outflow (0-1)
elev_source : Union[DataBlob, SCHISMDataBoundary], optional
Data source for elevation
vel_source : Union[DataBlob, SCHISMDataBoundary], optional
Data source for velocity
temp_source : Union[DataBlob, SCHISMDataBoundary], optional
Data source for temperature
salt_source : Union[DataBlob, SCHISMDataBoundary], optional
Data source for salinity
constituents : str or list, optional
Tidal constituents to include, by default "major"
tidal_database : str or Path, optional
Path to tidal database for pyTMD, by default None
tidal_model : str, optional
Tidal model to use, by default 'FES2014'
nodal_corrections : bool, optional
Whether to apply nodal corrections, by default True
tidal_potential : bool, optional
Whether to include tidal potential, by default True
cutoff_depth : float, optional
Depth threshold for tidal potential, by default 50.0
tide_interpolation_method : str, optional
Method for tide interpolation, by default "bilinear"
Returns
SCHISMDataBoundaryConditions
Configured boundary conditions
Source code in rompy_schism/boundary_core.py
| def create_nested_boundary_config(
with_tides: bool = True,
inflow_relax: float = 0.8,
outflow_relax: float = 0.2,
elev_source: Optional[Union[Any, Any]] = None,
vel_source: Optional[Union[Any, Any]] = None,
temp_source: Optional[Union[Any, Any]] = None,
salt_source: Optional[Union[Any, Any]] = None,
constituents: Union[str, List[str]] = "major",
tidal_database: Union[str, Path] = None,
tidal_model: Optional[str] = "FES2014",
nodal_corrections: bool = True,
tidal_potential: bool = True,
cutoff_depth: float = 50.0,
tide_interpolation_method: str = "bilinear",
):
"""
Create a configuration for nested model boundaries with external data.
Parameters
----------
with_tides : bool
Include tidal components
inflow_relax : float
Relaxation parameter for inflow (0-1)
outflow_relax : float
Relaxation parameter for outflow (0-1)
elev_source : Union[DataBlob, SCHISMDataBoundary], optional
Data source for elevation
vel_source : Union[DataBlob, SCHISMDataBoundary], optional
Data source for velocity
temp_source : Union[DataBlob, SCHISMDataBoundary], optional
Data source for temperature
salt_source : Union[DataBlob, SCHISMDataBoundary], optional
Data source for salinity
constituents : str or list, optional
Tidal constituents to include, by default "major"
tidal_database : str or Path, optional
Path to tidal database for pyTMD, by default None
tidal_model : str, optional
Tidal model to use, by default 'FES2014'
nodal_corrections : bool, optional
Whether to apply nodal corrections, by default True
tidal_potential : bool, optional
Whether to include tidal potential, by default True
cutoff_depth : float, optional
Depth threshold for tidal potential, by default 50.0
tide_interpolation_method : str, optional
Method for tide interpolation, by default "bilinear"
Returns
-------
SCHISMDataBoundaryConditions
Configured boundary conditions
"""
from rompy_schism.data import BoundarySetupWithSource, SCHISMDataBoundaryConditions
from rompy_schism.tides_enhanced import TidalDataset
# Create tidal dataset if both paths are provided and needed
tidal_data = None
if with_tides:
tidal_data = TidalDataset(
constituents=constituents,
tidal_database=tidal_database,
tidal_model=tidal_model,
nodal_corrections=nodal_corrections,
tidal_potential=tidal_potential,
cutoff_depth=cutoff_depth,
tide_interpolation_method=tide_interpolation_method,
)
# Create the basic config
config = SCHISMDataBoundaryConditions(
tidal_data=tidal_data,
setup_type="nested",
hotstart_config=None,
)
# Determine elevation type based on tides setting
elev_type = ElevationType.HARMONICEXTERNAL if with_tides else ElevationType.EXTERNAL
# Add the nested boundary configuration
config.boundaries[0] = BoundarySetupWithSource(
elev_type=elev_type,
vel_type=VelocityType.RELAXED,
temp_type=TracerType.EXTERNAL if temp_source else TracerType.NONE,
salt_type=TracerType.EXTERNAL if salt_source else TracerType.NONE,
inflow_relax=inflow_relax,
outflow_relax=outflow_relax,
elev_source=elev_source,
vel_source=vel_source,
temp_source=temp_source,
salt_source=salt_source,
)
return config
|
Example Usage:
from rompy_schism.boundary_conditions import create_nested_boundary_config
from rompy_schism.data import SCHISMDataBoundary
from rompy.core.source import SourceFile
# Nested boundary with tides and parent model data
bc = create_nested_boundary_config(
with_tides=True,
inflow_relax=0.9,
outflow_relax=0.1,
constituents=["M2", "S2"],
elev_source=SCHISMDataBoundary(
source=SourceFile(uri="/path/to/parent_model.nc"),
variables=["ssh"]
),
vel_source=SCHISMDataBoundary(
source=SourceFile(uri="/path/to/parent_model.nc"),
variables=["u", "v"]
)
)
# Nested boundary without tides
bc = create_nested_boundary_config(
with_tides=False,
inflow_relax=0.8,
outflow_relax=0.2,
elev_source=elev_data,
vel_source=vel_data
)
Low-Level Boundary Creation Functions
These functions create BoundaryHandler objects for direct grid-based boundary manipulation:
create_tidal_boundary
create_tidal_boundary(grid_path: Union[str, Path], constituents: Union[str, List[str]] = 'major', tidal_database: Union[str, Path] = None, tidal_model: Optional[str] = 'FES2014', nodal_corrections: bool = True, tidal_potential: bool = True, cutoff_depth: float = 50.0, tide_interpolation_method: str = 'bilinear') -> BoundaryHandler
Create a tidal-only boundary.
Parameters
grid_path : str or Path
Path to SCHISM grid
constituents : str or list, optional
Tidal constituents, by default "major"
tidal_database : str or Path, optional
Tidal database path for pyTMD to use, by default None
tidal_model : str, optional
Tidal model to use, by default 'FES2014'
nodal_corrections : bool, optional
Whether to apply nodal corrections, by default True
tidal_potential : bool, optional
Whether to include tidal potential, by default True
cutoff_depth : float, optional
Depth threshold for tidal potential, by default 50.0
Returns
BoundaryHandler
Configured tidal boundary
Source code in rompy_schism/boundary_core.py
| def create_tidal_boundary(
grid_path: Union[str, Path],
constituents: Union[str, List[str]] = "major",
tidal_database: Union[str, Path] = None,
tidal_model: Optional[str] = "FES2014",
nodal_corrections: bool = True,
tidal_potential: bool = True,
cutoff_depth: float = 50.0,
tide_interpolation_method: str = "bilinear",
) -> BoundaryHandler:
"""Create a tidal-only boundary.
Parameters
----------
grid_path : str or Path
Path to SCHISM grid
constituents : str or list, optional
Tidal constituents, by default "major"
tidal_database : str or Path, optional
Tidal database path for pyTMD to use, by default None
tidal_model : str, optional
Tidal model to use, by default 'FES2014'
nodal_corrections : bool, optional
Whether to apply nodal corrections, by default True
tidal_potential : bool, optional
Whether to include tidal potential, by default True
cutoff_depth : float, optional
Depth threshold for tidal potential, by default 50.0
Returns
-------
BoundaryHandler
Configured tidal boundary
"""
tidal_data = TidalDataset(
constituents=constituents,
tidal_database=tidal_database,
tidal_model=tidal_model,
nodal_corrections=nodal_corrections,
tidal_potential=tidal_potential,
cutoff_depth=cutoff_depth,
tide_interpolation_method=tide_interpolation_method,
)
boundary = BoundaryHandler(
grid_path=grid_path,
tidal_data=tidal_data,
)
# Set default configuration for all boundaries: pure tidal
boundary.set_boundary_type(
0, # Will be applied to all boundaries
elev_type=ElevationType.HARMONIC,
vel_type=VelocityType.HARMONIC,
)
return boundary
|
create_hybrid_boundary
create_hybrid_boundary(grid_path: Union[str, Path], constituents: Union[str, List[str]] = 'major', tidal_database: Union[str, Path] = None, tidal_model: Optional[str] = 'FES2014', nodal_corrections: bool = True, tidal_potential: bool = True, cutoff_depth: float = 50.0, tide_interpolation_method: str = 'bilinear') -> BoundaryHandler
Create a hybrid boundary with tides + external data.
Parameters
grid_path : str or Path
Path to SCHISM grid
constituents : str or list, optional
Tidal constituents to include, by default "major"
tidal_database : str or Path, optional
Path to tidal database for pyTMD, by default None
tidal_model : str, optional
Tidal model to use, by default 'FES2014'
nodal_corrections : bool, optional
Whether to apply nodal corrections, by default True
tidal_potential : bool, optional
Whether to include tidal potential, by default True
cutoff_depth : float, optional
Depth threshold for tidal potential, by default 50.0
tide_interpolation_method : str, optional
Method for tide interpolation, by default "bilinear"
Returns
BoundaryHandler
Configured hybrid boundary
Source code in rompy_schism/boundary_core.py
| def create_hybrid_boundary(
grid_path: Union[str, Path],
constituents: Union[str, List[str]] = "major",
tidal_database: Union[str, Path] = None,
tidal_model: Optional[str] = "FES2014",
nodal_corrections: bool = True,
tidal_potential: bool = True,
cutoff_depth: float = 50.0,
tide_interpolation_method: str = "bilinear",
) -> BoundaryHandler:
"""Create a hybrid boundary with tides + external data.
Parameters
----------
grid_path : str or Path
Path to SCHISM grid
constituents : str or list, optional
Tidal constituents to include, by default "major"
tidal_database : str or Path, optional
Path to tidal database for pyTMD, by default None
tidal_model : str, optional
Tidal model to use, by default 'FES2014'
nodal_corrections : bool, optional
Whether to apply nodal corrections, by default True
tidal_potential : bool, optional
Whether to include tidal potential, by default True
cutoff_depth : float, optional
Depth threshold for tidal potential, by default 50.0
tide_interpolation_method : str, optional
Method for tide interpolation, by default "bilinear"
Returns
-------
BoundaryHandler
Configured hybrid boundary
"""
tidal_data = TidalDataset(
constituents=constituents,
tidal_database=tidal_database,
tidal_model=tidal_model,
nodal_corrections=nodal_corrections,
tidal_potential=tidal_potential,
cutoff_depth=cutoff_depth,
tide_interpolation_method=tide_interpolation_method,
)
boundary = BoundaryHandler(grid_path=grid_path, tidal_data=tidal_data)
# Set default configuration for all boundaries: tidal + spacetime
boundary.set_boundary_type(
0, # Will be applied to all boundaries
elev_type=ElevationType.HARMONICEXTERNAL,
vel_type=VelocityType.HARMONICEXTERNAL,
)
return boundary
|
create_river_boundary
create_river_boundary(grid_path: Union[str, Path], river_flow: float = -100.0, river_boundary_index: int = 0) -> BoundaryHandler
Create a river boundary with constant flow.
Parameters
grid_path : str or Path
Path to SCHISM grid
river_flow : float, optional
River flow value (negative for inflow), by default -100.0
river_boundary_index : int, optional
Index of the river boundary, by default 0
Returns
BoundaryHandler
Configured river boundary
Source code in rompy_schism/boundary_core.py
| def create_river_boundary(
grid_path: Union[str, Path],
river_flow: float = -100.0, # Negative for inflow
river_boundary_index: int = 0,
) -> BoundaryHandler:
"""Create a river boundary with constant flow.
Parameters
----------
grid_path : str or Path
Path to SCHISM grid
river_flow : float, optional
River flow value (negative for inflow), by default -100.0
river_boundary_index : int, optional
Index of the river boundary, by default 0
Returns
-------
BoundaryHandler
Configured river boundary
"""
boundary = BoundaryHandler(grid_path=grid_path)
# Set river boundary
boundary.set_boundary_type(
river_boundary_index,
elev_type=ElevationType.NONE, # No elevation specified
vel_type=VelocityType.CONSTANT, # Constant flow
vthconst=river_flow, # Flow value
)
return boundary
|
create_nested_boundary
create_nested_boundary(grid_path: Union[str, Path], with_tides: bool = False, inflow_relax: float = 0.8, outflow_relax: float = 0.8, constituents: Union[str, List[str]] = 'major', tidal_database: Union[str, Path] = None, tidal_model: Optional[str] = 'FES2014', nodal_corrections: bool = True, tidal_potential: bool = True, cutoff_depth: float = 50.0, tide_interpolation_method: str = 'bilinear') -> BoundaryHandler
Create a nested boundary with optional tides.
Parameters
grid_path : str or Path
Path to SCHISM grid
with_tides : bool, optional
Whether to include tides, by default False
inflow_relax : float, optional
Relaxation factor for inflow, by default 0.8
outflow_relax : float, optional
Relaxation factor for outflow, by default 0.8
constituents : str or list, optional
Tidal constituents to include, by default "major"
tidal_database : str or Path, optional
Path to tidal database for pyTMD, by default None
tidal_model : str, optional
Tidal model to use, by default 'FES2014'
nodal_corrections : bool, optional
Whether to apply nodal corrections, by default True
tidal_potential : bool, optional
Whether to include tidal potential, by default True
cutoff_depth : float, optional
Depth threshold for tidal potential, by default 50.0
tide_interpolation_method : str, optional
Method for tide interpolation, by default "bilinear"
Returns
BoundaryHandler
Configured nested boundary
Source code in rompy_schism/boundary_core.py
| def create_nested_boundary(
grid_path: Union[str, Path],
with_tides: bool = False,
inflow_relax: float = 0.8,
outflow_relax: float = 0.8,
constituents: Union[str, List[str]] = "major",
tidal_database: Union[str, Path] = None,
tidal_model: Optional[str] = "FES2014",
nodal_corrections: bool = True,
tidal_potential: bool = True,
cutoff_depth: float = 50.0,
tide_interpolation_method: str = "bilinear",
) -> BoundaryHandler:
"""Create a nested boundary with optional tides.
Parameters
----------
grid_path : str or Path
Path to SCHISM grid
with_tides : bool, optional
Whether to include tides, by default False
inflow_relax : float, optional
Relaxation factor for inflow, by default 0.8
outflow_relax : float, optional
Relaxation factor for outflow, by default 0.8
constituents : str or list, optional
Tidal constituents to include, by default "major"
tidal_database : str or Path, optional
Path to tidal database for pyTMD, by default None
tidal_model : str, optional
Tidal model to use, by default 'FES2014'
nodal_corrections : bool, optional
Whether to apply nodal corrections, by default True
tidal_potential : bool, optional
Whether to include tidal potential, by default True
cutoff_depth : float, optional
Depth threshold for tidal potential, by default 50.0
tide_interpolation_method : str, optional
Method for tide interpolation, by default "bilinear"
Returns
-------
BoundaryHandler
Configured nested boundary
"""
tidal_data = None
if with_tides:
tidal_data = TidalDataset(
constituents=constituents,
tidal_database=tidal_database,
tidal_model=tidal_model,
nodal_corrections=nodal_corrections,
tidal_potential=tidal_potential,
cutoff_depth=cutoff_depth,
tide_interpolation_method=tide_interpolation_method,
)
boundary = BoundaryHandler(
grid_path=grid_path,
constituents=constituents if with_tides else None,
tidal_data=tidal_data,
)
if with_tides:
# Tides + external data with relaxation
boundary.set_boundary_type(
0, # Will be applied to all boundaries
elev_type=ElevationType.HARMONICEXTERNAL,
vel_type=VelocityType.RELAXED,
temp_type=TracerType.EXTERNAL,
salt_type=TracerType.EXTERNAL,
inflow_relax=inflow_relax,
outflow_relax=outflow_relax,
)
else:
# Just external data with relaxation
boundary.set_boundary_type(
0, # Will be applied to all boundaries
elev_type=ElevationType.EXTERNAL,
vel_type=VelocityType.RELAXED,
temp_type=TracerType.EXTERNAL,
salt_type=TracerType.EXTERNAL,
inflow_relax=inflow_relax,
outflow_relax=outflow_relax,
)
return boundary
|
Usage Examples
Tidal-Only Configuration
For simulations with purely tidal forcing:
from rompy_schism.boundary_conditions import create_tidal_only_boundary_config
from rompy_schism.data import SCHISMData
# Create tidal-only boundary configuration
boundary_conditions = create_tidal_only_boundary_config(
constituents=["M2", "S2", "N2", "K1", "O1"],
tidal_database="tpxo",
tidal_elevations="path/to/tidal_elevations.nc",
tidal_velocities="path/to/tidal_velocities.nc",
)
# Use in SCHISM configuration
schism_data = SCHISMData(
boundary_conditions=boundary_conditions,
)
Hybrid Tidal + Ocean Data
For simulations combining tidal forcing with external ocean data:
from rompy_schism.boundary_conditions import create_hybrid_boundary_config
from rompy.core.data import DataBlob
# Create hybrid boundary configuration
boundary_conditions = create_hybrid_boundary_config(
constituents=["M2", "S2"],
tidal_elevations="path/to/tidal_elevations.nc",
tidal_velocities="path/to/tidal_velocities.nc",
# Add ocean data sources
elev_source=DataBlob(source="path/to/elev2D.th.nc"),
vel_source=DataBlob(source="path/to/uv3D.th.nc"),
temp_source=DataBlob(source="path/to/TEM_3D.th.nc"),
salt_source=DataBlob(source="path/to/SAL_3D.th.nc"),
)
River Boundary Configuration
For simulations with river inputs:
from rompy_schism.boundary_conditions import create_river_boundary_config
# Create river boundary configuration
boundary_conditions = create_river_boundary_config(
river_boundary_index=1, # Index of the river boundary
river_flow=-100.0, # Negative for inflow (m³/s)
other_boundaries="tidal", # Other boundaries are tidal
constituents=["M2", "S2"],
tidal_elevations="path/to/tidal_elevations.nc",
tidal_velocities="path/to/tidal_velocities.nc",
)
Nested Model Configuration
For simulations nested within a larger model:
from rompy_schism.boundary_conditions import create_nested_boundary_config
from rompy_schism.data import SCHISMDataBoundary
from rompy.core.source import SourceFile
# Create nested boundary configuration
boundary_conditions = create_nested_boundary_config(
with_tides=True,
inflow_relax=0.8,
outflow_relax=0.2,
constituents=["M2", "S2"],
tidal_elevations="path/to/tidal_elevations.nc",
tidal_velocities="path/to/tidal_velocities.nc",
# Add parent model data sources
elev_source=SCHISMDataBoundary(
source=SourceFile(uri="path/to/parent_model.nc"),
variables=["ssh"],
),
vel_source=SCHISMDataBoundary(
source=SourceFile(uri="path/to/parent_model.nc"),
variables=["u", "v"],
),
)
Direct Boundary Handler Usage
For maximum control, use the BoundaryHandler class directly:
from rompy_schism.boundary_core import (
BoundaryHandler,
ElevationType,
VelocityType,
TracerType
)
# Create boundary handler
boundary = BoundaryHandler(
grid_path="path/to/hgrid.gr3",
constituents=["M2", "S2", "K1", "O1"],
tidal_database="tpxo",
tidal_elevations="path/to/h_tpxo9.nc",
tidal_velocities="path/to/uv_tpxo9.nc"
)
# Configure different boundary types
boundary.set_boundary_type(
0, # Ocean boundary with tides
elev_type=ElevationType.HARMONIC,
vel_type=VelocityType.HARMONIC
)
boundary.set_boundary_type(
1, # River boundary
elev_type=ElevationType.NONE,
vel_type=VelocityType.CONSTANT,
vthconst=-500.0 # River inflow
)
# Set simulation parameters and write output
boundary.set_run_parameters(start_time, run_days)
boundary.write_boundary_file("path/to/bctides.in")
Custom Boundary Configuration
For complex scenarios with mixed boundary types:
from rompy_schism.data import SCHISMDataBoundaryConditions, BoundarySetupWithSource
from rompy_schism.boundary_core import ElevationType, VelocityType, TracerType
from rompy.core.data import DataBlob
# Create custom boundary configuration
boundary_conditions = SCHISMDataBoundaryConditions(
constituents=["M2", "S2"],
tidal_database="tpxo",
boundaries={
# Ocean boundary (harmonic + external data)
0: BoundarySetupWithSource(
elev_type=ElevationType.HARMONICEXTERNAL,
vel_type=VelocityType.HARMONICEXTERNAL,
temp_type=TracerType.EXTERNAL,
salt_type=TracerType.EXTERNAL,
elev_source=DataBlob(source="path/to/elev2D.th.nc"),
vel_source=DataBlob(source="path/to/uv3D.th.nc"),
temp_source=DataBlob(source="path/to/TEM_3D.th.nc"),
salt_source=DataBlob(source="path/to/SAL_3D.th.nc"),
),
# River boundary (constant flow)
1: BoundarySetupWithSource(
elev_type=ElevationType.NONE,
vel_type=VelocityType.CONSTANT,
temp_type=TracerType.CONSTANT,
salt_type=TracerType.CONSTANT,
const_flow=-100.0, # m³/s, negative for inflow
const_temp=15.0, # °C
const_salt=0.5, # PSU
),
}
)
Boundary Types
The system supports various boundary condition types for different variables:
Elevation Types
- NONE - No elevation boundary condition
- TIMEHIST - Time history from elev.th
- CONSTANT - Constant elevation
- HARMONIC - Pure harmonic tidal elevation using tidal constituents
- EXTERNAL - Time-varying elevation from external data (elev2D.th.nc)
- HARMONICEXTERNAL - Combined harmonic and external elevation data
Velocity Types
- NONE - No velocity boundary condition
- TIMEHIST - Time history from flux.th
- CONSTANT - Constant velocity/flow rate
- HARMONIC - Pure harmonic tidal velocity using tidal constituents
- EXTERNAL - Time-varying velocity from external data (uv3D.th.nc)
- HARMONICEXTERNAL - Combined harmonic and external velocity data
- FLATHER - Flather type radiation boundary
- RELAXED - Relaxation boundary condition (for nesting)
Tracer Types
- NONE - No tracer boundary condition
- TIMEHIST - Time history from temp/salt.th
- CONSTANT - Constant tracer value
- INITIAL - Initial profile for inflow
- EXTERNAL - Time-varying tracer from external data
Data Sources
The system supports multiple data source types:
DataBlob
Simple file-based data source for pre-processed SCHISM input files:
from rompy.core.data import DataBlob
elev_source = DataBlob(source="path/to/elev2D.th.nc")
SCHISMDataBoundary
Advanced data source with variable mapping and coordinate transformation:
from rompy_schism.data import SCHISMDataBoundary
from rompy.core.source import SourceFile
vel_source = SCHISMDataBoundary(
source=SourceFile(uri="path/to/ocean_model.nc"),
variables=["u", "v"],
crop_coords={"lon": [-180, 180], "lat": [-90, 90]},
)
Configuration Files
The boundary conditions can also be configured via YAML files:
Tidal-Only Configuration:
boundary_conditions:
data_type: boundary_conditions
constituents: ["M2", "S2", "N2", "K1", "O1"]
tidal_database: tpxo
tidal_data:
elevations: path/to/h_tpxo9.nc
velocities: path/to/u_tpxo9.nc
setup_type: tidal
Hybrid Configuration:
boundary_conditions:
data_type: boundary_conditions
constituents: ["M2", "S2", "N2", "K1", "O1"]
tidal_database: tpxo
tidal_data:
elevations: path/to/h_tpxo9.nc
velocities: path/to/u_tpxo9.nc
setup_type: hybrid
boundaries:
0:
elev_type: HARMONICEXTERNAL
vel_type: HARMONICEXTERNAL
temp_type: EXTERNAL
salt_type: EXTERNAL
elev_source:
data_type: blob
source: path/to/elev2D.th.nc
vel_source:
data_type: blob
source: path/to/uv3D.th.nc
temp_source:
data_type: blob
source: path/to/TEM_3D.th.nc
salt_source:
data_type: blob
source: path/to/SAL_3D.th.nc
River Configuration:
boundary_conditions:
data_type: boundary_conditions
constituents: ["M2", "S2"]
tidal_database: tpxo
setup_type: river
boundaries:
0: # Tidal boundary
elev_type: HARMONIC
vel_type: HARMONIC
temp_type: NONE
salt_type: NONE
1: # River boundary
elev_type: NONE
vel_type: CONSTANT
temp_type: CONSTANT
salt_type: CONSTANT
const_flow: -500.0
const_temp: 15.0
const_salt: 0.1
Nested Configuration:
boundary_conditions:
data_type: boundary_conditions
constituents: ["M2", "S2"]
tidal_database: tpxo
tidal_data:
elevations: path/to/h_tpxo9.nc
velocities: path/to/u_tpxo9.nc
setup_type: nested
boundaries:
0:
elev_type: HARMONICEXTERNAL
vel_type: RELAXED
temp_type: EXTERNAL
salt_type: EXTERNAL
inflow_relax: 0.8
outflow_relax: 0.2
elev_source:
data_type: schism_boundary
source:
data_type: source_file
uri: path/to/parent_model.nc
variables: ["ssh"]
vel_source:
data_type: schism_boundary
source:
data_type: source_file
uri: path/to/parent_model.nc
variables: ["u", "v"]
Benefits of the New System
- Unified Interface - Single configuration object for all boundary types
- Flexible Configuration - Mix different boundary types per segment
- Factory Functions - Simplified setup for common scenarios
- Better Validation - Comprehensive validation of boundary configurations
- Data Source Integration - Seamless integration with data processing pipeline
- Backward Compatibility - Maintains compatibility with existing workflows where possible
- Clear Naming - Module and class names reflect actual functionality
- Consolidated Code - Eliminates duplication between modules
Advanced Features
Factory Function Parameters
All factory functions support additional parameters for fine-tuning:
Common Parameters:
constituents: List of tidal constituents (e.g., ["M2", "S2", "N2", "K1", "O1"])tidal_database: Database identifier ("tpxo", "fes2014", "got")tidal_elevations: Path to tidal elevation NetCDF filetidal_velocities: Path to tidal velocity NetCDF file
Tidal Potential:
bc = create_tidal_only_boundary_config(
constituents=["M2", "S2", "K1", "O1"],
ntip=1, # Enable tidal potential
tip_dp=1.0, # Depth threshold
cutoff_depth=50.0, # Cutoff depth
)
Relaxation Parameters:
bc = create_nested_boundary_config(
with_tides=True,
inflow_relax=0.8, # Strong relaxation for inflow
outflow_relax=0.2, # Weak relaxation for outflow
)
Multiple Tidal Databases:
bc = create_tidal_only_boundary_config(
tidal_database="fes2014", # Alternative: "tpxo", "got"
constituents=["M2", "S2", "N2", "K2", "K1", "O1", "P1", "Q1"],
)
Custom Boundary Types:
from rompy_schism.data import BoundarySetupWithSource
from rompy_schism.boundary_core import ElevationType, VelocityType, TracerType
# Custom boundary with specific types
custom_boundary = BoundarySetupWithSource(
elev_type=ElevationType.HARMONICEXTERNAL,
vel_type=VelocityType.HARMONICEXTERNAL,
temp_type=TracerType.EXTERNAL,
salt_type=TracerType.EXTERNAL,
inflow_relax=0.9,
outflow_relax=0.1
)
Flather Radiation Boundaries
Configure Flather radiation boundaries using the low-level BoundaryHandler:
from rompy_schism.boundary_core import BoundaryHandler, ElevationType, VelocityType
# Create boundary handler
boundary = BoundaryHandler(grid_path="path/to/hgrid.gr3")
# Configure Flather boundary
boundary.set_boundary_type(
boundary_index=1,
elev_type=ElevationType.NONE,
vel_type=VelocityType.FLATHER,
eta_mean=[0.0, 0.0, 0.0], # Mean elevation at each node
vn_mean=[[0.1], [0.1], [0.1]] # Mean normal velocity at each node
)
Common Tidal Constituents
Major Constituents (recommended for most applications):
Semi-diurnal:
- M2 (Principal lunar), S2 (Principal solar), N2 (Lunar elliptic), K2 (Lunisolar)
Diurnal:
- K1 (Lunar diurnal), O1 (Lunar principal), P1 (Solar principal), Q1 (Larger lunar elliptic)
Long Period:
- Mf (Lunisolar fortnightly), Mm (Lunar monthly), Ssa (Solar semiannual)
Full Set Example:
bc = create_tidal_only_boundary_config(
constituents=[
"M2", "S2", "N2", "K2", # Semi-diurnal
"K1", "O1", "P1", "Q1", # Diurnal
"Mf", "Mm", "Ssa" # Long period
]
)
Best Practices
- Start Simple: Begin with tidal-only configurations using major constituents
- Validate Data: Ensure tidal and external data files cover your model domain and time period
- Check Units: River flows are in m³/s (negative for inflow)
- Relaxation Values: Use 0.8-1.0 for strong nudging, 0.1-0.3 for weak nudging
- File Formats: Use NetCDF files for better performance and metadata
- Coordinate Systems: Ensure all data sources use consistent coordinate systems
- Time Coverage: External data must cover the entire simulation period plus spin-up
See Also
- Core Data - Core data handling classes
- Core Boundary - Base boundary condition classes
rompy_schism.data.SCHISMData - Main SCHISM configuration classrompy_schism.grid.SCHISMGrid - SCHISM grid handling- Hotstart - Hotstart configuration documentation