Skip to content

Advanced elements

Base models inherited by all pySWAP models.

A lot of functionality can be abstracted away in the base models. This way, the code is more DRY and easier to maintain. The base models are used to enforce the correct data types and structure of the input data. They also provide methods to convert the data to the format required by the SWAP model.

Classes defined here are based on Pydantic BaseModel and Pandera DataFrameModel. Both are meant to ensure the correct data types and structure of the input data, as successful validation means smooth execution of the SWAP model. Particularily important when run as a submitted job on an HPC.

Classes:

Name Description
BaseModel

Base class for pySWAP models. Inherits from Pydantic BaseModel.
BaseTableModel

Base class for pySWAP models that validate pandas DataFrames. Inherits from Pandera DataFrameModel.

BaseTableModel

Bases: DataFrameModel

Base model for pandas DataFrames.

Methods:

Name Description
create

Create a validated DataFrame from a dictionary.
Source code in pyswap/core/basemodel.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
class BaseTableModel(pa.DataFrameModel):
    """Base model for pandas DataFrames.

    Methods:
        create: Create a validated DataFrame from a dictionary.
    """

    class Config:
        coerce = True

    @classmethod
    def create(cls, data: dict, columns: list | None = None) -> DataFrame:
        df = pd.DataFrame(data)
        if columns:
            df.columns = columns
        else:
            df.columns = df.columns.str.upper()
        validated_df = cls.validate(df)
        return validated_df

PySWAPBaseModel

Bases: BaseModel

Base class for pySWAP models.

Methods:

Name Description
__setattr__

Overriden method to silently ignore assignment of frozen fields.
update

Update the model with new values from a dictionary.
Source code in pyswap/core/basemodel.py 
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
class PySWAPBaseModel(BaseModel):
    """Base class for pySWAP models.

    Methods:
        __setattr__: Overriden method to silently ignore assignment of frozen
            fields.
        update: Update the model with new values from a dictionary.
    """

    model_config = ConfigDict(
        arbitrary_types_allowed=True,
        validate_assignment=True,
        extra="ignore",
        populate_by_name=True,
    )

    def __setattr__(self, name, value):
        """Silently ignore assignment of frozen fields.

        This method is overridden to silently ignore assignment of frozen fields
        to avoid errors when an old swp files is read.
        """

        if name in self.model_fields and self.model_fields[name].frozen:
            return
        super().__setattr__(name, value)

    def update(self, new: dict, inplace: bool = False, no_validate: bool = False):
        """Update the model with new values.

        Given dictionary of values is first filtered to include only the fields
        that exist in the model. The model is then updated with the new values.
        The updated model is returned (either new or updated self).

        Parameters:
            new (dict): Dictionary with new values.
            inplace (bool): If True, update the model in place.
        """

        # filtered = {k: v for k, v in new.items() if k in self.model_fields}

        # updated_model = self.model_validate(dict(self) | filtered)
        updated_model = self.model_validate(dict(self) | new)

        if not inplace:
            # added this for the case when the user loads a model from the
            # classic ASCII files. Then the .update() method is used, but not
            # all the attributes will be available immediatelly. Full validation
            # will still be performed upon model run.
            if no_validate:
                updated_model._validation = False
            else:
                updated_model._validation = True
            updated_model.validate_with_yaml() if hasattr(
                updated_model, "validate_with_yaml"
            ) else None
            return updated_model

        else:
            for field, value in updated_model:
                setattr(self, field, value)
            if no_validate:
                updated_model._validation = False
            else:
                updated_model._validation = True
            self.validate_with_yaml() if hasattr(
                updated_model, "validate_with_yaml"
            ) else None

            return self

    @field_validator("*", mode="before")
    @classmethod
    def convert_switches(cls, value: Any, info: Any) -> Any:
        """Convert switch values to integers.

        This method was necessary to ensure that loading models from ASCII files
        would work. It could be improved to include literals that do not start
        with "sw" as well.
        """
        if (
            (info.field_name.startswith("sw") or info.field_name in ADDITIONAL_SWITCHES)
            and info.field_name != "swap_ver"
            and value
        ):
            try:
                return int(value)
            except ValueError:
                return value
        return value

__setattr__(name, value)

Silently ignore assignment of frozen fields.

This method is overridden to silently ignore assignment of frozen fields to avoid errors when an old swp files is read.

Source code in pyswap/core/basemodel.py 
51
52
53
54
55
56
57
58
59
60
def __setattr__(self, name, value):
    """Silently ignore assignment of frozen fields.

    This method is overridden to silently ignore assignment of frozen fields
    to avoid errors when an old swp files is read.
    """

    if name in self.model_fields and self.model_fields[name].frozen:
        return
    super().__setattr__(name, value)

convert_switches(value, info) classmethod

Convert switch values to integers.

This method was necessary to ensure that loading models from ASCII files would work. It could be improved to include literals that do not start with "sw" as well.

Source code in pyswap/core/basemodel.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
@field_validator("*", mode="before")
@classmethod
def convert_switches(cls, value: Any, info: Any) -> Any:
    """Convert switch values to integers.

    This method was necessary to ensure that loading models from ASCII files
    would work. It could be improved to include literals that do not start
    with "sw" as well.
    """
    if (
        (info.field_name.startswith("sw") or info.field_name in ADDITIONAL_SWITCHES)
        and info.field_name != "swap_ver"
        and value
    ):
        try:
            return int(value)
        except ValueError:
            return value
    return value

update(new, inplace=False, no_validate=False)

Update the model with new values.

Given dictionary of values is first filtered to include only the fields that exist in the model. The model is then updated with the new values. The updated model is returned (either new or updated self).

Parameters:

Name Type Description Default
new dict

Dictionary with new values.

 required
inplace bool

If True, update the model in place.

 False
Source code in pyswap/core/basemodel.py 
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def update(self, new: dict, inplace: bool = False, no_validate: bool = False):
    """Update the model with new values.

    Given dictionary of values is first filtered to include only the fields
    that exist in the model. The model is then updated with the new values.
    The updated model is returned (either new or updated self).

    Parameters:
        new (dict): Dictionary with new values.
        inplace (bool): If True, update the model in place.
    """

    # filtered = {k: v for k, v in new.items() if k in self.model_fields}

    # updated_model = self.model_validate(dict(self) | filtered)
    updated_model = self.model_validate(dict(self) | new)

    if not inplace:
        # added this for the case when the user loads a model from the
        # classic ASCII files. Then the .update() method is used, but not
        # all the attributes will be available immediatelly. Full validation
        # will still be performed upon model run.
        if no_validate:
            updated_model._validation = False
        else:
            updated_model._validation = True
        updated_model.validate_with_yaml() if hasattr(
            updated_model, "validate_with_yaml"
        ) else None
        return updated_model

    else:
        for field, value in updated_model:
            setattr(self, field, value)
        if no_validate:
            updated_model._validation = False
        else:
            updated_model._validation = True
        self.validate_with_yaml() if hasattr(
            updated_model, "validate_with_yaml"
        ) else None

        return self

Mixins

Reusable mixins enhancing functionality of specific PySWAPBaseModel.

To keep the main PySWAPBaseModel class and the components library clean and focused, mixins are used to add additional functionality to the classes that need it. The concept of the mixins was inspired by the Django framework and it really helps to keep the code clean and organized.

Should more functionality be needed in the future for one or more classes, it should be implemented as a mixin and then inherited by the classes that need it.

Classes:

FileMixin: Custom saving functionality for models that need file I/O.
SerializableMixin: Converting a model to a SWAP-formatted string.
YAMLValidatorMixin: Validating parameters using external YAML rules.
WOFOSTUpdateMixin: Interface for the WOFOST crop parameters database for
    pySWAP.

FileMixin

Custom saving functionality for models that need file I/O.

!!! note:

The _extension attribute should be set in the class that inherits
this mixin. It is recommended that pydantic's PrivateAttr is used to
hide this attribute from the user.

Methods:

Name Description
save_file

Saves a string to a file.
Source code in pyswap/utils/mixins.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
class FileMixin:
    """Custom saving functionality for models that need file I/O.

    !!! note:

        The _extension attribute should be set in the class that inherits
        this mixin. It is recommended that pydantic's PrivateAttr is used to
        hide this attribute from the user.

    Methods:
        save_file: Saves a string to a file.
    """

    def save_file(
        self,
        string: str,
        fname: str,
        path: Path,
    ) -> None:
        """Saves a string to a file.

        The extension should now be provided in each class inheriting this
        mixin as a private attribute.

        Parameters:
            string: The string to be saved to a file.
            fname: The name of the file.
            path: The path where the file should be saved.
        """

        if not hasattr(self, "_extension"):
            msg = "The _extension attribute should be set."
            raise AttributeError(msg)

        ext = self._extension
        fname = f"{fname}.{ext}" if ext else fname

        with open(f"{path}/{fname}", "w", encoding="ascii") as f:
            f.write(string)

        logger.info(f"{fname} saved successfully.")

        return None

save_file(string, fname, path)

Saves a string to a file.

The extension should now be provided in each class inheriting this mixin as a private attribute.

Parameters:

Name Type Description Default
string str

The string to be saved to a file.

 required
fname str

The name of the file.

 required
path Path

The path where the file should be saved.

 required
Source code in pyswap/utils/mixins.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def save_file(
    self,
    string: str,
    fname: str,
    path: Path,
) -> None:
    """Saves a string to a file.

    The extension should now be provided in each class inheriting this
    mixin as a private attribute.

    Parameters:
        string: The string to be saved to a file.
        fname: The name of the file.
        path: The path where the file should be saved.
    """

    if not hasattr(self, "_extension"):
        msg = "The _extension attribute should be set."
        raise AttributeError(msg)

    ext = self._extension
    fname = f"{fname}.{ext}" if ext else fname

    with open(f"{path}/{fname}", "w", encoding="ascii") as f:
        f.write(string)

    logger.info(f"{fname} saved successfully.")

    return None

SerializableMixin

Bases: BaseModel

Converting a model to a SWAP-formatted string.

This mixin is only inherited by classes that directly serialize to a SWAP-formatted string. The assumptions are that the inheriting classes:

  • do not contain nested classes.
  • if the class contains nested classes it should either use Subsection field types or override the model_string() method.

Methods:

Name Description
if_is_union_type

Check if the field type is a Union type.
is_annotated_exception_type

Check if the attribute type is Table, Arrays, or ObjectList.
serialize_model

Override the default serialization method.
model_string

Concatenate the formatted strings from dictionary to one string.
Source code in pyswap/utils/mixins.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
class SerializableMixin(BaseModel):
    """Converting a model to a SWAP-formatted string.

    This mixin is only inherited by classes that directly serialize to a
    SWAP-formatted string. The assumptions are that the inheriting classes:

    - do not contain nested classes.
    - if the class contains nested classes it should either use Subsection field
        types or override the `model_string()` method.

    Methods:
        if_is_union_type: Check if the field type is a Union type.
        is_annotated_exception_type: Check if the attribute type is Table,
            Arrays, or ObjectList.
        serialize_model: Override the default serialization method.
        model_string: Concatenate the formatted strings from dictionary to
            one string.
    """

    def if_is_union_type(self, field_info: FieldInfo) -> dict | None:
        """Check if the field type is a Union type.

        If it is, look for the json_schema_extra attribute in the field_info
        of the first argument of the Union type. If it is not found, return
        None. It was necessary in cases of, for example, optional classes like
        Union[Table, None].

        Parameters:
            field_info (FieldInfo): The FieldInfo object of the field.
        """

        field_type = field_info.annotation

        if get_origin(field_type) is Union:
            union_args = get_args(field_type)
            args = get_args(union_args[0])

            field_info = [item for item in args if isinstance(item, FieldInfo)]

            if not field_info:
                return None

            # Only return the json_schema_extra attribute. This is used in some
            # cases to pass addotional information from the serializer in
            # pyswap.core.fields module to the model_dump.
            return field_info[0].json_schema_extra
        return None

    def is_annotated_exception_type(self, field_name: str) -> bool:
        """Check if the attribute type is Table, Arrays, or ObjectList.

        For Table, Arrays, and ObjectList types True is returned, ensuring a
        separate serialization path.

        First try to assign the json_schema_extra from a Union type. If that
        fails, assign the json_schema_extra from the field_info. If the
        json_schema_extra is None, return False.
        """
        # Every special field will have a FieldInfo object
        field_info = self.model_fields.get(field_name, None)

        if field_info is None:
            return False

        json_schema_extra = (
            self.if_is_union_type(field_info) or field_info.json_schema_extra
        )

        if json_schema_extra is None:
            return False

        return json_schema_extra.get("is_annotated_exception_type", False)

    @model_serializer(when_used="json", mode="wrap")
    def serialize_model(self, handler: Any):
        """Override the default serialization method.

        In the intermediate step, a dictionary is created with SWAP formatted
        strings.
        """
        result = {}
        validated_self = handler(self)
        for field_name, field_value in validated_self.items():
            if self.is_annotated_exception_type(field_name):
                result[field_name] = field_value
            else:
                result[field_name] = f"{field_name.upper()} = {field_value}"
        return result

    def model_string(
        self, mode: Literal["str", "list"] = "string", **kwargs
    ) -> str | list[str]:
        """Concatenate the formatted strings from dictionary to one string.


        !!! note:
            By alias is True, because in some cases, particularily in the case
            of CropSettings, the WOFOST names of parameters in the database were
            different from those used in SWAP. This allows those parameters to
            be properly matched, yet serialized properly in SWAP input files.

        Parameters:
            mode (Literal["str", "list]): The output format.
            kwargs (dict): Additional keyword arguments passed to `model_dump()`.
        """
        dump = self.model_dump(
            mode="json", exclude_none=True, by_alias=True, **kwargs
        ).values()

        if mode == "list":
            return list(dump)
        else:
            return "\n".join(dump)

if_is_union_type(field_info)

Check if the field type is a Union type.

If it is, look for the json_schema_extra attribute in the field_info of the first argument of the Union type. If it is not found, return None. It was necessary in cases of, for example, optional classes like Union[Table, None].

Parameters:

Name Type Description Default
field_info FieldInfo

The FieldInfo object of the field.

 required
Source code in pyswap/utils/mixins.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def if_is_union_type(self, field_info: FieldInfo) -> dict | None:
    """Check if the field type is a Union type.

    If it is, look for the json_schema_extra attribute in the field_info
    of the first argument of the Union type. If it is not found, return
    None. It was necessary in cases of, for example, optional classes like
    Union[Table, None].

    Parameters:
        field_info (FieldInfo): The FieldInfo object of the field.
    """

    field_type = field_info.annotation

    if get_origin(field_type) is Union:
        union_args = get_args(field_type)
        args = get_args(union_args[0])

        field_info = [item for item in args if isinstance(item, FieldInfo)]

        if not field_info:
            return None

        # Only return the json_schema_extra attribute. This is used in some
        # cases to pass addotional information from the serializer in
        # pyswap.core.fields module to the model_dump.
        return field_info[0].json_schema_extra
    return None

is_annotated_exception_type(field_name)

Check if the attribute type is Table, Arrays, or ObjectList.

For Table, Arrays, and ObjectList types True is returned, ensuring a separate serialization path.

First try to assign the json_schema_extra from a Union type. If that fails, assign the json_schema_extra from the field_info. If the json_schema_extra is None, return False.

Source code in pyswap/utils/mixins.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
def is_annotated_exception_type(self, field_name: str) -> bool:
    """Check if the attribute type is Table, Arrays, or ObjectList.

    For Table, Arrays, and ObjectList types True is returned, ensuring a
    separate serialization path.

    First try to assign the json_schema_extra from a Union type. If that
    fails, assign the json_schema_extra from the field_info. If the
    json_schema_extra is None, return False.
    """
    # Every special field will have a FieldInfo object
    field_info = self.model_fields.get(field_name, None)

    if field_info is None:
        return False

    json_schema_extra = (
        self.if_is_union_type(field_info) or field_info.json_schema_extra
    )

    if json_schema_extra is None:
        return False

    return json_schema_extra.get("is_annotated_exception_type", False)

model_string(mode='string', **kwargs)

Concatenate the formatted strings from dictionary to one string.

!!! note: By alias is True, because in some cases, particularily in the case of CropSettings, the WOFOST names of parameters in the database were different from those used in SWAP. This allows those parameters to be properly matched, yet serialized properly in SWAP input files.

Parameters:

Name Type Description Default
mode Literal["str", "list]

The output format.

 'string'
kwargs dict

Additional keyword arguments passed to model_dump().

 {}
Source code in pyswap/utils/mixins.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
def model_string(
    self, mode: Literal["str", "list"] = "string", **kwargs
) -> str | list[str]:
    """Concatenate the formatted strings from dictionary to one string.


    !!! note:
        By alias is True, because in some cases, particularily in the case
        of CropSettings, the WOFOST names of parameters in the database were
        different from those used in SWAP. This allows those parameters to
        be properly matched, yet serialized properly in SWAP input files.

    Parameters:
        mode (Literal["str", "list]): The output format.
        kwargs (dict): Additional keyword arguments passed to `model_dump()`.
    """
    dump = self.model_dump(
        mode="json", exclude_none=True, by_alias=True, **kwargs
    ).values()

    if mode == "list":
        return list(dump)
    else:
        return "\n".join(dump)

serialize_model(handler)

Override the default serialization method.

In the intermediate step, a dictionary is created with SWAP formatted strings.

Source code in pyswap/utils/mixins.py 
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
@model_serializer(when_used="json", mode="wrap")
def serialize_model(self, handler: Any):
    """Override the default serialization method.

    In the intermediate step, a dictionary is created with SWAP formatted
    strings.
    """
    result = {}
    validated_self = handler(self)
    for field_name, field_value in validated_self.items():
        if self.is_annotated_exception_type(field_name):
            result[field_name] = field_value
        else:
            result[field_name] = f"{field_name.upper()} = {field_value}"
    return result

WOFOSTUpdateMixin

Interface for the WOFOST crop parameters database for pySWAP.

This mixin should be inherited by classes that share parameters with the WOFOST crop database.

Source code in pyswap/utils/mixins.py 
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
class WOFOSTUpdateMixin:
    """Interface for the WOFOST crop parameters database for pySWAP.

    This mixin should be inherited by classes that share parameters with the
    WOFOST crop database.
    """

    def update_from_wofost(self) -> None:
        """Update the model with the WOFOST variety settings."""
        from pyswap.utils.old_swap import create_array_objects

        # parameters attribute returns a dictionary with the key-value pairs and
        # tables as list of lists. Before updating, the tables should be
        # created.
        if not hasattr(self, "wofost_variety"):
            msg = "The model does not have the WOFOST variety settings."
            raise AttributeError(msg)

        variety_params = self.wofost_variety.parameters
        new_arrays = create_array_objects(variety_params)
        new = variety_params | new_arrays
        self.update(new, inplace=True)

update_from_wofost()

Update the model with the WOFOST variety settings.

Source code in pyswap/utils/mixins.py 
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
def update_from_wofost(self) -> None:
    """Update the model with the WOFOST variety settings."""
    from pyswap.utils.old_swap import create_array_objects

    # parameters attribute returns a dictionary with the key-value pairs and
    # tables as list of lists. Before updating, the tables should be
    # created.
    if not hasattr(self, "wofost_variety"):
        msg = "The model does not have the WOFOST variety settings."
        raise AttributeError(msg)

    variety_params = self.wofost_variety.parameters
    new_arrays = create_array_objects(variety_params)
    new = variety_params | new_arrays
    self.update(new, inplace=True)

YAMLValidatorMixin

Bases: BaseModel

A mixin class that provides YAML-based validation for models.

Initially, pySWAP had model serializers on each model component class which had a number of assertions to validate the parameters (i.e., require parameters rlwtb and wrtmax if swrd = 3). This created chaos in the code, and since none of it was used by inspection tools anyways, it was decided to leave the validation logic in the code and move the rules to a separate YAML file.

Methods:

Name Description
validate_parameters

Validate parameters against required rules.
validate_with_yaml

Pydantic validator executing validation logic.
Source code in pyswap/utils/mixins.py 
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
class YAMLValidatorMixin(BaseModel):
    """A mixin class that provides YAML-based validation for models.

    Initially, pySWAP had model serializers on each model component class which
    had a number of assertions to validate the parameters (i.e., require
    parameters rlwtb and wrtmax if swrd = 3). This created chaos
    in the code, and since none of it was used by inspection tools anyways, it
    was decided to leave the validation logic in the code and move the rules to
    a separate YAML file.

    Methods:
        validate_parameters: Validate parameters against required rules.
        validate_with_yaml: Pydantic validator executing validation logic.
    """

    _validation: bool = PrivateAttr(default=False)

    @staticmethod
    def validate_parameters(
        switch_name: str, switch_value: str, params: dict, rules: dict
    ):
        """Validate parameters against required rules.

        This method reads the rules for the model from the YAML file and checks
        if the required parameters are present. If not, it raises a ValueError.

        ```yaml
        SaltStress: # <--- Model name
            swsalinity:  # <--- Switch name (switch_name)
                1:  # <--- Switch value (switch_value)
                - saltmax  # <---| Required parameters
                - saltslope  # <--|
                2:
                - salthead
        ```

        Parameters:
            switch_name (str): The name of the switch (e.g., 'swcf').
            switch_value (Any): The value of the switch (e.g., 1 or 2).
            params (dict): Dictionary of parameters to check.
            rules (dict): Dictionary with validation rules.

        Raises:
            ValueError: If required parameters are missing.
        """

        required_params = rules.get(switch_name, {}).get(switch_value, [])

        if not required_params:
            return  # No rules for this switch value

        missing_params = [
            param for param in required_params if params.get(param) is None
        ]

        if missing_params:
            msg = f"The following parameters are required for {switch_name}={switch_value}: {', '.join(missing_params)}"
            raise ValueError(msg)

    @model_validator(mode="after")
    def validate_with_yaml(self) -> Self:
        """Pydantic validator executing validation logic.

        All validators defined on a model run on model instantiation. This
        method makes sure that YAML validation is postponed until the
        _validation parameter (required on all classes inheriting this mixin) is
        set to True. This state is done when all the required parameters are
        presumed to be set, e.g., when the user tries to run the model.
        """

        if not self._validation:
            return self

        rules = VALIDATIONRULES.get(self.__class__.__name__, {})

        for switch_name in rules:
            switch_value = getattr(self, switch_name, None)
            if switch_value is not None:  # Only validate if the switch is set
                self.validate_parameters(
                    switch_name, switch_value, self.__dict__, rules
                )

        self._validation = False
        return self

validate_parameters(switch_name, switch_value, params, rules) staticmethod

Validate parameters against required rules.

This method reads the rules for the model from the YAML file and checks if the required parameters are present. If not, it raises a ValueError.

SaltStress: # <--- Model name
    swsalinity:  # <--- Switch name (switch_name)
        1:  # <--- Switch value (switch_value)
        - saltmax  # <---| Required parameters
        - saltslope  # <--|
        2:
        - salthead

Parameters:

Name Type Description Default
switch_name str

The name of the switch (e.g., 'swcf').

 required
switch_value Any

The value of the switch (e.g., 1 or 2).

 required
params dict

Dictionary of parameters to check.

 required
rules dict

Dictionary with validation rules.

 required

Raises:

Type Description
ValueError

If required parameters are missing.
Source code in pyswap/utils/mixins.py 
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
@staticmethod
def validate_parameters(
    switch_name: str, switch_value: str, params: dict, rules: dict
):
    """Validate parameters against required rules.

    This method reads the rules for the model from the YAML file and checks
    if the required parameters are present. If not, it raises a ValueError.

    ```yaml
    SaltStress: # <--- Model name
        swsalinity:  # <--- Switch name (switch_name)
            1:  # <--- Switch value (switch_value)
            - saltmax  # <---| Required parameters
            - saltslope  # <--|
            2:
            - salthead
    ```

    Parameters:
        switch_name (str): The name of the switch (e.g., 'swcf').
        switch_value (Any): The value of the switch (e.g., 1 or 2).
        params (dict): Dictionary of parameters to check.
        rules (dict): Dictionary with validation rules.

    Raises:
        ValueError: If required parameters are missing.
    """

    required_params = rules.get(switch_name, {}).get(switch_value, [])

    if not required_params:
        return  # No rules for this switch value

    missing_params = [
        param for param in required_params if params.get(param) is None
    ]

    if missing_params:
        msg = f"The following parameters are required for {switch_name}={switch_value}: {', '.join(missing_params)}"
        raise ValueError(msg)

validate_with_yaml()

Pydantic validator executing validation logic.

All validators defined on a model run on model instantiation. This method makes sure that YAML validation is postponed until the _validation parameter (required on all classes inheriting this mixin) is set to True. This state is done when all the required parameters are presumed to be set, e.g., when the user tries to run the model.

Source code in pyswap/utils/mixins.py 
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
@model_validator(mode="after")
def validate_with_yaml(self) -> Self:
    """Pydantic validator executing validation logic.

    All validators defined on a model run on model instantiation. This
    method makes sure that YAML validation is postponed until the
    _validation parameter (required on all classes inheriting this mixin) is
    set to True. This state is done when all the required parameters are
    presumed to be set, e.g., when the user tries to run the model.
    """

    if not self._validation:
        return self

    rules = VALIDATIONRULES.get(self.__class__.__name__, {})

    for switch_name in rules:
        switch_value = getattr(self, switch_name, None)
        if switch_value is not None:  # Only validate if the switch is set
            self.validate_parameters(
                switch_name, switch_value, self.__dict__, rules
            )

    self._validation = False
    return self

Validation and serialization

Functions to parse SWAP formatted ascii files into pySWAP objects.

pySWAP has the ability to interact directly with the classic SWAP input files. Parsers defined in this module are used for the custom field validators defined in the pyswap.core.fields module. These functions convert (or deserialize) the SWAP formatted ascii files into pySWAP objects.

Parsers in this module

parse_string_list: Convert a SWAP string list to a list of strings. parse_quoted_string: Make sure to remove unnecessary quotes from source. parse_day_month: Convert a string to a date object with just the day and month.

parse_day_month(value)

Convert a string to a date object with just the day and month.

Source code in pyswap/core/parsers.py 
57
58
59
60
61
62
63
64
65
66
67
68
def parse_day_month(value: str | date) -> date:
    """Convert a string to a date object with just the day and month."""
    msg = "Invalid day-month format. Expected 'DD MM'"
    if isinstance(value, date):
        return value
    if isinstance(value, str):
        try:
            day, month = map(int, value.split())
            return date(date.today().year, month, day)
        except (ValueError, TypeError):
            raise ValueError(msg) from None
    raise ValueError(msg)

parse_decimal(value)

remove fortan style decimal point.

Source code in pyswap/core/parsers.py 
42
43
44
45
46
def parse_decimal(value: str) -> str:
    """remove fortan style decimal point."""
    if isinstance(value, str):
        value = value.lower().replace("d", "e")
    return float(value)

parse_float_list(value)

Convert a SWAP string list to a list of strings.

Source code in pyswap/core/parsers.py 
26
27
28
29
30
31
def parse_float_list(value: str) -> str:
    """Convert a SWAP string list to a list of strings."""
    if isinstance(value, list):
        return value
    if isinstance(value, str):
        return value.strip("'").split(" ")

parse_int_list(value)

Convert a SWAP string list to a list of strings.

Source code in pyswap/core/parsers.py 
34
35
36
37
38
39
def parse_int_list(value: str) -> str:
    """Convert a SWAP string list to a list of strings."""
    if isinstance(value, list):
        return value
    if isinstance(value, str):
        return value.strip("'").split(" ")

parse_quoted_string(value)

Make sure to remove unnecessary quotes from source.

Source code in pyswap/core/parsers.py 
49
50
51
52
53
54
def parse_quoted_string(value: str) -> str:
    """Make sure to remove unnecessary quotes from source."""
    if isinstance(value, str):
        return value.strip("'")
    msg = "Invalid type. Expected string"
    raise ValueError(msg)

parse_string_list(value)

Convert a SWAP string list to a list of strings.

Source code in pyswap/core/parsers.py 
18
19
20
21
22
23
def parse_string_list(value: str) -> str:
    """Convert a SWAP string list to a list of strings."""
    if isinstance(value, list):
        return value
    if isinstance(value, str):
        return value.strip("'").split(",")

Functions to fine tune the serializatino of pySWAP objects to SWAP formatted ASCII.

More complex serialization logic which would be unwieldy to implement directly in the Annotated field definitions (pyswap.core.fields module) as lambda functions are defined in the serializers module (pyswap.core.serializers). These are functions that convert objects to strings in the valid SWAP format.

Serializers in this module

serialize_table: Convert a DataFrame to a string. serialize_arrays: Convert a DataFrame to a string without headers and newline in front. serialize_csv_table: Convert a DataFrame to a string in CSV format. serialize_object_list: Convert a list of objects to a string. serialize_day_month: Convert a date object to a string with just the day and month.

serialize_arrays(table)

Convert the DataFrame to a string without headers and newline in front.

Arguments:
    table: The DataFrame to be serialized.

Result:
    >>> 'ARRAYS =

1 4 2 5 3 6

'

Source code in pyswap/core/serializers.py 
37
38
39
40
41
42
43
44
45
46
def serialize_arrays(table: DataFrame) -> str:
    """Convert the DataFrame to a string without headers and newline in front.

    Arguments:
        table: The DataFrame to be serialized.

    Result:
        >>> 'ARRAYS = \n1 4\n2 5\n3 6\n\n'
    """
    return f"\n{table.to_string(index=False, header=False)}\n"

serialize_csv_table(table)

Convert the DataFrame to a string in CSV format.

This serializer is specifically tailored to output the data in the format of the ,met files used in SWAP.

Parameters:

Name Type Description Default
table DataFrame

The DataFrame to be serialized.

 required
Source code in pyswap/core/serializers.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def serialize_csv_table(table: DataFrame) -> str:
    """Convert the DataFrame to a string in CSV format.

    This serializer is specifically tailored to output the data in the
    format of the ,met files used in SWAP.

    Arguments:
        table: The DataFrame to be serialized.
    """
    if isinstance(table.index, DatetimeIndex):
        table["DD"] = table.index.day
        table["MM"] = table.index.month
        table["YYYY"] = table.index.year
        required_order = [
            "Station",
            "DD",
            "MM",
            "YYYY",
            "RAD",
            "Tmin",
            "Tmax",
            "HUM",
            "WIND",
            "RAIN",
            "ETref",
            "WET",
        ]
        table = table[required_order]

    table.loc[:, "Station"] = table.Station.apply(
        lambda x: f"'{x}'" if not str(x).startswith("'") else x
    )
    return table.to_csv(index=False, lineterminator="\n")

serialize_day_month(value)

Serialize a date object to a string with just the day and month.

Parameters:

Name Type Description Default
value date

The date object to be serialized.

 required
Result

'01 01'

Source code in pyswap/core/serializers.py 
84
85
86
87
88
89
90
91
92
93
def serialize_day_month(value: date) -> str:
    """Serialize a date object to a string with just the day and month.

    Arguments:
        value: The date object to be serialized.

    Result:
        >>> '01 01'
    """
    return value.strftime("%d %m")

serialize_table(table)

Convert the DataFrame to a string.

Arguments:
    table: The DataFrame to be serialized.

Result:
    >>> ' A  B

1 4 2 5 3 6 '

Source code in pyswap/core/serializers.py 
25
26
27
28
29
30
31
32
33
34
def serialize_table(table: DataFrame) -> str:
    """Convert the DataFrame to a string.

    Arguments:
        table: The DataFrame to be serialized.

    Result:
        >>> ' A  B\n 1  4\n 2  5\n 3  6\n'
    """
    return f"{table.to_string(index=False)}\n"

I/O

Interact with the filesystem

All functions that interact with the filesystem are located in this subpackage.

Modules:

Name Description
io_ascii

Functions to interact with ASCII files.
io_yaml

Functions to interact with YAML files.
classic_swap

Functions to load with classic SWAP input files.

io_ascii

Interact with ASCII files.

Functions:

Name Description
open_ascii

Open an ASCII file and detect its encoding.
save_ascii

Save a string to an ASCII file.

open_ascii(file_path)

Open file and detect encoding.

Parameters:

Name Type Description Default
file_path str

Path to the file to be opened.

 required
Source code in pyswap/core/io/io_ascii.py 
13
14
15
16
17
18
19
20
21
22
23
def open_ascii(file_path: Path) -> str:
    """Open file and detect encoding.

    Arguments:
        file_path (str): Path to the file to be opened.
    """
    with open(file_path, "rb") as f:
        raw_data = f.read()
    encoding = chardet.detect(raw_data)["encoding"]

    return raw_data.decode(encoding)

save_ascii(string, fname, path, mode='w', extension=None, encoding='ascii')

Saves a string to a file with a given extension.

Parameters:

Name Type Description Default
string str

The string to be saved to a file.

 required
extension str

The extension that the file should have (e.g. 'txt', 'csv', etc.).

 None
fname str

The name of the file.

 required
path str

The path where the file should be saved.

 required
mode str

The mode in which the file should be opened (e.g. 'w' for write, 'a' for append, etc.).

 'w'
encoding str

The encoding to use for the file (default is 'ascii').

 'ascii'

Returns:

Type Description
None

None
Source code in pyswap/core/io/io_ascii.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
def save_ascii(
    string: str,
    fname: str,
    path: str,
    mode: str = "w",
    extension: str | None = None,
    encoding: str = "ascii",
) -> None:
    """
    Saves a string to a file with a given extension.

    Parameters:
        string (str): The string to be saved to a file.
        extension (str): The extension that the file should have (e.g. 'txt', 'csv', etc.).
        fname (str): The name of the file.
        path (str): The path where the file should be saved.
        mode (str): The mode in which the file should be opened (e.g. 'w' for write, 'a' for append, etc.).
        encoding (str): The encoding to use for the file (default is 'ascii').

    Returns:
        None
    """

    if extension is not None:
        fname = f"{fname}.{extension}"

    with open(f"{path}/{fname}", f"{mode}", encoding=f"{encoding}") as f:
        f.write(string)

io_yaml

Interact with YAML files.

Functions:

Name Description
load_yaml

Load a YAML file.

load_yaml(file)

Load a YAML file.

Parameters:

Name Type Description Default
file Path

Path to the YAML file.

 required
Source code in pyswap/core/io/io_yaml.py 
12
13
14
15
16
17
18
19
20
21
def load_yaml(file: Path) -> dict:
    """Load a YAML file.

    Arguments:
        file: Path to the YAML file.
    """
    with open(file) as file:
        content: dict = yaml.safe_load(file)

    return content

CLI

Command Line Interface for pySWAP.

This is a prototype subpackage for potential enhancement of pyswap's functionality. CLI tools can be very helpful in automating some tasks, like loading databases or classic SWAP models.

Note

At the moment only creating project structure was prototyped. More functionality will be added in the future if users express such need.

Example:

```cmd
pyswap init --notebook  # creates the project structure with a template .ipynb file.
pyswap init --script  # creates the project structure with a .py file.
```

After running the script, you will see the following folder created:

test project
├── README
├── __init__.py
├── data
├── models
│   ├── __init__.py
│   └── main.ipynb
└── scripts
    └── __init__.py

The __init__.py files are added to create a module structure. Now when you create a python file in scripts with some helper functions, you can import those functions to the main model script or notebook and use it there.

from ..scripts.helper_module import helper_function

var = helper_function(**kwargs)

By default, a git repository is also created along with the project structure.

cli

The cli module is supposed to help in structuring the direcotries of created models and enforce best practices in documenting. It creates a modular structure (with init.py files) what can be helpful when writing scripts. This way, modules from the scripts can be directly imported into the main.py or main.ipynb

init(script=False, notebook=True)

Prompt the user to enter their information and create a User class.

Source code in pyswap/core/cli/cli.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
@app.command()
def init(script: bool = False, notebook: bool = True):
    """Prompt the user to enter their information and create a User class."""
    attrs = {
        "project": typer.prompt("Project name"),
        "swap_ver": typer.prompt("SWAP version used"),
        "author": typer.prompt("Author first/last name"),
        "institution": typer.prompt("Your last institution"),
        "email": typer.prompt("Your email address"),
        "comment": typer.prompt("Any comments?", default=None),
    }

    folder_name = typer.prompt("Choose a folder name", default=attrs.get("project"))

    # Defining paths and creating folders.
    templates_path = Path(__file__).resolve().parent / "templates"
    project_root = Path.cwd() / folder_name

    basic_code_to_write_path = templates_path / "script.txt"
    basic_code_to_write = dict_to_custom_string(attrs)

    folders_to_create = ["models", "scripts", "data"]
    folders_to_create_paths = [project_root / folder for folder in folders_to_create]

    [folder.mkdir(parents=True, exist_ok=True) for folder in folders_to_create_paths]

    # Dealing with files.
    copy_readme(templates_path, project_root)
    create_inits(
        project_root=project_root,
        models_dir=folders_to_create_paths[0],
        scripts_dir=folders_to_create_paths[1],
    )

    if script:
        make_script(
            folders_to_create_paths[0], basic_code_to_write_path, basic_code_to_write
        )

    if notebook:
        make_notebook(
            folders_to_create_paths[0], basic_code_to_write, templates_path, attrs
        )

    init_git_repo(project_root)