swift2 code documentation

placeholder

module classes

CompositeParameteriser

Bases: HypercubeParameteriser

A parameteriser defined as the concatenation of several parameterisers

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

class CompositeParameteriser(HypercubeParameteriser):
    """A parameteriser defined as the concatenation of several parameterisers"""
    def __init__(
        self,
        handle: CffiData,
        release_native: Callable[[CffiData], None],
        type_id: Optional[str] = None,
        prior_ref_count: int = 0,
    ):
        super(CompositeParameteriser, self).__init__(
            handle, release_native, type_id, prior_ref_count
        )

    @staticmethod
    def empty_composite() -> "CompositeParameteriser":
        """Creates an empty parameteriser to be populated with other parameterisers

        Returns:
            CompositeParameteriser: composite parameteriser
        """        
        return swg.CreateCompositeParameterizer_py()

    @staticmethod
    def concatenate(
        *args: Sequence["HypercubeParameteriser"], strategy: str = ""
    ) -> "CompositeParameteriser":
        """Concatenates some hypercubes to a single parameteriser

        Args:
            strategy (str, optional): The strategy to contatenate. Defaults to "", equivalent to "composite", the only available. May have other options in the future.

        Returns:
            CompositeParameteriser: A concatenated parameteriser
        """
        return sp.concatenate_parameterisers(*args, strategy=strategy)

    def append(self, p: "HypercubeParameteriser"):
        """Append a parameteriser to this composite parameteriser

        Args:
            p (HypercubeParameteriser): hypercube to append to this
        """        
        swg.AddToCompositeParameterizer_py(self, p)

append(p)

Append a parameteriser to this composite parameteriser

Parameters:

Name	Type	Description	Default
p	HypercubeParameteriser	hypercube to append to this	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def append(self, p: "HypercubeParameteriser"):
    """Append a parameteriser to this composite parameteriser

    Args:
        p (HypercubeParameteriser): hypercube to append to this
    """        
    swg.AddToCompositeParameterizer_py(self, p)

concatenate(*args, strategy='') staticmethod

Concatenates some hypercubes to a single parameteriser

Parameters:

Name	Type	Description	Default
strategy	str	The strategy to contatenate. Defaults to "", equivalent to "composite", the only available. May have other options in the future.	''

Returns:

Name	Type	Description
CompositeParameteriser	CompositeParameteriser	A concatenated parameteriser

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

@staticmethod
def concatenate(
    *args: Sequence["HypercubeParameteriser"], strategy: str = ""
) -> "CompositeParameteriser":
    """Concatenates some hypercubes to a single parameteriser

    Args:
        strategy (str, optional): The strategy to contatenate. Defaults to "", equivalent to "composite", the only available. May have other options in the future.

    Returns:
        CompositeParameteriser: A concatenated parameteriser
    """
    return sp.concatenate_parameterisers(*args, strategy=strategy)

empty_composite() staticmethod

Creates an empty parameteriser to be populated with other parameterisers

Returns:

Name	Type	Description
CompositeParameteriser	CompositeParameteriser	composite parameteriser

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

@staticmethod
def empty_composite() -> "CompositeParameteriser":
    """Creates an empty parameteriser to be populated with other parameterisers

    Returns:
        CompositeParameteriser: composite parameteriser
    """        
    return swg.CreateCompositeParameterizer_py()

EnsembleSimulation

Bases: DeletableCffiNativeHandle

A simulation designed to facilitate model runs over ensemble of inputs

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

class EnsembleSimulation(DeletableCffiNativeHandle):
    """A simulation designed to facilitate model runs over ensemble of inputs"""
    def __init__(
        self,
        handle: CffiData,
        release_native: Callable[[CffiData], None],
        type_id: Optional[str] = None,
        prior_ref_count: int = 0,
    ):
        super(EnsembleSimulation, self).__init__(
            handle, release_native, type_id, prior_ref_count
        )

    def setup(self, forecast_start: datetime, ensemble_size: int, forecast_horizon_length: int) -> None:
        """Sets up this ensemble simulation

        Args:
            forecast_start (datetime): Start date for the simulation
            ensemble_size (int): size of the ensemble
            forecast_horizon_length (int): length of the simulation in numbers of time steps.
        """        
        swg.SetupEnsembleModelRunner_py(
            self, forecast_start, ensemble_size, forecast_horizon_length
        )

    def record(self, variable_id:str) -> None:
        """Records a state variable of the simualtion

        Args:
            variable_id (str): state variable identifier
        """        
        swg.RecordEnsembleModelRunner_py(self, variable_id)

    def get_simulation_span(self) -> Dict[str, Any]:
        """Gets the span of the simulation: start, end, time step

        Returns:
            Dict[str, Any]: simulation span
        """        
        return swc.get_simulation_span_pkg(self)

    def record_ensemble_state(
        self,
        var_ids: "VecStr" = CATCHMENT_FLOWRATE_VARID,
        recording_provider: Optional["TimeSeriesLibrary"] = None,
        data_ids: Optional["VecStr"] = None,
    ) -> None:
        """Records one or more state values from an ensemble simulation

        Args:
            var_ids (VecStr, optional): Model variable identierfier(s). Defaults to CATCHMENT_FLOWRATE_VARID.
            recording_provider (Optional[TimeSeriesLibrary], optional): An optional time series library to record to. Defaults to None.
            data_ids (Optional[VecStr], optional): Data identifier(s). Defaults to None.
        """    
        spr.record_ensemble_state(self, var_ids, recording_provider, data_ids)

get_simulation_span()

Gets the span of the simulation: start, end, time step

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: simulation span

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def get_simulation_span(self) -> Dict[str, Any]:
    """Gets the span of the simulation: start, end, time step

    Returns:
        Dict[str, Any]: simulation span
    """        
    return swc.get_simulation_span_pkg(self)

record(variable_id)

Records a state variable of the simualtion

Parameters:

Name	Type	Description	Default
variable_id	str	state variable identifier	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def record(self, variable_id:str) -> None:
    """Records a state variable of the simualtion

    Args:
        variable_id (str): state variable identifier
    """        
    swg.RecordEnsembleModelRunner_py(self, variable_id)

record_ensemble_state(var_ids=CATCHMENT_FLOWRATE_VARID, recording_provider=None, data_ids=None)

Records one or more state values from an ensemble simulation

Parameters:

Name	Type	Description	Default
var_ids	VecStr	Model variable identierfier(s). Defaults to CATCHMENT_FLOWRATE_VARID.	CATCHMENT_FLOWRATE_VARID
recording_provider	Optional[TimeSeriesLibrary]	An optional time series library to record to. Defaults to None.	None
data_ids	Optional[VecStr]	Data identifier(s). Defaults to None.	None

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def record_ensemble_state(
    self,
    var_ids: "VecStr" = CATCHMENT_FLOWRATE_VARID,
    recording_provider: Optional["TimeSeriesLibrary"] = None,
    data_ids: Optional["VecStr"] = None,
) -> None:
    """Records one or more state values from an ensemble simulation

    Args:
        var_ids (VecStr, optional): Model variable identierfier(s). Defaults to CATCHMENT_FLOWRATE_VARID.
        recording_provider (Optional[TimeSeriesLibrary], optional): An optional time series library to record to. Defaults to None.
        data_ids (Optional[VecStr], optional): Data identifier(s). Defaults to None.
    """    
    spr.record_ensemble_state(self, var_ids, recording_provider, data_ids)

setup(forecast_start, ensemble_size, forecast_horizon_length)

Sets up this ensemble simulation

Parameters:

Name	Type	Description	Default
forecast_start	datetime	Start date for the simulation	required
ensemble_size	int	size of the ensemble	required
forecast_horizon_length	int	length of the simulation in numbers of time steps.	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def setup(self, forecast_start: datetime, ensemble_size: int, forecast_horizon_length: int) -> None:
    """Sets up this ensemble simulation

    Args:
        forecast_start (datetime): Start date for the simulation
        ensemble_size (int): size of the ensemble
        forecast_horizon_length (int): length of the simulation in numbers of time steps.
    """        
    swg.SetupEnsembleModelRunner_py(
        self, forecast_start, ensemble_size, forecast_horizon_length
    )

HypercubeParameteriser

Bases: Parameteriser

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

class HypercubeParameteriser(Parameteriser):
    def __init__(
        self,
        handle: CffiData,
        release_native: Callable[[CffiData], None],
        type_id: Optional[str] = None,
        prior_ref_count: int = 0,
    ):
        super(HypercubeParameteriser, self).__init__(
            handle, release_native, type_id, prior_ref_count
        )

    def __str__(self):
        """string representation"""
        return str(self.as_dataframe())

    def __repr__(self):
        """representation"""
        return repr(self.as_dataframe())

    def as_dataframe(self) -> pd.DataFrame:
        """Convert this hypercube parameteriser to a pandas data frame representation

        Returns:
            pd.DataFrame: pandas data frame
        """
        return sp.parameteriser_as_dataframe(self)

    def num_free_parameters(self) -> int:
        """Number of free parameters in this hypercube parameteriser

        Returns:
            int: Number of free parameters
        """
        return sp.num_free_parameters(self)

    def set_parameter_value(self, variable_name: "VecStr", value: "VecScalars"):
        """Sets the value(s) of one or more parameter(s)

        Args:
            variable_name (VecStr): one or more parameter name(s)
            value (VecScalars): one or more parameter value(s)
        """
        sp.set_parameter_value(self, variable_name, value)

    def set_max_parameter_value(self, variable_name: "VecStr", value: "VecScalars"):
        """Sets the value(s) of the upper bound of one or more parameter(s)

        Args:
            variable_name (VecStr): one or more parameter name(s)
            value (VecScalars): one or more parameter value(s)
        """
        sp.set_max_parameter_value(self, variable_name, value)

    def set_min_parameter_value(self, variable_name: "VecStr", value: "VecScalars"):
        """Sets the value(s) of the lower bound of one or more parameter(s)

        Args:
            variable_name (VecStr): one or more parameter name(s)
            value (VecScalars): one or more parameter value(s)
        """
        sp.set_min_parameter_value(self, variable_name, value)

    def set_parameter_definition(
        self, variable_name: str, min: float, max: float, value: float
    ):
        """Sets the feasible range and value for a parameter

        Args:
            variable_name (str): parameter name
            min (float): min
            max (float): max
            value (float): value
        """    
        swg.SetParameterDefinition_py(self, variable_name, min, max, value)

    def create_parameter_sampler(self, seed: int = 0, type: str = "urs") -> "CandidateFactorySeed":
        """Creates a sampler for this parameteriser

        Args:
            seed (int, optional): a seed for the sampler. Defaults to 0.
            type (str, optional): the type of sampler. Defaults to "urs". Only option supported as of 2023-01.

        Returns:
            CandidateFactorySeed: a sampler, aka candidate factory
        """        
        return sp.create_parameter_sampler(seed, self, type)

    def make_state_init_parameteriser(self) -> "StateInitParameteriser":
        """Create a parameteriser used for model state initialisation

        This allows to define tied parameters where, for instance, pval = a * modelStateVal. 
        A more concrete use case is to define an initial soil moisture store 'S0',
        as a fraction of the model store capacity 'Smax'. 
        The model state to initialise is 'S'

        Note:
            See also [swift2.classes.ScalingParameteriser][] for typical joint usage.

        Returns:
            StateInitParameteriser: state initialisation parameteriser

        Examples:
            >>> todo()
        """
        return sp.make_state_init_parameteriser(self)

    def filtered_parameters(self) -> "FilteringParameteriser":
        """Wrap a parameteriser in a filter that can hide some parameters"""
        return sp.filtered_parameters(self)

    def hide_parameters(self, patterns, regex=False, starts_with=False, strict=False):
        """Hide some parameters (from the outside e.g. optimisers) in a filter parameteriser

        Args:
            patterns ([type]):  character, one or more pattern to match and hide matching parameters. Match according to other parameters.
            regex (bool, optional): logical, defaults False, should the patterns be used as regular expressions.. Defaults to False.
            starts_with (bool, optional): logical, defaults False. Ignored if regex is True. Should the patterns be used as starting strings in the parameter names.. Defaults to False.
            strict (bool, optional): logical, default False. Used only if regex and starts_with are False. If True, raises an error if one of the "patterns" has no exact match in the parameters.. Defaults to False.
        """
        sp.hide_parameters(self, patterns, regex, starts_with, strict)

    def show_parameters(self, patterns, regex=False, starts_with=False):
        """Show some parameters (from the outside e.g. optimisers) in a filter parameteriser

        Args:
            patterns ([type]):  character, one or more pattern to match and show matching parameters. Match according to other parameters
            regex (bool, optional): should the patterns be used as regular expressions. Defaults to False.
            starts_with (bool, optional): should the patterns be used as starting strings in the parameter names. Defaults to False.
        """
        sp.show_parameters(self, patterns, regex, starts_with)

    def wrap_transform(self) -> "TransformParameteriser":
        """Create a parameteriser for which parameter transformations can be defined.

        This allows to define e.g. a virtual parameter log_X instead of calibrating on the parameter X.

        Returns:
            TransformParameteriser: A new parameteriser (TransformParameteriser) which has methods to define parameter transforms
        """
        return sp.wrap_transform(self)

    def backtransform(self) -> "HypercubeParameteriser":
        """Get the parameteriser values in the untransformed space

        Get the parameteriser values in the untransformed space, i.e. remove any 
        transform added via [`HypercubeParameteriser.wrap_transform`][HypercubeParameteriser.wrap_transform].
        This allows to transform back e.g. from a virtual parameter log_X
        to the underlying model (or even virtual/meta) parameter X.

        Returns:
            HypercubeParameteriser: The parameters definitions without the transforms (if there are any)

        Examples:
            >>> ref_area = 250
            >>> time_span = 3600
            >>> ptrans = sdh.define_gr4j_scaled_parameter(ref_area, time_span)
            >>> ptrans
                Name     Value       Min       Max
            0    log_x4  0.305422  0.000000  2.380211
            1    log_x1  0.506690  0.000000  3.778151
            2    log_x3  0.315425  0.000000  3.000000
            3  asinh_x2  2.637752 -3.989327  3.989327
            >>> ptrans.backtransform()
            Name    Value   Min     Max
            0   x2  6.95511 -27.0    27.0
            1   x3  2.06740   1.0  1000.0
            2   x4  2.02033   1.0   240.0
            3   x1  3.21137   1.0  6000.0
            >>> 
        """
        return sp.backtransform(self)

    @staticmethod
    def from_dataframe(
        type="Generic subareas", definition: Optional[pd.DataFrame] = None
    ) -> "HypercubeParameteriser":
        """Create a parameteriser

        Args:
            type (str, optional): A string identifying the type of parameteriser to use. Recognised types are (case insensitive) 'log-likelihood', 'generic', 'subareas', 'links', 'nodes' and 'muskingum'. Defaults to "Generic subareas".
            specs (pd.DataFrame, optional): An optional data frame description of the parameter set, with at least columns Name, Min, Max, Value. Defaults to None.

        Returns:
            HypercubeParameteriser: new parameteriser

        Examples:
            >>> d = pd.DataFrame(
            ...     dict(
            ...         Name=c("alpha", "inverse_velocity"),
            ...         Value=c(1, 1),
            ...         Min=c(1e-3, 1e-3),
            ...         Max=c(1e2, 1e2),
            ...     )
            ... )
            meteriser.from_dataframe("Generic links", d)
            p
            >>> p = HypercubeParameteriser.from_dataframe("Generic links", d)
        """
        return sp.create_parameteriser(type, specs=definition)

    def add_to_hypercube(self, specs: pd.DataFrame):
        """Add entries to a hypercube. Must be a type of object that is expandable, otherwise may raise an exception.

        Args:
            specs (pd.DataFrame): An optional data frame description of the parameter set, with at least columns Name, Min, Max, Value.
        """
        sp.add_to_hypercube(self, specs)

    def add_parameter_to_hypercube(
        self, name: str, value: float, min: float, max: float
    ):
        """Add a parameter to a hypercube. Must be a type of object that is expandable, otherwise may raise an exception"""
        specs = parameter_df(name, value, min, max)
        sp.add_to_hypercube(self, specs)

    def set_hypercube(self, specs: pd.DataFrame):
        """Set the properties of a hypercube parameteriser

        Args:
            specs (pd.DataFrame): An optional data frame description of the parameter set, with at least columns Name, Min, Max, Value.
        """
        sp.set_hypercube(self, specs)

    def clone(self) -> "HypercubeParameteriser":
        return swg.CloneHypercubeParameterizer_py(self)

__repr__()

representation

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def __repr__(self):
    """representation"""
    return repr(self.as_dataframe())

__str__()

string representation

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def __str__(self):
    """string representation"""
    return str(self.as_dataframe())

add_parameter_to_hypercube(name, value, min, max)

Add a parameter to a hypercube. Must be a type of object that is expandable, otherwise may raise an exception

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def add_parameter_to_hypercube(
    self, name: str, value: float, min: float, max: float
):
    """Add a parameter to a hypercube. Must be a type of object that is expandable, otherwise may raise an exception"""
    specs = parameter_df(name, value, min, max)
    sp.add_to_hypercube(self, specs)

add_to_hypercube(specs)

Add entries to a hypercube. Must be a type of object that is expandable, otherwise may raise an exception.

Parameters:

Name	Type	Description	Default
specs	pd.DataFrame	An optional data frame description of the parameter set, with at least columns Name, Min, Max, Value.	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def add_to_hypercube(self, specs: pd.DataFrame):
    """Add entries to a hypercube. Must be a type of object that is expandable, otherwise may raise an exception.

    Args:
        specs (pd.DataFrame): An optional data frame description of the parameter set, with at least columns Name, Min, Max, Value.
    """
    sp.add_to_hypercube(self, specs)

as_dataframe()

Convert this hypercube parameteriser to a pandas data frame representation

Returns:

Type	Description
pd.DataFrame	pd.DataFrame: pandas data frame

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def as_dataframe(self) -> pd.DataFrame:
    """Convert this hypercube parameteriser to a pandas data frame representation

    Returns:
        pd.DataFrame: pandas data frame
    """
    return sp.parameteriser_as_dataframe(self)

backtransform()

Get the parameteriser values in the untransformed space

Get the parameteriser values in the untransformed space, i.e. remove any transform added via [HypercubeParameteriser.wrap_transform][HypercubeParameteriser.wrap_transform]. This allows to transform back e.g. from a virtual parameter log_X to the underlying model (or even virtual/meta) parameter X.

Returns:

Name	Type	Description
HypercubeParameteriser	HypercubeParameteriser	The parameters definitions without the transforms (if there are any)

Examples:

>>> ref_area = 250
>>> time_span = 3600
>>> ptrans = sdh.define_gr4j_scaled_parameter(ref_area, time_span)
>>> ptrans
    Name     Value       Min       Max
0    log_x4  0.305422  0.000000  2.380211
1    log_x1  0.506690  0.000000  3.778151
2    log_x3  0.315425  0.000000  3.000000
3  asinh_x2  2.637752 -3.989327  3.989327
>>> ptrans.backtransform()
Name    Value   Min     Max
0   x2  6.95511 -27.0    27.0
1   x3  2.06740   1.0  1000.0
2   x4  2.02033   1.0   240.0
3   x1  3.21137   1.0  6000.0
>>>

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def backtransform(self) -> "HypercubeParameteriser":
    """Get the parameteriser values in the untransformed space

    Get the parameteriser values in the untransformed space, i.e. remove any 
    transform added via [`HypercubeParameteriser.wrap_transform`][HypercubeParameteriser.wrap_transform].
    This allows to transform back e.g. from a virtual parameter log_X
    to the underlying model (or even virtual/meta) parameter X.

    Returns:
        HypercubeParameteriser: The parameters definitions without the transforms (if there are any)

    Examples:
        >>> ref_area = 250
        >>> time_span = 3600
        >>> ptrans = sdh.define_gr4j_scaled_parameter(ref_area, time_span)
        >>> ptrans
            Name     Value       Min       Max
        0    log_x4  0.305422  0.000000  2.380211
        1    log_x1  0.506690  0.000000  3.778151
        2    log_x3  0.315425  0.000000  3.000000
        3  asinh_x2  2.637752 -3.989327  3.989327
        >>> ptrans.backtransform()
        Name    Value   Min     Max
        0   x2  6.95511 -27.0    27.0
        1   x3  2.06740   1.0  1000.0
        2   x4  2.02033   1.0   240.0
        3   x1  3.21137   1.0  6000.0
        >>> 
    """
    return sp.backtransform(self)

create_parameter_sampler(seed=0, type='urs')

Creates a sampler for this parameteriser

Parameters:

Name	Type	Description	Default
seed	int	a seed for the sampler. Defaults to 0.	0
type	str	the type of sampler. Defaults to "urs". Only option supported as of 2023-01.	'urs'

Returns:

Name	Type	Description
CandidateFactorySeed	CandidateFactorySeed	a sampler, aka candidate factory

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def create_parameter_sampler(self, seed: int = 0, type: str = "urs") -> "CandidateFactorySeed":
    """Creates a sampler for this parameteriser

    Args:
        seed (int, optional): a seed for the sampler. Defaults to 0.
        type (str, optional): the type of sampler. Defaults to "urs". Only option supported as of 2023-01.

    Returns:
        CandidateFactorySeed: a sampler, aka candidate factory
    """        
    return sp.create_parameter_sampler(seed, self, type)

filtered_parameters()

Wrap a parameteriser in a filter that can hide some parameters

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def filtered_parameters(self) -> "FilteringParameteriser":
    """Wrap a parameteriser in a filter that can hide some parameters"""
    return sp.filtered_parameters(self)

from_dataframe(type='Generic subareas', definition=None) staticmethod

Create a parameteriser

Parameters:

Name	Type	Description	Default
type	str	A string identifying the type of parameteriser to use. Recognised types are (case insensitive) 'log-likelihood', 'generic', 'subareas', 'links', 'nodes' and 'muskingum'. Defaults to "Generic subareas".	'Generic subareas'
specs	pd.DataFrame	An optional data frame description of the parameter set, with at least columns Name, Min, Max, Value. Defaults to None.	required

Returns:

Name	Type	Description
HypercubeParameteriser	HypercubeParameteriser	new parameteriser

Examples:

>>> d = pd.DataFrame(
...     dict(
...         Name=c("alpha", "inverse_velocity"),
...         Value=c(1, 1),
...         Min=c(1e-3, 1e-3),
...         Max=c(1e2, 1e2),
...     )
... )
meteriser.from_dataframe("Generic links", d)
p
>>> p = HypercubeParameteriser.from_dataframe("Generic links", d)

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

@staticmethod
def from_dataframe(
    type="Generic subareas", definition: Optional[pd.DataFrame] = None
) -> "HypercubeParameteriser":
    """Create a parameteriser

    Args:
        type (str, optional): A string identifying the type of parameteriser to use. Recognised types are (case insensitive) 'log-likelihood', 'generic', 'subareas', 'links', 'nodes' and 'muskingum'. Defaults to "Generic subareas".
        specs (pd.DataFrame, optional): An optional data frame description of the parameter set, with at least columns Name, Min, Max, Value. Defaults to None.

    Returns:
        HypercubeParameteriser: new parameteriser

    Examples:
        >>> d = pd.DataFrame(
        ...     dict(
        ...         Name=c("alpha", "inverse_velocity"),
        ...         Value=c(1, 1),
        ...         Min=c(1e-3, 1e-3),
        ...         Max=c(1e2, 1e2),
        ...     )
        ... )
        meteriser.from_dataframe("Generic links", d)
        p
        >>> p = HypercubeParameteriser.from_dataframe("Generic links", d)
    """
    return sp.create_parameteriser(type, specs=definition)

hide_parameters(patterns, regex=False, starts_with=False, strict=False)

Hide some parameters (from the outside e.g. optimisers) in a filter parameteriser

Parameters:

Name	Type	Description	Default
patterns	[type]	character, one or more pattern to match and hide matching parameters. Match according to other parameters.	required
regex	bool	logical, defaults False, should the patterns be used as regular expressions.. Defaults to False.	False
starts_with	bool	logical, defaults False. Ignored if regex is True. Should the patterns be used as starting strings in the parameter names.. Defaults to False.	False
strict	bool	logical, default False. Used only if regex and starts_with are False. If True, raises an error if one of the "patterns" has no exact match in the parameters.. Defaults to False.	False

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def hide_parameters(self, patterns, regex=False, starts_with=False, strict=False):
    """Hide some parameters (from the outside e.g. optimisers) in a filter parameteriser

    Args:
        patterns ([type]):  character, one or more pattern to match and hide matching parameters. Match according to other parameters.
        regex (bool, optional): logical, defaults False, should the patterns be used as regular expressions.. Defaults to False.
        starts_with (bool, optional): logical, defaults False. Ignored if regex is True. Should the patterns be used as starting strings in the parameter names.. Defaults to False.
        strict (bool, optional): logical, default False. Used only if regex and starts_with are False. If True, raises an error if one of the "patterns" has no exact match in the parameters.. Defaults to False.
    """
    sp.hide_parameters(self, patterns, regex, starts_with, strict)

make_state_init_parameteriser()

Create a parameteriser used for model state initialisation

This allows to define tied parameters where, for instance, pval = a * modelStateVal. A more concrete use case is to define an initial soil moisture store 'S0', as a fraction of the model store capacity 'Smax'. The model state to initialise is 'S'

Note

See also swift2.classes.ScalingParameteriser for typical joint usage.

Returns:

Name	Type	Description
StateInitParameteriser	StateInitParameteriser	state initialisation parameteriser

Examples:

>>> todo()

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def make_state_init_parameteriser(self) -> "StateInitParameteriser":
    """Create a parameteriser used for model state initialisation

    This allows to define tied parameters where, for instance, pval = a * modelStateVal. 
    A more concrete use case is to define an initial soil moisture store 'S0',
    as a fraction of the model store capacity 'Smax'. 
    The model state to initialise is 'S'

    Note:
        See also [swift2.classes.ScalingParameteriser][] for typical joint usage.

    Returns:
        StateInitParameteriser: state initialisation parameteriser

    Examples:
        >>> todo()
    """
    return sp.make_state_init_parameteriser(self)

num_free_parameters()

Number of free parameters in this hypercube parameteriser

Returns:

Name	Type	Description
int	int	Number of free parameters

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def num_free_parameters(self) -> int:
    """Number of free parameters in this hypercube parameteriser

    Returns:
        int: Number of free parameters
    """
    return sp.num_free_parameters(self)

set_hypercube(specs)

Set the properties of a hypercube parameteriser

Parameters:

Name	Type	Description	Default
specs	pd.DataFrame	An optional data frame description of the parameter set, with at least columns Name, Min, Max, Value.	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def set_hypercube(self, specs: pd.DataFrame):
    """Set the properties of a hypercube parameteriser

    Args:
        specs (pd.DataFrame): An optional data frame description of the parameter set, with at least columns Name, Min, Max, Value.
    """
    sp.set_hypercube(self, specs)

set_max_parameter_value(variable_name, value)

Sets the value(s) of the upper bound of one or more parameter(s)

Parameters:

Name	Type	Description	Default
variable_name	VecStr	one or more parameter name(s)	required
value	VecScalars	one or more parameter value(s)	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def set_max_parameter_value(self, variable_name: "VecStr", value: "VecScalars"):
    """Sets the value(s) of the upper bound of one or more parameter(s)

    Args:
        variable_name (VecStr): one or more parameter name(s)
        value (VecScalars): one or more parameter value(s)
    """
    sp.set_max_parameter_value(self, variable_name, value)

set_min_parameter_value(variable_name, value)

Sets the value(s) of the lower bound of one or more parameter(s)

Parameters:

Name	Type	Description	Default
variable_name	VecStr	one or more parameter name(s)	required
value	VecScalars	one or more parameter value(s)	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def set_min_parameter_value(self, variable_name: "VecStr", value: "VecScalars"):
    """Sets the value(s) of the lower bound of one or more parameter(s)

    Args:
        variable_name (VecStr): one or more parameter name(s)
        value (VecScalars): one or more parameter value(s)
    """
    sp.set_min_parameter_value(self, variable_name, value)

set_parameter_definition(variable_name, min, max, value)

Sets the feasible range and value for a parameter

Parameters:

Name	Type	Description	Default
variable_name	str	parameter name	required
min	float	min	required
max	float	max	required
value	float	value	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def set_parameter_definition(
    self, variable_name: str, min: float, max: float, value: float
):
    """Sets the feasible range and value for a parameter

    Args:
        variable_name (str): parameter name
        min (float): min
        max (float): max
        value (float): value
    """    
    swg.SetParameterDefinition_py(self, variable_name, min, max, value)

set_parameter_value(variable_name, value)

Sets the value(s) of one or more parameter(s)

Parameters:

Name	Type	Description	Default
variable_name	VecStr	one or more parameter name(s)	required
value	VecScalars	one or more parameter value(s)	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def set_parameter_value(self, variable_name: "VecStr", value: "VecScalars"):
    """Sets the value(s) of one or more parameter(s)

    Args:
        variable_name (VecStr): one or more parameter name(s)
        value (VecScalars): one or more parameter value(s)
    """
    sp.set_parameter_value(self, variable_name, value)

show_parameters(patterns, regex=False, starts_with=False)

Show some parameters (from the outside e.g. optimisers) in a filter parameteriser

Parameters:

Name	Type	Description	Default
patterns	[type]	character, one or more pattern to match and show matching parameters. Match according to other parameters	required
regex	bool	should the patterns be used as regular expressions. Defaults to False.	False
starts_with	bool	should the patterns be used as starting strings in the parameter names. Defaults to False.	False

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def show_parameters(self, patterns, regex=False, starts_with=False):
    """Show some parameters (from the outside e.g. optimisers) in a filter parameteriser

    Args:
        patterns ([type]):  character, one or more pattern to match and show matching parameters. Match according to other parameters
        regex (bool, optional): should the patterns be used as regular expressions. Defaults to False.
        starts_with (bool, optional): should the patterns be used as starting strings in the parameter names. Defaults to False.
    """
    sp.show_parameters(self, patterns, regex, starts_with)

wrap_transform()

Create a parameteriser for which parameter transformations can be defined.

This allows to define e.g. a virtual parameter log_X instead of calibrating on the parameter X.

Returns:

Name	Type	Description
TransformParameteriser	TransformParameteriser	A new parameteriser (TransformParameteriser) which has methods to define parameter transforms

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def wrap_transform(self) -> "TransformParameteriser":
    """Create a parameteriser for which parameter transformations can be defined.

    This allows to define e.g. a virtual parameter log_X instead of calibrating on the parameter X.

    Returns:
        TransformParameteriser: A new parameteriser (TransformParameteriser) which has methods to define parameter transforms
    """
    return sp.wrap_transform(self)

ObjectiveEvaluator

Bases: DeletableCffiNativeHandle

Objective Evaluator

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

class ObjectiveEvaluator(DeletableCffiNativeHandle):
    """Objective Evaluator"""
    def __init__(
        self,
        handle: CffiData,
        release_native: Callable[[CffiData], None],
        type_id: Optional[str] = None,
        prior_ref_count: int = 0,
    ):
        super(ObjectiveEvaluator, self).__init__(
            handle, release_native, type_id, prior_ref_count
        )

    def create_sce_optim_swift(
        self,
        termination_criterion: Optional["SceTerminationCondition"] = None,
        sce_params: Optional[Dict[str, float]] = None,
        population_initialiser: Optional[Union["CandidateFactorySeed", "HypercubeParameteriser"]] = None,
    ) -> "Optimiser":
        """Creates a shuffled complex optimiser for this objective

        Args:
            termination_criterion (Optional["SceTerminationCondition"], optional): A termination criterion for the optimiser. Defaults to None, in which case an arbitrary "relative standard deviation" is set up.
            sce_params (Optional[Dict[str, float]], optional): hyperparameters controlling SCE. Defaults to None, in which case [`swift2.parameteriser.get_default_sce_parameters`][swift2.parameteriser.get_default_sce_parameters] is used.
            population_initialiser (Optional["CandidateFactorySeed"], optional): A candidate factory to initialise the population of parameters the optimiser starts from, or a hypercube. In the latter case, uniform random sampling is used. Defaults to None, which leads to an error (for legacy reasons). 

        Returns:
            Optimiser: SCE optimiser
        """    
        return sp.create_sce_optim_swift(
            self, termination_criterion, sce_params, population_initialiser
        )

    # CreateOptimizerWila?

    def get_score(self, p_set: "HypercubeParameteriser") -> Dict[str,Any]:
        """Evaluate this objective for a given parameterisation

        Args:
            p_set (HypercubeParameteriser): parameteriser

        Returns:
            Dict[str,Any]: score(s), and a data frame representation of the input parameters.
        """
        return ssf.get_score(self, p_set)

    def get_scores(self, p_set: "HypercubeParameteriser") -> Dict[str,float]:
        """Evaluate this objective for a given parameterisation

        Args:
            p_set (HypercubeParameteriser): parameteriser

        Returns:
            Dict[str,float]: score(s)
        """
        return swg.EvaluateScoresForParametersWila_py(self, p_set)

    # EvaluateScoreForParametersWilaInitState?

    @staticmethod
    def create_composite_objective(
        objectives: Sequence["ObjectiveEvaluator"],
        weights: Sequence[float],
        names: Sequence[str],
    ) -> "ObjectiveEvaluator":
        """Creates a composite objective, weighted average of several objectives

        Args:
            objectives (Sequence["ObjectiveEvaluator"]): objective evaluators, for instance measures at several points in the catchment
            weights (Sequence[float]): Weights to use to average the objectives. This may not add to one, but must not sum to zero
            names (Sequence[str]): Names of individual objectives

        Returns:
            ObjectiveEvaluator: An objective evaluator that can be use by an optimiser
        """    
        return ssf.create_composite_objective(objectives, weights, names)

create_composite_objective(objectives, weights, names) staticmethod

Creates a composite objective, weighted average of several objectives

Parameters:

Name	Type	Description	Default
objectives	Sequence["ObjectiveEvaluator"]	objective evaluators, for instance measures at several points in the catchment	required
weights	Sequence[float]	Weights to use to average the objectives. This may not add to one, but must not sum to zero	required
names	Sequence[str]	Names of individual objectives	required

Returns:

Name	Type	Description
ObjectiveEvaluator	ObjectiveEvaluator	An objective evaluator that can be use by an optimiser

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

@staticmethod
def create_composite_objective(
    objectives: Sequence["ObjectiveEvaluator"],
    weights: Sequence[float],
    names: Sequence[str],
) -> "ObjectiveEvaluator":
    """Creates a composite objective, weighted average of several objectives

    Args:
        objectives (Sequence["ObjectiveEvaluator"]): objective evaluators, for instance measures at several points in the catchment
        weights (Sequence[float]): Weights to use to average the objectives. This may not add to one, but must not sum to zero
        names (Sequence[str]): Names of individual objectives

    Returns:
        ObjectiveEvaluator: An objective evaluator that can be use by an optimiser
    """    
    return ssf.create_composite_objective(objectives, weights, names)

create_sce_optim_swift(termination_criterion=None, sce_params=None, population_initialiser=None)

Creates a shuffled complex optimiser for this objective

Parameters:

Name	Type	Description	Default
termination_criterion	Optional["SceTerminationCondition"]	A termination criterion for the optimiser. Defaults to None, in which case an arbitrary "relative standard deviation" is set up.	None
sce_params	Optional[Dict[str, float]]	hyperparameters controlling SCE. Defaults to None, in which case swift2.parameteriser.get_default_sce_parameters is used.	None
population_initialiser	Optional["CandidateFactorySeed"]	A candidate factory to initialise the population of parameters the optimiser starts from, or a hypercube. In the latter case, uniform random sampling is used. Defaults to None, which leads to an error (for legacy reasons).	None

Returns:

Name	Type	Description
Optimiser	Optimiser	SCE optimiser

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def create_sce_optim_swift(
    self,
    termination_criterion: Optional["SceTerminationCondition"] = None,
    sce_params: Optional[Dict[str, float]] = None,
    population_initialiser: Optional[Union["CandidateFactorySeed", "HypercubeParameteriser"]] = None,
) -> "Optimiser":
    """Creates a shuffled complex optimiser for this objective

    Args:
        termination_criterion (Optional["SceTerminationCondition"], optional): A termination criterion for the optimiser. Defaults to None, in which case an arbitrary "relative standard deviation" is set up.
        sce_params (Optional[Dict[str, float]], optional): hyperparameters controlling SCE. Defaults to None, in which case [`swift2.parameteriser.get_default_sce_parameters`][swift2.parameteriser.get_default_sce_parameters] is used.
        population_initialiser (Optional["CandidateFactorySeed"], optional): A candidate factory to initialise the population of parameters the optimiser starts from, or a hypercube. In the latter case, uniform random sampling is used. Defaults to None, which leads to an error (for legacy reasons). 

    Returns:
        Optimiser: SCE optimiser
    """    
    return sp.create_sce_optim_swift(
        self, termination_criterion, sce_params, population_initialiser
    )

get_score(p_set)

Evaluate this objective for a given parameterisation

Parameters:

Name	Type	Description	Default
p_set	HypercubeParameteriser	parameteriser	required

Returns:

Type	Description
Dict[str, Any]	Dict[str,Any]: score(s), and a data frame representation of the input parameters.

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def get_score(self, p_set: "HypercubeParameteriser") -> Dict[str,Any]:
    """Evaluate this objective for a given parameterisation

    Args:
        p_set (HypercubeParameteriser): parameteriser

    Returns:
        Dict[str,Any]: score(s), and a data frame representation of the input parameters.
    """
    return ssf.get_score(self, p_set)

get_scores(p_set)

Evaluate this objective for a given parameterisation

Parameters:

Name	Type	Description	Default
p_set	HypercubeParameteriser	parameteriser	required

Returns:

Type	Description
Dict[str, float]	Dict[str,float]: score(s)

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def get_scores(self, p_set: "HypercubeParameteriser") -> Dict[str,float]:
    """Evaluate this objective for a given parameterisation

    Args:
        p_set (HypercubeParameteriser): parameteriser

    Returns:
        Dict[str,float]: score(s)
    """
    return swg.EvaluateScoresForParametersWila_py(self, p_set)

Parameteriser

Bases: DeletableCffiNativeHandle

Wrapper around a native parameteriser.

Note

This is a parent class for more common types such as swift2.classes.HypercubeParameteriser

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

class Parameteriser(DeletableCffiNativeHandle):
    """Wrapper around a native parameteriser.

    Note:
        This is a parent class for more common types such as 
        [swift2.classes.HypercubeParameteriser][]

    """
    def __init__(
        self,
        handle: CffiData,
        release_native: Callable[[CffiData], None],
        type_id: Optional[str] = None,
        prior_ref_count: int = 0,
    ):
        super(Parameteriser, self).__init__(
            handle, release_native, type_id, prior_ref_count
        )

    def score_for_objective(self, objective: "ObjectiveEvaluator") -> Dict[str, Any]:
        """Computes the value of an objective for this given set of parameters"""
        return sp.evaluate_score_for_parameters(objective, self)

    # def _sapply_parameter_set(self, variable_name, value, api_func):
    #     sp._sapply_parameter_set(self, variable_name, value, api_func)

    def subcatchment_parameteriser(self, subcatchment):
        """Create a parameteriser that gets applied to a subset of a whole catchment

        Args:
            subcatchment (Simulation): the subcatchment, subset of a whole catchment, identifying which elements (subareas, nodes, links) will be parameterised by the new parameteriser. Other catchment elements are untouched.

        Returns:
            HypercubeParameteriser: New parameteriser whose application is limited to the subcatchment.

        Examples:
            >>> sub_cats = simulation.split_to_subcatchments(["node.node_7", "node.node_10"])
            >>> sc = sub_cats["node.node_7"]
            >>> p = sp.create_parameteriser('generic subarea')
            >>> p.add_parameter_to_hypercube("x1", 32, 1, 6e3)
            >>> sp = p.subcatchment_parameteriser(sc)
            >>> sp.apply_sys_config(simulation)

        """
        return sp.subcatchment_parameteriser(self, subcatchment)

    def apply_sys_config(self, simulation:"Simulation"):
        """Apply a model configuration to a simulation

        Args:
            simulation (Simulation): simulation
        """
        sp.apply_sys_config(self, simulation)

    def supports_thread_safe_cloning(self) -> bool:
        """Is this parameteriser clonable as a deep copy, safe for multi-threading?"""
        return swg.SupportsThreadSafeCloning_py(self)

apply_sys_config(simulation)

Apply a model configuration to a simulation

Parameters:

Name	Type	Description	Default
simulation	Simulation	simulation	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def apply_sys_config(self, simulation:"Simulation"):
    """Apply a model configuration to a simulation

    Args:
        simulation (Simulation): simulation
    """
    sp.apply_sys_config(self, simulation)

score_for_objective(objective)

Computes the value of an objective for this given set of parameters

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def score_for_objective(self, objective: "ObjectiveEvaluator") -> Dict[str, Any]:
    """Computes the value of an objective for this given set of parameters"""
    return sp.evaluate_score_for_parameters(objective, self)

subcatchment_parameteriser(subcatchment)

Create a parameteriser that gets applied to a subset of a whole catchment

Parameters:

Name	Type	Description	Default
subcatchment	Simulation	the subcatchment, subset of a whole catchment, identifying which elements (subareas, nodes, links) will be parameterised by the new parameteriser. Other catchment elements are untouched.	required

Returns:

Name	Type	Description
HypercubeParameteriser		New parameteriser whose application is limited to the subcatchment.

Examples:

>>> sub_cats = simulation.split_to_subcatchments(["node.node_7", "node.node_10"])
>>> sc = sub_cats["node.node_7"]
>>> p = sp.create_parameteriser('generic subarea')
>>> p.add_parameter_to_hypercube("x1", 32, 1, 6e3)
>>> sp = p.subcatchment_parameteriser(sc)
>>> sp.apply_sys_config(simulation)

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def subcatchment_parameteriser(self, subcatchment):
    """Create a parameteriser that gets applied to a subset of a whole catchment

    Args:
        subcatchment (Simulation): the subcatchment, subset of a whole catchment, identifying which elements (subareas, nodes, links) will be parameterised by the new parameteriser. Other catchment elements are untouched.

    Returns:
        HypercubeParameteriser: New parameteriser whose application is limited to the subcatchment.

    Examples:
        >>> sub_cats = simulation.split_to_subcatchments(["node.node_7", "node.node_10"])
        >>> sc = sub_cats["node.node_7"]
        >>> p = sp.create_parameteriser('generic subarea')
        >>> p.add_parameter_to_hypercube("x1", 32, 1, 6e3)
        >>> sp = p.subcatchment_parameteriser(sc)
        >>> sp.apply_sys_config(simulation)

    """
    return sp.subcatchment_parameteriser(self, subcatchment)

supports_thread_safe_cloning()

Is this parameteriser clonable as a deep copy, safe for multi-threading?

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def supports_thread_safe_cloning(self) -> bool:
    """Is this parameteriser clonable as a deep copy, safe for multi-threading?"""
    return swg.SupportsThreadSafeCloning_py(self)

ScalingParameteriser

Bases: TransformParameteriser

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

class ScalingParameteriser(TransformParameteriser):
    def __init__(
        self,
        handle: CffiData,
        release_native: Callable[[CffiData], None],
        type_id: Optional[str] = None,
        prior_ref_count: int = 0,
    ):
        super(ScalingParameteriser, self).__init__(
            handle, release_native, type_id, prior_ref_count
        )

    @staticmethod
    def linear_parameteriser_from(
        data_frame: pd.DataFrame, selector_type: str = "subareas"
    ):
        """Create a scaled linear parameteriser, tying by a linear transformation a (resp several) model state to a (resp several) parameter values
        This allows to define tied parameters where pval = a * modelStateVal + intercept. The intent in particular is to define virtual model parameters such as initial store value as a fraction of the maximum storage capacity.

        Args:
            data_frame (pd.DataFrame): data frame with columns "param_name", "state_name", "scaling_var_name", "min_value", "max_value", "value", "intercept",
            selector_type (str, optional): [description]. Defaults to "subareas".

        Returns:
            ScalingParameteriser: ScalingParameteriser
        """
        return sp.linear_parameteriser_from(data_frame, selector_type)

    @staticmethod
    def linear_parameteriser(
        param_name: "VecStr",
        state_name: "VecStr",
        scaling_var_name: "VecStr",
        min_p_val: "VecNum",
        max_p_val: "VecNum",
        value: "VecNum",
        selector_type: str = "subareas",
        intercept: "VecNum" = 0.0,
    ):
        """Create a scaled linear parameteriser, tying by a linear transformation a (resp several) model state to a (resp several) parameter values

        This allows to define tied parameters where pval = a * modelStateVal + intercept.
        The intent in particular is to define virtual model parameters such as initial store value as a fraction of the maximum storage capacity.

        Args:

            param_name (VecStr): the name of the meta-parameter. Note that it can be the same value as inner_param_name without interference, though this may be confusing a choice.
            state_name (VecStr): the name of the model state to modify, based on the value of the meta-parameter and the state found in 'scalingVarName'
            scaling_var_name (VecStr): the name of the parameter for each subarea model, to which to apply the area scaled value.
            min_p_val (VecNum): minimum value allowed for the meta-parameter
            max_p_val (VecNum): minimum value allowed for the meta-parameter
            value (VecNum): value for the meta parameter.
            selector_type (str, optional): an identifier to define to which catchment element(s) the parameteriser will be applied. Defaults to "subareas".
            intercept (VecNum, optional): [description]. Defaults to 0.0.

        Returns:
            ScalingParameteriser: new ScalingParameteriser
        """
        return sp.linear_parameteriser(
            param_name,
            state_name,
            scaling_var_name,
            min_p_val,
            max_p_val,
            value,
            selector_type,
            intercept,
        )

    def add_linear_scaled_parameter(
        self,
        param_name: str,
        state_name: str,
        scaling_var_name: str,
        min_p_val: float,
        max_p_val: float,
        value: float,
        intercept: float = 0.0,
    ):
        swg.AddLinearScalingParameterizer_py(
            self,
            param_name,
            state_name,
            scaling_var_name,
            intercept,
            min_p_val,
            max_p_val,
            value,
        )

linear_parameteriser(param_name, state_name, scaling_var_name, min_p_val, max_p_val, value, selector_type='subareas', intercept=0.0) staticmethod

Create a scaled linear parameteriser, tying by a linear transformation a (resp several) model state to a (resp several) parameter values

This allows to define tied parameters where pval = a * modelStateVal + intercept. The intent in particular is to define virtual model parameters such as initial store value as a fraction of the maximum storage capacity.

Parameters:

Name	Type	Description	Default
param_name	VecStr	the name of the meta-parameter. Note that it can be the same value as inner_param_name without interference, though this may be confusing a choice.	required
state_name	VecStr	the name of the model state to modify, based on the value of the meta-parameter and the state found in 'scalingVarName'	required
scaling_var_name	VecStr	the name of the parameter for each subarea model, to which to apply the area scaled value.	required
min_p_val	VecNum	minimum value allowed for the meta-parameter	required
max_p_val	VecNum	minimum value allowed for the meta-parameter	required
value	VecNum	value for the meta parameter.	required
selector_type	str	an identifier to define to which catchment element(s) the parameteriser will be applied. Defaults to "subareas".	'subareas'
intercept	VecNum	[description]. Defaults to 0.0.	0.0

Returns:

Name	Type	Description
ScalingParameteriser		new ScalingParameteriser

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

@staticmethod
def linear_parameteriser(
    param_name: "VecStr",
    state_name: "VecStr",
    scaling_var_name: "VecStr",
    min_p_val: "VecNum",
    max_p_val: "VecNum",
    value: "VecNum",
    selector_type: str = "subareas",
    intercept: "VecNum" = 0.0,
):
    """Create a scaled linear parameteriser, tying by a linear transformation a (resp several) model state to a (resp several) parameter values

    This allows to define tied parameters where pval = a * modelStateVal + intercept.
    The intent in particular is to define virtual model parameters such as initial store value as a fraction of the maximum storage capacity.

    Args:

        param_name (VecStr): the name of the meta-parameter. Note that it can be the same value as inner_param_name without interference, though this may be confusing a choice.
        state_name (VecStr): the name of the model state to modify, based on the value of the meta-parameter and the state found in 'scalingVarName'
        scaling_var_name (VecStr): the name of the parameter for each subarea model, to which to apply the area scaled value.
        min_p_val (VecNum): minimum value allowed for the meta-parameter
        max_p_val (VecNum): minimum value allowed for the meta-parameter
        value (VecNum): value for the meta parameter.
        selector_type (str, optional): an identifier to define to which catchment element(s) the parameteriser will be applied. Defaults to "subareas".
        intercept (VecNum, optional): [description]. Defaults to 0.0.

    Returns:
        ScalingParameteriser: new ScalingParameteriser
    """
    return sp.linear_parameteriser(
        param_name,
        state_name,
        scaling_var_name,
        min_p_val,
        max_p_val,
        value,
        selector_type,
        intercept,
    )

linear_parameteriser_from(data_frame, selector_type='subareas') staticmethod

Create a scaled linear parameteriser, tying by a linear transformation a (resp several) model state to a (resp several) parameter values This allows to define tied parameters where pval = a * modelStateVal + intercept. The intent in particular is to define virtual model parameters such as initial store value as a fraction of the maximum storage capacity.

Parameters:

Name	Type	Description	Default
data_frame	pd.DataFrame	data frame with columns "param_name", "state_name", "scaling_var_name", "min_value", "max_value", "value", "intercept",	required
selector_type	str	[description]. Defaults to "subareas".	'subareas'

Returns:

Name	Type	Description
ScalingParameteriser		ScalingParameteriser

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

@staticmethod
def linear_parameteriser_from(
    data_frame: pd.DataFrame, selector_type: str = "subareas"
):
    """Create a scaled linear parameteriser, tying by a linear transformation a (resp several) model state to a (resp several) parameter values
    This allows to define tied parameters where pval = a * modelStateVal + intercept. The intent in particular is to define virtual model parameters such as initial store value as a fraction of the maximum storage capacity.

    Args:
        data_frame (pd.DataFrame): data frame with columns "param_name", "state_name", "scaling_var_name", "min_value", "max_value", "value", "intercept",
        selector_type (str, optional): [description]. Defaults to "subareas".

    Returns:
        ScalingParameteriser: ScalingParameteriser
    """
    return sp.linear_parameteriser_from(data_frame, selector_type)

Simulation

Bases: DeletableCffiNativeHandle, SimulationMixin

Wrapper around single dimension simulation objects

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

class Simulation(DeletableCffiNativeHandle, SimulationMixin):
    """Wrapper around single dimension simulation objects"""

    def __init__(
        self,
        handle: CffiData,
        release_native: Callable[[CffiData], None],
        type_id: Optional[str] = None,
        prior_ref_count: int = 0,
    ):
        super(Simulation, self).__init__(
            handle, release_native, type_id, prior_ref_count
        )

    def clone(self) -> "Simulation":
        """Clone this simulation (deep copy)

        Returns:
            Simulation: A new simulation object
        """
        return swg.CloneModel_py(self)

    def __str__(self):
        """string representation"""
        tid = self.type_id if self.type_id is not None else ""
        return f'Simulation wrapper for a CFFI pointer handle to a native pointer of type id "{tid}"'

    def get_simulation_span(self) -> Dict[str, Any]:
        """Gets the simulation span of this simulation

        Returns:
            Dict[str,Any]: information on start, end, time step
        """
        return swc.get_simulation_span_pkg(self)

    def set_simulation_span(
        self, start: ConvertibleToTimestamp, end: ConvertibleToTimestamp
    ) -> None:
        """
        Sets the simulation span

        Args:
            start (ConvertibleToTimestamp): the start date of the simulation. The time zone will be forced to UTC.
            end (ConvertibleToTimestamp): the end date of the simulation. The time zone will be forced to UTC.
        """
        ss.set_simulation_span(self, start, end)

    def set_simulation_time_step(self, name: str) -> None:
        """
        Sets the time step of this simulation

        Args:
            name (str): a time step identifier, currently 'daily' or 'hourly' are supported. The identifier is made lower case in the function.
        """
        ss.set_simulation_time_step(self, name)

    def check_simulation(self) -> Dict:
        """
        Checks whether a simulation is configured to a state where it is executable
        """
        return ss.check_simulation(self)

    def swap_model(self, model_id: str, what: str = "runoff") -> "Simulation":
        """
        Clone and change a simulation, using another model

        Args:
            model_id (str): the identifier of the new model to use, e.g. 'GR4J'
            what (str): character identifying the type of structure replaced: 'runoff', 'channel_routing'

        Returns:
            Simulation: A SWIFT simulation object, clone of the simulation but with a new model type in use.

        """
        return ss.swap_model(self, model_id, what)

    def set_error_correction_model(
        self, model_id: str, element_id: str, length: int = 1, seed: int = 0
    ) -> None:
        """
        Add an error correction model to an element in a catchment

        Args:
            model_id (str): the identifier of the new model to use, e.g. 'ERRIS'
            element_id (str): the identifier of the catchment element (node, link, subcatchment) whose outflow rate is corrected.
            length (int): other parameters to pass to the creation of the error correction model. Currently length of the AR model only supported.
            seed (int): other parameters to pass to the creation of the error correction model. Currently length of the AR model only supported.

        """
        ss.set_error_correction_model(self, model_id, element_id, length, seed)

    def sort_by_execution_order(
        self, split_element_ids: Sequence[str], sorting_option: str = ""
    ) -> List[str]:
        """
        Sort the specified element ids according to the execution order of the simulation

        Args:
            split_element_ids (Sequence[str]): a character vector with element identifiers such as 'node.n1', 'link.linkId_2'
            sorting_option (str): a character - for future options. Ignored for now.

        Returns:
            List[str]: values in split_element_ids sorted by simulation execution order

        """
        return ss.sort_by_execution_order(self, split_element_ids, sorting_option)

    def get_link_names(self) -> List[str]:
        """
        Gets all the names of the links in the catchment
        """
        return ss.get_link_names(self)

    def get_node_names(self) -> List[str]:
        """
        Gets all the names of the nodes in the catchment
        """
        return ss.get_node_names(self)

    def get_subarea_names(self) -> List[str]:
        """
        Gets all the names of the subareas in the catchment
        """
        return ss.get_subarea_names(self)

    def get_link_ids(self) -> List[str]:
        """
        Gets all the identifiers of the links in the catchment
        """
        return ss.get_link_ids(self)

    def get_node_ids(self) -> List[str]:
        """
        Gets all the identifiers of the nodes in the catchment
        """
        return ss.get_node_ids(self)

    def get_subarea_ids(self) -> List[str]:
        """
        Gets all the identifiers of the subareas in the catchment
        """
        return ss.get_subarea_ids(self)

    def get_variable_ids(
        self, element_id: Optional[str] = None, full_id: bool = True
    ) -> List[str]:
        """
        Gets all the names of the variables of an element (link, node, subarea) within a catchment

        Args:
            element_id (Optional[str]): a character, identifier of the element within the catchment
            full_id (bool): boolean, if TRUE return the full hierarchical identifier

        """
        return ss.get_variable_ids(self, element_id, full_id)

    def is_variable_id(self, var_id: "VecStr") -> Union[Dict[str, bool], bool]:
        """Are one or more model state identifier(s) valid

        Args:
            var_id (VecStr): model identifier(s)

        Returns:
            Union[Dict[str, bool], bool]: whether the identifier(s) are valid. A dictionary is returned if the input is vectorised rather than scalar.
        """
        return ss.is_variable_id(self, var_id)

    def get_state_value(self, var_id: "VecStr") -> Union[Dict[str, float], float]:
        """
        Gets the value(s) of a model state(s)

        Args:
            var_id (VecStr): string or sequence of str, model variable state identifier(s)

        Returns:
            value(s) of the requested model states
        """
        return ss.get_state_value(self, var_id)

    def set_state_value(
        self,
        var_id: Union[str, Sequence[str]],
        value: Union[float, int, bool, Sequence] = None,
    ) -> None:
        """
        Sets the value of a model state

        Args:
            var_id (Any): character, model variable state identifier(s)
            value (Any): numeric value(s)

        """
        ss.set_state_value(self, var_id, value)

    def snapshot_state(self) -> "MemoryStates":
        """Take a snapshot of the memory states of a simulation

        Returns:
            MemoryStates: memory states, that can be stored and reapplied
        """
        return ss.snapshot_state(self)

    def set_states(self, states: "MemoryStates") -> None:
        """Apply memory states to a simulation

        Args:
            states (MemoryStates): memory states
        """
        ss.set_states(self, states)

    def set_reservoir_model(self, new_model_id: str, element_id: str) -> None:
        """Sets a new reservoir model on an element

        Args:
            new_model_id (str): Currently one of: "ControlledReleaseReservoir", "LevelVolumeAreaReservoir", "FarmDamReservoir";
            element_id (str): _description_
        """
        swg.SetReservoirModel_py(self, new_model_id, element_id)

    def set_reservoir_geometry(
        self, element_id: str, level: np.ndarray, storage: np.ndarray, area: np.ndarray
    ) -> None:
        """Sets the geometry of a reservoir

        Args:
            element_id (str): Element with a suitable reservoir supporting a geometry description
            level (np.ndarray): array of water surface levels, in S.I. units (m) TO BE CONFIRMED
            storage (np.ndarray): array of volume storages, in S.I. units (m3) TO BE CONFIRMED
            area (np.ndarray): array of surfce areas, in S.I. units (m2) TO BE CONFIRMED
        """
        num_entries = len(level)
        assert len(storage) == num_entries
        assert len(area) == num_entries
        swg.SetReservoirGeometry_py(self, element_id, num_entries, level, storage, area)

    def set_reservoir_min_discharge(
        self, element_id: str, level: np.ndarray, discharge: np.ndarray
    ) -> None:
        """Sets a reservoir operating curve, minimum release for a given level

        Args:
            element_id (str): Element with a suitable reservoir supporting a geometry description
            level (np.ndarray): array of levels (m)
            discharge (np.ndarray): array of minimum discharges (m3/s)
        """
        num_entries = len(level)
        swg.SetReservoirMinDischarge_py(self, element_id, num_entries, level, discharge)

    def set_reservoir_max_discharge(
        self, element_id: str, level: np.ndarray, discharge: np.ndarray
    ) -> None:
        """Sets a reservoir operating curve, maximum release for a given level

        Args:
            element_id (str): Element with a suitable reservoir supporting a geometry description
            level (np.ndarray): array of levels (m)
            discharge (np.ndarray): array of maximum discharges (m3/s)
        """
        num_entries = len(level)
        swg.SetReservoirMaxDischarge_py(self, element_id, num_entries, level, discharge)

    def use_state_initialises(self, state_initialiser: "StateInitialiser"):
        """Sets the state initialiser to use for a simulation. This forces the removal of any prior state initialiser.

        Args:
            state_initialiser (StateInitialiser): the new state initialiser to use
        """
        swg.UseStateInitializerModelRunner_py(self, state_initialiser)

    def remove_state_initialisers(self):
        """Forces the removal of any state initialiser."""
        swg.RemoveStateInitializerModelRunner_py(self)

    def add_state_initialiser(self, state_initialiser: "StateInitialiser"):
        """Adds a state initialiser to any prior list of state initialisers"""
        swg.AddStateInitializerModelRunner_py(self, state_initialiser)

    def reset_model_states(self) -> None:
        """Reset the model states of a simulation, and apply one or more state initialers if the simulation is configured with any."""
        ss.reset_model_states(self)

    def describe(self, verbosity=None) -> Dict:
        """Describe the catchment model structure using simple python representations"""
        return ss.describe(self, verbosity)

    def create_ensemble_forecast_simulation(
        self,
        data_library,
        start: ConvertibleToTimestamp,
        end: ConvertibleToTimestamp,
        input_map: Dict[str, List[str]],
        lead_time: int,
        ensemble_size: int,
        n_time_steps_between_forecasts: int,
    ) -> "EnsembleForecastSimulation":
        """
        Create an ensemble forecast simulation

        Args:
            data_library (Any): external pointer type ENSEMBLE_DATA_SET_PTR, or a Python class wrapper around it 
            start (ConvertibleToTimestamp): the start date of the simulation. The time zone will be forced to UTC.
            end (ConvertibleToTimestamp): the end date of the simulation. The time zone will be forced to UTC.
            input_map (dict): a named list were names are the data library data identifiers, and values are character vectors with model state identifiers.
            lead_time (int): integer, the length in time steps of the forecasts.
            ensemble_size (int): ensemble size
            n_time_steps_between_forecasts (int): nTimeStepsBetweenForecasts

        Returns:
            An external pointer

        """
        return ss.create_ensemble_forecast_simulation(
            self,
            data_library,
            start,
            end,
            input_map,
            lead_time,
            ensemble_size,
            n_time_steps_between_forecasts,
        )

    def ensemble_simulation(self, ensemble_size: int) -> "EnsembleSimulation":
        """Create an ensemble simulation templated from this simulation

        Args:
            ensemble_size (int): The size of the ensemble dimension

        Returns:
            EnsembleSimulation: Ensemble simulation (ensemble simulation runner)
        """        
        return swg.CreateEnsembleModelRunner_py(self, ensembleSize=ensemble_size)

    def erris_ensemble_simulation(
        self,
        warmup_start: ConvertibleToTimestamp,
        warmup_end: ConvertibleToTimestamp,
        observed_ts: TimeSeriesLike,
        error_model_element_id: str,
    ) -> "EnsembleSimulation":
        """Creates an ensemble simulation templated on this simulation, with an ERRIS model on one of the network element

        Args:
            warmup_start (ConvertibleToTimestamp): start time stamp for the warmup period
            warmup_end (ConvertibleToTimestamp): end time stamp for the warmup period
            observed_ts (TimeSeriesLike): Time series of observations to correct prediction against
            error_model_element_id (str): model element identifier where to set up an ERRIS correction model

        Returns:
            EnsembleSimulation: Ensemble simulation (ensemble simulation runner)
        """
        from swift2.internal import to_interop_univariate_series
        from cinterop.timeseries import as_pydatetime
        warmup_start = as_pydatetime(warmup_start)
        warmup_end = as_pydatetime(warmup_end)
        values, ts_geom = to_interop_univariate_series(observed_ts)
        return swg.PrepareEnsembleModelRunner_py(
            self, warmup_start, warmup_end, values, ts_geom, error_model_element_id
        )

    def play_inputs(
        self,
        data_library: uc.TimeSeriesLibrary,
        model_var_id: "VecStr",
        data_id: "VecStr",
        resample: "VecStr" = "",
    ) -> None:
        """
        Assign input time series from a time series library to a model simulation

        Args:
            data_library (TimeSeriesLibrary): external pointer type ENSEMBLE_DATA_SET_PTR, or a Python class wrapper around it 
            model_var_id (str or sequence of str): model state variable unique identifier(s)
            data_id (str or sequence of str): identifier(s) for data in the data_library. If length is not the same as model_var_id, the elements of data_id are reused to match it
            resample (str or sequence of str): identifier(s) for how the series is resampled (aggregated or disaggregated). If length is not the same as model_var_id, the elements of resample are reused to match it

        """
        spr.play_inputs(self, data_library, model_var_id, data_id, resample)

    def play_subarea_input(
        self, input: TimeSeriesLike, subarea_name: str, input_name: str
    ) -> None:
        """
        Sets time series as input to a simulation

        Args:
            input (TimeSeriesLike): univariate time series.
            subarea_name (str): a valid name of the subarea
            input_name (str): the name of the input variable to the model (i.e. 'P' for the precip of GR5H)

        """
        spr.play_subarea_input(self, input, subarea_name, input_name)

    def play_input(
        self, input_ts: TimeSeriesLike, var_ids: Optional["VecStr"] = None
    ) -> None:
        """
        Sets one or more time series as input(s) to a simulation

        Args:
            input_ts (TimeSeriesLike): univariate time series. If an xts time series column names must be valid model variable identifiers, unless explicitely provided via varIds
            var_ids (optional str or sequence of str): optional character, the variable identifiers to use, overriding the column names of the inputTs. If not NULL, must be of length equal to the number of columns in inputTs
        """
        spr.play_singular_simulation(self, input_ts, var_ids)

    def get_played(
        self,
        var_ids: Optional["VecStr"] = None,
        start_time: Optional[ConvertibleToTimestamp] = None,
        end_time: Optional[ConvertibleToTimestamp] = None,
    ) -> xr.DataArray:
        """
        Retrieves one or more played (input) time series from a simulation

        Args:
            var_ids (optional str or sequence of str): name(s) of the model variable(s) into which a time series is played as input. e.g. 'Catchment.StreamflowRate'. If missing, a multivariate time series of all played states is returned; this may be a large amount of data.
            start_time (datetime like): An optional parameter, the start of a period to subset the time series
            end_time (datetime like): An optional parameter, the end of a period to subset the time series

        Returns:
            xr.DataArray: a time series, possibly multivariate.

        """
        return spr.get_played(self, var_ids, start_time, end_time)

    def get_played_varnames(self) -> List[str]:
        """
        Gets all the names of model states fed an input time series
        """
        return spr.get_played_varnames(self)

    def get_recorded(
        self,
        var_ids: Optional["VecStr"] = None,
        start_time: Optional[ConvertibleToTimestamp] = None,
        end_time: Optional[ConvertibleToTimestamp] = None,
    ) -> xr.DataArray:
        """
        Retrieves a recorded time series from a simulation

        Args:
            var_ids (optional str or sequence of str): name(s) of the model variable(s) recorded to a time series. e.g. 'Catchment.StreamflowRate'. If missing, a multivariate time series of all played states is returned; this may be a large amount of data.
            start_time (datetime like): An optional parameter, the start of a period to subset the time series
            end_time (datetime like): An optional parameter, the end of a period to subset the time series

        Returns:
            xr.DataArray: a time series, possibly multivariate.

        """
        return spr.get_recorded(self, var_ids, start_time, end_time)

    def get_all_recorded(self) -> xr.DataArray:
        """Gets all the time series of models variables recorded from"""
        return spr.get_all_recorded(self)

    def get_all_played(self) -> xr.DataArray:
        """Gets all the time series of models variables into which input time sereis is/are played"""
        return spr.get_all_played(self)

    def apply_recording_function(
        self,
        recording_func: Optional["RecordToSignature"],
        var_ids: "VecStr",
        recording_provider,
        data_ids: "VecStr",
    ) -> None:
        """DRAFT Advanced/technical. Record states to a record provider using a callable function. 

        Likely not for end users. This is used by methods such as 
        [`EnsembleSimulation.record_ensemble_state`][swift2.classes.EnsembleSimulation.record_ensemble_state].

        """
        spr.apply_recording_function(
            self, recording_func, var_ids, recording_provider, data_ids
        )

    def record_singular_state(
        self,
        var_ids: "VecStr" = CATCHMENT_FLOWRATE_VARID,
        recording_provider: Optional["TimeSeriesLibrary"] = None,
        data_ids: Optional["VecStr"] = None,
    ) -> None:
        """DRAFT Advanced/technical. Record states to a record provider. 

        Likely not for end users.
        """
        spr.record_singular_state(self, var_ids, recording_provider, data_ids)

    def cookie_cut_dendritic_catchment(
        self, bottom_element_id: str, top_element_ids: "VecStr"
    ):
        """cookie cut a dendritic catchment (without confluences)

        Args:
            bottom_element_id (str): identifier of the most downstream element to keep
            top_element_ids (str): identifier(s) of the most upstream element(s) to keep

        Returns:
            Simulation: a subcatchment simulation, cookie cut from the base simulation. Deep clone of objects.
        """
        return smd.cookie_cut_dendritic_catchment(
            self, bottom_element_id, top_element_ids
        )

    def subset_catchment(self, element_id: str, action: str = "keep_above"):
        """Subsets a catchment, keeping only a portion above or below a node, link or subarea.

        Args:
            element_id (str): id of the element to cut at.
            action (str): how to cut; currently limited to 'keep_above'

        Returns:
            Simulation: a subcatchment simulation, cookie cut from the base simulation. Deep clone of objects.
        """
        return smd.subset_catchment(self, element_id, action)

    def split_to_subcatchments(
        self, split_element_ids: Sequence[str], include_upstream: Sequence[bool] = None
    ) -> OrderedDict[str, "Simulation"]:
        """Split a catchment in subcatchments, given a list of node/link element identifiers

        Args:
            split_element_ids (str): element identifiers such as 'node.n1', 'link.linkId_2'
            include_upstream (bool, optional): indicates whether for each element in split_element_ids it should be including in the upstream portion of the subcatchment. Defaults to None.

        Returns:
            OrderedDict: list of subcatchments resulting from the split

        Examples:
            >>> _, simulation = sdh.create_test_catchment_structure()
            >>> e_ids = ['node.n2', 'node.n4']
            >>> sub_sims = simulation.split_to_subcatchments(e_ids)
            >>> for k in sub_sims:
            >>>     print(k)
            >>>     print(sub_sims[k].get_node_ids())
            >>>     print(sub_sims[k].get_node_names())
            node.n4
            ['n4', 'n3', 'n1']
            ['n4_name', 'n3_name', 'n1_name']
            node.n2
            ['n2', 'n5']
            ['n2_name', 'n5_name']
            remainder
            ['n6']
            ['n6_name']
        """
        return smd.split_to_subcatchments(self, split_element_ids, include_upstream)

    def get_catchment_structure(self) -> Dict[str, Any]:
        """Gets the essential connective structure of a catchment

        Args:
            simulation (Simulation): base catchment simulation

        Returns:
            Dict[str, Any]: A nested dictionary describing the catchment connectivity of subareas, links, and nodes

        Examples:
            >>> _, simulation = sdh.create_test_catchment_structure()
            >>> simulation.get_catchment_structure()
            {'Node':    Id     Name
            0  n1  n1_name
            1  n2  n2_name
            2  n3  n3_name
            3  n4  n4_name
            4  n5  n5_name
            5  n6  n6_name, 'Link':      Id       Name  LengthMetres    f  ManningsN  Slope
            0  lnk1  lnk1_name           0.0  0.0        0.0    0.0
            1  lnk2  lnk2_name           0.0  0.0        0.0    0.0
            2  lnk3  lnk3_name           0.0  0.0        0.0    0.0
            3  lnk4  lnk4_name           0.0  0.0        0.0    0.0
            4  lnk5  lnk5_name           0.0  0.0        0.0    0.0, 'Subarea':      Id       Name  AreaKm2
            0  lnk1  lnk1_name      1.1
            1  lnk2  lnk2_name      2.2
            2  lnk3  lnk3_name      3.3
            3  lnk4  lnk4_name      4.4
            4  lnk5  lnk5_name      5.5, 'NodeLink':   DownstreamId UpstreamId LinkId
            0           n6         n2   lnk1
            1           n2         n5   lnk2
            2           n2         n4   lnk3
            3           n4         n3   lnk4
            4           n4         n1   lnk5, 'SubareaLink':   LinkId SubareaId
            0   lnk1      lnk1
            1   lnk2      lnk2
            2   lnk3      lnk3
            3   lnk4      lnk4
        """
        return smd.get_catchment_structure(self)

    def create_multisite_objective(
        self,
        statspec: pd.DataFrame,
        observations: Sequence[TimeSeriesLike],
        weights: Dict[str, float],
    ) -> "ObjectiveEvaluator":
        """
        Creates an objective that combines multiple statistics. Used for joined, "whole of catchment" calibration

        Args:
            statspec (pd.DataFrame): dataframe defining the objectives used. See function [`multi_statistic_definition`][swift2.statistics.multi_statistic_definition] to help build this dataframe.
            observations (Sequence[TimeSeriesLike]): A list of (time series) observations to calculated the statistics. Must be of same length as the number of rows of statspec.
            weights (Dict[str, float]): numeric vector of weights to ponderate each objective.

        Returns:
            ObjectiveEvaluator: an objective evaluator

        Examples:
            >>> _, ms = sdh.create_test_catchment_structure()
            >>> from swift2.utils import mk_full_data_id
            >>> 
            >>> nodeids = ['node.n2', 'node.n4']
            >>> mvids = mk_full_data_id(nodeids, 'OutflowRate')
            >>> 
            >>> sdh.configure_test_simulation(
            ...     ms,
            ...     data_id='MMH',
            ...     simul_start='1990-01-01',
            ...     simul_end='2005-12-31',
            ...     tstep='daily',
            ...     varname_rain='P',
            ...     varname_pet='E',
            ...     varname_data_rain='rain',
            ...     varname_data_pet='evap',
            ... )
            >>> 

            >>> ms.record_state(mvids)
            >>> ms.exec_simulation()
            >>> 
            >>> modFlows = ms.get_recorded()
            >>> 

            >>> w = dict(zip(mvids, [1.0, 2.0]))
            >>> w
            {'node.n2.OutflowRate': 1.0, 'node.n4.OutflowRate': 2.0}
            >>> span = ms.get_simulation_span()
            >>> 

            >>> from swift2.utils import rep
            >>> statspec = sst.multi_statistic_definition(mvids, rep('nse', 2), mvids, mvids, rep(span['start'], 2), rep(span['end'], 2) )
            >>> 
            >>> statspec
                        ModelVarId StatisticId          ObjectiveId        ObjectiveName      Start        End
            0  node.n2.OutflowRate         nse  node.n2.OutflowRate  node.n2.OutflowRate 1990-01-01 2005-12-31
            1  node.n4.OutflowRate         nse  node.n4.OutflowRate  node.n4.OutflowRate 1990-01-01 2005-12-31
            >>> 
            >>> # Create synthetic observations
            >>> observations = [
            ...     modFlows.sel(variable_identifiers=mvids[0]) * 0.33,
            ...     modFlows.sel(variable_identifiers=mvids[1]) * 0.77
            ... ]
            >>> 
            >>> obj = ms.create_multisite_objective(statspec, observations, w)
            >>> 

            >>> dummy = sp.create_parameteriser()
            >>> obj.get_scores(dummy)
            {'node.n2.OutflowRate': -4.152338377267432, 'node.n4.OutflowRate': 0.8884789439301954}
            >>> 

            >>> obj.get_score(dummy)
            {'scores': {'MultisiteObjectives': 0.7917934964690136}, 'sysconfig': Empty DataFrame
            Columns: [Name, Value, Min, Max]
            Index: []}
            >>> 
        """
        return ssf.create_multisite_objective(self, statspec, observations, weights)

    def create_objective(
        self,
        state_name: str,
        observation: TimeSeriesLike,
        statistic: str,
        start_date: ConvertibleToTimestamp,
        end_date: ConvertibleToTimestamp,
    ) -> "ObjectiveEvaluator":
        """
        Creates an objective calculator

        Args:
            state_name (str): The name identifying the model state variable to calibrate against the observation
            observation (TimeSeriesLike): an xts
            statistic (str): statistic identifier, e.g. "NSE"
            start_date (ConvertibleToTimestamp): start date of the period to calculate statistics on
            end_date (ConvertibleToTimestamp): end date of the period to calculate statistics on

        Returns:
            ObjectiveEvaluator: single objective evaluator
        """
        return ssf.create_objective(
            self, state_name, observation, statistic, start_date, end_date
        )

    def muskingum_param_constraints(
        self,
        inner_parameters: "HypercubeParameteriser",
        delta_t: float = 1.0,
        param_name_k: str = "K",
        param_name_x: str = "X",
    ) -> "ConstraintParameteriser":
        """Create a parameteriser with Muskingum-type constraints. 

        Given an existing parameteriser, create a wrapper that adds constraints on two of its parameters.

        Args:
            inner_parameters (HypercubeParameteriser): A SWIFT parameteriser object that contains two Muskingum-type attenuation  and delay parameters.
            delta_t (int, optional): the simulation time step in HOURS. Defaults to 1.
            param_name_k (str, optional): the variable identifier to use for the delay parameter of the Muskingum routing. Defaults to "K".
            param_name_x (str, optional): the variable identifier to use for the attenuation parameter of the Muskingum routing. Defaults to "X".

        Returns:
            ConstraintParameteriser: A parameteriser with constraints on the feasibility of the attenuation / delay parameters

        Examples:
           >>> todo()
        """
        return sp.create_muskingum_param_constraints(
            inner_parameters, delta_t, param_name_k, param_name_x, self
        )

    def prepare_erris_forecasting(
        self,
        observation: TimeSeriesLike,
        error_model_element_id: str,
        warmup_start: ConvertibleToTimestamp,
        warmup_end: ConvertibleToTimestamp,
    ) -> 'EnsembleSimulation':
        """Create an ensemble simulation for forecasting with ERRIS

        Args:
            observation (TimeSeriesLike): Time series of observations to correct prediction against
            error_model_element_id (str): model element identifier where to set up an ERRIS correction model
            warmup_start (ConvertibleToTimestamp): start time stamp for the warmup period
            warmup_end (ConvertibleToTimestamp): end time stamp for the warmup period

        Returns:
            EnsembleSimulation: Ensemble simulation (ensemble simulation runner)
        """
        from swift2.internal import to_interop_univariate_series

        values, ts_geom = to_interop_univariate_series(
            observation, warmup_start, warmup_end
        )
        return swg.PrepareERRISForecasting_py(
            self, values, ts_geom, error_model_element_id, warmup_start, warmup_end
        )

    def prepare_dual_pass_forecasting(
        self,
        observation: TimeSeriesLike,
        error_model_element_id: str,
        warmup_start: ConvertibleToTimestamp,
        warmup_end: ConvertibleToTimestamp,
        required_windows_percentage: float,
    ) -> 'EnsembleSimulation':
        """Create an ensemble simulation for forecasting with the Dual Pass error correction method 

        Args:
            observation (TimeSeriesLike): Time series of observations to correct prediction against
            error_model_element_id (str): model element identifier where to set up an ERRIS correction model
            warmup_start (ConvertibleToTimestamp): start time stamp for the warmup period
            warmup_end (ConvertibleToTimestamp): end time stamp for the warmup period
            required_windows_percentage (float): required_windows_percentage

        Returns:
            EnsembleSimulation: Ensemble simulation (ensemble simulation runner)
        """    
        from swift2.internal import to_interop_univariate_series

        values, ts_geom = to_interop_univariate_series(
            observation, warmup_start, warmup_end
        )
        return swg.PrepareDualPassForecasting_py(
            self,
            values,
            ts_geom,
            error_model_element_id,
            warmup_start,
            warmup_end,
            required_windows_percentage,
        )

__str__()

string representation

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def __str__(self):
    """string representation"""
    tid = self.type_id if self.type_id is not None else ""
    return f'Simulation wrapper for a CFFI pointer handle to a native pointer of type id "{tid}"'

add_state_initialiser(state_initialiser)

Adds a state initialiser to any prior list of state initialisers

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def add_state_initialiser(self, state_initialiser: "StateInitialiser"):
    """Adds a state initialiser to any prior list of state initialisers"""
    swg.AddStateInitializerModelRunner_py(self, state_initialiser)

apply_recording_function(recording_func, var_ids, recording_provider, data_ids)

DRAFT Advanced/technical. Record states to a record provider using a callable function.

Likely not for end users. This is used by methods such as EnsembleSimulation.record_ensemble_state.

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def apply_recording_function(
    self,
    recording_func: Optional["RecordToSignature"],
    var_ids: "VecStr",
    recording_provider,
    data_ids: "VecStr",
) -> None:
    """DRAFT Advanced/technical. Record states to a record provider using a callable function. 

    Likely not for end users. This is used by methods such as 
    [`EnsembleSimulation.record_ensemble_state`][swift2.classes.EnsembleSimulation.record_ensemble_state].

    """
    spr.apply_recording_function(
        self, recording_func, var_ids, recording_provider, data_ids
    )

check_simulation()

Checks whether a simulation is configured to a state where it is executable

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def check_simulation(self) -> Dict:
    """
    Checks whether a simulation is configured to a state where it is executable
    """
    return ss.check_simulation(self)

clone()

Clone this simulation (deep copy)

Returns:

Name	Type	Description
Simulation	Simulation	A new simulation object

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def clone(self) -> "Simulation":
    """Clone this simulation (deep copy)

    Returns:
        Simulation: A new simulation object
    """
    return swg.CloneModel_py(self)

cookie_cut_dendritic_catchment(bottom_element_id, top_element_ids)

cookie cut a dendritic catchment (without confluences)

Parameters:

Name	Type	Description	Default
bottom_element_id	str	identifier of the most downstream element to keep	required
top_element_ids	str	identifier(s) of the most upstream element(s) to keep	required

Returns:

Name	Type	Description
Simulation		a subcatchment simulation, cookie cut from the base simulation. Deep clone of objects.

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def cookie_cut_dendritic_catchment(
    self, bottom_element_id: str, top_element_ids: "VecStr"
):
    """cookie cut a dendritic catchment (without confluences)

    Args:
        bottom_element_id (str): identifier of the most downstream element to keep
        top_element_ids (str): identifier(s) of the most upstream element(s) to keep

    Returns:
        Simulation: a subcatchment simulation, cookie cut from the base simulation. Deep clone of objects.
    """
    return smd.cookie_cut_dendritic_catchment(
        self, bottom_element_id, top_element_ids
    )

create_ensemble_forecast_simulation(data_library, start, end, input_map, lead_time, ensemble_size, n_time_steps_between_forecasts)

Create an ensemble forecast simulation

Parameters:

Name	Type	Description	Default
data_library	Any	external pointer type ENSEMBLE_DATA_SET_PTR, or a Python class wrapper around it	required
start	ConvertibleToTimestamp	the start date of the simulation. The time zone will be forced to UTC.	required
end	ConvertibleToTimestamp	the end date of the simulation. The time zone will be forced to UTC.	required
input_map	dict	a named list were names are the data library data identifiers, and values are character vectors with model state identifiers.	required
lead_time	int	integer, the length in time steps of the forecasts.	required
ensemble_size	int	ensemble size	required
n_time_steps_between_forecasts	int	nTimeStepsBetweenForecasts	required

Returns:

Type	Description
EnsembleForecastSimulation	An external pointer

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def create_ensemble_forecast_simulation(
    self,
    data_library,
    start: ConvertibleToTimestamp,
    end: ConvertibleToTimestamp,
    input_map: Dict[str, List[str]],
    lead_time: int,
    ensemble_size: int,
    n_time_steps_between_forecasts: int,
) -> "EnsembleForecastSimulation":
    """
    Create an ensemble forecast simulation

    Args:
        data_library (Any): external pointer type ENSEMBLE_DATA_SET_PTR, or a Python class wrapper around it 
        start (ConvertibleToTimestamp): the start date of the simulation. The time zone will be forced to UTC.
        end (ConvertibleToTimestamp): the end date of the simulation. The time zone will be forced to UTC.
        input_map (dict): a named list were names are the data library data identifiers, and values are character vectors with model state identifiers.
        lead_time (int): integer, the length in time steps of the forecasts.
        ensemble_size (int): ensemble size
        n_time_steps_between_forecasts (int): nTimeStepsBetweenForecasts

    Returns:
        An external pointer

    """
    return ss.create_ensemble_forecast_simulation(
        self,
        data_library,
        start,
        end,
        input_map,
        lead_time,
        ensemble_size,
        n_time_steps_between_forecasts,
    )

create_multisite_objective(statspec, observations, weights)

Creates an objective that combines multiple statistics. Used for joined, "whole of catchment" calibration

Parameters:

Name	Type	Description	Default
statspec	pd.DataFrame	dataframe defining the objectives used. See function multi_statistic_definition to help build this dataframe.	required
observations	Sequence[TimeSeriesLike]	A list of (time series) observations to calculated the statistics. Must be of same length as the number of rows of statspec.	required
weights	Dict[str, float]	numeric vector of weights to ponderate each objective.	required

Returns:

Name	Type	Description
ObjectiveEvaluator	ObjectiveEvaluator	an objective evaluator

Examples:

>>> _, ms = sdh.create_test_catchment_structure()
>>> from swift2.utils import mk_full_data_id
>>> 
>>> nodeids = ['node.n2', 'node.n4']
>>> mvids = mk_full_data_id(nodeids, 'OutflowRate')
>>> 
>>> sdh.configure_test_simulation(
...     ms,
...     data_id='MMH',
...     simul_start='1990-01-01',
...     simul_end='2005-12-31',
...     tstep='daily',
...     varname_rain='P',
...     varname_pet='E',
...     varname_data_rain='rain',
...     varname_data_pet='evap',
... )
>>> 

>>> ms.record_state(mvids)
>>> ms.exec_simulation()
>>> 
>>> modFlows = ms.get_recorded()
>>> 

>>> w = dict(zip(mvids, [1.0, 2.0]))
>>> w
{'node.n2.OutflowRate': 1.0, 'node.n4.OutflowRate': 2.0}
>>> span = ms.get_simulation_span()
>>> 

>>> from swift2.utils import rep
>>> statspec = sst.multi_statistic_definition(mvids, rep('nse', 2), mvids, mvids, rep(span['start'], 2), rep(span['end'], 2) )
>>> 
>>> statspec
            ModelVarId StatisticId          ObjectiveId        ObjectiveName      Start        End
0  node.n2.OutflowRate         nse  node.n2.OutflowRate  node.n2.OutflowRate 1990-01-01 2005-12-31
1  node.n4.OutflowRate         nse  node.n4.OutflowRate  node.n4.OutflowRate 1990-01-01 2005-12-31
>>> 
>>> # Create synthetic observations
>>> observations = [
...     modFlows.sel(variable_identifiers=mvids[0]) * 0.33,
...     modFlows.sel(variable_identifiers=mvids[1]) * 0.77
... ]
>>> 
>>> obj = ms.create_multisite_objective(statspec, observations, w)
>>> 

>>> dummy = sp.create_parameteriser()
>>> obj.get_scores(dummy)
{'node.n2.OutflowRate': -4.152338377267432, 'node.n4.OutflowRate': 0.8884789439301954}
>>> 

>>> obj.get_score(dummy)
{'scores': {'MultisiteObjectives': 0.7917934964690136}, 'sysconfig': Empty DataFrame
Columns: [Name, Value, Min, Max]
Index: []}
>>>

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def create_multisite_objective(
    self,
    statspec: pd.DataFrame,
    observations: Sequence[TimeSeriesLike],
    weights: Dict[str, float],
) -> "ObjectiveEvaluator":
    """
    Creates an objective that combines multiple statistics. Used for joined, "whole of catchment" calibration

    Args:
        statspec (pd.DataFrame): dataframe defining the objectives used. See function [`multi_statistic_definition`][swift2.statistics.multi_statistic_definition] to help build this dataframe.
        observations (Sequence[TimeSeriesLike]): A list of (time series) observations to calculated the statistics. Must be of same length as the number of rows of statspec.
        weights (Dict[str, float]): numeric vector of weights to ponderate each objective.

    Returns:
        ObjectiveEvaluator: an objective evaluator

    Examples:
        >>> _, ms = sdh.create_test_catchment_structure()
        >>> from swift2.utils import mk_full_data_id
        >>> 
        >>> nodeids = ['node.n2', 'node.n4']
        >>> mvids = mk_full_data_id(nodeids, 'OutflowRate')
        >>> 
        >>> sdh.configure_test_simulation(
        ...     ms,
        ...     data_id='MMH',
        ...     simul_start='1990-01-01',
        ...     simul_end='2005-12-31',
        ...     tstep='daily',
        ...     varname_rain='P',
        ...     varname_pet='E',
        ...     varname_data_rain='rain',
        ...     varname_data_pet='evap',
        ... )
        >>> 

        >>> ms.record_state(mvids)
        >>> ms.exec_simulation()
        >>> 
        >>> modFlows = ms.get_recorded()
        >>> 

        >>> w = dict(zip(mvids, [1.0, 2.0]))
        >>> w
        {'node.n2.OutflowRate': 1.0, 'node.n4.OutflowRate': 2.0}
        >>> span = ms.get_simulation_span()
        >>> 

        >>> from swift2.utils import rep
        >>> statspec = sst.multi_statistic_definition(mvids, rep('nse', 2), mvids, mvids, rep(span['start'], 2), rep(span['end'], 2) )
        >>> 
        >>> statspec
                    ModelVarId StatisticId          ObjectiveId        ObjectiveName      Start        End
        0  node.n2.OutflowRate         nse  node.n2.OutflowRate  node.n2.OutflowRate 1990-01-01 2005-12-31
        1  node.n4.OutflowRate         nse  node.n4.OutflowRate  node.n4.OutflowRate 1990-01-01 2005-12-31
        >>> 
        >>> # Create synthetic observations
        >>> observations = [
        ...     modFlows.sel(variable_identifiers=mvids[0]) * 0.33,
        ...     modFlows.sel(variable_identifiers=mvids[1]) * 0.77
        ... ]
        >>> 
        >>> obj = ms.create_multisite_objective(statspec, observations, w)
        >>> 

        >>> dummy = sp.create_parameteriser()
        >>> obj.get_scores(dummy)
        {'node.n2.OutflowRate': -4.152338377267432, 'node.n4.OutflowRate': 0.8884789439301954}
        >>> 

        >>> obj.get_score(dummy)
        {'scores': {'MultisiteObjectives': 0.7917934964690136}, 'sysconfig': Empty DataFrame
        Columns: [Name, Value, Min, Max]
        Index: []}
        >>> 
    """
    return ssf.create_multisite_objective(self, statspec, observations, weights)

create_objective(state_name, observation, statistic, start_date, end_date)

Creates an objective calculator

Parameters:

Name	Type	Description	Default
state_name	str	The name identifying the model state variable to calibrate against the observation	required
observation	TimeSeriesLike	an xts	required
statistic	str	statistic identifier, e.g. "NSE"	required
start_date	ConvertibleToTimestamp	start date of the period to calculate statistics on	required
end_date	ConvertibleToTimestamp	end date of the period to calculate statistics on	required

Returns:

Name	Type	Description
ObjectiveEvaluator	ObjectiveEvaluator	single objective evaluator

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def create_objective(
    self,
    state_name: str,
    observation: TimeSeriesLike,
    statistic: str,
    start_date: ConvertibleToTimestamp,
    end_date: ConvertibleToTimestamp,
) -> "ObjectiveEvaluator":
    """
    Creates an objective calculator

    Args:
        state_name (str): The name identifying the model state variable to calibrate against the observation
        observation (TimeSeriesLike): an xts
        statistic (str): statistic identifier, e.g. "NSE"
        start_date (ConvertibleToTimestamp): start date of the period to calculate statistics on
        end_date (ConvertibleToTimestamp): end date of the period to calculate statistics on

    Returns:
        ObjectiveEvaluator: single objective evaluator
    """
    return ssf.create_objective(
        self, state_name, observation, statistic, start_date, end_date
    )

describe(verbosity=None)

Describe the catchment model structure using simple python representations

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def describe(self, verbosity=None) -> Dict:
    """Describe the catchment model structure using simple python representations"""
    return ss.describe(self, verbosity)

ensemble_simulation(ensemble_size)

Create an ensemble simulation templated from this simulation

Parameters:

Name	Type	Description	Default
ensemble_size	int	The size of the ensemble dimension	required

Returns:

Name	Type	Description
EnsembleSimulation	EnsembleSimulation	Ensemble simulation (ensemble simulation runner)

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def ensemble_simulation(self, ensemble_size: int) -> "EnsembleSimulation":
    """Create an ensemble simulation templated from this simulation

    Args:
        ensemble_size (int): The size of the ensemble dimension

    Returns:
        EnsembleSimulation: Ensemble simulation (ensemble simulation runner)
    """        
    return swg.CreateEnsembleModelRunner_py(self, ensembleSize=ensemble_size)

erris_ensemble_simulation(warmup_start, warmup_end, observed_ts, error_model_element_id)

Creates an ensemble simulation templated on this simulation, with an ERRIS model on one of the network element

Parameters:

Name	Type	Description	Default
warmup_start	ConvertibleToTimestamp	start time stamp for the warmup period	required
warmup_end	ConvertibleToTimestamp	end time stamp for the warmup period	required
observed_ts	TimeSeriesLike	Time series of observations to correct prediction against	required
error_model_element_id	str	model element identifier where to set up an ERRIS correction model	required

Returns:

Name	Type	Description
EnsembleSimulation	EnsembleSimulation	Ensemble simulation (ensemble simulation runner)

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def erris_ensemble_simulation(
    self,
    warmup_start: ConvertibleToTimestamp,
    warmup_end: ConvertibleToTimestamp,
    observed_ts: TimeSeriesLike,
    error_model_element_id: str,
) -> "EnsembleSimulation":
    """Creates an ensemble simulation templated on this simulation, with an ERRIS model on one of the network element

    Args:
        warmup_start (ConvertibleToTimestamp): start time stamp for the warmup period
        warmup_end (ConvertibleToTimestamp): end time stamp for the warmup period
        observed_ts (TimeSeriesLike): Time series of observations to correct prediction against
        error_model_element_id (str): model element identifier where to set up an ERRIS correction model

    Returns:
        EnsembleSimulation: Ensemble simulation (ensemble simulation runner)
    """
    from swift2.internal import to_interop_univariate_series
    from cinterop.timeseries import as_pydatetime
    warmup_start = as_pydatetime(warmup_start)
    warmup_end = as_pydatetime(warmup_end)
    values, ts_geom = to_interop_univariate_series(observed_ts)
    return swg.PrepareEnsembleModelRunner_py(
        self, warmup_start, warmup_end, values, ts_geom, error_model_element_id
    )

get_all_played()

Gets all the time series of models variables into which input time sereis is/are played

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def get_all_played(self) -> xr.DataArray:
    """Gets all the time series of models variables into which input time sereis is/are played"""
    return spr.get_all_played(self)

get_all_recorded()

Gets all the time series of models variables recorded from

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def get_all_recorded(self) -> xr.DataArray:
    """Gets all the time series of models variables recorded from"""
    return spr.get_all_recorded(self)

get_catchment_structure()

Gets the essential connective structure of a catchment

Parameters:

Name	Type	Description	Default
simulation	Simulation	base catchment simulation	required

Returns:

Type	Description
Dict[str, Any]	Dict[str, Any]: A nested dictionary describing the catchment connectivity of subareas, links, and nodes

Examples:

>>> _, simulation = sdh.create_test_catchment_structure()
>>> simulation.get_catchment_structure()
{'Node':    Id     Name
0  n1  n1_name
1  n2  n2_name
2  n3  n3_name
3  n4  n4_name
4  n5  n5_name
5  n6  n6_name, 'Link':      Id       Name  LengthMetres    f  ManningsN  Slope
0  lnk1  lnk1_name           0.0  0.0        0.0    0.0
1  lnk2  lnk2_name           0.0  0.0        0.0    0.0
2  lnk3  lnk3_name           0.0  0.0        0.0    0.0
3  lnk4  lnk4_name           0.0  0.0        0.0    0.0
4  lnk5  lnk5_name           0.0  0.0        0.0    0.0, 'Subarea':      Id       Name  AreaKm2
0  lnk1  lnk1_name      1.1
1  lnk2  lnk2_name      2.2
2  lnk3  lnk3_name      3.3
3  lnk4  lnk4_name      4.4
4  lnk5  lnk5_name      5.5, 'NodeLink':   DownstreamId UpstreamId LinkId
0           n6         n2   lnk1
1           n2         n5   lnk2
2           n2         n4   lnk3
3           n4         n3   lnk4
4           n4         n1   lnk5, 'SubareaLink':   LinkId SubareaId
0   lnk1      lnk1
1   lnk2      lnk2
2   lnk3      lnk3
3   lnk4      lnk4

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def get_catchment_structure(self) -> Dict[str, Any]:
    """Gets the essential connective structure of a catchment

    Args:
        simulation (Simulation): base catchment simulation

    Returns:
        Dict[str, Any]: A nested dictionary describing the catchment connectivity of subareas, links, and nodes

    Examples:
        >>> _, simulation = sdh.create_test_catchment_structure()
        >>> simulation.get_catchment_structure()
        {'Node':    Id     Name
        0  n1  n1_name
        1  n2  n2_name
        2  n3  n3_name
        3  n4  n4_name
        4  n5  n5_name
        5  n6  n6_name, 'Link':      Id       Name  LengthMetres    f  ManningsN  Slope
        0  lnk1  lnk1_name           0.0  0.0        0.0    0.0
        1  lnk2  lnk2_name           0.0  0.0        0.0    0.0
        2  lnk3  lnk3_name           0.0  0.0        0.0    0.0
        3  lnk4  lnk4_name           0.0  0.0        0.0    0.0
        4  lnk5  lnk5_name           0.0  0.0        0.0    0.0, 'Subarea':      Id       Name  AreaKm2
        0  lnk1  lnk1_name      1.1
        1  lnk2  lnk2_name      2.2
        2  lnk3  lnk3_name      3.3
        3  lnk4  lnk4_name      4.4
        4  lnk5  lnk5_name      5.5, 'NodeLink':   DownstreamId UpstreamId LinkId
        0           n6         n2   lnk1
        1           n2         n5   lnk2
        2           n2         n4   lnk3
        3           n4         n3   lnk4
        4           n4         n1   lnk5, 'SubareaLink':   LinkId SubareaId
        0   lnk1      lnk1
        1   lnk2      lnk2
        2   lnk3      lnk3
        3   lnk4      lnk4
    """
    return smd.get_catchment_structure(self)

get_link_ids()

Gets all the identifiers of the links in the catchment

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def get_link_ids(self) -> List[str]:
    """
    Gets all the identifiers of the links in the catchment
    """
    return ss.get_link_ids(self)

get_link_names()

Gets all the names of the links in the catchment

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def get_link_names(self) -> List[str]:
    """
    Gets all the names of the links in the catchment
    """
    return ss.get_link_names(self)

get_node_ids()

Gets all the identifiers of the nodes in the catchment

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def get_node_ids(self) -> List[str]:
    """
    Gets all the identifiers of the nodes in the catchment
    """
    return ss.get_node_ids(self)

get_node_names()

Gets all the names of the nodes in the catchment

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def get_node_names(self) -> List[str]:
    """
    Gets all the names of the nodes in the catchment
    """
    return ss.get_node_names(self)

get_played(var_ids=None, start_time=None, end_time=None)

Retrieves one or more played (input) time series from a simulation

Parameters:

Name	Type	Description	Default
var_ids	optional str or sequence of str	name(s) of the model variable(s) into which a time series is played as input. e.g. 'Catchment.StreamflowRate'. If missing, a multivariate time series of all played states is returned; this may be a large amount of data.	None
start_time	datetime like	An optional parameter, the start of a period to subset the time series	None
end_time	datetime like	An optional parameter, the end of a period to subset the time series	None

Returns:

Type	Description
xr.DataArray	xr.DataArray: a time series, possibly multivariate.

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def get_played(
    self,
    var_ids: Optional["VecStr"] = None,
    start_time: Optional[ConvertibleToTimestamp] = None,
    end_time: Optional[ConvertibleToTimestamp] = None,
) -> xr.DataArray:
    """
    Retrieves one or more played (input) time series from a simulation

    Args:
        var_ids (optional str or sequence of str): name(s) of the model variable(s) into which a time series is played as input. e.g. 'Catchment.StreamflowRate'. If missing, a multivariate time series of all played states is returned; this may be a large amount of data.
        start_time (datetime like): An optional parameter, the start of a period to subset the time series
        end_time (datetime like): An optional parameter, the end of a period to subset the time series

    Returns:
        xr.DataArray: a time series, possibly multivariate.

    """
    return spr.get_played(self, var_ids, start_time, end_time)

get_played_varnames()

Gets all the names of model states fed an input time series

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def get_played_varnames(self) -> List[str]:
    """
    Gets all the names of model states fed an input time series
    """
    return spr.get_played_varnames(self)

get_recorded(var_ids=None, start_time=None, end_time=None)

Retrieves a recorded time series from a simulation

Parameters:

Name	Type	Description	Default
var_ids	optional str or sequence of str	name(s) of the model variable(s) recorded to a time series. e.g. 'Catchment.StreamflowRate'. If missing, a multivariate time series of all played states is returned; this may be a large amount of data.	None
start_time	datetime like	An optional parameter, the start of a period to subset the time series	None
end_time	datetime like	An optional parameter, the end of a period to subset the time series	None

Returns:

Type	Description
xr.DataArray	xr.DataArray: a time series, possibly multivariate.

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def get_recorded(
    self,
    var_ids: Optional["VecStr"] = None,
    start_time: Optional[ConvertibleToTimestamp] = None,
    end_time: Optional[ConvertibleToTimestamp] = None,
) -> xr.DataArray:
    """
    Retrieves a recorded time series from a simulation

    Args:
        var_ids (optional str or sequence of str): name(s) of the model variable(s) recorded to a time series. e.g. 'Catchment.StreamflowRate'. If missing, a multivariate time series of all played states is returned; this may be a large amount of data.
        start_time (datetime like): An optional parameter, the start of a period to subset the time series
        end_time (datetime like): An optional parameter, the end of a period to subset the time series

    Returns:
        xr.DataArray: a time series, possibly multivariate.

    """
    return spr.get_recorded(self, var_ids, start_time, end_time)

get_simulation_span()

Gets the simulation span of this simulation

Returns:

Type	Description
Dict[str, Any]	Dict[str,Any]: information on start, end, time step

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def get_simulation_span(self) -> Dict[str, Any]:
    """Gets the simulation span of this simulation

    Returns:
        Dict[str,Any]: information on start, end, time step
    """
    return swc.get_simulation_span_pkg(self)

get_state_value(var_id)

Gets the value(s) of a model state(s)

Parameters:

Name	Type	Description	Default
var_id	VecStr	string or sequence of str, model variable state identifier(s)	required

Returns:

Type	Description
Union[Dict[str, float], float]	value(s) of the requested model states

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def get_state_value(self, var_id: "VecStr") -> Union[Dict[str, float], float]:
    """
    Gets the value(s) of a model state(s)

    Args:
        var_id (VecStr): string or sequence of str, model variable state identifier(s)

    Returns:
        value(s) of the requested model states
    """
    return ss.get_state_value(self, var_id)

get_subarea_ids()

Gets all the identifiers of the subareas in the catchment

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def get_subarea_ids(self) -> List[str]:
    """
    Gets all the identifiers of the subareas in the catchment
    """
    return ss.get_subarea_ids(self)

get_subarea_names()

Gets all the names of the subareas in the catchment

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def get_subarea_names(self) -> List[str]:
    """
    Gets all the names of the subareas in the catchment
    """
    return ss.get_subarea_names(self)

get_variable_ids(element_id=None, full_id=True)

Gets all the names of the variables of an element (link, node, subarea) within a catchment

Parameters:

Name	Type	Description	Default
element_id	Optional[str]	a character, identifier of the element within the catchment	None
full_id	bool	boolean, if TRUE return the full hierarchical identifier	True

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def get_variable_ids(
    self, element_id: Optional[str] = None, full_id: bool = True
) -> List[str]:
    """
    Gets all the names of the variables of an element (link, node, subarea) within a catchment

    Args:
        element_id (Optional[str]): a character, identifier of the element within the catchment
        full_id (bool): boolean, if TRUE return the full hierarchical identifier

    """
    return ss.get_variable_ids(self, element_id, full_id)

is_variable_id(var_id)

Are one or more model state identifier(s) valid

Parameters:

Name	Type	Description	Default
var_id	VecStr	model identifier(s)	required

Returns:

Type	Description
Union[Dict[str, bool], bool]	Union[Dict[str, bool], bool]: whether the identifier(s) are valid. A dictionary is returned if the input is vectorised rather than scalar.

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def is_variable_id(self, var_id: "VecStr") -> Union[Dict[str, bool], bool]:
    """Are one or more model state identifier(s) valid

    Args:
        var_id (VecStr): model identifier(s)

    Returns:
        Union[Dict[str, bool], bool]: whether the identifier(s) are valid. A dictionary is returned if the input is vectorised rather than scalar.
    """
    return ss.is_variable_id(self, var_id)

muskingum_param_constraints(inner_parameters, delta_t=1.0, param_name_k='K', param_name_x='X')

Create a parameteriser with Muskingum-type constraints.

Given an existing parameteriser, create a wrapper that adds constraints on two of its parameters.

Parameters:

Name	Type	Description	Default
inner_parameters	HypercubeParameteriser	A SWIFT parameteriser object that contains two Muskingum-type attenuation and delay parameters.	required
delta_t	int	the simulation time step in HOURS. Defaults to 1.	1.0
param_name_k	str	the variable identifier to use for the delay parameter of the Muskingum routing. Defaults to "K".	'K'
param_name_x	str	the variable identifier to use for the attenuation parameter of the Muskingum routing. Defaults to "X".	'X'

Returns:

Name	Type	Description
ConstraintParameteriser	ConstraintParameteriser	A parameteriser with constraints on the feasibility of the attenuation / delay parameters

Examples:

>>> todo()

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def muskingum_param_constraints(
    self,
    inner_parameters: "HypercubeParameteriser",
    delta_t: float = 1.0,
    param_name_k: str = "K",
    param_name_x: str = "X",
) -> "ConstraintParameteriser":
    """Create a parameteriser with Muskingum-type constraints. 

    Given an existing parameteriser, create a wrapper that adds constraints on two of its parameters.

    Args:
        inner_parameters (HypercubeParameteriser): A SWIFT parameteriser object that contains two Muskingum-type attenuation  and delay parameters.
        delta_t (int, optional): the simulation time step in HOURS. Defaults to 1.
        param_name_k (str, optional): the variable identifier to use for the delay parameter of the Muskingum routing. Defaults to "K".
        param_name_x (str, optional): the variable identifier to use for the attenuation parameter of the Muskingum routing. Defaults to "X".

    Returns:
        ConstraintParameteriser: A parameteriser with constraints on the feasibility of the attenuation / delay parameters

    Examples:
       >>> todo()
    """
    return sp.create_muskingum_param_constraints(
        inner_parameters, delta_t, param_name_k, param_name_x, self
    )

play_input(input_ts, var_ids=None)

Sets one or more time series as input(s) to a simulation

Parameters:

Name	Type	Description	Default
input_ts	TimeSeriesLike	univariate time series. If an xts time series column names must be valid model variable identifiers, unless explicitely provided via varIds	required
var_ids	optional str or sequence of str	optional character, the variable identifiers to use, overriding the column names of the inputTs. If not NULL, must be of length equal to the number of columns in inputTs	None

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def play_input(
    self, input_ts: TimeSeriesLike, var_ids: Optional["VecStr"] = None
) -> None:
    """
    Sets one or more time series as input(s) to a simulation

    Args:
        input_ts (TimeSeriesLike): univariate time series. If an xts time series column names must be valid model variable identifiers, unless explicitely provided via varIds
        var_ids (optional str or sequence of str): optional character, the variable identifiers to use, overriding the column names of the inputTs. If not NULL, must be of length equal to the number of columns in inputTs
    """
    spr.play_singular_simulation(self, input_ts, var_ids)

play_inputs(data_library, model_var_id, data_id, resample='')

Assign input time series from a time series library to a model simulation

Parameters:

Name	Type	Description	Default
data_library	TimeSeriesLibrary	external pointer type ENSEMBLE_DATA_SET_PTR, or a Python class wrapper around it	required
model_var_id	str or sequence of str	model state variable unique identifier(s)	required
data_id	str or sequence of str	identifier(s) for data in the data_library. If length is not the same as model_var_id, the elements of data_id are reused to match it	required
resample	str or sequence of str	identifier(s) for how the series is resampled (aggregated or disaggregated). If length is not the same as model_var_id, the elements of resample are reused to match it	''

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def play_inputs(
    self,
    data_library: uc.TimeSeriesLibrary,
    model_var_id: "VecStr",
    data_id: "VecStr",
    resample: "VecStr" = "",
) -> None:
    """
    Assign input time series from a time series library to a model simulation

    Args:
        data_library (TimeSeriesLibrary): external pointer type ENSEMBLE_DATA_SET_PTR, or a Python class wrapper around it 
        model_var_id (str or sequence of str): model state variable unique identifier(s)
        data_id (str or sequence of str): identifier(s) for data in the data_library. If length is not the same as model_var_id, the elements of data_id are reused to match it
        resample (str or sequence of str): identifier(s) for how the series is resampled (aggregated or disaggregated). If length is not the same as model_var_id, the elements of resample are reused to match it

    """
    spr.play_inputs(self, data_library, model_var_id, data_id, resample)

play_subarea_input(input, subarea_name, input_name)

Sets time series as input to a simulation

Parameters:

Name	Type	Description	Default
input	TimeSeriesLike	univariate time series.	required
subarea_name	str	a valid name of the subarea	required
input_name	str	the name of the input variable to the model (i.e. 'P' for the precip of GR5H)	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def play_subarea_input(
    self, input: TimeSeriesLike, subarea_name: str, input_name: str
) -> None:
    """
    Sets time series as input to a simulation

    Args:
        input (TimeSeriesLike): univariate time series.
        subarea_name (str): a valid name of the subarea
        input_name (str): the name of the input variable to the model (i.e. 'P' for the precip of GR5H)

    """
    spr.play_subarea_input(self, input, subarea_name, input_name)

prepare_dual_pass_forecasting(observation, error_model_element_id, warmup_start, warmup_end, required_windows_percentage)

Create an ensemble simulation for forecasting with the Dual Pass error correction method

Parameters:

Name	Type	Description	Default
observation	TimeSeriesLike	Time series of observations to correct prediction against	required
error_model_element_id	str	model element identifier where to set up an ERRIS correction model	required
warmup_start	ConvertibleToTimestamp	start time stamp for the warmup period	required
warmup_end	ConvertibleToTimestamp	end time stamp for the warmup period	required
required_windows_percentage	float	required_windows_percentage	required

Returns:

Name	Type	Description
EnsembleSimulation	EnsembleSimulation	Ensemble simulation (ensemble simulation runner)

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def prepare_dual_pass_forecasting(
    self,
    observation: TimeSeriesLike,
    error_model_element_id: str,
    warmup_start: ConvertibleToTimestamp,
    warmup_end: ConvertibleToTimestamp,
    required_windows_percentage: float,
) -> 'EnsembleSimulation':
    """Create an ensemble simulation for forecasting with the Dual Pass error correction method 

    Args:
        observation (TimeSeriesLike): Time series of observations to correct prediction against
        error_model_element_id (str): model element identifier where to set up an ERRIS correction model
        warmup_start (ConvertibleToTimestamp): start time stamp for the warmup period
        warmup_end (ConvertibleToTimestamp): end time stamp for the warmup period
        required_windows_percentage (float): required_windows_percentage

    Returns:
        EnsembleSimulation: Ensemble simulation (ensemble simulation runner)
    """    
    from swift2.internal import to_interop_univariate_series

    values, ts_geom = to_interop_univariate_series(
        observation, warmup_start, warmup_end
    )
    return swg.PrepareDualPassForecasting_py(
        self,
        values,
        ts_geom,
        error_model_element_id,
        warmup_start,
        warmup_end,
        required_windows_percentage,
    )

prepare_erris_forecasting(observation, error_model_element_id, warmup_start, warmup_end)

Create an ensemble simulation for forecasting with ERRIS

Parameters:

Name	Type	Description	Default
observation	TimeSeriesLike	Time series of observations to correct prediction against	required
error_model_element_id	str	model element identifier where to set up an ERRIS correction model	required
warmup_start	ConvertibleToTimestamp	start time stamp for the warmup period	required
warmup_end	ConvertibleToTimestamp	end time stamp for the warmup period	required

Returns:

Name	Type	Description
EnsembleSimulation	EnsembleSimulation	Ensemble simulation (ensemble simulation runner)

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def prepare_erris_forecasting(
    self,
    observation: TimeSeriesLike,
    error_model_element_id: str,
    warmup_start: ConvertibleToTimestamp,
    warmup_end: ConvertibleToTimestamp,
) -> 'EnsembleSimulation':
    """Create an ensemble simulation for forecasting with ERRIS

    Args:
        observation (TimeSeriesLike): Time series of observations to correct prediction against
        error_model_element_id (str): model element identifier where to set up an ERRIS correction model
        warmup_start (ConvertibleToTimestamp): start time stamp for the warmup period
        warmup_end (ConvertibleToTimestamp): end time stamp for the warmup period

    Returns:
        EnsembleSimulation: Ensemble simulation (ensemble simulation runner)
    """
    from swift2.internal import to_interop_univariate_series

    values, ts_geom = to_interop_univariate_series(
        observation, warmup_start, warmup_end
    )
    return swg.PrepareERRISForecasting_py(
        self, values, ts_geom, error_model_element_id, warmup_start, warmup_end
    )

record_singular_state(var_ids=CATCHMENT_FLOWRATE_VARID, recording_provider=None, data_ids=None)

DRAFT Advanced/technical. Record states to a record provider.

Likely not for end users.

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def record_singular_state(
    self,
    var_ids: "VecStr" = CATCHMENT_FLOWRATE_VARID,
    recording_provider: Optional["TimeSeriesLibrary"] = None,
    data_ids: Optional["VecStr"] = None,
) -> None:
    """DRAFT Advanced/technical. Record states to a record provider. 

    Likely not for end users.
    """
    spr.record_singular_state(self, var_ids, recording_provider, data_ids)

remove_state_initialisers()

Forces the removal of any state initialiser.

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def remove_state_initialisers(self):
    """Forces the removal of any state initialiser."""
    swg.RemoveStateInitializerModelRunner_py(self)

reset_model_states()

Reset the model states of a simulation, and apply one or more state initialers if the simulation is configured with any.

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def reset_model_states(self) -> None:
    """Reset the model states of a simulation, and apply one or more state initialers if the simulation is configured with any."""
    ss.reset_model_states(self)

set_error_correction_model(model_id, element_id, length=1, seed=0)

Add an error correction model to an element in a catchment

Parameters:

Name	Type	Description	Default
model_id	str	the identifier of the new model to use, e.g. 'ERRIS'	required
element_id	str	the identifier of the catchment element (node, link, subcatchment) whose outflow rate is corrected.	required
length	int	other parameters to pass to the creation of the error correction model. Currently length of the AR model only supported.	1
seed	int	other parameters to pass to the creation of the error correction model. Currently length of the AR model only supported.	0

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def set_error_correction_model(
    self, model_id: str, element_id: str, length: int = 1, seed: int = 0
) -> None:
    """
    Add an error correction model to an element in a catchment

    Args:
        model_id (str): the identifier of the new model to use, e.g. 'ERRIS'
        element_id (str): the identifier of the catchment element (node, link, subcatchment) whose outflow rate is corrected.
        length (int): other parameters to pass to the creation of the error correction model. Currently length of the AR model only supported.
        seed (int): other parameters to pass to the creation of the error correction model. Currently length of the AR model only supported.

    """
    ss.set_error_correction_model(self, model_id, element_id, length, seed)

set_reservoir_geometry(element_id, level, storage, area)

Sets the geometry of a reservoir

Parameters:

Name	Type	Description	Default
element_id	str	Element with a suitable reservoir supporting a geometry description	required
level	np.ndarray	array of water surface levels, in S.I. units (m) TO BE CONFIRMED	required
storage	np.ndarray	array of volume storages, in S.I. units (m3) TO BE CONFIRMED	required
area	np.ndarray	array of surfce areas, in S.I. units (m2) TO BE CONFIRMED	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def set_reservoir_geometry(
    self, element_id: str, level: np.ndarray, storage: np.ndarray, area: np.ndarray
) -> None:
    """Sets the geometry of a reservoir

    Args:
        element_id (str): Element with a suitable reservoir supporting a geometry description
        level (np.ndarray): array of water surface levels, in S.I. units (m) TO BE CONFIRMED
        storage (np.ndarray): array of volume storages, in S.I. units (m3) TO BE CONFIRMED
        area (np.ndarray): array of surfce areas, in S.I. units (m2) TO BE CONFIRMED
    """
    num_entries = len(level)
    assert len(storage) == num_entries
    assert len(area) == num_entries
    swg.SetReservoirGeometry_py(self, element_id, num_entries, level, storage, area)

set_reservoir_max_discharge(element_id, level, discharge)

Sets a reservoir operating curve, maximum release for a given level

Parameters:

Name	Type	Description	Default
element_id	str	Element with a suitable reservoir supporting a geometry description	required
level	np.ndarray	array of levels (m)	required
discharge	np.ndarray	array of maximum discharges (m3/s)	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def set_reservoir_max_discharge(
    self, element_id: str, level: np.ndarray, discharge: np.ndarray
) -> None:
    """Sets a reservoir operating curve, maximum release for a given level

    Args:
        element_id (str): Element with a suitable reservoir supporting a geometry description
        level (np.ndarray): array of levels (m)
        discharge (np.ndarray): array of maximum discharges (m3/s)
    """
    num_entries = len(level)
    swg.SetReservoirMaxDischarge_py(self, element_id, num_entries, level, discharge)

set_reservoir_min_discharge(element_id, level, discharge)

Sets a reservoir operating curve, minimum release for a given level

Parameters:

Name	Type	Description	Default
element_id	str	Element with a suitable reservoir supporting a geometry description	required
level	np.ndarray	array of levels (m)	required
discharge	np.ndarray	array of minimum discharges (m3/s)	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def set_reservoir_min_discharge(
    self, element_id: str, level: np.ndarray, discharge: np.ndarray
) -> None:
    """Sets a reservoir operating curve, minimum release for a given level

    Args:
        element_id (str): Element with a suitable reservoir supporting a geometry description
        level (np.ndarray): array of levels (m)
        discharge (np.ndarray): array of minimum discharges (m3/s)
    """
    num_entries = len(level)
    swg.SetReservoirMinDischarge_py(self, element_id, num_entries, level, discharge)

set_reservoir_model(new_model_id, element_id)

Sets a new reservoir model on an element

Parameters:

Name	Type	Description	Default
new_model_id	str	Currently one of: "ControlledReleaseReservoir", "LevelVolumeAreaReservoir", "FarmDamReservoir";	required
element_id	str	description	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def set_reservoir_model(self, new_model_id: str, element_id: str) -> None:
    """Sets a new reservoir model on an element

    Args:
        new_model_id (str): Currently one of: "ControlledReleaseReservoir", "LevelVolumeAreaReservoir", "FarmDamReservoir";
        element_id (str): _description_
    """
    swg.SetReservoirModel_py(self, new_model_id, element_id)

set_simulation_span(start, end)

Sets the simulation span

Parameters:

Name	Type	Description	Default
start	ConvertibleToTimestamp	the start date of the simulation. The time zone will be forced to UTC.	required
end	ConvertibleToTimestamp	the end date of the simulation. The time zone will be forced to UTC.	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def set_simulation_span(
    self, start: ConvertibleToTimestamp, end: ConvertibleToTimestamp
) -> None:
    """
    Sets the simulation span

    Args:
        start (ConvertibleToTimestamp): the start date of the simulation. The time zone will be forced to UTC.
        end (ConvertibleToTimestamp): the end date of the simulation. The time zone will be forced to UTC.
    """
    ss.set_simulation_span(self, start, end)

set_simulation_time_step(name)

Sets the time step of this simulation

Parameters:

Name	Type	Description	Default
name	str	a time step identifier, currently 'daily' or 'hourly' are supported. The identifier is made lower case in the function.	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def set_simulation_time_step(self, name: str) -> None:
    """
    Sets the time step of this simulation

    Args:
        name (str): a time step identifier, currently 'daily' or 'hourly' are supported. The identifier is made lower case in the function.
    """
    ss.set_simulation_time_step(self, name)

set_state_value(var_id, value=None)

Sets the value of a model state

Parameters:

Name	Type	Description	Default
var_id	Any	character, model variable state identifier(s)	required
value	Any	numeric value(s)	None

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def set_state_value(
    self,
    var_id: Union[str, Sequence[str]],
    value: Union[float, int, bool, Sequence] = None,
) -> None:
    """
    Sets the value of a model state

    Args:
        var_id (Any): character, model variable state identifier(s)
        value (Any): numeric value(s)

    """
    ss.set_state_value(self, var_id, value)

set_states(states)

Apply memory states to a simulation

Parameters:

Name	Type	Description	Default
states	MemoryStates	memory states	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def set_states(self, states: "MemoryStates") -> None:
    """Apply memory states to a simulation

    Args:
        states (MemoryStates): memory states
    """
    ss.set_states(self, states)

snapshot_state()

Take a snapshot of the memory states of a simulation

Returns:

Name	Type	Description
MemoryStates	MemoryStates	memory states, that can be stored and reapplied

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def snapshot_state(self) -> "MemoryStates":
    """Take a snapshot of the memory states of a simulation

    Returns:
        MemoryStates: memory states, that can be stored and reapplied
    """
    return ss.snapshot_state(self)

sort_by_execution_order(split_element_ids, sorting_option='')

Sort the specified element ids according to the execution order of the simulation

Parameters:

Name	Type	Description	Default
split_element_ids	Sequence[str]	a character vector with element identifiers such as 'node.n1', 'link.linkId_2'	required
sorting_option	str	a character - for future options. Ignored for now.	''

Returns:

Type	Description
List[str]	List[str]: values in split_element_ids sorted by simulation execution order

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def sort_by_execution_order(
    self, split_element_ids: Sequence[str], sorting_option: str = ""
) -> List[str]:
    """
    Sort the specified element ids according to the execution order of the simulation

    Args:
        split_element_ids (Sequence[str]): a character vector with element identifiers such as 'node.n1', 'link.linkId_2'
        sorting_option (str): a character - for future options. Ignored for now.

    Returns:
        List[str]: values in split_element_ids sorted by simulation execution order

    """
    return ss.sort_by_execution_order(self, split_element_ids, sorting_option)

split_to_subcatchments(split_element_ids, include_upstream=None)

Split a catchment in subcatchments, given a list of node/link element identifiers

Parameters:

Name	Type	Description	Default
split_element_ids	str	element identifiers such as 'node.n1', 'link.linkId_2'	required
include_upstream	bool	indicates whether for each element in split_element_ids it should be including in the upstream portion of the subcatchment. Defaults to None.	None

Returns:

Name	Type	Description
OrderedDict	OrderedDict[str, Simulation]	list of subcatchments resulting from the split

Examples:

>>> _, simulation = sdh.create_test_catchment_structure()
>>> e_ids = ['node.n2', 'node.n4']
>>> sub_sims = simulation.split_to_subcatchments(e_ids)
>>> for k in sub_sims:
>>>     print(k)
>>>     print(sub_sims[k].get_node_ids())
>>>     print(sub_sims[k].get_node_names())
node.n4
['n4', 'n3', 'n1']
['n4_name', 'n3_name', 'n1_name']
node.n2
['n2', 'n5']
['n2_name', 'n5_name']
remainder
['n6']
['n6_name']

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def split_to_subcatchments(
    self, split_element_ids: Sequence[str], include_upstream: Sequence[bool] = None
) -> OrderedDict[str, "Simulation"]:
    """Split a catchment in subcatchments, given a list of node/link element identifiers

    Args:
        split_element_ids (str): element identifiers such as 'node.n1', 'link.linkId_2'
        include_upstream (bool, optional): indicates whether for each element in split_element_ids it should be including in the upstream portion of the subcatchment. Defaults to None.

    Returns:
        OrderedDict: list of subcatchments resulting from the split

    Examples:
        >>> _, simulation = sdh.create_test_catchment_structure()
        >>> e_ids = ['node.n2', 'node.n4']
        >>> sub_sims = simulation.split_to_subcatchments(e_ids)
        >>> for k in sub_sims:
        >>>     print(k)
        >>>     print(sub_sims[k].get_node_ids())
        >>>     print(sub_sims[k].get_node_names())
        node.n4
        ['n4', 'n3', 'n1']
        ['n4_name', 'n3_name', 'n1_name']
        node.n2
        ['n2', 'n5']
        ['n2_name', 'n5_name']
        remainder
        ['n6']
        ['n6_name']
    """
    return smd.split_to_subcatchments(self, split_element_ids, include_upstream)

subset_catchment(element_id, action='keep_above')

Subsets a catchment, keeping only a portion above or below a node, link or subarea.

Parameters:

Name	Type	Description	Default
element_id	str	id of the element to cut at.	required
action	str	how to cut; currently limited to 'keep_above'	'keep_above'

Returns:

Name	Type	Description
Simulation		a subcatchment simulation, cookie cut from the base simulation. Deep clone of objects.

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def subset_catchment(self, element_id: str, action: str = "keep_above"):
    """Subsets a catchment, keeping only a portion above or below a node, link or subarea.

    Args:
        element_id (str): id of the element to cut at.
        action (str): how to cut; currently limited to 'keep_above'

    Returns:
        Simulation: a subcatchment simulation, cookie cut from the base simulation. Deep clone of objects.
    """
    return smd.subset_catchment(self, element_id, action)

swap_model(model_id, what='runoff')

Clone and change a simulation, using another model

Parameters:

Name	Type	Description	Default
model_id	str	the identifier of the new model to use, e.g. 'GR4J'	required
what	str	character identifying the type of structure replaced: 'runoff', 'channel_routing'	'runoff'

Returns:

Name	Type	Description
Simulation	Simulation	A SWIFT simulation object, clone of the simulation but with a new model type in use.

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def swap_model(self, model_id: str, what: str = "runoff") -> "Simulation":
    """
    Clone and change a simulation, using another model

    Args:
        model_id (str): the identifier of the new model to use, e.g. 'GR4J'
        what (str): character identifying the type of structure replaced: 'runoff', 'channel_routing'

    Returns:
        Simulation: A SWIFT simulation object, clone of the simulation but with a new model type in use.

    """
    return ss.swap_model(self, model_id, what)

use_state_initialises(state_initialiser)

Sets the state initialiser to use for a simulation. This forces the removal of any prior state initialiser.

Parameters:

Name	Type	Description	Default
state_initialiser	StateInitialiser	the new state initialiser to use	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def use_state_initialises(self, state_initialiser: "StateInitialiser"):
    """Sets the state initialiser to use for a simulation. This forces the removal of any prior state initialiser.

    Args:
        state_initialiser (StateInitialiser): the new state initialiser to use
    """
    swg.UseStateInitializerModelRunner_py(self, state_initialiser)

SimulationMixin

A parent class for simulation objects. Most users are unlikely to explicitly use it.

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

class SimulationMixin:
    """A parent class for simulation objects. Most users are unlikely to explicitly use it."""

    def __init__(
        self,
    ):
        super(SimulationMixin, self).__init__()

    def record_state(
        self,
        var_ids: "VecStr" = CATCHMENT_FLOWRATE_VARID,
        recording_provider: Optional["TimeSeriesLibrary"] = None,
        data_ids: Optional["VecStr"] = None,
    ) -> None:
        """
        Record a time series of one of the state of the model

        Args:
            var_ids (VecStr, optional): identifier(s) of the output variable recorded to a time series, e.g. 'Catchment|StreamflowRate' or 'subcatchment.Subarea.runoff'. Defaults to CATCHMENT_FLOWRATE_VARID.
            recording_provider (TimeSeriesLibrary, optional): _description_. Defaults to None.
            data_ids (VecStr, optional): _description_. Defaults to None.

        Raises:
            ValueError: _description_
        """
        spr.record_state(self, var_ids, recording_provider, data_ids)

    def get_recorded_varnames(self) -> List[str]:
        """
        Gets all the names of the recorded states

        Returns:
            List[str]: The names of the state variables being recorded into time series
        """
        return spr.get_recorded_varnames(self)

    def get_played_varnames(self) -> List[str]:
        """
        Gets all the names of states fed an input time series

        Returns:
            List[str]: The names of the state variables fed over the simulation with values from a time series
        """
        return spr.get_played_varnames(self)

    def exec_simulation(self, reset_initial_states: bool = True) -> None:
        """
        Execute a simulation

        Args:
            reset_initial_states (bool): logical, should the states of the model be reinitialized before the first time step.

        """
        ss.exec_simulation(self, reset_initial_states)

exec_simulation(reset_initial_states=True)

Execute a simulation

Parameters:

Name	Type	Description	Default
reset_initial_states	bool	logical, should the states of the model be reinitialized before the first time step.	True

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def exec_simulation(self, reset_initial_states: bool = True) -> None:
    """
    Execute a simulation

    Args:
        reset_initial_states (bool): logical, should the states of the model be reinitialized before the first time step.

    """
    ss.exec_simulation(self, reset_initial_states)

get_played_varnames()

Gets all the names of states fed an input time series

Returns:

Type	Description
List[str]	List[str]: The names of the state variables fed over the simulation with values from a time series

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def get_played_varnames(self) -> List[str]:
    """
    Gets all the names of states fed an input time series

    Returns:
        List[str]: The names of the state variables fed over the simulation with values from a time series
    """
    return spr.get_played_varnames(self)

get_recorded_varnames()

Gets all the names of the recorded states

Returns:

Type	Description
List[str]	List[str]: The names of the state variables being recorded into time series

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def get_recorded_varnames(self) -> List[str]:
    """
    Gets all the names of the recorded states

    Returns:
        List[str]: The names of the state variables being recorded into time series
    """
    return spr.get_recorded_varnames(self)

record_state(var_ids=CATCHMENT_FLOWRATE_VARID, recording_provider=None, data_ids=None)

Record a time series of one of the state of the model

Parameters:

Name	Type	Description	Default
var_ids	VecStr	identifier(s) of the output variable recorded to a time series, e.g. 'Catchment|StreamflowRate' or 'subcatchment.Subarea.runoff'. Defaults to CATCHMENT_FLOWRATE_VARID.	CATCHMENT_FLOWRATE_VARID
recording_provider	TimeSeriesLibrary	description. Defaults to None.	None
data_ids	VecStr	description. Defaults to None.	None

Raises:

Type	Description
ValueError	description

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def record_state(
    self,
    var_ids: "VecStr" = CATCHMENT_FLOWRATE_VARID,
    recording_provider: Optional["TimeSeriesLibrary"] = None,
    data_ids: Optional["VecStr"] = None,
) -> None:
    """
    Record a time series of one of the state of the model

    Args:
        var_ids (VecStr, optional): identifier(s) of the output variable recorded to a time series, e.g. 'Catchment|StreamflowRate' or 'subcatchment.Subarea.runoff'. Defaults to CATCHMENT_FLOWRATE_VARID.
        recording_provider (TimeSeriesLibrary, optional): _description_. Defaults to None.
        data_ids (VecStr, optional): _description_. Defaults to None.

    Raises:
        ValueError: _description_
    """
    spr.record_state(self, var_ids, recording_provider, data_ids)

TransformParameteriser

Bases: HypercubeParameteriser

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

class TransformParameteriser(HypercubeParameteriser):
    def __init__(
        self,
        handle: CffiData,
        release_native: Callable[[CffiData], None],
        type_id: Optional[str] = None,
        prior_ref_count: int = 0,
    ):
        super(TransformParameteriser, self).__init__(
            handle, release_native, type_id, prior_ref_count
        )

    def add_transform(
        self,
        param_name: str,
        inner_param_name: str,
        transform_id: str,
        a: float = 1.0,
        b: float = 0.0,
    ):
        """Create a parameteriser for which parameter transformations can be defined

            This allows to define e.g. a virtual parameter log_X instead of calibrating on the parameter X.

        Args:
            param_name (str): the name of the meta-parameter. Note that it can be the same value as inner_param_name, but this is NOT recommended.
            inner_param_name (str): the name of the parameter being transformed
            transform_id (str): identifier for a known bijective univariate function
            a (float, optional): parameter in Y = F(ax+b). Defaults to 1.0.
            b (float, optional): parameter in Y = F(ax+b). Defaults to 0.0.
        """
        sp.add_transform(self, param_name, inner_param_name, transform_id, a, b)

add_transform(param_name, inner_param_name, transform_id, a=1.0, b=0.0)

Create a parameteriser for which parameter transformations can be defined

This allows to define e.g. a virtual parameter log_X instead of calibrating on the parameter X.

Parameters:

Name	Type	Description	Default
param_name	str	the name of the meta-parameter. Note that it can be the same value as inner_param_name, but this is NOT recommended.	required
inner_param_name	str	the name of the parameter being transformed	required
transform_id	str	identifier for a known bijective univariate function	required
a	float	parameter in Y = F(ax+b). Defaults to 1.0.	1.0
b	float	parameter in Y = F(ax+b). Defaults to 0.0.	0.0

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/classes.py

def add_transform(
    self,
    param_name: str,
    inner_param_name: str,
    transform_id: str,
    a: float = 1.0,
    b: float = 0.0,
):
    """Create a parameteriser for which parameter transformations can be defined

        This allows to define e.g. a virtual parameter log_X instead of calibrating on the parameter X.

    Args:
        param_name (str): the name of the meta-parameter. Note that it can be the same value as inner_param_name, but this is NOT recommended.
        inner_param_name (str): the name of the parameter being transformed
        transform_id (str): identifier for a known bijective univariate function
        a (float, optional): parameter in Y = F(ax+b). Defaults to 1.0.
        b (float, optional): parameter in Y = F(ax+b). Defaults to 0.0.
    """
    sp.add_transform(self, param_name, inner_param_name, transform_id, a, b)

module common

module const

module doc_helper

configure_hourly_gr4j(simulation)

Configure a simulation with GR4J models for hourly time step modelling

Configure a simulation with GR4J models for hourly time step modelling

Parameters:

Name	Type	Description	Default
simulation	Simulation	A swift simulation object	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/doc_helper.py

def configure_hourly_gr4j(simulation):
    """
    Configure a simulation with GR4J models for hourly time step modelling

    Configure a simulation with GR4J models for hourly time step modelling

    Args:
        simulation (Simulation): A swift simulation object

    """
    pGr4jHourly = create_gr4jh_parameters()
    apply_sys_config(pGr4jHourly, simulation)

create_gr4jh_parameters()

Get a parameter set that configures GR4J for hourly operations

Get a parameter set that configures GR4J for hourly operations

Returns:

Name	Type	Description
HyperCubeParameteriser		a parameter set that can be applied to SWIFT systems with GR4J

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/doc_helper.py

def create_gr4jh_parameters():
    """
    Get a parameter set that configures GR4J for hourly operations

    Get a parameter set that configures GR4J for hourly operations

    Returns:
        HyperCubeParameteriser: a parameter set that can be applied to SWIFT systems with GR4J

    """
    # Configure for GR4J but with changed parameters, supposed to be OK for hourly operations.
    return create_parameteriser(
        specs=_df_from_dict(
            Name=["PercFactor", "UHExponent"],
            Value=_npf([4, 5 / 4]),
            Min=_npf([4, 5 / 4]),
            Max=_npf([4, 5 / 4]),
        )
    )

define_gr4j_scaled_parameter(ref_area=250, time_span=3600, pspec_gr4j=None)

define a scaled and transformed parameterizer for GR4J

define a scaled and transformed parameterizer for GR4J

Parameters:

Name	Type	Description	Default
ref_area	float	the reference area in square kilometres	250
time_span	int	the time span of the simulation intented for this model, in seconds	3600
pspec_gr4j	pd.DataFrame	optional - data frame specifying the feasible parameter space for parameters x1 to x2 of GR4J	None

Returns:

Name	Type	Description
TransformParameteriser		a parameterizer for GR4J, combining time and area scaling and superimposed with log10 transforms for x1, x3, x4 and arc-sinh for x2

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/doc_helper.py

def define_gr4j_scaled_parameter(ref_area:float=250, time_span:int=3600, pspec_gr4j:Optional[pd.DataFrame]=None):
    """
    define a scaled and transformed parameterizer for GR4J

    define a scaled and transformed parameterizer for GR4J

    Args:
        ref_area (float): the reference area in square kilometres
        time_span (int): the time span of the simulation intented for this model, in seconds
        pspec_gr4j (pd.DataFrame): optional - data frame specifying the feasible parameter space for parameters x1 to x2 of GR4J

    Returns:
        TransformParameteriser: a parameterizer for GR4J, combining time and area scaling and superimposed with log10 transforms for x1, x3, x4 and arc-sinh for x2
    """
    time_span = int(time_span)
    parameteriser = gr4j_scaled_parameteriser(ref_area, time_span)
    if pspec_gr4j is None:
        pspec_gr4j = _df_from_dict(
            Name=["x1", "x2", "x3", "x4"],
            Value=_npf([3.21137e00, 6.95511e00, 2.06740e00, 2.02033e00]),
            Min=_npf([1.0e00, -27.0, 1.0e00, 1.0e00]),
            Max=_npf([6.0e03, 27.0, 1.0e03, 2.4e02]),
        )
    set_hypercube(parameteriser, pspec_gr4j)
    parameteriser = wrap_transform(parameteriser)
    add_transform(parameteriser, "log_x4", "x4", "log10")
    add_transform(parameteriser, "log_x1", "x1", "log10")
    add_transform(parameteriser, "log_x3", "x3", "log10")
    add_transform(parameteriser, "asinh_x2", "x2", "asinh")

    return parameteriser

get_catchment_dot_graph(simulation)

Gets a catchment representation in Graphviz DOT format

Gets a catchment representation in Graphviz DOT format

Parameters:

Name	Type	Description	Default
simulation	Simulation	A swift simulation object	required

Returns:

Type	Description
	a string in a notation usable by the DiagrammeR package.

Examples:

TODO

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/doc_helper.py

def get_catchment_dot_graph(simulation):
    """
    Gets a catchment representation in Graphviz DOT format

    Gets a catchment representation in Graphviz DOT format

    Args:
        simulation (Simulation): A swift simulation object

    Returns:
        a string in a notation usable by the DiagrammeR package.

    Examples:
        TODO

    """
    # Examples:
    # >>> # library(swift)
    # >>> # library('DiagrammeR')
    # >>> # dataId <- 'MMH'
    # >>> # simulation <- createTestCatchment(nSubareas=4, dataId=dataId, varNameDataRain='rain', varNameDataPet='evap')
    # >>> # DiagrammeR(getCatchmentDotGraph(simulation))
    return swg.GetCatchmentDOTGraph_py(simulation)

get_free_params(model_id)

Get a default parameter set for models

Get a default parameter set for models, as a data frame

Parameters:

Name	Type	Description	Default
model_id	Any	an identifier for the model, e.g. 'GR5H'	required

Returns:

Type	Description
	a data frame with Min, Max, Value, Name

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/doc_helper.py

def get_free_params(model_id):
    """
    Get a default parameter set for models

    Get a default parameter set for models, as a data frame

    Args:
        model_id (Any): an identifier for the model, e.g. 'GR5H'

    Returns:
        a data frame with Min, Max, Value, Name

    """
    pSpec = None
    if model_id == "GR5H":
        pSpec = _df_from_dict(
            Name=["x1", "x2", "x3", "x4", "x5"],
            Value=_npf([44.6, 30.0, 10.0, 14.0, 200.0]),
            Min=_npf([1, 1, 0, 1, 1]),
            Max=_npf([1000, 400, 1000, 240, 1000.0]),
        )
    elif model_id == "GR6J":
        # per202 2014-09-25
        # I take some values for the parameters from the unit tests, but the min/max bounds are PURE guesses. Q/A TBC.
        pSpec = _df_from_dict(
            Name=["x1", "x2", "x3", "x4", "x5", "x6"],
            Value=_npf([20, -2, 10, 2, 0, 1]),
            Min=_npf([1, -5, 0, 1, 0, 0]),
            Max=_npf([1000, 400, 1000, 240, 1, 1]),
        )
    elif model_id == "PDM":
        # per202 2014-11-13
        #
        pSpec = _df_from_dict(
            Name=[
                "cmax",
                "cminrat",
                "b",
                "be",
                "kg",
                "bg",
                "Strat",
                "k1",
                "k2rat",
                "kb",
            ],
            Value=_npf([400, 0.5, 1.8, 1.0, 1300.0, 1.0, 0.5, 35.0, 0.3, 2.4]),
            Min=_npf([1, 0, 0.001, 1.0, 1.0, 1.0, 0.0, 1.0, 0.0001, 1.0]),
            Max=_npf([3000, 1, 2.0, 2.0, 50000.0, 1.0, 1.0, 300.0, 1.0000, 2000.0]),
        )
    elif model_id == "SAC" or model_id == "SACSMA":
        # rob635 2015-01-11
        #
        pSpec = _df_from_dict(
            Name=[
                "UZTWM",
                "UZFWM",
                "UZK",
                "PCTIM",
                "ADIMP",
                "RIVA",
                "REXP",
                "LZTWM",
                "LZFSM",
                "LZFPM",
                "LZSK",
                "LZPK",
                "PFREE",
                "RSERV",
                "SIDE",
            ],
            Value=_npf(
                [
                    50,
                    40,
                    0.245,
                    0.01,
                    0.00,
                    0.01,
                    2.00,
                    130.0,
                    23.0,
                    40.0,
                    0.043,
                    0.009,
                    0.10,
                    0.30,
                    0.00,
                ]
            ),
            Min=_npf(
                [
                    10,
                    5,
                    0.100,
                    0.00,
                    0.00,
                    0.00,
                    1.00,
                    10.0,
                    5.0,
                    10.0,
                    0.001,
                    0.001,
                    0.00,
                    0.00,
                    0.00,
                ]
            ),
            Max=_npf(
                [
                    300,
                    150,
                    0.750,
                    1.00,
                    1.00,
                    1.00,
                    5.00,
                    500.0,
                    400.0,
                    1000.0,
                    0.350,
                    0.050,
                    0.80,
                    1.00,
                    1.00,
                ]
            ),
        )
    elif model_id == "GR4J":
        # \\OSM-01-MEL.it.csiro.au\CLW_HYDROFORECAST_common\Projects\WIRADA_4_1\Work\2013_2014_Act2_Flood_Cal\SWIFT_hourly\C-F2+F3+F5\South_Esk\Output\Out_CaliPara.txt
        # \\OSM-01-MEL.it.csiro.au\CLW_HYDROFORECAST_common\Projects\WIRADA_4_1\Work\2013_2014_Act2_Flood_Cal\SWIFT_hourly\C-F2+F3+F5\South_Esk\Setup\Model_Parameters.txt
        pSpec = _df_from_dict(
            # Name                 X1             X2             X3             X4
            # Number                1              2              3              4
            # Fit(Y\N)              1              1              1              1
            # LowBound    1.00000E+00   -2.70000E+01    1.00000E+00    1.00000E+00
            # UpBound     3.00000E+03    2.70000E+01    6.60000E+02    2.40000E+02
            # #-Calibrated parameters-----------------------------------------------------------------
            # Chmt_001    6.50488E+02   -2.80648E-01    7.89123E+00    1.89172E+01
            Name=["x1", "x2", "x3", "x4"],
            Value=_npf([6.50488e02, -2.80648e-01, 7.89123e00, 1.89172e01]),
            Min=_npf([1, -27, 1, 1]),
            Max=_npf([3.00000e03, 2.70000e01, 6.60000e02, 2.40000e02]),
        )
    elif model_id == "SAC" or model_id == "SACSMA":
        # rob635 2015-01-11
        #
        pSpec = _df_from_dict(
            Name=[
                "UZTWM",
                "UZFWM",
                "UZK",
                "PCTIM",
                "ADIMP",
                "RIVA",
                "REXP",
                "LZTWM",
                "LZFSM",
                "LZFPM",
                "LZSK",
                "LZPK",
                "PFREE",
                "RSERV",
                "SIDE",
            ],
            Value=_npf(
                [
                    50,
                    40,
                    0.245,
                    0.01,
                    0.00,
                    0.01,
                    2.00,
                    130.0,
                    23.0,
                    40.0,
                    0.043,
                    0.009,
                    0.10,
                    0.30,
                    0.00,
                ]
            ),
            Min=_npf(
                [
                    10,
                    5,
                    0.100,
                    0.00,
                    0.00,
                    0.00,
                    1.00,
                    10.0,
                    5.0,
                    10.0,
                    0.001,
                    0.001,
                    0.00,
                    0.00,
                    0.00,
                ]
            ),
            Max=_npf(
                [
                    300,
                    150,
                    0.750,
                    1.00,
                    1.00,
                    1.00,
                    5.00,
                    500.0,
                    400.0,
                    1000.0,
                    0.350,
                    0.050,
                    0.80,
                    1.00,
                    1.00,
                ]
            ),
        )
    else:
        raise Exception("Model identifier not recognised: " + model_id)
    return pSpec

gr4j_scaled_parameteriser(reference_area=240, t_step_seconds=3600)

Get a time step and area scaled parameterizer for the GR4 model structure

Get a time step and area scaled parameterizer for the GR4 model structure

Parameters:

Name	Type	Description	Default
reference_area	Any	The reference area in km^2 for the parameter x4	240
t_step_seconds	Any	The simulation time step in seconds.	3600

Returns:

Type	Description
	A SWIFT catchment parameterizer for GR4 model structures

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/doc_helper.py

def gr4j_scaled_parameteriser(reference_area: float = 240, t_step_seconds: int = 3600):
    """
    Get a time step and area scaled parameterizer for the GR4 model structure

    Get a time step and area scaled parameterizer for the GR4 model structure

    Args:
        reference_area (Any): The reference area in km^2 for the parameter x4
        t_step_seconds (Any): The simulation time step in seconds.

    Returns:
        A SWIFT catchment parameterizer for GR4 model structures

    """
    return swg.CreateGr4ScaledParameterizer_py(reference_area, t_step_seconds)

inspect(simulation, element='link', id='1', full_names=False)

Inspect an element of a catchment model

Inspect the current state values of an element of a catchment model

Parameters:

Name	Type	Description	Default
simulation	Simulation	A swift simulation object	required
element	Any	type of top level element, within c('link','node','subarea')	'link'
id	Any	SWIFT simulation	'1'
full_names	Any	if TRUE returns the full names of state variable ids (e.g. link.link_1.OutlfowRate)	False

Returns:

Type	Description
	named numeric vector, the current state values of the catchment model element

Examples:

TODO

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/doc_helper.py

def inspect(simulation: "Simulation", element="link", id="1", full_names=False):
    """
    Inspect an element of a catchment model

    Inspect the current state values of an element of a catchment model

    Args:
        simulation (Simulation): A swift simulation object
        element (Any): type of top level element, within c('link','node','subarea')
        id (Any): SWIFT simulation
        full_names (Any): if TRUE returns the full names of state variable ids (e.g. link.link_1.OutlfowRate)

    Returns:
        named numeric vector, the current state values of the catchment model element

    Examples:
        TODO
    """
    # Examples:
    #     >>> # ms <- createTestCatchmentStructure()$model
    #     >>> # inspect(ms, element='link', id='lnk1')
    #     >>> # inspect(ms, element='subarea', id='lnk1')
    #     >>> # inspect(ms, element='subarea', id='lnk1', fullNames=TRUE)
    all_f_ids = get_variable_ids(simulation, mk_full_data_id(element, id), full_id=True)
    all_f_ids = np.sort([s for s in set(all_f_ids)])
    x = get_state_value(simulation, all_f_ids)

    def last_in_vec(v):
        if len(v) > 0:
            return v[-1]
        else:
            return ""

    if not full_names:
        new_k = dict([(k, last_in_vec(k.split("."))) for k in x.keys()])
        # if len(new_k) != len(set(new_k)): raise Exception()
        new_x = dict([(new_k[k], x[k]) for k in x.keys()])
        x = new_x
    return x

sample_catchment_model(site_id='South_Esk', config_id='catchment')

Deserialize a basic catchment structure from the package sample data

Deserialize a basic catchment structure from the package sample data. This function is mostly for documentation purposes.

Parameters:

Name	Type	Description	Default
site_id	Any	a site identifier	'South_Esk'
config_id	Any	a variable identifier for a model structure valid for the given site_id	'catchment'

Returns:

Type	Description
	a model simulation

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/doc_helper.py

def sample_catchment_model(site_id="South_Esk", config_id="catchment"):
    """
    Deserialize a basic catchment structure from the package sample data

    Deserialize a basic catchment structure from the package sample data. This function is mostly for documentation purposes.

    Args:
        site_id (Any): a site identifier
        config_id (Any): a variable identifier for a model structure valid for the given site_id

    Returns:
        a model simulation

    """
    # if not exists('swiftSampleData')) data(swift_sample_data)
    s = _sample_data()
    site_data: Dict = s[site_id]
    if config_id not in site_data.keys():
        raise ValueError(
            "Sample catchment model for site "
            + site_id
            + " identified by "
            + config_id
            + " not found"
        )
    cat_structure = site_data[config_id]
    return create_catchment_model_from_structure(cat_structure)

sample_series(site_id='MMH', var_name='rain')

Deserialize information to a UTC time series

Deserialize information to a UTC time series. This function is overcoming some behaviors in saving/loading xts series to/from binary RData format. Usage is not intended for most users.

Parameters:

Name	Type	Description	Default
site_id	Any	a site identifier	'MMH'
var_name	Any	a variable identifier valid for the given site_id	'rain'

Returns:

Type	Description
	an xts time series with UTC time indexing

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/doc_helper.py

def sample_series(site_id="MMH", var_name="rain"):
    """
    Deserialize information to a UTC time series

    Deserialize information to a UTC time series. This function is overcoming some behaviors in saving/loading xts series to/from binary RData format. Usage is not intended for most users.

    Args:
        site_id (Any): a site identifier
        var_name (Any): a variable identifier valid for the given site_id

    Returns:
        an xts time series with UTC time indexing

    """
    s = _sample_data()
    siteData = s[site_id]
    varData = siteData[var_name]
    # if (varData)) stop(paste('Sample data for site', site_id, 'and variable name', var_name, 'not found'))
    return deserialise_sample_series(varData)

sce_parameter(nparams, nshuffle=40)

Create SCE parameters suited for a given number of parameters.

Parameters:

Name	Type	Description	Default
nparams	int	number of free model parameters	required
nshuffle	int	maximum number of shuffles to do, if no other termination criterion. Defaults to 40.	40

Returns:

Type	Description
Dict[str, float]	Dict[str,float]: SCE hyperparameters

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/doc_helper.py

def sce_parameter(nparams: int, nshuffle: int = 40) -> Dict[str, float]:
    """Create SCE parameters suited for a given number of parameters.

    Args:
        nparams (int): number of free model parameters
        nshuffle (int, optional): maximum number of shuffles to do, if no other termination criterion. Defaults to 40.

    Returns:
        Dict[str,float]: SCE hyperparameters
    """
    sce_params = get_default_sce_parameters()
    sce_params["P"] = nparams + 2
    sce_params["Pmin"] = nparams + 2
    sce_params["M"] = 2 * nparams + 1
    sce_params["Q"] = max(sce_params["M"] - 2, 2)
    sce_params["NumShuffle"] = nshuffle
    return sce_params

set_loglik_param_keys(a='a', b='b', m='m', s='s', maxobs='maxobs', ct='ct', censopt='CensOpt')

Specify the global parameter names to use in the log-likelihood calculation

Specify the global parameter names to use in the log-likelihood calculation. Consequence of prehistoric legacy.

Parameters:

Name	Type	Description	Default
a	str	the name of the a parameter	'a'
b	str	the name of the b parameter	'b'
m	str	the name of the m parameter	'm'
s	str	the name of the s parameter	's'
maxobs	str	the name of the maxobs parameter	'maxobs'
ct	str	the name of the ct parameter	'ct'
censopt	str	the name of the censopt parameter	'CensOpt'

Examples:

TODO

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/doc_helper.py

def set_loglik_param_keys(
    a="a", b="b", m="m", s="s", maxobs="maxobs", ct="ct", censopt="CensOpt"
):
    """
    Specify the global parameter names to use in the log-likelihood calculation

    Specify the global parameter names to use in the log-likelihood calculation. Consequence of prehistoric legacy.

    Args:
        a (str): the name of the a parameter
        b (str): the name of the b parameter
        m (str): the name of the m parameter
        s (str): the name of the s parameter
        maxobs (str): the name of the maxobs parameter
        ct (str): the name of the ct parameter
        censopt (str): the name of the censopt parameter

    Examples:
        TODO
    """
    # Examples:
    #     >>> # set_loglik_param_keys('lambda', 'epsilon', 'm', 's')
    #     >>> # pSpecGr4j = get_free_params('GR4J')
    #     >>> # # pSpecGr4j$Value = c(542.1981111, -0.4127542, 7.7403390, 1.2388548)
    #     >>> # # pSpecGr4j$Min = c(1,-30, 1,1)
    #     >>> # # pSpecGr4j$Max = c(3000, 30, 1000, 240)
    #     >>> # # pSpecGr4j$Name = paste0(rootId, pSpecGr4j$Name)
    #     >>> # pgr4 = create_parameterizer(type='Generic', specs=pSpecGr4j)
    #     >>> # p = create_parameterizer(type='log-likelihood')
    #     >>> # p.add("epsilon", -30, 0, -7)
    #     >>> # p.add("m", 0, 0, 0)
    #     >>> # p.add("s", 1, 1000, 100)
    #     >>> # p.add("lambda", -30, 1, -10)
    #     >>> # parameterizer = concatenate_parameterisers(pgr4, p)
    #     >>> # print(parameterizer)
    swg.SetLogLikelihoodVariableNames_py(a, b, m, s, maxobs, ct, censopt)

set_sample_data(model_sim, site_id='MMH', rain_data_var='rain', evap_data_var='evap', rain_model_var='P', evap_model_var='E', t_step='daily')

Sets sample input data into a model simulation

Sets input data from the included sample data into a model simulation

Parameters:

Name	Type	Description	Default
model_sim	Any	an S4 object 'ExternalObjRef' [package "cinterop"] with external pointer type "MODEL_SIMULATION_PTR"	required
site_id	Any	sample data site identifier	'MMH'
rain_data_var	Any	time series ID for the rainfall in the sample data	'rain'
evap_data_var	Any	time series ID for the evaporation in the sample data	'evap'
rain_model_var	Any	sub-area runoff model state identifier for the rainfall, e.g. 'P'	'P'
evap_model_var	Any	sub-area runoff model state identifier for the evaporation, e.g. 'E'	'E'
t_step	Any	identifier for the time step to set the simulation to.	'daily'

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/doc_helper.py

def set_sample_data(
    model_sim,
    site_id="MMH",
    rain_data_var="rain",
    evap_data_var="evap",
    rain_model_var="P",
    evap_model_var="E",
    t_step="daily",
):
    """
    Sets sample input data into a model simulation

    Sets input data from the included sample data into a model simulation

    Args:
        model_sim (Any): an S4 object 'ExternalObjRef' [package "cinterop"] with external pointer type "MODEL_SIMULATION_PTR"
        site_id (Any): sample data site identifier
        rain_data_var (Any): time series ID for the rainfall in the sample data
        evap_data_var (Any): time series ID for the evaporation in the sample data
        rain_model_var (Any): sub-area runoff model state identifier for the rainfall, e.g. 'P'
        evap_model_var (Any): sub-area runoff model state identifier for the evaporation, e.g. 'E'
        t_step (Any): identifier for the time step to set the simulation to.

    """

    rain = sample_series(site_id, var_name=rain_data_var)
    evap = sample_series(site_id, var_name=evap_data_var)
    sa_ids = get_subarea_ids(model_sim)
    for sa_id in sa_ids:
        play_subarea_input(model_sim, rain, sa_id, rain_model_var)
        play_subarea_input(model_sim, evap, sa_id, evap_model_var)
    set_simulation_time_step(model_sim, t_step)
    set_simulation_span(model_sim, start_ts(rain), end_ts(rain))

short_var_id(var_ids)

Shorten long model variable identifiers to the short model variable name

Shorten long model variable identifiers to the short model variable name. This is useful for instance to prepare time series names for multi-plot displays.

Parameters:

Name	Type	Description	Default
var_ids	Any	character vector	required

Returns:

Type	Description
VecStr	the short model variable identifiers

Examples:

>>> short_var_id('elementtype|elementname|varid')
>>> short_var_id('elementtype.elementname.varid')

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/doc_helper.py

def short_var_id(var_ids: "VecStr") -> "VecStr":
    """
    Shorten long model variable identifiers to the short model variable name

    Shorten long model variable identifiers to the short model variable name. This is useful for instance to prepare time series names for multi-plot displays.

    Args:
        var_ids (Any): character vector

    Returns:
        the short model variable identifiers

    Examples:
        >>> short_var_id('elementtype|elementname|varid')
        >>> short_var_id('elementtype.elementname.varid')

    """
    if is_common_iterable(var_ids):
        return [short_var_id(v) for v in var_ids]
    else:
        return var_ids.split("|")[-1].split["."][-1]

module helpers

module internal

simplify_time_series(input_ts)

simplify a 1D time series object to a representation suitable for portable serialisation.

Parameters:

Name	Type	Description	Default
input_ts	TimeSeriesLike	time series	required

Returns:

Type	Description
Dict[str, Any]	Dict[str,Any]: dictionary with keys "tsgeom" for the time series geometry, and "tsvalues" for its values.

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/internal.py

def simplify_time_series(input_ts: TimeSeriesLike) -> Dict[str, Any]:
    """simplify a 1D time series object to a representation suitable for portable serialisation.

    Args:
        input_ts (TimeSeriesLike): time series

    Returns:
        Dict[str,Any]: dictionary with keys "tsgeom" for the time series geometry, and "tsvalues" for its values.

    """
    # from cinterop.timeseries import getTsGeometry
    # stopifnot(is.xts(input_ts))
    # def getSeriesColumn(k):
    #     return(as.numeric(input_ts[,k]))
    return {
        TS_INTEROP_GEOM_KEY: marshal.as_native_tsgeom(get_tsgeom(input_ts)),
        TS_INTEROP_VALUES_KEY: input_ts.values.squeeze(),  # lapply(1:ncol(input_ts), FUN=getSeriesColumn)
    }

to_interop_univariate_series(ts, from_date=None, to_date=None)

Convert an univariate python time series to a representation suitable for interoperability with a C API

Parameters:

Name	Type	Description	Default
ts	TimeSeriesLike	Python native time series	required
from_date	ConvertibleToTimestamp	start timestamp of the time series to subset to. Defaults to None.	None
to_date	ConvertibleToTimestamp	end timestamp of the time series to subset to. Defaults to None.	None

Returns:

Type	Description
Tuple[np.ndarray, TimeSeriesGeometryNative]	Tuple[np.ndarray, TimeSeriesGeometryNative]: univeriate data and time series geometry for interop.

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/internal.py

def to_interop_univariate_series(
    ts: TimeSeriesLike,
    from_date: ConvertibleToTimestamp = None,
    to_date: ConvertibleToTimestamp = None,
) -> Tuple[np.ndarray, TimeSeriesGeometryNative]:
    """Convert an univariate python time series to a representation suitable for interoperability with a C API

    Args:
        ts (TimeSeriesLike): Python native time series
        from_date (ConvertibleToTimestamp, optional): start timestamp of the time series to subset to. Defaults to None.
        to_date (ConvertibleToTimestamp, optional): end timestamp of the time series to subset to. Defaults to None.

    Returns:
        Tuple[np.ndarray, TimeSeriesGeometryNative]: univeriate data and time series geometry for interop.
    """
    from cinterop.timeseries import ts_window
    from swift2.internal import simplify_time_series

    from_date = as_timestamp(from_date) if not from_date is None else None
    to_date = as_timestamp(to_date) if not from_date is None else None
    observation = ts_window(ts, from_date, to_date)
    simple_ts = simplify_time_series(observation)
    return (simple_ts[TS_INTEROP_VALUES_KEY], simple_ts[TS_INTEROP_GEOM_KEY])

module model_definitions

cookie_cut_dendritic_catchment(simulation, bottom_element_id, top_element_ids)

cookie cut a dendritic catchment (without confluences)

Parameters:

Name	Type	Description	Default
simulation	Simulation	base catchment simulation	required
bottom_element_id	str	identifier of the most downstream element to keep	required
top_element_ids	str	identifier(s) of the most upstream element(s) to keep	required

Returns:

Name	Type	Description
Simulation	Simulation	a subcatchment simulation, cookie cut from the base simulation. Deep clone of objects.

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/model_definitions.py

def cookie_cut_dendritic_catchment(
    simulation: "Simulation", bottom_element_id: str, top_element_ids: str
) -> "Simulation":
    """cookie cut a dendritic catchment (without confluences)

    Args:
        simulation (Simulation): base catchment simulation
        bottom_element_id (str): identifier of the most downstream element to keep
        top_element_ids (str): identifier(s) of the most upstream element(s) to keep

    Returns:
        Simulation: a subcatchment simulation, cookie cut from the base simulation. Deep clone of objects.
    """
    # stopifnot(element_id %in% as.character(swift_cal_element_ids))
    select_network_above_element = True
    include_element_in_selection = True
    invert_selection = False
    subset_simul = swg.SubsetModel_py(
        simulation,
        bottom_element_id,
        select_network_above_element,
        include_element_in_selection,
        invert_selection,
        top_element_ids,
        len(top_element_ids),
    )
    return subset_simul

get_catchment_structure(simulation)

Gets the essential connective structure of a catchment

Parameters:

Name	Type	Description	Default
simulation	Simulation	base catchment simulation	required

Returns:

Type	Description
Dict

Examples:

>>> _, simulation = sdh.create_test_catchment_structure()
>>> smd.get_catchment_structure(simulation)
{'Node':    Id     Name
0  n1  n1_name
1  n2  n2_name
2  n3  n3_name
3  n4  n4_name
4  n5  n5_name
5  n6  n6_name, 'Link':      Id       Name  LengthMetres    f  ManningsN  Slope
0  lnk1  lnk1_name           0.0  0.0        0.0    0.0
1  lnk2  lnk2_name           0.0  0.0        0.0    0.0
2  lnk3  lnk3_name           0.0  0.0        0.0    0.0
3  lnk4  lnk4_name           0.0  0.0        0.0    0.0
4  lnk5  lnk5_name           0.0  0.0        0.0    0.0, 'Subarea':      Id       Name  AreaKm2
0  lnk1  lnk1_name      1.1
1  lnk2  lnk2_name      2.2
2  lnk3  lnk3_name      3.3
3  lnk4  lnk4_name      4.4
4  lnk5  lnk5_name      5.5, 'NodeLink':   DownstreamId UpstreamId LinkId
0           n6         n2   lnk1
1           n2         n5   lnk2
2           n2         n4   lnk3
3           n4         n3   lnk4
4           n4         n1   lnk5, 'SubareaLink':   LinkId SubareaId
0   lnk1      lnk1
1   lnk2      lnk2
2   lnk3      lnk3
3   lnk4      lnk4

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/model_definitions.py

def get_catchment_structure(simulation) -> Dict:
    """Gets the essential connective structure of a catchment

    Args:
        simulation (Simulation): base catchment simulation

    Returns:
        [type]: [description]

    Examples:
        >>> _, simulation = sdh.create_test_catchment_structure()
        >>> smd.get_catchment_structure(simulation)
        {'Node':    Id     Name
        0  n1  n1_name
        1  n2  n2_name
        2  n3  n3_name
        3  n4  n4_name
        4  n5  n5_name
        5  n6  n6_name, 'Link':      Id       Name  LengthMetres    f  ManningsN  Slope
        0  lnk1  lnk1_name           0.0  0.0        0.0    0.0
        1  lnk2  lnk2_name           0.0  0.0        0.0    0.0
        2  lnk3  lnk3_name           0.0  0.0        0.0    0.0
        3  lnk4  lnk4_name           0.0  0.0        0.0    0.0
        4  lnk5  lnk5_name           0.0  0.0        0.0    0.0, 'Subarea':      Id       Name  AreaKm2
        0  lnk1  lnk1_name      1.1
        1  lnk2  lnk2_name      2.2
        2  lnk3  lnk3_name      3.3
        3  lnk4  lnk4_name      4.4
        4  lnk5  lnk5_name      5.5, 'NodeLink':   DownstreamId UpstreamId LinkId
        0           n6         n2   lnk1
        1           n2         n5   lnk2
        2           n2         n4   lnk3
        3           n4         n3   lnk4
        4           n4         n1   lnk5, 'SubareaLink':   LinkId SubareaId
        0   lnk1      lnk1
        1   lnk2      lnk2
        2   lnk3      lnk3
        3   lnk4      lnk4
    """
    return swc.get_catchment_structure_pkg(simulation)

split_to_subcatchments(simulation, split_element_ids, include_upstream=None)

Split a catchment in subcatchments, given a list of node/link element identifiers

Parameters:

Name	Type	Description	Default
simulation	Simulation	base catchment simulation	required
split_element_ids	str	element identifiers such as 'node.n1', 'link.linkId_2'	required
include_upstream	bool	indicates whether for each element in split_element_ids it should be including in the upstream portion of the subcatchment. Defaults to None.	None

Returns:

Name	Type	Description
OrderedDict	OrderedDict[str, Simulation]	list of subcatchments resulting from the split

Examples:

>>> _, simulation = sdh.create_test_catchment_structure()
>>> e_ids = ['node.n2', 'node.n4']
>>> sub_sims = smd.split_to_subcatchments(simulation, e_ids)
>>>
>>> for k in sub_sims:
...     print(k)
...     print(sub_sims[k].get_node_ids())
...     print(sub_sims[k].get_node_names())
...
node.n4
['n4', 'n3', 'n1']
['n4_name', 'n3_name', 'n1_name']
node.n2
['n2', 'n5']
['n2_name', 'n5_name']
remainder
['n6']
['n6_name']

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/model_definitions.py

def split_to_subcatchments(
    simulation, split_element_ids, include_upstream=None
) -> OrderedDict[str, "Simulation"]:
    """Split a catchment in subcatchments, given a list of node/link element identifiers

    Args:
        simulation (Simulation): base catchment simulation
        split_element_ids (str): element identifiers such as 'node.n1', 'link.linkId_2'
        include_upstream (bool, optional): indicates whether for each element in split_element_ids it should be including in the upstream portion of the subcatchment. Defaults to None.

    Returns:
        OrderedDict: list of subcatchments resulting from the split

    Examples:
        >>> _, simulation = sdh.create_test_catchment_structure()
        >>> e_ids = ['node.n2', 'node.n4']
        >>> sub_sims = smd.split_to_subcatchments(simulation, e_ids)
        >>>
        >>> for k in sub_sims:
        ...     print(k)
        ...     print(sub_sims[k].get_node_ids())
        ...     print(sub_sims[k].get_node_names())
        ...
        node.n4
        ['n4', 'n3', 'n1']
        ['n4_name', 'n3_name', 'n1_name']
        node.n2
        ['n2', 'n5']
        ['n2_name', 'n5_name']
        remainder
        ['n6']
        ['n6_name']
    """
    from swift2.utils import rep

    if include_upstream is None:
        include_upstream = rep(True, len(split_element_ids))
    assert len(split_element_ids) == len(include_upstream)
    # It is conceivable that we'd cut out on the same link or node (i.e. keep one link+its catchment or just a node
    # , but for the time being not supported - unnecessary complication
    if len(split_element_ids) != len(set(split_element_ids)):
        raise ValueError(
            "split_element_ids has some duplicate elements - they must be unique"
        )
    # TODO check that all elements are valid identifiers

    e_ids_sorted = sort_by_execution_order(simulation, split_element_ids)
    # include_upstreamSorted = sort_by(include_upstream, split_element_ids, e_ids_sorted)

    n = len(e_ids_sorted)
    result = OrderedDict()

    remaining_catchment = simulation
    for i in range(n):
        e_id = e_ids_sorted[i]
        action_up = (
            "keep_above_inclusive" if include_upstream[i] else "keep_above_exclusive"
        )
        action_down = (
            "keep_below_exclusive" if include_upstream[i] else "keep_below_inclusive"
        )
        up_stream = subset_catchment(remaining_catchment, e_id, action=action_up)
        remaining_catchment = subset_catchment(
            remaining_catchment, e_id, action=action_down
        )
        result[e_ids_sorted[i]] = up_stream
    # We have n to n+1 outputs, depending on whether we have a remainder subcatchment at the bottom of the catchment.
    if len(get_node_ids(remaining_catchment) + get_link_ids(remaining_catchment)) > 0:
        if "remainder" in result.keys():
            raise ValueError(
                "There is already a key 'remainder'; cannot add the remaining downstream catchment"
            )
        result["remainder"] = remaining_catchment
    return result

subset_catchment(simulation, element_id, action='keep_above')

Subsets a catchment, keeping only a portion above or below a node, link or subarea.

Parameters:

Name	Type	Description	Default
simulation	Simulation	an S4 object 'ExternalObjRef' [package "cinterop"] with external pointer type "MODEL_SIMULATION_PTR"	required
element_id	str	id of the element to cut at.	required
action	str	how to cut; currently limited to 'keep_above'	'keep_above'

Returns:

Name	Type	Description
Simulation	Simulation	a subcatchment simulation, cookie cut from the base simulation. Deep clone of objects.

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/model_definitions.py

def subset_catchment(
    simulation: "Simulation", element_id: str, action: str = "keep_above"
) -> "Simulation":
    """Subsets a catchment, keeping only a portion above or below a node, link or subarea.

    Args:
        simulation (Simulation): an S4 object 'ExternalObjRef' [package "cinterop"] with external pointer type "MODEL_SIMULATION_PTR"
        element_id (str): id of the element to cut at.
        action (str): how to cut; currently limited to 'keep_above'

    Returns:
        Simulation: a subcatchment simulation, cookie cut from the base simulation. Deep clone of objects.
    """
    # if (action != 'keep_above'):stop('Only subset upstream of a node or link is supported for now')}
    # 'keep above'
    # 'keep above exclusive' # above element but element is not in the result;
    # 'keep below'
    # 'keep below exclusive'

    action = action.lower()
    action = action.replace("_", " ")

    select_network_above_element = "above" in action
    include_element_in_selection = not "exclusive" in action
    invert_selection = False  # TOCHECK
    return swc.subset_model_pkg(
        simulation,
        element_id,
        select_network_above_element,
        include_element_in_selection,
        invert_selection,
    )

module parameteriser

MhData

Data log from metaheuristic calibration processes

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

class MhData:
    """Data log from metaheuristic calibration processes"""

    def __init__(
        self,
        data: pd.DataFrame,
        fitness: str = "NSE",
        messages: str = "Message",
        categories: str = "Category",
    ) -> None:
        self._fitness = fitness
        self._messages: str = messages
        self._categories: str = categories
        self._data: pd.DataFrame = data

    @property
    def data(self):
        """The inner data of this data log"""
        return self._data

    def rename_columns(self, colnames_map: Dict[str, str]) -> None:
        """Rename the columns of the data log according to a mapping. 

        This is handy for instance to change fully qualified parameter names 
        such as 'subarea.Wolf_Creek.x1' to just 'x1' to produce more legible plots.

        Args:
            colnames_map (Dict[str, str]): mapping
        """        
        d = self._data.rename(colnames_map, axis=1, inplace=False)
        self._data = d

    def subset_by_pattern(self, colname: str, pattern: str) -> "MhData":
        """Subset the log by filtering an attribute by a regexp pattern

        Args:
            colname (str): column name to filter on
            pattern (str): regexp pattern, filter definition

        Returns:
            Any: New MhData object with subset data
        """
        criterion: pd.DataFrame = self._data[[colname]]
        indices = criterion.squeeze().str.match(pattern)
        data = self._data.loc[indices.values]
        return MhData(data, self._fitness, self._messages, self._categories)

    def bound_fitness(self, obj_lims: Sequence[float] = None) -> pd.DataFrame:
        """Return a copy of the log data with the fitness measure bound by min/max limits

        Args:
            obj_lims (Sequence[float], optional): min/max limits, length 2. Defaults to None.

        Returns:
            pd.DataFrame: log data with bound fitness
        """
        if obj_lims is None:
            return self._data
        d = self._data.copy()
        d = bound_values_df(d, self._fitness, obj_lims)
        return d

    #' Subset an mhData object based on string pattern matching
    #'
    #' Subset an mhData object based on string pattern matching on the column for 'messages'
    #'
    #' @param log_mh an mhData object
    #' @param pattern a pattern suitable for use by \code{\link{str_detect}}, for instance 'Initial.*|Reflec.*|Contrac.*|Add.*'
    #' @return an mhData object
    #' @export
    def subset_by_message(
        self, pattern: str = "Initial.*|Reflec.*|Contrac.*|Add.*"
    ) -> "MhData":
        """Subset the log by filtering the 'Message' column by a regexp pattern

        Args:
            pattern (str): regexp pattern, filter definition

        Returns:
            Any: New MhData object with subset data
        """
        return self.subset_by_pattern(self._messages, pattern)

data property

The inner data of this data log

bound_fitness(obj_lims=None)

Return a copy of the log data with the fitness measure bound by min/max limits

Parameters:

Name	Type	Description	Default
obj_lims	Sequence[float]	min/max limits, length 2. Defaults to None.	None

Returns:

Type	Description
pd.DataFrame	pd.DataFrame: log data with bound fitness

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def bound_fitness(self, obj_lims: Sequence[float] = None) -> pd.DataFrame:
    """Return a copy of the log data with the fitness measure bound by min/max limits

    Args:
        obj_lims (Sequence[float], optional): min/max limits, length 2. Defaults to None.

    Returns:
        pd.DataFrame: log data with bound fitness
    """
    if obj_lims is None:
        return self._data
    d = self._data.copy()
    d = bound_values_df(d, self._fitness, obj_lims)
    return d

rename_columns(colnames_map)

Rename the columns of the data log according to a mapping.

This is handy for instance to change fully qualified parameter names such as 'subarea.Wolf_Creek.x1' to just 'x1' to produce more legible plots.

Parameters:

Name	Type	Description	Default
colnames_map	Dict[str, str]	mapping	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def rename_columns(self, colnames_map: Dict[str, str]) -> None:
    """Rename the columns of the data log according to a mapping. 

    This is handy for instance to change fully qualified parameter names 
    such as 'subarea.Wolf_Creek.x1' to just 'x1' to produce more legible plots.

    Args:
        colnames_map (Dict[str, str]): mapping
    """        
    d = self._data.rename(colnames_map, axis=1, inplace=False)
    self._data = d

subset_by_message(pattern='Initial.*|Reflec.*|Contrac.*|Add.*')

Subset the log by filtering the 'Message' column by a regexp pattern

Parameters:

Name	Type	Description	Default
pattern	str	regexp pattern, filter definition	'Initial.*|Reflec.*|Contrac.*|Add.*'

Returns:

Name	Type	Description
Any	MhData	New MhData object with subset data

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def subset_by_message(
    self, pattern: str = "Initial.*|Reflec.*|Contrac.*|Add.*"
) -> "MhData":
    """Subset the log by filtering the 'Message' column by a regexp pattern

    Args:
        pattern (str): regexp pattern, filter definition

    Returns:
        Any: New MhData object with subset data
    """
    return self.subset_by_pattern(self._messages, pattern)

subset_by_pattern(colname, pattern)

Subset the log by filtering an attribute by a regexp pattern

Parameters:

Name	Type	Description	Default
colname	str	column name to filter on	required
pattern	str	regexp pattern, filter definition	required

Returns:

Name	Type	Description
Any	MhData	New MhData object with subset data

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def subset_by_pattern(self, colname: str, pattern: str) -> "MhData":
    """Subset the log by filtering an attribute by a regexp pattern

    Args:
        colname (str): column name to filter on
        pattern (str): regexp pattern, filter definition

    Returns:
        Any: New MhData object with subset data
    """
    criterion: pd.DataFrame = self._data[[colname]]
    indices = criterion.squeeze().str.match(pattern)
    data = self._data.loc[indices.values]
    return MhData(data, self._fitness, self._messages, self._categories)

add_to_hypercube(parameteriser, specs)

Add entries to a hypercube

Parameters:

Name	Type	Description	Default
parameteriser	HypercubeParameteriser	A HypercubeParameteriser wrapper, or a type inheriting from it	required
specs	pd.DataFrame	An optional data frame description of the parameter set, with at least columns Name, Min, Max, Value.	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def add_to_hypercube(parameteriser, specs):
    """Add entries to a hypercube

    Args:
        parameteriser (HypercubeParameteriser): A HypercubeParameteriser wrapper, or a type inheriting from it
        specs (pd.DataFrame): An optional data frame description of the parameter set, with at least columns Name, Min, Max, Value.
    """
    swc.add_parameters_pkg(parameteriser, specs)

add_transform(parameteriser, param_name, inner_param_name, transform_id, a=1.0, b=0.0)

Create a parameteriser for which parameter transformations can be defined

This allows to define e.g. a virtual parameter log_X instead of calibrating on the parameter X.

Parameters:

Name	Type	Description	Default
parameteriser	TransformParameteriser	A TransformParameteriser wrapper, or a type inheriting from it	required
param_name	str	the name of the meta-parameter. Note that it can be the same value as inner_param_name, but this is NOT recommended.	required
inner_param_name	str	the name of the parameter being transformed	required
transform_id	str	identifier for a known bijective univariate function	required
a	float	parameter in Y = F(ax+b). Defaults to 1.0.	1.0
b	float	parameter in Y = F(ax+b). Defaults to 0.0.	0.0

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def add_transform(
    parameteriser: "TransformParameteriser",
    param_name: str,
    inner_param_name: str,
    transform_id: str,
    a=1.0,
    b=0.0,
):
    """Create a parameteriser for which parameter transformations can be defined

        This allows to define e.g. a virtual parameter log_X instead of calibrating on the parameter X.

    Args:
        parameteriser (TransformParameteriser): A TransformParameteriser wrapper, or a type inheriting from it
        param_name (str): the name of the meta-parameter. Note that it can be the same value as inner_param_name, but this is NOT recommended.
        inner_param_name (str): the name of the parameter being transformed
        transform_id (str): identifier for a known bijective univariate function
        a (float, optional): parameter in Y = F(ax+b). Defaults to 1.0.
        b (float, optional): parameter in Y = F(ax+b). Defaults to 0.0.
    """
    swg.AddParameterTransform_py(
        parameteriser, param_name, inner_param_name, transform_id, a, b
    )

apply_sys_config(parameteriser, simulation)

Apply a model configuration to a simulation

Parameters:

Name	Type	Description	Default
parameteriser	HypercubeParameteriser	A HypercubeParameteriser wrapper, or a type inheriting from it	required
simulation	Simulation	simulation	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def apply_sys_config(parameteriser, simulation):
    """Apply a model configuration to a simulation

    Args:
        parameteriser (HypercubeParameteriser): A HypercubeParameteriser wrapper, or a type inheriting from it
        simulation (Simulation): simulation
    """
    if is_score(parameteriser):
        parameteriser = swg.GetSystemConfigurationWila_py(parameteriser)
    assert is_hypercube(parameteriser)
    swg.ApplyConfiguration_py(parameteriser, simulation)

as_py_structure(x)

Try to convert an external pointer to a native python representation

Parameters:

Name	Type	Description	Default
x	Any	object, presumably wrapper around an Xptr, to convert to a 'pure' python representation	required

Returns:

Type	Description

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def as_py_structure(x: Any):
    """Try to convert an external pointer to a native python representation

    Args:
        x (Any): object, presumably wrapper around an Xptr, to convert to a 'pure' python representation

    Returns:
        [type]: [description]
    """
    if not is_cffi_native_handle(x):
        return x
    if is_score(x):
        return swc.scores_as_rpy_dict_pkg(x)
    elif is_set_of_scores(x):
        return scores_as_dataframe(x)
    else:
        return x

backtransform(parameteriser)

Get the parameteriser values in the untransformed space

Get the parameteriser values in the untransformed space, i.e. remove any transform added via wrapTransform. This allows to transform back e.g. from a virtual parameter log_X to the underlying model (or even virtual/meta) parameter X.

Parameters:

Name	Type	Description	Default
parameteriser	HypercubeParameteriser	A HypercubeParameteriser wrapper, or a type inheriting from it	required

Returns:

Type	Description
	[HypercubeParameteriser]: The parameters definitions without the transforms (if there are any)

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def backtransform(parameteriser):
    """Get the parameteriser values in the untransformed space

    Get the parameteriser values in the untransformed space, i.e. remove any transform added via wrapTransform.
    This allows to transform back e.g. from a virtual parameter log_X to the underlying model (or even virtual/meta) parameter X.

    Args:
        parameteriser (HypercubeParameteriser): A HypercubeParameteriser wrapper, or a type inheriting from it

    Returns:
        [HypercubeParameteriser]: The parameters definitions without the transforms (if there are any)
    """
    return swg.UntransformHypercubeParameterizer_py(parameteriser)

bound_values_df(x, colname, lim=None)

min/max bound a column in a data frame

Parameters:

Name	Type	Description	Default
x	[type]	a data frame	required
colname	[type]	a character vector, name of the column to bound	required
lim	[type]	a num vector of the min/max limits to apply, for instance c(0, 1). Defaults to None.	None

Returns:

Type	Description

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def bound_values_df(x, colname, lim=None):
    """min/max bound a column in a data frame

    Args:
        x ([type]):   a data frame
        colname ([type]): a character vector, name of the column to bound
        lim ([type], optional): a num vector of the min/max limits to apply, for instance c(0, 1). Defaults to None.

    Returns:
        [type]: [description]
    """
    if lim is None:
        return x
    return x.assign(**{colname: bound_values(x[[colname]], lim)})

concatenate_parameterisers(*args, strategy='')

Concatenate hypercubes to a single parameteriser

Parameters:

Name	Type	Description	Default
strategy	str	The strategy to contatenate. Defaults to "", equivalent to "composite", the only available. May have other options in the future.	''

Returns:

Name	Type	Description
CompositeParameteriser	CompositeParameteriser	A concatenated parameteriser

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def concatenate_parameterisers(
    *args: Sequence['HypercubeParameteriser'], strategy: str = ""
) -> "CompositeParameteriser":
    """Concatenate hypercubes to a single parameteriser

    Args:
        strategy (str, optional): The strategy to contatenate. Defaults to "", equivalent to "composite", the only available. May have other options in the future.

    Returns:
        CompositeParameteriser: A concatenated parameteriser
    """
    parameterisers = [x for x in args]
    return swc.aggregate_parameterisers_pkg(strategy, parameterisers)

create_multisite_obj_parameteriser(func_parameterisers, func_identifiers, prefixes=None, mix_func_parameteriser=None, hydro_parameteriser=None)

Builds a parameteriser usable with a multisite multiobjective calculator.

Parameters:

Name	Type	Description	Default
func_parameterisers	[type]	list of external pointers, parameterisers for each function of a multiobjective calculation.	required
func_identifiers	[type]	character, identifiers for each of the objectives defined in an multisite objective definition.	required
prefixes	[type]	Optional prefixes to use to disambiguate short parameter names used in each function of a multiobjective calculator.. Defaults to None.	None
mix_func_parameteriser	[type]	parameteriser, default None. (FUTURE) Optional parameteriser used in mixing the multiple objectives.. Defaults to None.	None
hydro_parameteriser	[type]	parameteriser, default None. Optional parameteriser applied to the simulation model.. Defaults to None.	None

Returns:

Type	Description

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def create_multisite_obj_parameteriser(
    func_parameterisers,
    func_identifiers,
    prefixes=None,
    mix_func_parameteriser=None,
    hydro_parameteriser=None,
):
    """Builds a parameteriser usable with a multisite multiobjective calculator.

    Args:
        func_parameterisers ([type]): list of external pointers, parameterisers for each function of a multiobjective calculation.
        func_identifiers ([type]): character, identifiers for each of the objectives defined in an multisite objective definition.
        prefixes ([type], optional): Optional prefixes to use to disambiguate short parameter names used in each function of a multiobjective calculator.. Defaults to None.
        mix_func_parameteriser ([type], optional): parameteriser, default None. (FUTURE) Optional parameteriser used in mixing the multiple objectives.. Defaults to None.
        hydro_parameteriser ([type], optional): parameteriser, default None. Optional parameteriser applied to the simulation model.. Defaults to None.

    Returns:
        [FunctionsParameteriser]: [description]
    """
    # stopifnot(is.list(func_parameterisers))
    # stopifnot(len(func_parameterisers) == len(func_identifiers))
    if not prefixes is None:
        assert len(func_parameterisers) == len(prefixes)
    cp = swg.CreateCompositeParameterizer_py()
    n = len(func_parameterisers)
    for i in range(n):
        swg.TagParameterizer_py(func_parameterisers[i], func_identifiers[i])
        if not prefixes is None:
            pp = swg.CreatePrefixingParameterizer_py(
                func_parameterisers[i], prefixes[i]
            )
            swg.AddToCompositeParameterizer_py(cp, pp)
        # else:
        #     TODO: what? forgot and the R implementation had a minor bug I think
        #     swg.AddToCompositeParameterizer_py(cp, pp)
    if not mix_func_parameteriser is None:
        swg.TagParameterizer_py(mix_func_parameteriser, "mixing_function")
        pmix_func_parameteriser = swg.CreatePrefixingParameterizer_py(
            mix_func_parameteriser, "mixing_function."
        )
        swg.AddToCompositeParameterizer_py(cp, pmix_func_parameteriser)
    if hydro_parameteriser is None:  # create a dummy
        hydro_parameteriser = swg.CreateHypercubeParameterizer_py("no apply")
    fp = swg.CreateFunctionsParameterizer_py(hydro_parameteriser, cp)
    return fp

create_muskingum_param_constraints(inner_parameters, delta_t=1, param_name_k='K', param_name_x='X', simulation=None)

Create a parameteriser with Muskingum-type constraints. Given an existing parameteriser, create a wrapper that adds constraints on two of its parameters.

Parameters:

Name	Type	Description	Default
inner_parameters	[HypercubeParameteriser]	A SWIFT parameteriser object.	required
delta_t	int	the simulation time step in HOURS. Defaults to 1.	1
param_name_k	str	the variable identifier to use for the delay parameter of the Muskingum routing. Defaults to "K".	'K'
param_name_x	str	the variable identifier to use for the attenuation parameter of the Muskingum routing. Defaults to "X".	'X'
simulation	[Simulation]	the model simulation from which link properties are inspected to define constraints. The links' parameters must already be set.. Defaults to None.	None

Raises:

Type	Description
ValueError	[description]

Returns:

Type	Description

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def create_muskingum_param_constraints(
    inner_parameters, delta_t=1, param_name_k="K", param_name_x="X", simulation=None
):
    """Create a parameteriser with Muskingum-type constraints. Given an existing parameteriser, create a wrapper that adds constraints on two of its parameters.

    Args:
        inner_parameters ([HypercubeParameteriser]): A SWIFT parameteriser object.
        delta_t (int, optional): the simulation time step in HOURS. Defaults to 1.
        param_name_k (str, optional): the variable identifier to use for the delay parameter of the Muskingum routing. Defaults to "K".
        param_name_x (str, optional): the variable identifier to use for the attenuation parameter of the Muskingum routing. Defaults to "X".
        simulation ([Simulation], optional): the model simulation from which link properties are inspected to define constraints. The links' parameters must already be set.. Defaults to None.

    Raises:
        ValueError: [description]

    Returns:
        [ConstraintParameteriser]: [description]
    """
    if simulation is None:
        raise ValueError("simulation argument must not be None")
    p = swg.CreateMuskingumConstraint_py(
        inner_parameters, delta_t, param_name_k, param_name_x, simulation
    )
    return p

create_parameter_sampler(seed, parameteriser, type)

Parameters:

Name	Type	Description	Default
seed	[type]	seed integer, the seed to use for the sampler	required
parameteriser	HypercubeParameteriser	A HypercubeParameteriser wrapper, or a type inheriting from it	required
type	str	identifying a method such as 'urs' for uniform random sampling.	required

Returns:

Type	Description

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def create_parameter_sampler(seed, parameteriser, type: str):
    """

    Args:
        seed ([type]): seed integer, the seed to use for the sampler
        parameteriser (HypercubeParameteriser): A HypercubeParameteriser wrapper, or a type inheriting from it
        type (str): identifying a method such as 'urs' for uniform random sampling.

    Returns:
        [type]: [description]
    """
    return swg.CreateCandidateFactorySeedWila_py(parameteriser, type, seed)

create_parameteriser(type='Generic subareas', specs=None)

Create a SWIFT parameteriser

Parameters:

Name	Type	Description	Default
type	str	A string identifying the (likely SWIFT-specific) type of parameteriser to use. Recognised types are (case insensitive) 'log-likelihood', 'generic', 'subareas', 'links', 'nodes' and 'muskingum'. Defaults to "Generic subareas".	'Generic subareas'
specs	pd.DataFrame	An optional data frame description of the parameter set, with at least columns Name, Min, Max, Value. Defaults to None.	None

Returns:

Type	Description
	[HypercubeParameteriser]: new parameteriser

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def create_parameteriser(type="Generic subareas", specs: pd.DataFrame = None):
    """Create a SWIFT parameteriser

    Args:
        type (str, optional): A string identifying the (likely SWIFT-specific) type of parameteriser to use. Recognised types are (case insensitive) 'log-likelihood', 'generic', 'subareas', 'links', 'nodes' and 'muskingum'. Defaults to "Generic subareas".
        specs (pd.DataFrame, optional): An optional data frame description of the parameter set, with at least columns Name, Min, Max, Value. Defaults to None.

    Returns:
        [HypercubeParameteriser]: new parameteriser
    """
    p = swg.CreateHypercubeParameterizer_py(strategy=type)
    # TODO: consider how to reuse mh::setHyperCube without introducing an undesirable package dependency
    # Maybe pass a function to a function in the mh package
    if specs is not None:
        add_to_hypercube(p, specs)
    return p

create_sce_optim_swift(objective, termination_criterion, sce_params, population_initialiser)

Build an SCE optimiser for a SWIFT model

Parameters:

Name	Type	Description	Default
objective	ObjectiveEvaluator	an objective calculator	required
termination_criterion	SceTerminationCondition	An object that can be passed to SCE for testing the completion of the algorithm.	required
sce_params	dict	optional; parameters controlling the behavior of the SCE optimisers.	required
population_initialiser	CandidateFactorySeed	an S4 object 'ExternalObjRef' [package "cinterop"] with external pointer type HYPERCUBE_PTR or coercible to it, or a type of object that can seed a sampler i.e. coercible to a type CANDIDATE_FACTORY_SEED_WILA_PTR. If the argument is a hypercube, a uniform random sampler is created.	required

Returns:

Type	Description

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def create_sce_optim_swift(
    objective, termination_criterion, sce_params, population_initialiser
):
    """Build an SCE optimiser for a SWIFT model

    Args:
        objective ('ObjectiveEvaluator'):  an objective calculator
        termination_criterion ('SceTerminationCondition'):  An object that can be passed to SCE for testing the completion of the algorithm.
        sce_params (dict):  optional; parameters controlling the behavior of the SCE optimisers.
        population_initialiser ('CandidateFactorySeed'):  an S4 object 'ExternalObjRef' [package "cinterop"] with external pointer type HYPERCUBE_PTR or coercible to it, or a type of object that can seed a sampler i.e. coercible to a type CANDIDATE_FACTORY_SEED_WILA_PTR. If the argument is a hypercube, a uniform random sampler is created.

    Returns:
        [Optimiser]: [description]
    """

    if is_sampler_seeding(population_initialiser):
        # nothing to do.
        pass
    elif is_hypercube(population_initialiser):
        population_initialiser = create_parameter_sampler(
            0, population_initialiser, "urs"
        )
    else:
        raise ValueError(
            "population_initialiser must be provided (can be a sampler or a hypercube)"
        )
    # if(missing(terminationCriterion)) terminationCriterion = maxWallTimeTermination()
    # if(missing(SCEpars)) SCEpars = getDefaultSceParameters()
    if termination_criterion is None:
        max_hours = str(10 / 3600)
        termination_criterion = create_sce_termination_wila(
            "relative standard deviation", ["0.002", max_hours]
        )
    if sce_params is None:
        sce_params = get_default_sce_parameters()
    return swg.CreateShuffledComplexEvolutionWila_py(
        objective, termination_criterion, sce_params, population_initialiser
    )

create_sce_termination_wila(type, arguments)

Create a type of termination criteria suitable for the SCE algorithm.

Parameters:

Name	Type	Description	Default
type	str	A type of termination criterion; currently at least "relative standard deviation" and "maximum evaluations" are valid options	required
arguments	Sequence[str]	Arguments, in string forms even for numeric values, options for the selected type.	required

Returns:

Name	Type	Description
SceTerminationCondition	SceTerminationCondition	[description]

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def create_sce_termination_wila(
    type: str, arguments: Sequence[str]
) -> "SceTerminationCondition":
    """Create a type of termination criteria suitable for the SCE algorithm.

    Args:
        type (str): A type of termination criterion; currently at least "relative standard deviation" and "maximum evaluations" are valid options
        arguments (Sequence[str]): Arguments, in string forms even for numeric values, options for the selected type.

    Returns:
        SceTerminationCondition: [description]
    """
    return swg.CreateSceTerminationWila_py(type, arguments, len(arguments))

evaluate_score_for_parameters(objective, parameteriser)

Computes the value of an objective for a given set of parameters

Parameters:

Name	Type	Description	Default
objective	[type]	an objective calculator	required
parameteriser	HypercubeParameteriser	A HypercubeParameteriser wrapper, or a type inheriting from it	required

Returns:

Type	Description

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def evaluate_score_for_parameters(objective, parameteriser):
    """Computes the value of an objective for a given set of parameters

    Args:
        objective ([type]): an objective calculator
        parameteriser (HypercubeParameteriser): A HypercubeParameteriser wrapper, or a type inheriting from it

    Returns:
        [type]: [description]
    """
    return swc.evaluate_score_wila_pkg(objective, parameteriser)

example_parameteriser(type, strict=False)

Get examples of typical parameterisers

Parameters:

Name	Type	Description	Default
type	str	identifier for a type of parameteriser including 'log-likelihood'	required
strict	bool	If True an error is raised if the type is not found, otherwise a dummy empty parameteriser is returned.. Defaults to False.	False

Returns:

Type	Description

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def example_parameteriser(type: str, strict=False):
    """Get examples of typical parameterisers

    Args:
        type (str): identifier for a type of parameteriser including 'log-likelihood'
        strict (bool, optional): If True an error is raised if the type is not found, otherwise a dummy empty parameteriser is returned.. Defaults to False.

    Returns:
        [HypercubeParameteriser]: [description]
    """
    type = type.lower()
    if type == "log-likelihood":
        p = create_parameteriser(type="no apply")
        calc_m_and_s = 1.0  # meaning true
        censopt = 0.0
        add_to_hypercube(
            p,
            _df_from_dict(
                Name=["b", "m", "s", "a", "maxobs", "ct", "censopt", "calc_mod_m_s"],
                Min=_npf([-30, 0, 1, -30, 100.0, 0.01, censopt, calc_m_and_s]),
                Max=_npf([0, 0, 1000, 1, 100.0, 0.01, censopt, calc_m_and_s]),
                Value=_npf([-7, 0, 100, -10, 100.0, 0.01, censopt, calc_m_and_s]),
            ),
        )
        return p
    if not strict:
        return create_parameteriser(type="Generic")
    else:
        raise Exception("No example parameteriser yet for type " + type)

execute_optimisation(optimiser)

Launch an optimization task, as defined by the object passed as an argument

Parameters:

Name	Type	Description	Default
optimiser	Optimiser	the instance of the optimiser that has been created for the optimisation task about to be launched.	required

Returns:

Type	Description

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def execute_optimisation(optimiser):
    """Launch an optimization task, as defined by the object passed as an argument

    Args:
        optimiser (Optimiser): the instance of the optimiser that has been created for the optimisation task about to be launched.

    Returns:
        [VectorObjectiveScores]: [description]
    """
    return swg.ExecuteOptimizerWila_py(optimiser)

feasible_muskingum_bounds(simulation, delta_t_hours=1)

[summary]

Parameters:

Name	Type	Description	Default
simulation	Simulation	[description]	required
delta_t_hours	int	[description]. Defaults to 1.	1

Returns:

Type	Description

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def feasible_muskingum_bounds(simulation: "Simulation", delta_t_hours=1):
    """[summary]

    Args:
        simulation (Simulation): [description]
        delta_t_hours (int, optional): [description]. Defaults to 1.

    Returns:
        [type]: [description]
    """
    return swg.GetFeasibleMuskingumBounds_py(simulation, delta_t_hours)

filtered_parameters(parameteriser)

Wrap a parameteriser in a filter that can hide some parameters

Parameters:

Name	Type	Description	Default
parameteriser	HypercubeParameteriser	A HypercubeParameteriser wrapper, or a type inheriting from it. A deep copy of the input is taken.	required

Returns:

Type	Description

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def filtered_parameters(parameteriser):
    """Wrap a parameteriser in a filter that can hide some parameters

    Args:
        parameteriser (HypercubeParameteriser): A HypercubeParameteriser wrapper, or a type inheriting from it. A deep copy of the input is taken.

    Returns:
        [FilteringParameteriser]: [description]
    """
    return swg.CreateFilteringParameterizer_py(parameteriser)

get_best_score(scores_population, score_name='NSE', convert_to_py=False)

Gets the best score in a population for a given objective

Parameters:

Name	Type	Description	Default
scores_population	[type]	an S4 object 'ExternalObjRef' [package "cinterop"] with external pointer type VEC_OBJECTIVE_SCORES_PTR	required
score_name	str	name of the objective to use for sorting. Defaults to "NSE".	'NSE'
convert_to_py	bool	should the returned score be converted to an R representation. Default False. Defaults to False.	False

Returns:

Type	Description

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def get_best_score(scores_population, score_name="NSE", convert_to_py=False):
    """Gets the best score in a population for a given objective

    Args:
        scores_population ([type]): an S4 object 'ExternalObjRef' [package "cinterop"] with external pointer type VEC_OBJECTIVE_SCORES_PTR
        score_name (str, optional): name of the objective to use for sorting. Defaults to "NSE".
        convert_to_py (bool, optional): should the returned score be converted to an R representation. Default False. Defaults to False.

    Returns:
        [ObjectiveScores or Dict]: [description]
    """
    sorted_results = sort_by_score(scores_population, score_name)
    s = get_score_at_index(sorted_results, 1)
    if convert_to_py:
        return swc.scores_as_rpy_dict_pkg(s)
    else:
        return s

get_default_sce_parameters()

[summary]

Returns:

Type	Description

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def get_default_sce_parameters():
    """[summary]

    Returns:
        [type]: [description]
    """
    from swift2.wrap.swift_wrap_custom import default_sce_parameters_pkg

    return default_sce_parameters_pkg()

get_logger_content(optimiser, add_numbering=False)

Gets logger content on an optimiser, recorded detail of the optimisation process for post-optimisation analysis.

Parameters:

Name	Type	Description	Default
optimiser	[type]	the instance of the optimiser that has been created for the optimisation task about to be launched.	required
add_numbering	bool	Add an explicit column for numbering the lines of the log. Defaults to False.	False

Returns:

Type	Description
pd.DataFrame	pd.DataFrame: The data log of the optimiser

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def get_logger_content(optimiser:DeletableCffiNativeHandle, add_numbering:bool=False) -> pd.DataFrame:
    """Gets logger content on an optimiser, recorded detail of the optimisation process for post-optimisation analysis.

    Args:
        optimiser ([type]): the instance of the optimiser that has been created for the optimisation task about to be launched.
        add_numbering (bool, optional): Add an explicit column for numbering the lines of the log. Defaults to False.

    Returns:
        pd.DataFrame: The data log of the optimiser
    """
    # coercion to data.frame is a workaround for https://jira.csiro.au/browse/WIRADA-245
    if is_cffi_native_handle(optimiser, type_id="ERRIS_STAGED_CALIBRATION_PTR"):
        log_data = swg.GetERRISCalibrationLog_py(optimiser)
    else:
        log_data = swg.GetOptimizerLogDataWila_py(optimiser)
    return convert_optimisation_logger(log_data, add_numbering)

get_marginal_termination(tolerance=1e-06, cutoff_no_improvement=10, max_hours=0.05)

Create an termination criterion based on the rate of marginal fitness improvement

Parameters:

Name	Type	Description	Default
tolerance	[type]	the increment in the objective below which the improvement is considered negligible. Defaults to 1e-06.	1e-06
cutoff_no_improvement	int	the maximum number of successive times the algorithm fails to improve the objective function.. Defaults to 10.	10
max_hours	float	the maximum wall time runtime for the optimisation. Defaults to 0.05.	0.05

Returns:

Type	Description

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def get_marginal_termination(tolerance=1e-06, cutoff_no_improvement=10, max_hours=0.05):
    """Create an termination criterion based on the rate of marginal fitness improvement

    Args:
        tolerance ([type], optional): the increment in the objective below which the improvement is considered negligible. Defaults to 1e-06.
        cutoff_no_improvement (int, optional): the maximum number of successive times the algorithm fails to improve the objective function.. Defaults to 10.
        max_hours (float, optional): the maximum wall time runtime for the optimisation. Defaults to 0.05.

    Returns:
        [SceTerminationCondition]: [description]
    """
    return swg.CreateSceMarginalTerminationWila_py(
        tolerance=tolerance,
        cutoffNoImprovement=cutoff_no_improvement,
        maxHours=max_hours,
    )

get_max_iteration_termination(max_iterations=1000)

Create an termination criterion based on the number of objective evaluations

Parameters:

Name	Type	Description	Default
max_iterations	int	number of iterations, which, if less than total count of optim objective evaluations, defines optim termination.. Defaults to 1000.	1000

Returns:

Type	Description

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def get_max_iteration_termination(max_iterations=1000):
    """Create an termination criterion based on the number of objective evaluations

    Args:
        max_iterations (int, optional): number of iterations, which, if less than total count of optim objective evaluations, defines optim termination.. Defaults to 1000.

    Returns:
        [SceTerminationCondition]: [description]
    """
    return swg.CreateSceMaxIterationTerminationWila_py(maxIterations=max_iterations)

get_max_runtime_termination(max_hours=0.05)

Create an termination criterion based on the wall clock runtime

Parameters:

Name	Type	Description	Default
max_hours	float	the maximum wall time runtime in hours for the optimisation. Defaults to 0.05.	0.05

Returns:

Type	Description

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def get_max_runtime_termination(max_hours=0.05):
    """Create an termination criterion based on the wall clock runtime

    Args:
        max_hours (float, optional): the maximum wall time runtime in hours for the optimisation. Defaults to 0.05.

    Returns:
        [SceTerminationCondition]: [description]
    """
    return swg.CreateSceMaxRuntimeTerminationWila_py(maxHours=max_hours)

get_score_at_index(scores_population, index)

Get an objective scores in a vector thereof

Parameters:

Name	Type	Description	Default
scores_population	[type]	an S4 object 'ExternalObjRef' [package "cinterop"] with external pointer type VEC_OBJECTIVE_SCORES_PTR	required
index	int	one-based index in the population	required

Returns:

Type	Description

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def get_score_at_index(scores_population, index: int):
    """Get an objective scores in a vector thereof

    Args:
        scores_population ([type]): an S4 object 'ExternalObjRef' [package "cinterop"] with external pointer type VEC_OBJECTIVE_SCORES_PTR
        index (int): one-based index in the population

    Returns:
        [ObjectiveScores]: [description]
    """
    return swg.GetScoresAtIndex_py(scores_population, index - 1)

hide_parameters(parameteriser, patterns, regex=False, starts_with=False, strict=False)

Hide some parameters (from the outside e.g. optimisers) in a filter parameteriser

Parameters:

Name	Type	Description	Default
parameteriser	HypercubeParameteriser	A HypercubeParameteriser wrapper, or a type inheriting from it	required
patterns	[type]	character, one or more pattern to match and hide matching parameters. Match according to other parameters.	required
regex	bool	logical, defaults False, should the patterns be used as regular expressions.. Defaults to False.	False
starts_with	bool	logical, defaults False. Ignored if regex is True. Should the patterns be used as starting strings in the parameter names.. Defaults to False.	False
strict	bool	logical, default False. Used only if regex and starts_with are False. If True, raises an error if one of the "patterns" has no exact match in the parameters.. Defaults to False.	False

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def hide_parameters(
    parameteriser, patterns, regex=False, starts_with=False, strict=False
):
    """Hide some parameters (from the outside e.g. optimisers) in a filter parameteriser

    Args:
        parameteriser (HypercubeParameteriser): A HypercubeParameteriser wrapper, or a type inheriting from it
        patterns ([type]):  character, one or more pattern to match and hide matching parameters. Match according to other parameters.
        regex (bool, optional): logical, defaults False, should the patterns be used as regular expressions.. Defaults to False.
        starts_with (bool, optional): logical, defaults False. Ignored if regex is True. Should the patterns be used as starting strings in the parameter names.. Defaults to False.
        strict (bool, optional): logical, default False. Used only if regex and starts_with are False. If True, raises an error if one of the "patterns" has no exact match in the parameters.. Defaults to False.
    """
    swg.HideParameters_py(parameteriser, patterns, regex, starts_with, strict)

is_hypercube(p_set)

Is the object a native parameteriser that can be cast as a hypercube?

Parameters:

Name	Type	Description	Default
p_set	CffiNativeHandle	[description]	required

Returns:

Type	Description

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def is_hypercube(p_set: CffiNativeHandle):
    """Is the object a native parameteriser that can be cast as a hypercube?

    Args:
        p_set (CffiNativeHandle): [description]

    Returns:
        [type]: [description]
    """
    # TODO: implement a SWIFT API function to check this.
    # KLUDGE:
    from refcount.interop import is_cffi_native_handle

    return is_cffi_native_handle(p_set) and p_set.type_id in [
        "HYPERCUBE_PTR",
        "COMPOSITE_PARAMETERIZER_PTR",
        "FUNCTIONS_PARAMETERIZER_PTR",
        "CONSTRAINT_PARAMETERIZER_PTR",
        "SCALING_PARAMETERIZER_PTR",
        "STATE_INIT_PARAMETERIZER_PTR",
        "TRANSFORM_PARAMETERIZER_PTR",
        "STATE_INITIALIZER_PTR",
        "SUBAREAS_SCALING_PARAMETERIZER_PTR",
        "HYPERCUBE_WILA_PTR",
        "XPtr<OpaquePointer>",  # TODO this is to circumvent issues now that some functions are generated from Rcpp code rather than API. See e.g. aggregate_parameterisers_pkg
    ]

is_sampler_seeding(obj)

Is the argument a native object that is a seeded candidate parameteriser factory

Parameters:

Name	Type	Description	Default
obj	CffiNativeHandle	[description]	required

Returns:

Type	Description

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def is_sampler_seeding(obj: CffiNativeHandle):
    """Is the argument a native object that is a seeded candidate parameteriser factory

    Args:
        obj (CffiNativeHandle): [description]

    Returns:
        [type]: [description]
    """
    # KLUDGE:
    from refcount.interop import is_cffi_native_handle

    return is_cffi_native_handle(obj, "CANDIDATE_FACTORY_SEED_WILA_PTR")

is_score(x)

OBJECTIVE_SCORES_WILA_PTR

Parameters:

Name	Type	Description	Default
x	[type]	[description]	required

Returns:

Type	Description

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def is_score(x):
    """OBJECTIVE_SCORES_WILA_PTR

    Args:
        x ([type]): [description]

    Returns:
        [type]: [description]
    """
    # TODO: implement a SWIFT API function to check this.
    # KLUDGE?:
    return is_cffi_native_handle(x, type_id="OBJECTIVE_SCORES_WILA_PTR")

is_set_of_scores(x)

VEC_OBJECTIVE_SCORES_PTR

Parameters:

Name	Type	Description	Default
x	[type]	[description]	required

Returns:

Type	Description

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def is_set_of_scores(x):
    """VEC_OBJECTIVE_SCORES_PTR

    Args:
        x ([type]): [description]

    Returns:
        [type]: [description]
    """
    return is_cffi_native_handle(x, type_id="VEC_OBJECTIVE_SCORES_PTR")

linear_parameteriser(param_name, state_name, scaling_var_name, min_p_val, max_p_val, value, selector_type='subareas', intercept=0.0)

Create a scaled linear parameteriser, tying by a linear transformation a (resp several) model state to a (resp several) parameter values

This allows to define tied parameters where pval = a * modelStateVal + intercept. The intent in particular is to define virtual model parameters such as initial store value as a fraction of the maximum storage capacity.

Parameters:

Name	Type	Description	Default
param_name	VecStr	the name of the meta-parameter. Note that it can be the same value as inner_param_name without interference, though this may be confusing a choice.	required
state_name	VecStr	the name of the model state to modify, based on the value of the meta-parameter and the state found in 'scalingVarName'	required
scaling_var_name	VecStr	the name of the parameter for each subarea model, to which to apply the area scaled value.	required
min_p_val	VecNum	minimum value allowed for the meta-parameter	required
max_p_val	VecNum	minimum value allowed for the meta-parameter	required
value	VecNum	value for the meta parameter.	required
selector_type	str	an identifier to define to which catchment element(s) the parameteriser will be applied. Defaults to "subareas".	'subareas'
intercept	VecNum	intercepts in the linear relationship(s). Defaults to 0.0.	0.0

Returns:

Type	Description
	[ScalingParameteriser]: new ScalingParameteriser

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def linear_parameteriser(
    param_name: VecStr,
    state_name: VecStr,
    scaling_var_name: VecStr,
    min_p_val: VecNum,
    max_p_val: VecNum,
    value: VecNum,
    selector_type: str = "subareas",
    intercept: VecNum = 0.0,
):
    """Create a scaled linear parameteriser, tying by a linear transformation a (resp several) model state to a (resp several) parameter values

    This allows to define tied parameters where pval = a * modelStateVal + intercept.
    The intent in particular is to define virtual model parameters such as initial store value as a fraction of the maximum storage capacity.

    Args:

        param_name (VecStr): the name of the meta-parameter. Note that it can be the same value as inner_param_name without interference, though this may be confusing a choice.
        state_name (VecStr): the name of the model state to modify, based on the value of the meta-parameter and the state found in 'scalingVarName'
        scaling_var_name (VecStr): the name of the parameter for each subarea model, to which to apply the area scaled value.
        min_p_val (VecNum): minimum value allowed for the meta-parameter
        max_p_val (VecNum): minimum value allowed for the meta-parameter
        value (VecNum): value for the meta parameter.
        selector_type (str, optional): an identifier to define to which catchment element(s) the parameteriser will be applied. Defaults to "subareas".
        intercept (VecNum, optional): intercepts in the linear relationship(s). Defaults to 0.0.

    Returns:
        [ScalingParameteriser]: new ScalingParameteriser
    """
    # stopifnot(len(selectorType) == 1)
    param_name, state_name, scaling_var_name, min_p_val, max_p_val, value = listify(
        param_name, state_name, scaling_var_name, min_p_val, max_p_val, value
    )
    lengths = [
        len(x)
        for x in [param_name, state_name, scaling_var_name, min_p_val, max_p_val, value]
    ]
    if len(set(lengths)) != 1:
        raise Exception(
            "arguments must all be vectors of same length: param_name, state_name, scalingVarName, minPval, maxPval, value"
        )
    if not is_common_iterable(intercept):
        intercept = np.repeat(intercept, lengths[0])
    elif len(intercept) != lengths[0]:
        raise Exception(
            'argument "intercept" be of length 1 or the same as: param_name, state_name, scalingVarName, minPval, maxPval, value'
        )
    p = swg.CreateTargetScalingParameterizer_py(selector_type)
    for i in range(lengths[0]):
        swg.AddLinearScalingParameterizer_py(
            p,
            param_name[i],
            state_name[i],
            scaling_var_name[i],
            intercept[i],
            min_p_val[i],
            max_p_val[i],
            value[i],
        )
    return p

linear_parameteriser_from(data_frame, selector_type='subareas')

Create a scaled linear parameteriser, tying by a linear transformation a (resp several) model state to a (resp several) parameter values

This allows to define tied parameters where pval = a * modelStateVal + intercept. The intent in particular is to define virtual model parameters such as initial store value as a fraction of the maximum storage capacity.

Parameters:

Name	Type	Description	Default
data_frame	pd.DataFrame	data frame with columns "param_name", "state_name", "scaling_var_name", "min_value", "max_value", "value", "intercept",	required
selector_type	str	[description]. Defaults to "subareas".	'subareas'

Returns:

Type	Description

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def linear_parameteriser_from(
    data_frame: pd.DataFrame, selector_type: str = "subareas"
):
    """Create a scaled linear parameteriser, tying by a linear transformation a (resp several) model state to a (resp several) parameter values

    This allows to define tied parameters where pval = a * modelStateVal + intercept.
    The intent in particular is to define virtual model parameters such as initial store value as a fraction of the maximum storage capacity.
    Args:
        data_frame (pd.DataFrame): data frame with columns "param_name", "state_name", "scaling_var_name", "min_value", "max_value", "value", "intercept",
        selector_type (str, optional): [description]. Defaults to "subareas".

    Returns:
        [ScalingParameteriser]: ScalingParameteriser
    """
    return linear_parameteriser(
        param_name=data_frame[[PARAM_NAME_COL]],
        state_name=data_frame[[STATE_NAME_COL]],
        scaling_var_name=data_frame[[SCALING_VAR_NAME_COL]],
        min_p_val=data_frame[[MIN_VALUE_COL]],
        max_p_val=data_frame[[MAX_VALUE_COL]],
        value=data_frame[[VALUE_COL]],
        selector_type=selector_type,
        intercept=data_frame[[INTERCEPT_COL]],
    )

make_state_init_parameteriser(parameteriser)

[summary]

Parameters:

Name	Type	Description	Default
parameteriser	HypercubeParameteriser	A HypercubeParameteriser wrapper, or a type inheriting from it	required

Returns:

Type	Description
	[StateInitParameteriser]: new state initialisation parameteriser

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def make_state_init_parameteriser(parameteriser):
    """[summary]

    Args:
        parameteriser (HypercubeParameteriser): A HypercubeParameteriser wrapper, or a type inheriting from it

    Returns:
        [StateInitParameteriser]: new state initialisation parameteriser
    """
    return swg.CreateStateInitParameterizer_py(parameteriser)

parameteriser_as_dataframe(parameteriser)

Convert an external object hypercube parameteriser to a pandas data frame

Parameters:

Name	Type	Description	Default
parameteriser	HypercubeParameteriser	A HypercubeParameteriser wrapper, or a type inheriting from it	required

Returns:

Type	Description
	[type]: [a data frame]

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def parameteriser_as_dataframe(parameteriser):
    """Convert an external object hypercube parameteriser to a pandas data frame

    Args:
        parameteriser (HypercubeParameteriser): A HypercubeParameteriser wrapper, or a type inheriting from it

    Returns:
        [type]: [a data frame]
    """
    return swc.parameteriser_to_data_frame_pkg(parameteriser)

parameteriser_for_score(score)

Gets the parameteriser for a score

Parameters:

Name	Type	Description	Default
score	[type]	[description]	required

Returns:

Type	Description

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def parameteriser_for_score(score:'ObjectiveScores'):
    """Gets the parameteriser for a score

    Args:
        score ([type]): [description]

    Returns:
        [HypercubeParameteriser]: [description]
    """
    return swg.GetSystemConfigurationWila_py(score)

scores_as_dataframe(scores_population)

Convert objective scores to a pandas data frame representation

Parameters:

Name	Type	Description	Default
scores_population	[type]	[description]	required

Returns:

Type	Description

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def scores_as_dataframe(scores_population):
    """Convert objective scores to a pandas data frame representation

    Args:
        scores_population ([type]): [description]

    Returns:
        [type]: [description]
    """
    return swc.vec_scores_as_dataframe_pkg(scores_population)

set_calibration_logger(optimiser, type='')

Sets logging on an optimiser, so as to record a detail of the optimisation process for post-optimisation analysis.

Parameters:

Name	Type	Description	Default
optimiser	[type]	[description]	required
type	str	[description]. Defaults to "".	''

Returns:

Type	Description

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def set_calibration_logger(optimiser, type=""):
    """Sets logging on an optimiser, so as to record a detail of the optimisation process for post-optimisation analysis.

    Args:
        optimiser ([type]): [description]
        type (str, optional): [description]. Defaults to "".

    Returns:
        [type]: [description]
    """
    return swg.SetOptimizerLoggerWila_py(optimiser, type)

set_hypercube(parameteriser, specs)

Set the properties of a hypercube parameteriser

Parameters:

Name	Type	Description	Default
parameteriser	HypercubeParameteriser	A HypercubeParameteriser wrapper, or a type inheriting from it	required
specs	pd.DataFrame	An optional data frame description of the parameter set, with at least columns Name, Min, Max, Value.	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def set_hypercube(parameteriser: "HypercubeParameteriser", specs: pd.DataFrame):
    """Set the properties of a hypercube parameteriser

    Args:
        parameteriser (HypercubeParameteriser): A HypercubeParameteriser wrapper, or a type inheriting from it
        specs (pd.DataFrame): An optional data frame description of the parameter set, with at least columns Name, Min, Max, Value.
    """
    swc.set_parameters_pkg(parameteriser, specs)

set_max_parameter_value(parameteriser, variable_name, value)

Sets the maximum value of a model parameter value

Parameters:

Name	Type	Description	Default
parameteriser	HypercubeParameteriser	A HypercubeParameteriser wrapper, or a type inheriting from it	required
variable_name	str or iterable of str	model variable state identifier(s)	required
value	numeric or iterable of numeric	value(s)	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def set_max_parameter_value(parameteriser, variable_name, value):
    """Sets the maximum value of a model parameter value

    Args:
        parameteriser (HypercubeParameteriser): A HypercubeParameteriser wrapper, or a type inheriting from it
        variable_name (str or iterable of str): model variable state identifier(s)
        value (numeric or iterable of numeric): value(s)
    """
    _sapply_parameter_set(
        parameteriser, variable_name, value, swg.SetMaxParameterValue_py
    )

set_min_parameter_value(parameteriser, variable_name, value)

Sets the minimum value of a model parameter value

Parameters:

Name	Type	Description	Default
parameteriser	HypercubeParameteriser	A HypercubeParameteriser wrapper, or a type inheriting from it	required
variable_name	str or iterable of str	model variable state identifier(s)	required
value	numeric or iterable of numeric	value(s)	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def set_min_parameter_value(parameteriser, variable_name, value):
    """Sets the minimum value of a model parameter value

    Args:
        parameteriser (HypercubeParameteriser): A HypercubeParameteriser wrapper, or a type inheriting from it
        variable_name (str or iterable of str): model variable state identifier(s)
        value (numeric or iterable of numeric): value(s)
    """
    _sapply_parameter_set(
        parameteriser, variable_name, value, swg.SetMinParameterValue_py
    )

set_parameter_value(parameteriser, variable_name, value)

Sets the value of a model parameter value

Parameters:

Name	Type	Description	Default
parameteriser	HypercubeParameteriser	A HypercubeParameteriser wrapper, or a type inheriting from it	required
variable_name	str or iterable of str	model variable state identifier(s)	required
value	numeric or iterable of numeric	value(s)	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def set_parameter_value(parameteriser, variable_name, value):
    """Sets the value of a model parameter value

    Args:
        parameteriser (HypercubeParameteriser): A HypercubeParameteriser wrapper, or a type inheriting from it
        variable_name (str or iterable of str): model variable state identifier(s)
        value (numeric or iterable of numeric): value(s)
    """
    _sapply_parameter_set(parameteriser, variable_name, value, swg.SetParameterValue_py)

show_parameters(parameteriser, patterns, regex=False, starts_with=False)

Show some parameters (from the outside e.g. optimisers) in a filter parameteriser

Parameters:

Name	Type	Description	Default
parameteriser	HypercubeParameteriser	A HypercubeParameteriser wrapper, or a type inheriting from it	required
patterns	[type]	character, one or more pattern to match and show matching parameters. Match according to other parameters	required
regex	bool	should the patterns be used as regular expressions. Defaults to False.	False
starts_with	bool	should the patterns be used as starting strings in the parameter names. Defaults to False.	False

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def show_parameters(parameteriser, patterns, regex=False, starts_with=False):
    """Show some parameters (from the outside e.g. optimisers) in a filter parameteriser

    Args:
        parameteriser (HypercubeParameteriser): A HypercubeParameteriser wrapper, or a type inheriting from it
        patterns ([type]):  character, one or more pattern to match and show matching parameters. Match according to other parameters
        regex (bool, optional): should the patterns be used as regular expressions. Defaults to False.
        starts_with (bool, optional): should the patterns be used as starting strings in the parameter names. Defaults to False.
    """
    swg.ShowParameters_py(parameteriser, patterns, regex, starts_with)

sort_by_score(scores_population, score_name='NSE')

Sort objective scores according to one of the objective values

Parameters:

Name	Type	Description	Default
scores_population	[type]	an S4 object 'ExternalObjRef' [package "cinterop"] with external pointer type VEC_OBJECTIVE_SCORES_PTR	required
score_name	str	name of the objective to use for sorting. Defaults to "NSE".	'NSE'

Returns:

Name	Type	Description
VectorObjectiveScores		an S4 object 'ExternalObjRef' [package "cinterop"] with external pointer type VEC_OBJECTIVE_SCORES_PTR

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def sort_by_score(scores_population, score_name="NSE"):
    """Sort objective scores according to one of the objective values

    Args:
        scores_population ([type]): an S4 object 'ExternalObjRef' [package "cinterop"] with external pointer type VEC_OBJECTIVE_SCORES_PTR
        score_name (str, optional): name of the objective to use for sorting. Defaults to "NSE".

    Returns:
        VectorObjectiveScores: an S4 object 'ExternalObjRef' [package "cinterop"] with external pointer type VEC_OBJECTIVE_SCORES_PTR
    """
    return swg.SortSetOfScoresBy_py(scores_population, score_name)

subcatchment_parameteriser(parameteriser, subcatchment)

Create a parameteriser that gets applied to a subset of a whole catchment

Parameters:

Name	Type	Description	Default
parameteriser	HypercubeParameteriser	A HypercubeParameteriser wrapper, or a type inheriting from it	required
subcatchment	Simulation	the subcatchment, subset of a whole catchment, identifying which elements (subareas, nodes, links) will be parameterised by the new parameteriser. Other catchment elements are untouched.	required

Returns:

Type	Description
	[HypercubeParameteriser]: New parameteriser whose application is limited to the subcatchment.

Examples:

>>> sub_cats = simulation.split_to_subcatchments(["node.node_7", "node.node_10"])
>>> sc = sub_cats["node.node_7"]
>>> p = sp.create_parameteriser('generic subarea')
>>> p.add_parameter_to_hypercube("x1", 32, 1, 6e3)
>>> sp = p.subcatchment_parameteriser(sc)
>>> sp.apply_sys_config(simulation)

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def subcatchment_parameteriser(parameteriser, subcatchment):
    """Create a parameteriser that gets applied to a subset of a whole catchment

    Args:
        parameteriser (HypercubeParameteriser): A HypercubeParameteriser wrapper, or a type inheriting from it
        subcatchment (Simulation): the subcatchment, subset of a whole catchment, identifying which elements (subareas, nodes, links) will be parameterised by the new parameteriser. Other catchment elements are untouched.

    Returns:
        [HypercubeParameteriser]: New parameteriser whose application is limited to the subcatchment.

    Examples:
        >>> sub_cats = simulation.split_to_subcatchments(["node.node_7", "node.node_10"])
        >>> sc = sub_cats["node.node_7"]
        >>> p = sp.create_parameteriser('generic subarea')
        >>> p.add_parameter_to_hypercube("x1", 32, 1, 6e3)
        >>> sp = p.subcatchment_parameteriser(sc)
        >>> sp.apply_sys_config(simulation)

    """
    p = swg.CreateSubcatchmentHypercubeParameterizer_py(parameteriser, subcatchment)
    return p

wrap_transform(parameteriser)

Create a parameteriser for which parameter transformations can be defined.

This allows to define e.g. a virtual parameter log_X instead of calibrating on the parameter X.

Parameters:

Name	Type	Description	Default
parameteriser	HypercubeParameteriser	A HypercubeParameteriser wrapper, or a type inheriting from it	required

Returns:

Name	Type	Description
TransformParameteriser		A new parameteriser (TransformParameteriser) which has methods to define parameter transforms

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/parameteriser.py

def wrap_transform(parameteriser):
    """Create a parameteriser for which parameter transformations can be defined.

    This allows to define e.g. a virtual parameter log_X instead of calibrating on the parameter X.

    Args:
        parameteriser (HypercubeParameteriser): A HypercubeParameteriser wrapper, or a type inheriting from it

    Returns:
        TransformParameteriser: A new parameteriser (TransformParameteriser) which has methods to define parameter transforms
    """
    return swg.CreateTransformParameterizer_py(parameteriser)

module play_record

get_played(simulation, var_ids=None, start_time=None, end_time=None)

Retrieves a played time series from a simulation

Retrieves a played time series from a simulation.

Parameters:

Name	Type	Description	Default
simulation	Simulation	A swift simulation object	required
var_ids	Any	name of the output variable played to a time series. 'Catchment|StreamflowRate'. If missing, a multivariate time series of all played states is returned; this may be a large amount of data.	None
start_time	Any	An optional parameter, the start of a period to subset the time series	None
end_time	Any	An optional parameter, the end of a period to subset the time series	None

Returns:

Type	Description
	an xts time series, possibly multivariate.

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/play_record.py

def get_played(simulation: "Simulation", var_ids=None, start_time=None, end_time=None):
    """
    Retrieves a played time series from a simulation

    Retrieves a played time series from a simulation.

    Args:
        simulation (Simulation): A swift simulation object
        var_ids (Any): name of the output variable played to a time series. 'Catchment|StreamflowRate'. If missing, a multivariate time series of all played states is returned; this may be a large amount of data.
        start_time (Any): An optional parameter, the start of a period to subset the time series
        end_time (Any): An optional parameter, the end of a period to subset the time series

    Returns:
        an xts time series, possibly multivariate.

    """

    series = None
    if var_ids is None:
        var_ids = get_played_varnames(simulation)
    series = si.internal_get_played_tts(simulation, var_ids)
    return si.get_ts_window(series, start_time, end_time)

get_played_varnames(simulation)

Gets all the names of states fed an input time series

Gets all the names of states fed an input time series

Parameters:

Name	Type	Description	Default
simulation	Simulation	A swift simulation object	required

Returns:

Type	Description
	The names of the state variables fed over the simulation with values from a time series

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/play_record.py

def get_played_varnames(simulation):
    """
    Gets all the names of states fed an input time series

    Gets all the names of states fed an input time series

    Args:
        simulation (Simulation): A swift simulation object

    Returns:
        The names of the state variables fed over the simulation with values from a time series

    """
    return swg.GetPlayedVariableNames_py(simulation)

get_recorded(simulation, var_ids=None, start_time=None, end_time=None)

Retrieves a recorded time series from a simulation

Retrieves a recorded time series from a simulation.

Parameters:

Name	Type	Description	Default
simulation	Simulation	A swift simulation object	required
var_ids	Any	name of the output variable recorded to a time series. 'Catchment|StreamflowRate'. If missing, a multivariate time series of all recorded states is returned; this may be a large amount of data.	None
start_time	Any	An optional parameter, the start of a period to subset the time series	None
end_time	Any	An optional parameter, the end of a period to subset the time series	None

Returns:

Type	Description
	an xts time series, possibly multivariate.

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/play_record.py

def get_recorded(
    simulation: "Simulation", var_ids=None, start_time=None, end_time=None
):
    """
    Retrieves a recorded time series from a simulation

    Retrieves a recorded time series from a simulation.

    Args:
        simulation (Simulation): A swift simulation object
        var_ids (Any): name of the output variable recorded to a time series. 'Catchment|StreamflowRate'. If missing, a multivariate time series of all recorded states is returned; this may be a large amount of data.
        start_time (Any): An optional parameter, the start of a period to subset the time series
        end_time (Any): An optional parameter, the end of a period to subset the time series

    Returns:
        an xts time series, possibly multivariate.

    """
    if si.is_ensemble_forecast_simulation(simulation):
        return get_recorded_ensemble_forecast(simulation, var_ids, start_time, end_time)
    else:
        series = None
        if var_ids is None:
            var_ids = get_recorded_varnames(simulation)
        series = si.internal_get_recorded_tts(simulation, var_ids)
        return si.get_ts_window(series, start_time, end_time)

get_recorded_ensemble_forecast(simulation, var_id, start_time=None, end_time=None)

Retrieves a recorded time series from a simulation

Retrieves a recorded time series from a simulation.

Parameters:

Name	Type	Description	Default
simulation	Simulation	A swift simulation object	required
var_ids	Any	name of the output variable recorded to a time series. 'Catchment|StreamflowRate'. If missing, a multivariate time series of all recorded states is returned; this may be a large amount of data.	required
start_time	Any	NOT USED YET An optional parameter, the start of a period to subset the time series	None
end_time	Any	NOT USED YET An optional parameter, the end of a period to subset the time series	None

Returns:

Type	Description
	an xts time series, possibly multivariate.

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/play_record.py

def get_recorded_ensemble_forecast(
    simulation, var_id: str, start_time=None, end_time=None
):
    """
    Retrieves a recorded time series from a simulation

    Retrieves a recorded time series from a simulation.

    Args:
        simulation (Simulation): A swift simulation object
        var_ids (Any): name of the output variable recorded to a time series. 'Catchment|StreamflowRate'. If missing, a multivariate time series of all recorded states is returned; this may be a large amount of data.
        start_time (Any): NOT USED YET An optional parameter, the start of a period to subset the time series
        end_time (Any): NOT USED YET An optional parameter, the end of a period to subset the time series

    Returns:
        an xts time series, possibly multivariate.

    """
    si.check_ensemble_forecast_simulation(simulation)
    return swg.GetRecordedEnsembleForecastTimeSeries_py(simulation, var_id)

get_recorded_varnames(simulation)

Gets all the names of the recorded states

Gets all the names of the recorded states

Parameters:

Name	Type	Description	Default
simulation	Simulation	A swift simulation object	required

Returns:

Type	Description
	The names of the state variables being recorded into time series

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/play_record.py

def get_recorded_varnames(simulation):
    """
    Gets all the names of the recorded states

    Gets all the names of the recorded states

    Args:
        simulation (Simulation): A swift simulation object

    Returns:
        The names of the state variables being recorded into time series

    """
    return swg.GetRecordedVariableNames_py(simulation)

play_ensemble_forecast_input(simulation, input_ts, var_id)

Sets time series as input to a simulation

Sets time series as input to a simulation

Parameters:

Name	Type	Description	Default
simulation	Any	an S4 object 'ExternalObjRef' [package "cinterop"] with external pointer type "ENSEMBLE_FORECAST_SIMULATION_PTR"	required
input_ts	Any	an S4 object 'ExternalObjRef' [package "cinterop"] with external pointer type "ENSEMBLE_FORECAST_TIME_SERIES_PTR"	required
var_id	Any	character of length one, the variable identifier to use	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/play_record.py

def play_ensemble_forecast_input(
    simulation: "EnsembleForecastSimulation",
    input_ts: "EnsembleForecastTimeSeries",
    var_id: str,
) -> None:
    """
    Sets time series as input to a simulation

    Sets time series as input to a simulation

    Args:
        simulation (Any): an S4 object 'ExternalObjRef' [package "cinterop"] with external pointer type "ENSEMBLE_FORECAST_SIMULATION_PTR"
        input_ts (Any): an S4 object 'ExternalObjRef' [package "cinterop"] with external pointer type "ENSEMBLE_FORECAST_TIME_SERIES_PTR"
        var_id (Any): character of length one, the variable identifier to use

    """
    si.check_ensemble_forecast_simulation(simulation)
    si.check_ensemble_forecast_time_series(input_ts)
    assert isinstance(var_id, str)
    swg.PlayEnsembleForecastTimeSeries_py(simulation, input_ts, var_id)

play_input(simulation, input_ts, var_ids=None)

Sets time series as input to a simulation

Sets time series as input to a simulation

Parameters:

Name	Type	Description	Default
simulation	Simulation	A swift simulation object	required
input_ts	Any	an xts time series, or an S4 object 'ExternalObjRef' [package "cinterop"] with external pointer type "ENSEMBLE_FORECAST_TIME_SERIES_PTR". if an xts time series column names must be valid model variable identifiers, unless explicitely provided via varIds	required
var_ids	Any	optional character, the variable identifiers to use, overriding the column names of the inputTs. If not NULL, must be of length equal to the number of columns in inputTs	None

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/play_record.py

def play_input(
    simulation: "NdSimulation", input_ts: "TimeSeriesLike", var_ids: VecStr = None
) -> None:
    """
    Sets time series as input to a simulation

    Sets time series as input to a simulation

    Args:
        simulation (Simulation): A swift simulation object
        input_ts (Any): an xts time series, or an S4 object 'ExternalObjRef' [package "cinterop"] with external pointer type "ENSEMBLE_FORECAST_TIME_SERIES_PTR". if an xts time series column names must be valid model variable identifiers, unless explicitely provided via varIds
        var_ids (Any): optional character, the variable identifiers to use, overriding the column names of the inputTs. If not NULL, must be of length equal to the number of columns in inputTs

    """
    if si.is_ensemble_forecast_simulation(
        simulation
    ) and si.is_ensemble_forecast_time_series(input_ts):
        play_ensemble_forecast_input(simulation, input_ts, var_ids)
    elif si.is_ensemble_simulation(
        simulation
    ):  # ??? and si.is_ensemble_forecast_time_series(input_ts):
        raise NotImplementedError("play input into ensemble simulation not supported")
    # } else {
    # stopifnot(is.xts(input_ts))
    else:
        play_singular_simulation(simulation, input_ts, var_ids)

play_inputs(simulation, data_library, model_var_id, data_id, resample='')

Assign input time series from a time series library to a model simulation

Parameters:

Name	Type	Description	Default
simulation	Simulation	A swift simulation object	required
data_library	TimeSeriesLibrary	external pointer type ENSEMBLE_DATA_SET_PTR, or a Python class wrapper around it	required
model_var_id	str or sequence of str	model state variable unique identifier(s)	required
data_id	str or sequence of str	identifier(s) for data in the data_library. If length is not the same as model_var_id, the elements of data_id are reused to match it	required
resample	str or sequence of str	identifier(s) for how the series is resampled (aggregated or disaggregated). If length is not the same as model_var_id, the elements of resample are reused to match it	''

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/play_record.py

def play_inputs(
    simulation: "Simulation", data_library:uc.TimeSeriesLibrary, model_var_id:"VecStr", data_id:"VecStr", resample:"VecStr"=""
):
    """
    Assign input time series from a time series library to a model simulation

    Args:
        simulation (Simulation): A swift simulation object
        data_library (TimeSeriesLibrary): external pointer type ENSEMBLE_DATA_SET_PTR, or a Python class wrapper around it 
        model_var_id (str or sequence of str): model state variable unique identifier(s)
        data_id (str or sequence of str): identifier(s) for data in the data_library. If length is not the same as model_var_id, the elements of data_id are reused to match it
        resample (str or sequence of str): identifier(s) for how the series is resampled (aggregated or disaggregated). If length is not the same as model_var_id, the elements of resample are reused to match it

    """
    # model_var_id <- as.character(model_var_id)
    # data_id <- as.character(data_id)
    # resample <- as.character(resample)
    # TODO: match the behavior of R, perhaps.
    assert len(model_var_id) == len(data_id)
    assert len(model_var_id) == len(resample)
    # if(len(model_var_id) != len(data_id)):
    #     # warning('Reusing argument `data_id` to match length of `model_var_id`')
    #     data_id = np.repeat(data_id, length.out=length(model_var_id))

    # if(length(resample)!=length(model_var_id)) {
    # if(length(resample) != 1 || resample[1] != '') warning('Reusing argument `resample` to match the length of `model_var_id`')
    # resample <- rep(resample, length.out=length(model_var_id))
    # }
    swg.PlayDatasetInputs_py(
        simulation, data_library, model_var_id, data_id, resample, len(model_var_id)
    )

play_subarea_input(simulation, input, subarea_name, input_name)

Sets time series as input to a simulation

Sets time series as input to a simulation

Parameters:

Name	Type	Description	Default
simulation	Simulation	A swift simulation object	required
input	Any	an xts time series.	required
subarea_name	Any	a valid name of the subarea	required
input_name	Any	the name of the input variable to the model (i.e. 'P' for the precip of GR5H)	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/play_record.py

def play_subarea_input(simulation: "Simulation", input, subarea_name, input_name):
    """
    Sets time series as input to a simulation

    Sets time series as input to a simulation

    Args:
        simulation (Simulation): A swift simulation object
        input (Any): an xts time series.
        subarea_name (Any): a valid name of the subarea
        input_name (Any): the name of the input variable to the model (i.e. 'P' for the precip of GR5H)

    """
    play_input(simulation, input, _mkid("subarea", subarea_name, input_name))

record_state(simulation, var_ids=CATCHMENT_FLOWRATE_VARID, recording_provider=None, data_ids=None)

Record a time series of one of the state of the model

Record a time series of one of the state of the model

Parameters:

Name	Type	Description	Default
simulation	Any	an S4 object 'ExternalObjRef' [package "cinterop"] with external pointer type "MODEL_SIMULATION_PTR", "ENSEMBLE_SIMULATION_PTR" or "ENSEMBLE_FORECAST_SIMULATION_PTR"	required
var_ids	VecStr	identifier(s) of the output variable recorded to a time series, e.g. 'Catchment|StreamflowRate' or 'subcatchment.Subarea.runoff'. Defaults to CATCHMENT_FLOWRATE_VARID.	CATCHMENT_FLOWRATE_VARID
recording_provider	TimeSeriesLibrary	description. Defaults to None.	None
data_ids	VecStr	description. Defaults to None.	None

Raises:

Type	Description
ValueError	description

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/play_record.py

def record_state(
    simulation,
    var_ids: VecStr = CATCHMENT_FLOWRATE_VARID,
    recording_provider: "TimeSeriesLibrary" = None,
    data_ids: "VecStr" = None,
):
    """
    Record a time series of one of the state of the model

    Record a time series of one of the state of the model

    Args:
        simulation (Any): an S4 object 'ExternalObjRef' [package "cinterop"] with external pointer type "MODEL_SIMULATION_PTR", "ENSEMBLE_SIMULATION_PTR" or "ENSEMBLE_FORECAST_SIMULATION_PTR"
        var_ids (VecStr, optional): identifier(s) of the output variable recorded to a time series, e.g. 'Catchment|StreamflowRate' or 'subcatchment.Subarea.runoff'. Defaults to CATCHMENT_FLOWRATE_VARID.
        recording_provider (TimeSeriesLibrary, optional): _description_. Defaults to None.
        data_ids (VecStr, optional): _description_. Defaults to None.

    Raises:
        ValueError: _description_
    """
    # checkRecordingConsistency(var_ids, recording_provider, data_ids)
    # var_ids <- as.character(var_ids)
    # data_ids <- as.character(data_ids)
    # recordSingularState(simulation, var_ids, recording_provider, data_ids)Err
    if si.is_ensemble_forecast_simulation(simulation):
        record_ensemble_forecast_state(
            simulation, var_ids, recording_provider, data_ids
        )
    elif si.is_ensemble_simulation(simulation):
        record_ensemble_state(simulation, var_ids, recording_provider, data_ids)
    elif si.is_singular_simulation(simulation):
        record_singular_state(simulation, var_ids, recording_provider, data_ids)
    else:
        raise ValueError("Unknown type of simulation")

module proto

Prototypes

module prototypes

estimate_transformation_parameters(calib_obs, estimation_start, estimation_end, censor_threshold, exclusion, exclusion_start, exclusion_end, termination_condition=None)

Estimate the transformation parameters for a log-likelihood for a series of observations

Parameters:

Name	Type	Description	Default
calib_obs	TimeSeriesLike	An timeseries of observed data	required
estimation_start	datetime	Start of estimation period	required
estimation_end	datetime	End of estimation period	required
censor_threshold	float	The value below which observations are treated a censored data (Default=0.0)	required
exclusion	bool	Start of period exclued from estimation	required
exclusion_start	datetime	End of period exclued from estimation	required
exclusion_end	datetime	Use the exclusion period (bool)	required
termination_condition	SceTerminationCondition	A SWIFT termination condition used by the optimisation. Default max runtime of ~3 minutes if None.	None

Returns:

Name	Type	Description
HypercubeParameteriser	HypercubeParameteriser	transformation parameters

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/prototypes.py

def estimate_transformation_parameters(
    calib_obs: TimeSeriesLike,
    estimation_start: datetime,
    estimation_end: datetime,
    censor_threshold: float,
    exclusion: bool,
    exclusion_start: datetime,
    exclusion_end: datetime,
    termination_condition: "SceTerminationCondition" = None,
) -> "HypercubeParameteriser":
    """Estimate the transformation parameters for a log-likelihood for a series of observations

    Args:
        calib_obs (TimeSeriesLike):  An timeseries of observed data
        estimation_start (datetime): Start of estimation period
        estimation_end (datetime): End of estimation period
        censor_threshold (float): The value below which observations are treated a censored data (Default=0.0)
        exclusion (bool): Start of period exclued from estimation
        exclusion_start (datetime): End of period exclued from estimation
        exclusion_end (datetime): Use the exclusion period (bool)
        termination_condition (SceTerminationCondition): A SWIFT termination condition used by the optimisation. Default max runtime of ~3 minutes if None.

    Returns:
        HypercubeParameteriser: transformation parameters
    """
    import swift2.internal as si

    CENS_OPTION = 2
    simple_ts = si.simplify_time_series(calib_obs)
    if termination_condition is None:
        termination_condition = get_max_runtime_termination()

    return swg.EstimateTransformationParameters_py(
        obsValues=simple_ts[si.TS_INTEROP_VALUES_KEY],
        obsGeom=simple_ts[si.TS_INTEROP_GEOM_KEY],
        estimationStart=estimation_start,
        estimationEnd=estimation_end,
        censThr=censor_threshold,
        censOpt=CENS_OPTION,
        exclusionStart=exclusion_start,
        exclusionEnd=exclusion_end,
        exclusion=exclusion,
        terminationCondition=termination_condition,
    )

module simulation

check_simulation(simulation)

Checks whether a simulation is configured to a state where it is executable

Checks whether a simulation is configured to a state where it is executable

Parameters:

Name	Type	Description	Default
simulation	Simulation	A swift simulation object	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/simulation.py

def check_simulation(simulation) -> Dict:
    """
    Checks whether a simulation is configured to a state where it is executable

    Checks whether a simulation is configured to a state where it is executable

    Args:
        simulation (Simulation): A swift simulation object

    """
    # Trying to design this such that we can have several types of messages
    return {"errors": swg.CheckSimulationErrors_py(simulation)}

create_catchment(node_ids, node_names, link_ids, link_names, link_from_node, link_to_node, runoff_model_name='GR4J', areas_km2=None)

Create a SWIFT catchment with a specified hydrologic model

Create a SWIFT catchment with a specified hydrologic model. This function is intended mostly for testing, not for usual modelling code.

Parameters:

Name	Type	Description	Default
node_ids	Any	character, node unique identifiers	required
node_names	Any	character, node display names	required
link_ids	Any	character, links unique identifiers	required
link_names	Any	character, links display names	required
link_from_node	Any	character, identifier of the links' upstream node	required
link_to_node	Any	character, identifier of the links' downstream node	required
runoff_model_name	Any	A valid, known SWIFT model name (e.g. 'GR5H')	'GR4J'
areas_km2	Any	The areas in square kilometres	None

Returns:

Type	Description
	A SWIFT simulation object (i.e. a model runner)

Examples:

TODO

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/simulation.py

def create_catchment(
    node_ids: List,
    node_names: List,
    link_ids: List,
    link_names: List,
    link_from_node: List,
    link_to_node: List,
    runoff_model_name: str = "GR4J",
    areas_km2: List[float] = None,
):
    """
    Create a SWIFT catchment with a specified hydrologic model

    Create a SWIFT catchment with a specified hydrologic model.
    This function is intended mostly for testing, not for usual modelling code.

    Args:
        node_ids (Any): character, node unique identifiers
        node_names (Any): character, node display names
        link_ids (Any): character, links unique identifiers
        link_names (Any): character, links display names
        link_from_node (Any): character, identifier of the links' upstream node
        link_to_node (Any): character, identifier of the links' downstream node
        runoff_model_name (Any): A valid, known SWIFT model name (e.g. 'GR5H')
        areas_km2 (Any): The areas in square kilometres

    Returns:
        A SWIFT simulation object (i.e. a model runner)

    Examples:
        TODO

    """
    # >>> # nodeIds=paste0('n', 1:6)
    # >>> # linkIds = paste0('lnk', 1:5)
    # >>> # defn <- list(
    # >>> # nodeIds=nodeIds,
    # >>> # nodeNames = paste0(nodeIds, '_name'),
    # >>> # linkIds=linkIds,
    # >>> # linkNames = paste0(linkIds, '_name'),
    # >>> # fromNode = paste0('n', c(2,5,4,3,1)),
    # >>> # toNode = paste0('n', c(6,2,2,4,4)),
    # >>> # areasKm2 = c(1.2, 2.3, 4.4, 2.2, 1.5),
    # >>> # runoffModel = 'GR4J'
    # >>> # )
    # >>> # ms <- createCatchment(defn$nodeIds, defn$nodeNames, defn$linkIds, defn$linkNames, defn$fromNode, defn$toNode, defn$runoffModel, defn$areasKm2)
    if areas_km2 is None:
        areas_km2 = rep(1.0, len(link_ids))
    return swg.CreateCatchment_py(
        numNodes=len(node_ids),
        nodeIds=node_ids,
        nodeNames=node_names,
        numLinks=len(link_ids),
        linkIds=link_ids,
        linkNames=link_names,
        linkFromNode=link_from_node,
        linkToNode=link_to_node,
        runoffModelName=runoff_model_name,
        areasKm2=areas_km2,
    )

create_ensemble_forecast_simulation(simulation, data_library, start, end, input_map, lead_time, ensemble_size, n_time_steps_between_forecasts)

Create an ensemble forecast simulation

Create an ensemble forecast simulation

Parameters:

Name	Type	Description	Default
simulation	Simulation	A swift simulation object	required
data_library	Any	external pointer type ENSEMBLE_DATA_SET_PTR, or a Python class wrapper around it	required
start	Any	the start date of the simulation. The time zone will be forced to UTC.	required
end	Any	the end date of the simulation. The time zone will be forced to UTC.	required
input_map	Any	a named list were names are the data library data identifiers, and values are character vectors with model state identifiers.	required
lead_time	Any	integer, the length in time steps of the forecasts.	required
ensemble_size	Any	ensemble size	required
n_time_steps_between_forecasts	Any	nTimeStepsBetweenForecasts	required

Returns:

Type	Description
	An external pointer

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/simulation.py

def create_ensemble_forecast_simulation(
    simulation,
    data_library,
    start,
    end,
    input_map: Dict[str, List[str]],
    lead_time,
    ensemble_size,
    n_time_steps_between_forecasts,
):
    """
    Create an ensemble forecast simulation

    Create an ensemble forecast simulation

    Args:
        simulation (Simulation): A swift simulation object
        data_library (Any): external pointer type ENSEMBLE_DATA_SET_PTR, or a Python class wrapper around it 
        start (Any): the start date of the simulation. The time zone will be forced to UTC.
        end (Any): the end date of the simulation. The time zone will be forced to UTC.
        input_map (Any): a named list were names are the data library data identifiers, and values are character vectors with model state identifiers.
        lead_time (Any): integer, the length in time steps of the forecasts.
        ensemble_size (Any): ensemble size
        n_time_steps_between_forecasts (Any): nTimeStepsBetweenForecasts

    Returns:
        An external pointer

    """
    s = as_timestamp(start)
    e = as_timestamp(end)
    simulation_length = swg.GetNumStepsForTimeSpan_py(simulation, s, e)
    ef_simulation = swg.CreateEnsembleForecastSimulation_py(
        simulation,
        s,
        lead_time,
        ensemble_size,
        simulation_length,
        nTimeStepsBetweenForecasts=n_time_steps_between_forecasts,
    )
    data_ids = input_map.keys()
    for data_id in data_ids:
        identifiers = input_map[data_id]
        swg.PlayDatasetEnsembleForecastInput_py(
            ef_simulation,
            data_library,
            identifiers,
            rep(data_id, len(identifiers)),
            len(identifiers),
        )
    return ef_simulation

create_subarea(model_name, area_km2)

Create a SWIFT subarea with a specified hydrologic model

Create a SWIFT subarea with a specified hydrologic model

Parameters:

Name	Type	Description	Default
model_name	Any	A valid, known SWIFT model name (e.g. 'GR5H')	required
area_km2	Any	The area in square kilometres	required

Returns:

Type	Description
	A SWIFT simulation object (i.e. a model runner)

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/simulation.py

def create_subarea(model_name, area_km2):
    """
    Create a SWIFT subarea with a specified hydrologic model

    Create a SWIFT subarea with a specified hydrologic model

    Args:
        model_name (Any): A valid, known SWIFT model name (e.g. 'GR5H')
        area_km2 (Any): The area in square kilometres

    Returns:
        A SWIFT simulation object (i.e. a model runner)

    """
    return swg.CreateSubarea_py(model_name, area_km2)

create_subarea_simulation(data_id='MMH', simul_start='1990-01-01', simul_end='2005-12-31', model_id='GR4J', tstep='daily', varname_rain='P', varname_pet='E')

Creates a one sub-catchment simulation

Creates a one sub-catchment simulation. This function is intended for creating sample simulations, not for use in production.

Parameters:

Name	Type	Description	Default
dataId	Any	data identifier in swift_sample_data	required
simulStart	Any	ISO string for the simulation start date time	required
simulEnd	Any	ISO string for the simulation end date time	required
modelId	Any	model identifier	required
tstep	Any	character, 'daily' or 'hourly'	'daily'
varNameRain	Any	variable name to assign rainfall to	required
varNamePet	Any	variable name to assign PET to	required

Returns:

Type	Description
	A SWIFT simulation object, clone of the simulation but with a new model type in use.

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/simulation.py

def create_subarea_simulation(
    data_id="MMH",
    simul_start="1990-01-01",
    simul_end="2005-12-31",
    model_id="GR4J",
    tstep="daily",
    varname_rain="P",
    varname_pet="E",
):
    """
    Creates a one sub-catchment simulation

    Creates a one sub-catchment simulation. This function is intended for creating sample simulations, not for use in production.

    Args:
        dataId (Any): data identifier in swift_sample_data
        simulStart (Any): ISO string for the simulation start date time
        simulEnd (Any): ISO string for the simulation end date time
        modelId (Any): model identifier
        tstep (Any): character, 'daily' or 'hourly'
        varNameRain (Any): variable name to assign rainfall to
        varNamePet (Any): variable name to assign PET to

    Returns:
        A SWIFT simulation object, clone of the simulation but with a new model type in use.

    """
    from swift2.doc_helper import sample_series
    from swift2.wrap.swift_wrap_generated import CreateSubarea_py

    s_span = slice(simul_start, simul_end)
    rain = sample_series(data_id, "rain")[s_span]
    evap = sample_series(data_id, "evap")[s_span]

    ms = CreateSubarea_py(model_id, 1.0)
    s = rain.index[0]
    e = rain.index[-1]
    set_simulation_span(ms, s, e)
    set_simulation_time_step(ms, tstep)

    sa_name = swg.GetSubareaNames_py(ms)[0]
    play_subarea_input(ms, rain, sa_name, varname_rain)
    play_subarea_input(ms, evap, sa_name, varname_pet)
    return ms

exec_simulation(simulation, reset_initial_states=True)

Execute a simulation

Execute a simulation

Parameters:

Name	Type	Description	Default
simulation	Simulation	A swift simulation object	required
reset_initial_states	Any	logical, should the states of the model be reinitialized before the first time step.	True

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/simulation.py

def exec_simulation(simulation: "Simulation", reset_initial_states=True):
    """
    Execute a simulation

    Execute a simulation

    Args:
        simulation (Simulation): A swift simulation object
        reset_initial_states (Any): logical, should the states of the model be reinitialized before the first time step.

    """
    if simulation.type_id == "ENSEMBLE_FORECAST_SIMULATION_PTR":
        swg.ExecuteEnsembleForecastSimulation_py(simulation)
    elif simulation.type_id == "ENSEMBLE_SIMULATION_PTR":
        raise NotImplementedError(
            "execution api entry point for 'ENSEMBLE_SIMULATION_PTR' is not available??"
        )
    else:
        swg.ExecuteSimulation_py(simulation, reset_initial_states)

get_link_ids(simulation)

Gets all the identifiers of the links in the catchment

Gets all the identifiers of the links in the catchment

Parameters:

Name	Type	Description	Default
simulation	Simulation	A swift simulation object	required

Returns:

Type	Description
	The identifiers of the links in the catchment

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/simulation.py

def get_link_ids(simulation):
    """
    Gets all the identifiers of the links in the catchment

    Gets all the identifiers of the links in the catchment

    Args:
        simulation (Simulation): A swift simulation object

    Returns:
        The identifiers of the links in the catchment

    """
    return swg.GetLinkIdentifiers_py(simulation)

get_link_names(simulation)

Gets all the names of the links in the catchment

Gets all the names of the links in the catchment

Parameters:

Name	Type	Description	Default
simulation	Simulation	A swift simulation object	required

Returns:

Type	Description
	The names of the links in the catchment

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/simulation.py

def get_link_names(simulation):
    """
    Gets all the names of the links in the catchment

    Gets all the names of the links in the catchment

    Args:
        simulation (Simulation): A swift simulation object

    Returns:
        The names of the links in the catchment

    """

    return swg.GetLinkNames_py(simulation)

get_node_ids(simulation)

Gets all the identifiers of the nodes in the catchment

Gets all the identifiers of the nodes in the catchment

Parameters:

Name	Type	Description	Default
simulation	Simulation	A swift simulation object	required

Returns:

Type	Description
	The identifiers of the nodes in the catchment

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/simulation.py

def get_node_ids(simulation):
    """
    Gets all the identifiers of the nodes in the catchment

    Gets all the identifiers of the nodes in the catchment

    Args:
        simulation (Simulation): A swift simulation object

    Returns:
        The identifiers of the nodes in the catchment

    """
    return swg.GetNodeIdentifiers_py(simulation)

get_node_names(simulation)

Gets all the names of the nodes in the catchment

Gets all the names of the nodes in the catchment

Parameters:

Name	Type	Description	Default
simulation	Simulation	A swift simulation object	required

Returns:

Type	Description
	The names of the nodes in the catchment

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/simulation.py

def get_node_names(simulation):
    """
    Gets all the names of the nodes in the catchment

    Gets all the names of the nodes in the catchment

    Args:
        simulation (Simulation): A swift simulation object

    Returns:
        The names of the nodes in the catchment

    """
    return swg.GetNodeNames_py(simulation)

get_state_value(simulation, var_id)

Gets the value(s) of a model state(s)

Parameters:

Name	Type	Description	Default
simulation	Simulation	A swift simulation object	required
var_id	VecStr	string or sequence of str, model variable state identifier(s)	required

Returns:

Type	Description
	numeric vector, value(s) of the requested model states

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/simulation.py

def get_state_value(simulation: "Simulation", var_id: "VecStr"):
    """
    Gets the value(s) of a model state(s)

    Args:
        simulation (Simulation): A swift simulation object
        var_id (VecStr): string or sequence of str, model variable state identifier(s)

    Returns:
        numeric vector, value(s) of the requested model states

    """
    if isinstance(var_id, str):
        var_ids = [var_id]
    else:
        var_ids = var_id
    #  if(is.numeric(value)):
    f = swg.GetVariable_py
    #  } else if(is.integer(value)):
    #    f = GetVariableInt_R
    #  } else if(is.logical(value)):
    #    f = GetVariableBool_R
    #  else:
    #    stop(paste('value type', type(value), 'is not supported by getStateValue'))
    #  }
    return dict([(v, f(simulation, v)) for v in var_ids])

get_subarea_ids(simulation)

Gets all the identifiers of the sub-areas in the catchment

Gets all the identifiers of the sub-areas in the catchment

Parameters:

Name	Type	Description	Default
simulation	Simulation	A swift simulation object	required

Returns:

Type	Description
	The identifiers of the sub-areas in the catchment

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/simulation.py

def get_subarea_ids(simulation):
    """
    Gets all the identifiers of the sub-areas in the catchment

    Gets all the identifiers of the sub-areas in the catchment

    Args:
        simulation (Simulation): A swift simulation object

    Returns:
        The identifiers of the sub-areas in the catchment

    """
    return swg.GetSubareaIdentifiers_py(simulation)

get_subarea_names(simulation)

Gets all the names of the sub-areas in the catchment

Gets all the names of the sub-areas in the catchment

Parameters:

Name	Type	Description	Default
simulation	Simulation	A swift simulation object	required

Returns:

Type	Description
	The names of the sub-areas in the catchment

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/simulation.py

def get_subarea_names(simulation):
    """
    Gets all the names of the sub-areas in the catchment

    Gets all the names of the sub-areas in the catchment

    Args:
        simulation (Simulation): A swift simulation object

    Returns:
        The names of the sub-areas in the catchment

    """

    return swg.GetSubareaNames_py(simulation)

get_variable_ids(simulation, element_id=None, full_id=True)

Gets all the names of the variables of an element within a catchment

Gets all the names of the variables of an element (link, node, subarea) within a catchment

Parameters:

Name	Type	Description	Default
simulation	Simulation	A swift simulation object	required
element_id	Any	a character, identifier of the element within the catchment	None
full_id	Any	boolean, if TRUE return the full hierarchical identifier	True

Returns:

Type	Description
	character vector, names (identifiers) of model states in the element

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/simulation.py

def get_variable_ids(simulation: "Simulation", element_id=None, full_id=True):
    """
    Gets all the names of the variables of an element within a catchment

    Gets all the names of the variables of an element (link, node, subarea) within a catchment

    Args:
        simulation (Simulation): A swift simulation object
        element_id (Any): a character, identifier of the element within the catchment
        full_id (Any): boolean, if TRUE return the full hierarchical identifier

    Returns:
        character vector, names (identifiers) of model states in the element

    """
    s = swg.GetElementVarIdentifiers_py(simulation, element_id)
    if full_id and element_id is not None:
        s = [_mkid(element_id, v) for v in s]
    return s

is_variable_id(simulation, var_id)

Is a variable identifier valid for a simulation

Is a variable identifier valid for a simulation

Parameters:

Name	Type	Description	Default
simulation	Simulation	A swift simulation object	required
var_id	Any	a character, identifier(s) of the variable(s)	required

Returns:

Type	Description
	logical vector

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/simulation.py

def is_variable_id(simulation: "Simulation", var_id: "VecStr"):
    """
    Is a variable identifier valid for a simulation

    Is a variable identifier valid for a simulation

    Args:
        simulation (Simulation): A swift simulation object
        var_id (Any): a character, identifier(s) of the variable(s)

    Returns:
        logical vector

    """
    if is_common_iterable(var_id):
        return dict(
            [(v, swg.IsValidVariableIdentifier_py(simulation, v)) for v in var_id]
        )
    else:
        return swg.IsValidVariableIdentifier_py(simulation, var_id)

reset_model_states(simulation)

Reset the model states of a simulation, and apply one or more state initialers if the simulation is configured with any.

Parameters:

Name	Type	Description	Default
simulation	Simulation	simulation	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/simulation.py

def reset_model_states(simulation: "Simulation"):
    """Reset the model states of a simulation, and apply one or more state initialers if the simulation is configured with any.

    Args:
        simulation (Simulation): simulation
    """
    swg.ResetModelStates_py(simulation)

set_error_correction_model(simulation, model_id, element_id, length=1, seed=0)

Add an error correction model to an element in a catchment

Add an error correction model to an element in a catchment

Parameters:

Name	Type	Description	Default
simulation	Simulation	A swift simulation object	required
model_id	str	the identifier of the new model to use, e.g. 'ERRIS'	required
element_id	str	the identifier of the catchment element (node, link, subcatchment) whose outflow rate is corrected.	required
length	int	other parameters to pass to the creation of the error correction model. Currently length of the AR model only supported.	1
seed	int	other parameters to pass to the creation of the error correction model. Currently length of the AR model only supported.	0

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/simulation.py

def set_error_correction_model(
    simulation, model_id, element_id, length=1, seed=0
):  # TODO: revert to use ... if other kind of ECM
    """
    Add an error correction model to an element in a catchment

    Add an error correction model to an element in a catchment

    Args:
        simulation (Simulation): A swift simulation object
        model_id (str): the identifier of the new model to use, e.g. 'ERRIS'
        element_id (str): the identifier of the catchment element (node, link, subcatchment) whose outflow rate is corrected.
        length (int): other parameters to pass to the creation of the error correction model. Currently length of the AR model only supported.
        seed (int): other parameters to pass to the creation of the error correction model. Currently length of the AR model only supported.

    """
    swg.SetErrorCorrectionModel_py(simulation, model_id, element_id, length, seed)

set_simulation_span(simulation, start, end)

Sets simulation span

Sets the simulation span

Parameters:

Name	Type	Description	Default
simulation	Simulation	A swift simulation object	required
start	Any	the start date of the simulation. The time zone will be forced to UTC.	required
end	Any	the end date of the simulation. The time zone will be forced to UTC.	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/simulation.py

def set_simulation_span(simulation: "Simulation", start, end):
    """
    Sets simulation span

    Sets the simulation span

    Args:
        simulation (Simulation): A swift simulation object
        start (Any): the start date of the simulation. The time zone will be forced to UTC.
        end (Any): the end date of the simulation. The time zone will be forced to UTC.

    """
    s = as_timestamp(start)  # = as_timestamp(start, tz='UTC')
    e = as_timestamp(end)  # = as_timestamp(end, tz='UTC')
    swg.SetSpan_py(simulation, s, e)

set_simulation_time_step(simulation, name)

Sets the time step of a SWIFT simulation

Sets the time step of a SWIFT simulation

Parameters:

Name	Type	Description	Default
simulation	Simulation	A swift simulation object	required
name	Any	a time step identifier, currently 'daily' or 'hourly' are supported. The identifier is made lower case in the function.	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/simulation.py

def set_simulation_time_step(simulation: "Simulation", name: str):
    """
    Sets the time step of a SWIFT simulation

    Sets the time step of a SWIFT simulation

    Args:
        simulation (Simulation): A swift simulation object
        name (Any): a time step identifier, currently 'daily' or 'hourly' are supported. The identifier is made lower case in the function.

    """
    name = name.lower()
    # assert name in set(['daily','hourly'])
    swg.SetTimeStep_py(simulation, name)

set_state_value(simulation, var_id, value)

Sets the value of a model state

Sets the value of a model state

Parameters:

Name	Type	Description	Default
simulation	Simulation	A swift simulation object	required
var_id	(str, Sequence[str])	character, model variable state identifier(s)	required
value	(float, int, bool, Sequence)	numeric value(s)	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/simulation.py

def set_state_value(
    simulation: "Simulation",
    var_id: Union[str, Sequence[str]],
    value: Union[float, int, bool, Sequence],
) -> None:
    """
    Sets the value of a model state

    Sets the value of a model state

    Args:
        simulation (Simulation): A swift simulation object
        var_id (str, Sequence[str]): character, model variable state identifier(s)
        value (float, int, bool, Sequence): numeric value(s)

    """

    def func_set_for(v):
        if isinstance(v, float): # or isinstance(v, np.float): # np obsolete?
            return swg.SetVariable_py
        elif isinstance(v, int): # or isinstance(v, np.int):
            return swg.SetVariableInt_py
        elif isinstance(v, bool): # or isinstance(v, np.bool8):
            return swg.SetVariableBool_py
        else:
            raise TypeError(
                "value type " + str(type(v)) + "is not supported by setStateValue"
            )

    if isinstance(var_id, str):
        f = func_set_for(value)
        f(simulation, var_id, value)
        return
    if isinstance(var_id, dict):
        d = [(k, v) for k, v in var_id.items()]
        var_id = [x[0] for x in d]
        value = [x[1] for x in d]
    if not is_common_iterable(var_id):
        raise ValueError("var_id must be a string, or an iterable of strings")
    if not is_common_iterable(value):
        value = rep(value, len(var_id))
    else:
        assert len(var_id) == len(value)

    for i in range(len(var_id)):
        k = var_id[i]
        v = value[i]
        f = func_set_for(v)
        f(simulation, k, v)

set_states(simulation, states)

Apply memory states to a simulation

Parameters:

Name	Type	Description	Default
simulation	Simulation	simulation	required
states	MemoryStates	memory states	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/simulation.py

def set_states(simulation: "Simulation", states: "MemoryStates"):
    """Apply memory states to a simulation

    Args:
        simulation (Simulation): simulation
        states (MemoryStates): memory states
    """
    swg.ApplyMemoryStates_py(simulation, states)

snapshot_state(simulation)

Take a snapshot of the memory states of a simulation

Parameters:

Name	Type	Description	Default
simulation	Simulation	model simulation	required

Returns:

Name	Type	Description
MemoryStates	MemoryStates	memory states, that can be stored and reapplied

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/simulation.py

def snapshot_state(simulation: "Simulation") -> "MemoryStates":
    """Take a snapshot of the memory states of a simulation

    Args:
        simulation (Simulation): model simulation

    Returns:
        MemoryStates: memory states, that can be stored and reapplied
    """
    return swg.SnapshotMemoryStates_py(simulation)

sort_by_execution_order(simulation, split_element_ids, sorting_option='')

Sort the specified element ids according to the execution order of the simulation

Sort the specified element ids according to the execution order of the simulation

Parameters:

Name	Type	Description	Default
simulation	Simulation	A swift simulation object	required
split_element_ids	Any	a character vector with element identifiers such as 'node.n1', 'link.linkId_2'	required
sorting_option	Any	a character - for future options. Ignored for now.	''

Returns:

Type	Description
	values in split_element_ids sorted by simulation execution order

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/simulation.py

def sort_by_execution_order(
    simulation: "Simulation", split_element_ids, sorting_option=""
):
    """
    Sort the specified element ids according to the execution order of the simulation

    Sort the specified element ids according to the execution order of the simulation

    Args:
        simulation (Simulation): A swift simulation object
        split_element_ids (Any): a character vector with element identifiers such as 'node.n1', 'link.linkId_2'
        sorting_option (Any): a character - for future options. Ignored for now.

    Returns:
        values in split_element_ids sorted by simulation execution order

    """
    return swc.sort_simulation_elements_by_run_order_pkg(
        simulation, split_element_ids, sorting_option
    )

swap_model(simulation, model_id, what='runoff')

Clone and change a simulation, using another runoff model

Clone and change a simulation, using another runoff model

Parameters:

Name	Type	Description	Default
simulation	Simulation	A swift simulation object	required
model_id	Any	the identifier of the new model to use, e.g. 'GR4J'	required
what	Any	character identifying the type of structure: 'runoff', 'channel_routing'	'runoff'

Returns:

Type	Description
	A SWIFT simulation object, clone of the simulation but with a new model type in use.

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/simulation.py

def swap_model(simulation: "Simulation", model_id, what="runoff"):
    """
    Clone and change a simulation, using another runoff model

    Clone and change a simulation, using another runoff model

    Args:
        simulation (Simulation): A swift simulation object
        model_id (Any): the identifier of the new model to use, e.g. 'GR4J'
        what (Any): character identifying the type of structure: 'runoff', 'channel_routing'

    Returns:
        A SWIFT simulation object, clone of the simulation but with a new model type in use.

    """
    if what == "runoff":
        cloned = swg.SwapRunoffModel_py(simulation, model_id)
    elif what == "channel_routing":
        cloned = swg.CloneModel_py(simulation)
        swg.SetChannelRoutingModel_py(cloned, model_id)
    else:
        raise ValueError(f"option not supported: {what}")
    return cloned

module statistics

createCompositeObjective(simulation, state_name, observation, yamlstring_statistic, start_date, end_date)

Creates a composite objective calculator

Creates a composite objective calculator

Parameters:

Name	Type	Description	Default
simulation	Simulation	A SWIFT simulation object (i.e. a model runner)	required
state_name	Any	The name identifying the model state variable to calibrate against the observation	required
observation	Any	an xts	required
yamlstring_statistic	Any	a yaml string representing objective functions and weights eg...	required
start_date	Any	start date of the period to calculate statistics on	required
end_date	Any	end date of the period to calculate statistics on	required

Returns:

Type	Description
	objective evaluator

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/statistics.py

def createCompositeObjective(
    simulation, state_name, observation, yamlstring_statistic, start_date, end_date
):
    """
    Creates a composite objective calculator

    Creates a composite objective calculator

    Args:
        simulation (Simulation): A SWIFT simulation object (i.e. a model runner)
        state_name (Any): The name identifying the model state variable to calibrate against the observation
        observation (Any): an xts
        yamlstring_statistic (Any): a yaml string representing objective functions and weights eg...
        start_date (Any): start date of the period to calculate statistics on
        end_date (Any): end date of the period to calculate statistics on

    Returns:
        objective evaluator

    """
    import xarray as xr
    from cinterop.timeseries import ts_window

    # if not isinstance(observation, xr.DataArray):
    #     raise ValueError('observation must be an xts object')
    # if(ncol(observation) != 1):
    #     stop('observation must have exactly one series')
    # }
    observation = ts_window(observation, start_date, end_date)
    simple_ts = simplify_time_series(observation)
    return swg.CreateCompositeObservationObjectiveEvaluator_py(
        simulation,
        state_name,
        simple_ts[TS_INTEROP_VALUES_KEY],
        simple_ts[TS_INTEROP_GEOM_KEY],
        yamlstring_statistic,
    )

create_multisite_objective(simulation, statspec, observations, weights)

Creates an objective that combines multiple statistics

Creates an objective that combines multiple statistics. Used for joined, "whole of catchment" calibration

Parameters:

Name	Type	Description	Default
simulation	Simulation	A SWIFT simulation object (i.e. a model runner)	required
statspec	pd.DataFrame	dataframe defining the objectives used. See function multi_statistic_definition to help build this dataframe.	required
observations	Sequence[TimeSeriesLike]	A list of (time series) observations to calculated the statistics. Must be of same length as the number of rows of statspec.	required
weights	Dict[str, float]	numeric vector of weights to ponderate each objective.	required
Examples		todo()	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/statistics.py

def create_multisite_objective(
    simulation: "Simulation", 
    statspec: pd.DataFrame,
    observations: Sequence[TimeSeriesLike],
    weights: Dict[str, float],
):
    """
    Creates an objective that combines multiple statistics

    Creates an objective that combines multiple statistics. Used for joined, "whole of catchment" calibration

    Args:
        simulation (Simulation): A SWIFT simulation object (i.e. a model runner)
        statspec (pd.DataFrame): dataframe defining the objectives used. See function [`multi_statistic_definition`][swift2.statistics.multi_statistic_definition] to help build this dataframe.
        observations (Sequence[TimeSeriesLike]): A list of (time series) observations to calculated the statistics. Must be of same length as the number of rows of statspec.
        weights (Dict[str, float]): numeric vector of weights to ponderate each objective.

        Examples:
            >>> todo()
    """
    from swift2.wrap.ffi_interop import marshal

    # stopifnot(is.list(observations))
    # stopifnot(nrow(statspec) == length(observations))
    # stopifnot(nrow(statspec) == length(weights))
    interop_obs = [marshal.as_native_time_series(x) for x in observations]
    defn = {
        "Length": len(statspec),
        "Statistics": statspec,
        "Observations": interop_obs,
    }
    obj = swg.CreateMultisiteObjectiveEvaluator_py(simulation, defn, weights)
    return swg.WrapObjectiveEvaluatorWila_py(obj, True)

create_objective(simulation, state_name, observation, statistic, start_date, end_date)

Creates an objective calculator

Parameters:

Name	Type	Description	Default
simulation	Simulation	A SWIFT simulation object (i.e. a model runner)	required
state_name	Any	The name identifying the model state variable to calibrate against the observation	required
observation	TimeSeriesLike	an xts	required
statistic	str	statistic identifier, e.g. "NSE"	required
start_date	ConvertibleToTimestamp	start date of the period to calculate statistics on	required
end_date	ConvertibleToTimestamp	end date of the period to calculate statistics on	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/statistics.py

def create_objective(
    simulation: "Simulation",
    state_name: str,
    observation: TimeSeriesLike,
    statistic: str,
    start_date: ConvertibleToTimestamp,
    end_date: ConvertibleToTimestamp,
):
    """
    Creates an objective calculator

    Args:
        simulation (Simulation): A SWIFT simulation object (i.e. a model runner)
        state_name (Any): The name identifying the model state variable to calibrate against the observation
        observation (TimeSeriesLike): an xts
        statistic (str): statistic identifier, e.g. "NSE"
        start_date (ConvertibleToTimestamp): start date of the period to calculate statistics on
        end_date (ConvertibleToTimestamp): end date of the period to calculate statistics on
    """
    from swift2.internal import simplify_time_series
    from cinterop.timeseries import ts_window, as_timestamp

    start_date = as_timestamp(start_date)
    end_date = as_timestamp(end_date)

    # if(!is.xts(observation)):
    #     stop('observation must be an xts object')
    # }
    # if(ncol(observation) != 1):
    #     stop('observation must have exactly one series')
    # }
    observation = ts_window(observation, start_date, end_date)
    simple_ts = simplify_time_series(observation)
    return swg.CreateSingleObservationObjectiveEvaluatorWila_py(
        simulation,
        state_name,
        simple_ts[TS_INTEROP_VALUES_KEY],
        simple_ts[TS_INTEROP_GEOM_KEY],
        statistic,
    )

get_score(objective_evaluator, p_set)

Evaluate an objective for a given parameterisation

Parameters:

Name	Type	Description	Default
objective_evaluator	ObjectiveEvaluator	objective evaluator	required
p_set	HypercubeParameteriser	parameteriser	required

Returns:

Type	Description
Dict[str, Any]	Dict[str,Any]: score(s), and a data frame representation of the input parameters.

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/statistics.py

def get_score(
    objective_evaluator: "ObjectiveEvaluator", p_set: "HypercubeParameteriser"
) -> Dict[str, Any]:
    """Evaluate an objective for a given parameterisation

    Args:
        objective_evaluator (ObjectiveEvaluator): objective evaluator
        p_set (HypercubeParameteriser): parameteriser

    Returns:
        Dict[str,Any]: score(s), and a data frame representation of the input parameters.
    """
    #  if (p_set.type_id == "HYPERCUBE_PTR"):
    #    p_set = ToHypercubeWila_py(p_set);
    #  }
    #  stopifnot(p_set.type_id=='HYPERCUBE_WILA_PTR')
    return swc.evaluate_score_wila_pkg(objective_evaluator, p_set)

multi_statistic_definition(model_var_ids, statistic_ids, objective_ids, objective_names, starts, ends)

Collate information for use in multisite multiobjective definition

Collate information for use in multisite multiobjective definition

Parameters:

Name	Type	Description	Default
model_var_ids	Any	character vector, model state identifiers where statistics are calculated	required
statistic_ids	Any	character vector, identifiers for bivariate statistics (e.g. nse, lognse, et.)	required
objective_ids	Any	character vector, identifiers for the objectives. Can be the same as modelVarIds.	required
objective_names	Any	character vector, display names for the objectives. Can be the same as modelVarIds.	required
starts	Any	POSIXct vector of start dates for statistics	required
ends	Any	POSIXct vector of end dates for statistics	required

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/statistics.py

def multi_statistic_definition(
    model_var_ids, statistic_ids, objective_ids, objective_names, starts, ends
):
    """
    Collate information for use in multisite multiobjective definition

    Collate information for use in multisite multiobjective definition

    Args:
        model_var_ids (Any): character vector, model state identifiers where statistics are calculated
        statistic_ids (Any): character vector, identifiers for bivariate statistics (e.g. nse, lognse, et.)
        objective_ids (Any): character vector, identifiers for the objectives. Can be the same as modelVarIds.
        objective_names (Any): character vector, display names for the objectives. Can be the same as modelVarIds.
        starts (Any): POSIXct vector of start dates for statistics
        ends (Any): POSIXct vector of end dates for statistics

    """
    # stopifnot(is.character(model_var_ids))
    # stopifnot(is.character(statistic_ids))
    # stopifnot(is.character(objective_ids))
    # stopifnot(is.character(objective_names))
    # stopifnot(is.numeric(starts)) # TODO but not the right check. POSIXct
    # stopifnot(is.numeric(ends))

    return pd.DataFrame.from_dict(
        {
            "ModelVarId": model_var_ids,
            "StatisticId": statistic_ids,
            "ObjectiveId": objective_ids,
            "ObjectiveName": objective_names,
            "Start": starts,
            "End": ends,
        }
    )

module system

get_last_swift_error()

Retrieve the message for the last known error in SWIFT

Retrieve the message for the last known error in SWIFT. Error means here that an exception was thrown by the core SWIFT library. The SWIFT C API intercepts these messages to make them available to users for diagnosis.

Returns:

Type	Description
	A character, the message for the last known error in SWIFT.

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/system.py

def get_last_swift_error():
    """
    Retrieve the message for the last known error in SWIFT

    Retrieve the message for the last known error in SWIFT. Error means here that an exception was thrown by the core
    SWIFT library. The SWIFT C API intercepts these messages to make them available to users for diagnosis.

    Returns:
        A character, the message for the last known error in SWIFT.
    """

    return swg.GetLastStdExceptionMessage_py()

runoff_model_ids()

Gets all the names of known runoff models

Gets all the names of known runoff models

Returns:

Type	Description
	character vector, names (identifiers) of runoff models

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/system.py

def runoff_model_ids():
    """
    Gets all the names of known runoff models

    Gets all the names of known runoff models

    Returns:
        character vector, names (identifiers) of runoff models

    """
    return swg.GetRunoffModelIdentifiers_py()

runoff_model_var_ids(model_id)

Gets all the names of the variables a runoff model exposes

Gets all the names of the variables a runoff model exposes for dynamic query.

Parameters:

Name	Type	Description	Default
model_id	Any	character; A recognized model identifier	required

Returns:

Type	Description
	a character vector, the known model variable that can be set/gotten

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/system.py

def runoff_model_var_ids(model_id):
    """
    Gets all the names of the variables a runoff model exposes

    Gets all the names of the variables a runoff model exposes for dynamic query.

    Args:
        model_id (Any): character; A recognized model identifier

    Returns:
        a character vector, the known model variable that can be set/gotten

    """
    if model_id not in runoff_model_ids():
        raise ValueError("Unrecognized runoff model identifier: " + model_id)
    return swg.GetRunoffModelVarIdentifiers_py(model_id)

set_default_max_parallelism_threads(n_threads=-1)

Sets the level of thread parallelism to use by default for new objects such as optimisers. May be overwritten for each instance afterwards.

Parameters:

Name	Type	Description	Default
n_threads	int	number of threads. Positive, or -1 to mean "as many as available"	-1

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/system.py

def set_default_max_parallelism_threads(n_threads: int = -1):
    """Sets the level of thread parallelism to use by default for new objects such as optimisers. May be overwritten for each instance afterwards.

    Args:
        n_threads (int): number of threads. Positive, or -1 to mean "as many as available"
    """
    swg.SetDefaultMaxThreadsWila_py(n_threads)

set_maximum_threads(optimiser, n_threads=-1)

Sets the maximum level of parallelism of an optimizer

Sets the maximum level of threading of an optimizer. NOTE: this also modifies a global default for further optimizers, which is a hack for ERRIS, acceptable but still likely to change in the future. It is VERY important to use this function prior to running calibrations on some systems such as clusters, as the default hardware detection may not be appropriate if the cluster node is not dedicated.

Parameters:

Name	Type	Description	Default
optimizer	Any	an S4 object 'ExternalObjRef' [package "cinterop"] with external pointer type "OPTIMIZER_PTR"	required
n_threads	Any	integer, maximum number of threads allowed. If -1, the system defaults to using all but one of the CPU cores detected on the hardware.	-1

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/system.py

def set_maximum_threads(optimiser, n_threads=-1):
    """
    Sets the maximum level of parallelism of an optimizer

    Sets the maximum level of threading of an optimizer. NOTE: this also modifies a global default for further optimizers,
    which is a hack for ERRIS, acceptable but still likely to change in the future.
    It is VERY important to use this function prior to running calibrations on some systems such as clusters,
    as the default hardware detection may not be appropriate if the cluster node is not dedicated.

    Args:
        optimizer (Any): an S4 object 'ExternalObjRef' [package "cinterop"] with external pointer type "OPTIMIZER_PTR"
        n_threads (Any): integer, maximum number of threads allowed. If -1, the system defaults to using all but one of the CPU cores detected on the hardware.

    """
    swg.SetMaxThreadsOptimizerWila_py(optimiser, n_threads)

module utils

c(*args)

Emulate the R c (concatenate) function, somewhat.

Returns:

Type	Description
np.ndarray	np.ndarray: [description]

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/utils.py

def c(*args) -> np.ndarray:
    """Emulate the R c (concatenate) function, somewhat.

    Returns:
        np.ndarray: [description]
    """
    return np.array([x for x in args])

is_common_iterable(obj)

True if an object is iterable but not a string (str)

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/utils.py

def is_common_iterable(obj: Any) -> bool:
    """True if an object is iterable but not a string (str)"""
    if isinstance(obj, str):
        return False
    # if isinstance(obj, np.ndarray) and obj.size == 1:
    #     return False # otherwise likely to get error "len() of unsized object"
    return hasattr(type(obj), "__iter__")

mk_full_data_id(*args)

Create swift IDs (dot separated hierarchical naming scheme)

Create swift IDs (dot separated hierarchical naming scheme). Note that the behavior is different than 'paste' for empty characters.

Parameters:

Name	Type	Description	Default
args	Any	one or more character vectors.	()

Examples:

TODO

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/utils.py

def mk_full_data_id(*args):
    """
    Create swift IDs (dot separated hierarchical naming scheme)

    Create swift IDs (dot separated hierarchical naming scheme). Note that the behavior is different than 'paste' for empty characters.

    Args:
        args (Any): one or more character vectors.

    Examples:
        TODO

    """
    #    >>> # (obsVarname <- mkFullDataId('link','corin_dam','ec','Observed'))
    #    >>> # subareaFullIds <- mkFullDataId( 'subarea', getSubareaIds(simulation))
    #    >>> # mkFullDataId(character(0),'corin_dam','ec','Observed')
    ids = list(args)
    if len(ids) == 0:
        return None
    else:
        lengths = [len(x) for x in ids if not isinstance(x, str)]
        if len(lengths) > 0:
            minl = min(lengths)
            if minl < 1:
                return None
        return paste(*args, sep=".")

paste(*lists, sep=' ', collapse=None)

Port of R paste function

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/utils.py

def paste(*lists, sep=" ", collapse=None):
    """Port of R paste function"""
    result = reduce_concat(lists, sep=sep)
    if collapse is not None:
        return reduce_concat(result, sep=collapse)
    return result

paste0(*lists, collapse=None)

Port of R paste0 function

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/utils.py

def paste0(*lists, collapse=None):
    """Port of R paste0 function"""
    return paste(*lists, sep="", collapse=collapse)

paste_2(x, y, sep=' ')

Port of R vectorised paste, for 2 elements

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/utils.py

def paste_2(x: VecScalars, y: VecScalars, sep: str = " "):
    """Port of R vectorised paste, for 2 elements"""
    if is_common_iterable(x):
        if is_common_iterable(y):
            return paste_lists(x, y, sep)
        else:
            return paste_list_scalar(x, y, sep)
    else:
        if is_common_iterable(y):
            return paste_scalar_list(x, y, sep)
        else:
            return paste_scalar_scalar(x, y, sep)

sort_by(x, unsorted_reference, sorted_reference)

Sort one vector according to the known reordering of another

Parameters:

Name	Type	Description	Default
x	Any	values to sort	required
unsorted_reference	Any	unique 'keys' corresponding to each element in x	required
sorted_reference	Any	set of 'keys', identical as a set to unsorted_reference, but sorted	required

Returns:

Type	Description
	the values in x reordered such that the same reordering of unsorted_reference matches sorted_reference

Examples:

TODO

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/utils.py

def sort_by(x, unsorted_reference, sorted_reference):
    """
    Sort one vector according to the known reordering of another

    Args:
        x (Any): values to sort
        unsorted_reference (Any): unique 'keys' corresponding to each element in x
        sorted_reference (Any): set of 'keys', identical as a set to unsorted_reference, but sorted

    Returns:
        the values in x reordered such that the same reordering of unsorted_reference matches sorted_reference

    Examples:
        TODO

    """
    # >>> # set.seed(12)
    # >>> # (x = sample(1:5))
    # >>> # (unsorted_reference = letters[x])
    # >>> # (sorted_reference = letters[1:5])
    # >>> # (x[order(match(unsorted_reference,sorted_reference))])
    # >>> # sortBy(x, unsorted_reference, sorted_reference)
    assert len(x) == len(unsorted_reference)
    assert len(sorted_reference) == len(unsorted_reference)
    assert set(unsorted_reference) == set(sorted_reference)
    # There may be something more elegant
    def find_index(shuffled, sorted):
        if isinstance(shuffled, list):
            shuffled = np.array(shuffled)
        s = [np.argwhere(x == shuffled) for x in sorted]
        for v in s:
            assert v.shape == (1, 1)
        return [v[0][0] for v in s]

    # TODO unit test of course:
    # x = np.array([2, 5, 3, 1, 4])
    # unsorted_reference = np.array(["b","e","c","a","d"])
    # sorted_reference = np.array(["a","b","c","d","e"])
    # indx = find_index(unsorted_reference, sorted_reference)
    # [x[i] for i in indx]
    indx = find_index(unsorted_reference, sorted_reference)
    return [x[i] for i in indx]

vpaste(root, vars)

vectorised paste for 2 elements; Port of R paste0 in spirit

Parameters:

Name	Type	Description	Default
root	VecScalars	left hand side(s) of the paste	required
vars	VecScalars	right hand side(s) of the paste	required

Returns:

Type	Description
Union[str, Sequence[str]]	Union[str,Sequence[str]]: pasted scalars

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/utils.py

def vpaste(root: VecScalars, vars: VecScalars) -> Union[str, Sequence[str]]:
    """vectorised paste for 2 elements; Port of R paste0 in spirit

    Args:
        root (VecScalars): left hand side(s) of the paste
        vars (VecScalars): right hand side(s) of the paste

    Returns:
        Union[str,Sequence[str]]: pasted scalars
    """
    return paste_2(root, vars, sep="")

module vis

OptimisationPlots

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/vis.py

class OptimisationPlots:
    def __init__(self, optim_geom: MhData) -> None:
        self._optim_geom = optim_geom

    def parameter_evolution(
        self,
        param_name: str,
        obj_lims: Sequence[float] = None,
        title: str = "Evolution of parameter values",
        xlab="Logged point",
        ylab=None,
        **kwargs
    ):
        d = self._optim_geom.bound_fitness(obj_lims)
        if ylab is None:
            ylab = param_name
        ax = d.plot.scatter(
            x="PointNumber",
            y=param_name,
            c=self._optim_geom._fitness,
            colormap="viridis",
            **kwargs
        )
        if xlab is not None:
            ax.set_xlabel(xlab)
        if ylab is not None:
            ax.set_ylabel(ylab)
        if title is not None:
            ax.set_title(title)
        return ax

    def shuffles(self, x: str, y: str, obj_lims: Sequence[float] = None) -> Any:
        """Facetted bi-parameter scatter plots of the value of a parameter along the optimisation process

        Plot the value of a parameter along the optimisation process.
        The color scale is the objective score. Useful to check the behavior of the optimisation process.

        Args:
            x (str): the exact name of one of the model parameters
            y (str): the exact name of a second model parameter
            obj_lims (Sequence[float], optional): min/max limits to plot the fitness, for example min 0 for NSE. Defaults to None.

        Returns:
            sns.FacetGrid: FacetGrid object
        """
        import matplotlib.pyplot as plt

        d = self._optim_geom.bound_fitness(obj_lims)
        # matplotlib makes it difficult to use continuous color scales for num values...
        # https://stackoverflow.com/a/44642014/2752565
        g = sns.FacetGrid(
            d,
            col=self._optim_geom._categories,
            hue=self._optim_geom._fitness,
            col_wrap=3,
        )

        def facet_scatter(x, y, c, **kwargs):
            """Draw scatterplot with point colors from a faceted DataFrame columns."""
            kwargs.pop("color")
            plt.scatter(x, y, c=c, **kwargs)

        values = d[self._optim_geom._fitness].values
        vmin, vmax = np.min(values), np.max(values)
        # cmap = sns.diverging_palette(240, 10, l=65, center="light", as_cmap=True)
        cmap = sns.color_palette(
            palette="viridis", n_colors=None, desat=None, as_cmap=True
        )

        g = g.map(
            facet_scatter,
            x,
            y,
            self._optim_geom._fitness,
            s=100,
            alpha=0.5,
            vmin=vmin,
            vmax=vmax,
            cmap=cmap,
        )

        # Make space for the colorbar
        g.fig.subplots_adjust(right=0.92)

        # Define a new Axes where the colorbar will go
        cax = g.fig.add_axes([0.94, 0.25, 0.02, 0.6])

        # Get a mappable object with the same colormap as the data
        points = plt.scatter([], [], c=[], vmin=vmin, vmax=vmax, cmap=cmap)

        # Draw the colorbar
        g.fig.colorbar(points, cax=cax)

        g.fig.set_size_inches(15, 10)  # Sorry, only silly imperial units available
        # grid.add_legend()
        return g

shuffles(x, y, obj_lims=None)

Facetted bi-parameter scatter plots of the value of a parameter along the optimisation process

Plot the value of a parameter along the optimisation process. The color scale is the objective score. Useful to check the behavior of the optimisation process.

Parameters:

Name	Type	Description	Default
x	str	the exact name of one of the model parameters	required
y	str	the exact name of a second model parameter	required
obj_lims	Sequence[float]	min/max limits to plot the fitness, for example min 0 for NSE. Defaults to None.	None

Returns:

Type	Description
Any	sns.FacetGrid: FacetGrid object

Source code in /home/per202/src/swift/bindings/python/swift2/swift2/vis.py

def shuffles(self, x: str, y: str, obj_lims: Sequence[float] = None) -> Any:
    """Facetted bi-parameter scatter plots of the value of a parameter along the optimisation process

    Plot the value of a parameter along the optimisation process.
    The color scale is the objective score. Useful to check the behavior of the optimisation process.

    Args:
        x (str): the exact name of one of the model parameters
        y (str): the exact name of a second model parameter
        obj_lims (Sequence[float], optional): min/max limits to plot the fitness, for example min 0 for NSE. Defaults to None.

    Returns:
        sns.FacetGrid: FacetGrid object
    """
    import matplotlib.pyplot as plt

    d = self._optim_geom.bound_fitness(obj_lims)
    # matplotlib makes it difficult to use continuous color scales for num values...
    # https://stackoverflow.com/a/44642014/2752565
    g = sns.FacetGrid(
        d,
        col=self._optim_geom._categories,
        hue=self._optim_geom._fitness,
        col_wrap=3,
    )

    def facet_scatter(x, y, c, **kwargs):
        """Draw scatterplot with point colors from a faceted DataFrame columns."""
        kwargs.pop("color")
        plt.scatter(x, y, c=c, **kwargs)

    values = d[self._optim_geom._fitness].values
    vmin, vmax = np.min(values), np.max(values)
    # cmap = sns.diverging_palette(240, 10, l=65, center="light", as_cmap=True)
    cmap = sns.color_palette(
        palette="viridis", n_colors=None, desat=None, as_cmap=True
    )

    g = g.map(
        facet_scatter,
        x,
        y,
        self._optim_geom._fitness,
        s=100,
        alpha=0.5,
        vmin=vmin,
        vmax=vmax,
        cmap=cmap,
    )

    # Make space for the colorbar
    g.fig.subplots_adjust(right=0.92)

    # Define a new Axes where the colorbar will go
    cax = g.fig.add_axes([0.94, 0.25, 0.02, 0.6])

    # Get a mappable object with the same colormap as the data
    points = plt.scatter([], [], c=[], vmin=vmin, vmax=vmax, cmap=cmap)

    # Draw the colorbar
    g.fig.colorbar(points, cax=cax)

    g.fig.set_size_inches(15, 10)  # Sorry, only silly imperial units available
    # grid.add_legend()
    return g