Skip to content

TxtinoutReader

TxtinoutReader

TxtinoutReader

Source code in pySWATPlus/TxtinoutReader.py
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 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
 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
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
200
201
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
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
class TxtinoutReader:

    RESERVED_PARAMS: typing.Final[list[str]] = ['has_units']
    IGNORED_FILE_PATTERNS: typing.Final[tuple[str, ...]] = tuple(
        f'_{suffix}.{ext}'
        for suffix in ('day', 'mon', 'yr', 'aa')
        for ext in ('txt', 'csv')
    )

    def __init__(
        self,
        path: str | Path
    ) -> None:
        """
        Initialize a TxtinoutReader instance for working with SWAT model data.

        Args:
            path (str or Path): The path to the SWAT model folder.

        Raises:
            TypeError: If the provided path is not a string or Path object,
                    if more than one .exe file is found,
                    or if no .exe file is found.
            FileNotFoundError: If the folder does not exist.
        """

        # check if path is a string or a path
        if not isinstance(path, (str, Path)):
            raise TypeError("path must be a string or os.PathLike object")

        path = Path(path).resolve()

        # check if folder exists
        if not path.is_dir():
            raise FileNotFoundError("Folder does not exist")

        # check .exe files in the directory
        exe_list = [file for file in path.iterdir() if file.suffix == ".exe"]

        # raise error if empty list
        if not exe_list:
            raise TypeError(".exe not found in parent folder")

        # raise error if more than one .exe file
        if len(exe_list) > 1:
            raise TypeError("More than one .exe file found in the parent folder")

        # find parent directory
        self.root_folder: Path = path
        self.swat_exe_path: Path = path / exe_list[0]

    def enable_object_in_print_prt(
        self,
        obj: str,
        daily: bool,
        monthly: bool,
        yearly: bool,
        avann: bool
    ) -> None:
        """
        Enable or update an object in the 'print.prt' file. If obj is not a default identifier, it will be added at the end of the file.

        Args:
            obj (str): The object name or identifier.
            daily (bool): Flag for daily print frequency.
            monthly (bool): Flag for monthly print frequency.
            yearly (bool): Flag for yearly print frequency.
            avann (bool): Flag for average annual print frequency.

        Returns:
            None
        """

        # check if obj is object itself or file
        if Path(obj).suffix:
            arg_to_add = obj.rsplit('_', maxsplit=1)[0]
        else:
            arg_to_add = obj

        # read all print_prt file, line by line
        print_prt_path = self.root_folder / 'print.prt'
        new_print_prt = ""
        found = False
        with open(print_prt_path) as file:
            for line in file:
                if not line.startswith(arg_to_add + ' '):  # Line must start exactly with arg_to_add, not a word that starts with arg_to_add
                    new_print_prt += line
                else:
                    # obj already exist, replace it in same position
                    new_print_prt += _build_line_to_add(arg_to_add, daily, monthly, yearly, avann)
                    found = True

        if not found:
            new_print_prt += _build_line_to_add(arg_to_add, daily, monthly, yearly, avann)

        # store new print_prt
        with open(print_prt_path, 'w') as file:
            file.write(new_print_prt)

    def set_begin_and_end_year(
        self,
        begin: int,
        end: int
    ) -> None:
        """
        Modify the beginning and end year in the 'time.sim' file.

        Parameters:
            beginning (int): The new beginning year.
            end (int): The new end year.

        Returns:
            None
        """

        nth_line = 3

        time_sim_path = self.root_folder / 'time.sim'

        # Open the file in read mode and read its contents
        with open(time_sim_path, 'r') as file:
            lines = file.readlines()

        year_line = lines[nth_line - 1]

        # Split the input string by spaces
        elements = year_line.split()

        # insert years
        elements[1] = str(begin)
        elements[3] = str(end)

        # Reconstruct the result string while maintaining spaces
        result_string = '{: >8} {: >10} {: >10} {: >10} {: >10} \n'.format(*elements)

        lines[nth_line - 1] = result_string

        with open(time_sim_path, 'w') as file:
            file.writelines(lines)

    def set_warmup_year(
        self,
        warmup: int
    ) -> None:
        """
        Modify the warmup period in the 'time.sim' file.

        Args:
            warmup (int): The new warmup period value.

        Returns:
            None
        """
        time_sim_path = self.root_folder / 'print.prt'

        # Open the file in read mode and read its contents
        with open(time_sim_path, 'r') as file:
            lines = file.readlines()

        nth_line = 3
        year_line = lines[nth_line - 1]

        # Split the input string by spaces
        elements = year_line.split()

        # Modify warmup year
        elements[0] = str(warmup)

        # Reconstruct the result string while maintaining spaces
        result_string = '{: <12} {: <11} {: <11} {: <10} {: <10} {: <10} \n'.format(*elements)

        lines[nth_line - 1] = result_string

        with open(time_sim_path, 'w') as file:
            file.writelines(lines)

    def _enable_disable_csv_print(
        self,
        enable: bool = True
    ) -> None:
        """
        Enable or disable CSV print in the 'print.prt' file.

        Parameters:
            enable (bool): True to enable CSV print, False to disable (default is True).

        Returns:
            None
        """

        # read
        nth_line = 7

        # time_sim_path = f"{self.root_folder}\\{'time.sim'}"
        print_prt_path = self.root_folder / 'print.prt'

        # Open the file in read mode and read its contents
        with open(print_prt_path, 'r') as file:
            lines = file.readlines()

        if enable:
            lines[nth_line - 1] = 'y' + lines[nth_line - 1][1:]
        else:
            lines[nth_line - 1] = 'n' + lines[nth_line - 1][1:]

        with open(print_prt_path, 'w') as file:
            file.writelines(lines)

    def enable_csv_print(
        self
    ) -> None:
        """
        Enable CSV print in the 'print.prt' file.

        Returns:
            None
        """

        self._enable_disable_csv_print(enable=True)

    def disable_csv_print(
        self
    ) -> None:
        """
        Disable CSV print in the 'print.prt' file.

        Returns:
            None
        """

        self._enable_disable_csv_print(enable=False)

    def register_file(
        self,
        filename: str,
        has_units: bool = False,
        usecols: typing.Optional[list[str]] = None,
        filter_by: typing.Optional[str] = None
    ) -> FileReader:
        """
        Register a file to work with in the SWAT model.

        Parameters:
            filename (str): The name of the file to register.
            has_units (bool): Indicates if the file has units information (default is False).
            usecols (List[str], optional): A list of column names to read (default is None).
            filter_by (str, optional): Pandas query string to select applicable rows (default is None).

        Returns:
            FileReader: A FileReader instance for the registered file.
        """
        file_path = self.root_folder / filename

        return FileReader(file_path, has_units, usecols, filter_by)

    def _copy_swat(
        self,
        target_dir: str | Path,
    ) -> str:
        """
        Prepare a working directory containing the necessary SWAT model files.

        This function copies the contents of the SWAT model input folder (`self.root_folder`)
        to a target directory.

        Parameters:
            target_dir (str or Path): Destination directory for the SWAT model files.

        Returns:
            str: The path to the directory where the SWAT files were copied.
        """

        dest_path = Path(target_dir)

        # Copy files from source folder
        for file in self.root_folder.iterdir():
            if file.is_dir() or file.name.endswith(self.IGNORED_FILE_PATTERNS):
                continue
            shutil.copy2(file, dest_path / file.name)

        return str(dest_path)

    def _run_swat(
        self,
    ) -> None:
        """
        Run the SWAT simulation.

        Returns:
            None

        Raises:
            subprocess.CalledProcessError: If the SWAT executable fails
            FileNotFoundError: If the SWAT executable is not found
        """

        # Run simulation
        try:
            process = subprocess.Popen(
                [str(self.swat_exe_path.resolve())],
                cwd=str(self.root_folder.resolve()),  # Sets working dir just for this subprocess
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                bufsize=1,  # Line buffered
                text=True   # Handles text output
            )

            # Real-time output handling
            for line in process.stdout:
                clean_line = line.strip()
                if clean_line:
                    logger.info(clean_line)

            # Wait for process and check for errors
            return_code = process.wait()
            if return_code != 0:
                stderr = process.stderr.read()
                raise subprocess.CalledProcessError(
                    return_code,
                    process.args,
                    stderr=stderr
                )

        except FileNotFoundError:
            logger.error(f"SWAT executable not found: {self.swat_exe_path}")
            raise
        except Exception as e:
            logger.error(f"Failed to run SWAT: {str(e)}")
            raise

    def run_swat(
        self,
        params: ParamsType = None,
    ) -> str:
        """
        Run the SWAT simulation with optional parameter changes.

        Args:
            params (ParamsType, optional): Nested dictionary specifying parameter changes to apply.

                The `params` dictionary should follow this structure:

                ```python
                params = {
                    "<input_file>": {
                        "has_units": bool,              # Optional. Whether the file has units information (default is False)
                        "<parameter_name>": [           # One or more changes to apply to the parameter
                            {
                                "value": float,         # New value to assign
                                "change_type": str,     # (Optional) One of: 'absval' (default), 'abschg', 'pctchg'
                                "filter_by": str        # (Optional) pandas `.query()` filter string to select rows
                            },
                            # ... more changes
                        ]
                    },
                    # ... more input files
                }
                ```

        Returns:
            str: The path where the SWAT simulation was executed.

        Example:
            ```python
            params = {
                'plants.plt': {
                    'has_units': False,
                    'bm_e': [
                        {'value': 100, 'change_type': 'absval', 'filter_by': 'name == "agrl"'},
                        {'value': 110, 'change_type': 'absval', 'filter_by': 'name == "almd"'},
                    ],
                },
            }

            reader.run_swat(params)
            ```
        """
        _params = params or {}

        _validate_params(_params)

        # Modify files for simulation
        for filename, file_params in _params.items():
            has_units = file_params.get('has_units', False)
            file = self.register_file(
                filename,
                has_units=has_units,
            )
            df = file.df

            for param_name, param_spec in file_params.items():
                if param_name in self.RESERVED_PARAMS:
                    continue  # Skip reserved parameters

                # Normalize to list of changes
                changes = param_spec if isinstance(param_spec, list) else [param_spec]

                # Process each parameter change
                for change in changes:
                    _apply_param_change(df, param_name, change)

            # Store the modified file
            file.overwrite_file()

        # run simulation
        self._run_swat()
        return self.root_folder

    def run_swat_in_other_dir(
        self,
        target_dir: str | Path,
        params: ParamsType = None,
    ) -> str:
        """
        Copy the SWAT model files to a specified directory, modify input parameters, and run the simulation.

        Args:
            target_dir (str or Path): Path to the directory where the SWAT model files will be copied.

            params (ParamsType, optional): Nested dictionary specifying parameter changes.

                The `params` dictionary should follow this structure:

                ```python
                params = {
                    "<input_file>": {
                        "has_units": bool,              # Optional. Whether the file has units information (default is False)
                        "<parameter_name>": [           # One or more changes to apply to the parameter
                            {
                                "value": float,         # New value to assign
                                "change_type": str,     # (Optional) One of: 'absval' (default), 'abschg', 'pctchg'
                                "filter_by": str        # (Optional) pandas `.query()` filter string to select rows
                            },
                            # ... more changes
                        ]
                    },
                    # ... more input files
                }
                ```

        Returns:
            str: The path to the directory where the SWAT simulation was executed.

        Example:
            ```python
            params = {
                'plants.plt': {
                    'has_units': False,
                    'bm_e': [
                        {'value': 100, 'change_type': 'absval', 'filter_by': 'name == "agrl"'},
                        {'value': 110, 'change_type': 'absval', 'filter_by': 'name == "almd"'},
                    ],
                },
            }

            with tempfile.TemporaryDirectory() as tmp_dir:
                simulation = pySWATPlus.TxtinoutReader.run_swat_in_other_dir(
                    target_dir=tmp_dir,
                    params=params
                )
            ```
        """
        # Validate target_dir
        if not isinstance(target_dir, (str, Path)):
            raise TypeError("target_dir must be a string or Path object")

        target_dir = Path(target_dir).resolve()

        # Create the directory if it does not exist
        target_dir.mkdir(parents=True, exist_ok=True)

        tmp_path = self._copy_swat(target_dir=target_dir)
        reader = TxtinoutReader(tmp_path)

        return reader.run_swat(params)

__init__(path: str | Path) -> None

Initialize a TxtinoutReader instance for working with SWAT model data.

Parameters:

Name Type Description Default
path str or Path

The path to the SWAT model folder.

required

Raises:

Type Description
TypeError

If the provided path is not a string or Path object, if more than one .exe file is found, or if no .exe file is found.

FileNotFoundError

If the folder does not exist.

Source code in pySWATPlus/TxtinoutReader.py
22
23
24
25
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
54
55
56
57
58
59
60
61
62
def __init__(
    self,
    path: str | Path
) -> None:
    """
    Initialize a TxtinoutReader instance for working with SWAT model data.

    Args:
        path (str or Path): The path to the SWAT model folder.

    Raises:
        TypeError: If the provided path is not a string or Path object,
                if more than one .exe file is found,
                or if no .exe file is found.
        FileNotFoundError: If the folder does not exist.
    """

    # check if path is a string or a path
    if not isinstance(path, (str, Path)):
        raise TypeError("path must be a string or os.PathLike object")

    path = Path(path).resolve()

    # check if folder exists
    if not path.is_dir():
        raise FileNotFoundError("Folder does not exist")

    # check .exe files in the directory
    exe_list = [file for file in path.iterdir() if file.suffix == ".exe"]

    # raise error if empty list
    if not exe_list:
        raise TypeError(".exe not found in parent folder")

    # raise error if more than one .exe file
    if len(exe_list) > 1:
        raise TypeError("More than one .exe file found in the parent folder")

    # find parent directory
    self.root_folder: Path = path
    self.swat_exe_path: Path = path / exe_list[0]

disable_csv_print() -> None

Disable CSV print in the 'print.prt' file.

Returns:

Type Description
None

None

Source code in pySWATPlus/TxtinoutReader.py
233
234
235
236
237
238
239
240
241
242
243
def disable_csv_print(
    self
) -> None:
    """
    Disable CSV print in the 'print.prt' file.

    Returns:
        None
    """

    self._enable_disable_csv_print(enable=False)

enable_csv_print() -> None

Enable CSV print in the 'print.prt' file.

Returns:

Type Description
None

None

Source code in pySWATPlus/TxtinoutReader.py
221
222
223
224
225
226
227
228
229
230
231
def enable_csv_print(
    self
) -> None:
    """
    Enable CSV print in the 'print.prt' file.

    Returns:
        None
    """

    self._enable_disable_csv_print(enable=True)

enable_object_in_print_prt(obj: str, daily: bool, monthly: bool, yearly: bool, avann: bool) -> None

Enable or update an object in the 'print.prt' file. If obj is not a default identifier, it will be added at the end of the file.

Parameters:

Name Type Description Default
obj str

The object name or identifier.

required
daily bool

Flag for daily print frequency.

required
monthly bool

Flag for monthly print frequency.

required
yearly bool

Flag for yearly print frequency.

required
avann bool

Flag for average annual print frequency.

required

Returns:

Type Description
None

None

Source code in pySWATPlus/TxtinoutReader.py
 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
def enable_object_in_print_prt(
    self,
    obj: str,
    daily: bool,
    monthly: bool,
    yearly: bool,
    avann: bool
) -> None:
    """
    Enable or update an object in the 'print.prt' file. If obj is not a default identifier, it will be added at the end of the file.

    Args:
        obj (str): The object name or identifier.
        daily (bool): Flag for daily print frequency.
        monthly (bool): Flag for monthly print frequency.
        yearly (bool): Flag for yearly print frequency.
        avann (bool): Flag for average annual print frequency.

    Returns:
        None
    """

    # check if obj is object itself or file
    if Path(obj).suffix:
        arg_to_add = obj.rsplit('_', maxsplit=1)[0]
    else:
        arg_to_add = obj

    # read all print_prt file, line by line
    print_prt_path = self.root_folder / 'print.prt'
    new_print_prt = ""
    found = False
    with open(print_prt_path) as file:
        for line in file:
            if not line.startswith(arg_to_add + ' '):  # Line must start exactly with arg_to_add, not a word that starts with arg_to_add
                new_print_prt += line
            else:
                # obj already exist, replace it in same position
                new_print_prt += _build_line_to_add(arg_to_add, daily, monthly, yearly, avann)
                found = True

    if not found:
        new_print_prt += _build_line_to_add(arg_to_add, daily, monthly, yearly, avann)

    # store new print_prt
    with open(print_prt_path, 'w') as file:
        file.write(new_print_prt)

register_file(filename: str, has_units: bool = False, usecols: typing.Optional[list[str]] = None, filter_by: typing.Optional[str] = None) -> FileReader

Register a file to work with in the SWAT model.

Parameters:

Name Type Description Default
filename str

The name of the file to register.

required
has_units bool

Indicates if the file has units information (default is False).

False
usecols List[str]

A list of column names to read (default is None).

None
filter_by str

Pandas query string to select applicable rows (default is None).

None

Returns:

Name Type Description
FileReader FileReader

A FileReader instance for the registered file.

Source code in pySWATPlus/TxtinoutReader.py
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
def register_file(
    self,
    filename: str,
    has_units: bool = False,
    usecols: typing.Optional[list[str]] = None,
    filter_by: typing.Optional[str] = None
) -> FileReader:
    """
    Register a file to work with in the SWAT model.

    Parameters:
        filename (str): The name of the file to register.
        has_units (bool): Indicates if the file has units information (default is False).
        usecols (List[str], optional): A list of column names to read (default is None).
        filter_by (str, optional): Pandas query string to select applicable rows (default is None).

    Returns:
        FileReader: A FileReader instance for the registered file.
    """
    file_path = self.root_folder / filename

    return FileReader(file_path, has_units, usecols, filter_by)

run_swat(params: ParamsType = None) -> str

Run the SWAT simulation with optional parameter changes.

Parameters:

Name Type Description Default
params ParamsType

Nested dictionary specifying parameter changes to apply.

The params dictionary should follow this structure:

params = {
    "<input_file>": {
        "has_units": bool,              # Optional. Whether the file has units information (default is False)
        "<parameter_name>": [           # One or more changes to apply to the parameter
            {
                "value": float,         # New value to assign
                "change_type": str,     # (Optional) One of: 'absval' (default), 'abschg', 'pctchg'
                "filter_by": str        # (Optional) pandas `.query()` filter string to select rows
            },
            # ... more changes
        ]
    },
    # ... more input files
}
None

Returns:

Name Type Description
str str

The path where the SWAT simulation was executed.

Example
params = {
    'plants.plt': {
        'has_units': False,
        'bm_e': [
            {'value': 100, 'change_type': 'absval', 'filter_by': 'name == "agrl"'},
            {'value': 110, 'change_type': 'absval', 'filter_by': 'name == "almd"'},
        ],
    },
}

reader.run_swat(params)
Source code in pySWATPlus/TxtinoutReader.py
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
def run_swat(
    self,
    params: ParamsType = None,
) -> str:
    """
    Run the SWAT simulation with optional parameter changes.

    Args:
        params (ParamsType, optional): Nested dictionary specifying parameter changes to apply.

            The `params` dictionary should follow this structure:

            ```python
            params = {
                "<input_file>": {
                    "has_units": bool,              # Optional. Whether the file has units information (default is False)
                    "<parameter_name>": [           # One or more changes to apply to the parameter
                        {
                            "value": float,         # New value to assign
                            "change_type": str,     # (Optional) One of: 'absval' (default), 'abschg', 'pctchg'
                            "filter_by": str        # (Optional) pandas `.query()` filter string to select rows
                        },
                        # ... more changes
                    ]
                },
                # ... more input files
            }
            ```

    Returns:
        str: The path where the SWAT simulation was executed.

    Example:
        ```python
        params = {
            'plants.plt': {
                'has_units': False,
                'bm_e': [
                    {'value': 100, 'change_type': 'absval', 'filter_by': 'name == "agrl"'},
                    {'value': 110, 'change_type': 'absval', 'filter_by': 'name == "almd"'},
                ],
            },
        }

        reader.run_swat(params)
        ```
    """
    _params = params or {}

    _validate_params(_params)

    # Modify files for simulation
    for filename, file_params in _params.items():
        has_units = file_params.get('has_units', False)
        file = self.register_file(
            filename,
            has_units=has_units,
        )
        df = file.df

        for param_name, param_spec in file_params.items():
            if param_name in self.RESERVED_PARAMS:
                continue  # Skip reserved parameters

            # Normalize to list of changes
            changes = param_spec if isinstance(param_spec, list) else [param_spec]

            # Process each parameter change
            for change in changes:
                _apply_param_change(df, param_name, change)

        # Store the modified file
        file.overwrite_file()

    # run simulation
    self._run_swat()
    return self.root_folder

run_swat_in_other_dir(target_dir: str | Path, params: ParamsType = None) -> str

Copy the SWAT model files to a specified directory, modify input parameters, and run the simulation.

Parameters:

Name Type Description Default
target_dir str or Path

Path to the directory where the SWAT model files will be copied.

required
params ParamsType

Nested dictionary specifying parameter changes.

The params dictionary should follow this structure:

params = {
    "<input_file>": {
        "has_units": bool,              # Optional. Whether the file has units information (default is False)
        "<parameter_name>": [           # One or more changes to apply to the parameter
            {
                "value": float,         # New value to assign
                "change_type": str,     # (Optional) One of: 'absval' (default), 'abschg', 'pctchg'
                "filter_by": str        # (Optional) pandas `.query()` filter string to select rows
            },
            # ... more changes
        ]
    },
    # ... more input files
}
None

Returns:

Name Type Description
str str

The path to the directory where the SWAT simulation was executed.

Example
params = {
    'plants.plt': {
        'has_units': False,
        'bm_e': [
            {'value': 100, 'change_type': 'absval', 'filter_by': 'name == "agrl"'},
            {'value': 110, 'change_type': 'absval', 'filter_by': 'name == "almd"'},
        ],
    },
}

with tempfile.TemporaryDirectory() as tmp_dir:
    simulation = pySWATPlus.TxtinoutReader.run_swat_in_other_dir(
        target_dir=tmp_dir,
        params=params
    )
Source code in pySWATPlus/TxtinoutReader.py
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
def run_swat_in_other_dir(
    self,
    target_dir: str | Path,
    params: ParamsType = None,
) -> str:
    """
    Copy the SWAT model files to a specified directory, modify input parameters, and run the simulation.

    Args:
        target_dir (str or Path): Path to the directory where the SWAT model files will be copied.

        params (ParamsType, optional): Nested dictionary specifying parameter changes.

            The `params` dictionary should follow this structure:

            ```python
            params = {
                "<input_file>": {
                    "has_units": bool,              # Optional. Whether the file has units information (default is False)
                    "<parameter_name>": [           # One or more changes to apply to the parameter
                        {
                            "value": float,         # New value to assign
                            "change_type": str,     # (Optional) One of: 'absval' (default), 'abschg', 'pctchg'
                            "filter_by": str        # (Optional) pandas `.query()` filter string to select rows
                        },
                        # ... more changes
                    ]
                },
                # ... more input files
            }
            ```

    Returns:
        str: The path to the directory where the SWAT simulation was executed.

    Example:
        ```python
        params = {
            'plants.plt': {
                'has_units': False,
                'bm_e': [
                    {'value': 100, 'change_type': 'absval', 'filter_by': 'name == "agrl"'},
                    {'value': 110, 'change_type': 'absval', 'filter_by': 'name == "almd"'},
                ],
            },
        }

        with tempfile.TemporaryDirectory() as tmp_dir:
            simulation = pySWATPlus.TxtinoutReader.run_swat_in_other_dir(
                target_dir=tmp_dir,
                params=params
            )
        ```
    """
    # Validate target_dir
    if not isinstance(target_dir, (str, Path)):
        raise TypeError("target_dir must be a string or Path object")

    target_dir = Path(target_dir).resolve()

    # Create the directory if it does not exist
    target_dir.mkdir(parents=True, exist_ok=True)

    tmp_path = self._copy_swat(target_dir=target_dir)
    reader = TxtinoutReader(tmp_path)

    return reader.run_swat(params)

set_begin_and_end_year(begin: int, end: int) -> None

Modify the beginning and end year in the 'time.sim' file.

Parameters:

Name Type Description Default
beginning int

The new beginning year.

required
end int

The new end year.

required

Returns:

Type Description
None

None

Source code in pySWATPlus/TxtinoutReader.py
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
def set_begin_and_end_year(
    self,
    begin: int,
    end: int
) -> None:
    """
    Modify the beginning and end year in the 'time.sim' file.

    Parameters:
        beginning (int): The new beginning year.
        end (int): The new end year.

    Returns:
        None
    """

    nth_line = 3

    time_sim_path = self.root_folder / 'time.sim'

    # Open the file in read mode and read its contents
    with open(time_sim_path, 'r') as file:
        lines = file.readlines()

    year_line = lines[nth_line - 1]

    # Split the input string by spaces
    elements = year_line.split()

    # insert years
    elements[1] = str(begin)
    elements[3] = str(end)

    # Reconstruct the result string while maintaining spaces
    result_string = '{: >8} {: >10} {: >10} {: >10} {: >10} \n'.format(*elements)

    lines[nth_line - 1] = result_string

    with open(time_sim_path, 'w') as file:
        file.writelines(lines)

set_warmup_year(warmup: int) -> None

Modify the warmup period in the 'time.sim' file.

Parameters:

Name Type Description Default
warmup int

The new warmup period value.

required

Returns:

Type Description
None

None

Source code in pySWATPlus/TxtinoutReader.py
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
def set_warmup_year(
    self,
    warmup: int
) -> None:
    """
    Modify the warmup period in the 'time.sim' file.

    Args:
        warmup (int): The new warmup period value.

    Returns:
        None
    """
    time_sim_path = self.root_folder / 'print.prt'

    # Open the file in read mode and read its contents
    with open(time_sim_path, 'r') as file:
        lines = file.readlines()

    nth_line = 3
    year_line = lines[nth_line - 1]

    # Split the input string by spaces
    elements = year_line.split()

    # Modify warmup year
    elements[0] = str(warmup)

    # Reconstruct the result string while maintaining spaces
    result_string = '{: <12} {: <11} {: <11} {: <10} {: <10} {: <10} \n'.format(*elements)

    lines[nth_line - 1] = result_string

    with open(time_sim_path, 'w') as file:
        file.writelines(lines)