commandtools

This module implements some main features for using HydPy from your command line tools via script hyd.

Module commandtools implements the following members:

hydpy.exe.commandtools.run_subprocess(command: str, verbose: bool = True, blocking: bool = True) → Union[subprocess.CompletedProcess[str], subprocess.Popen[str]][source]

Execute the given command in a new process.

Only when both verbose and blocking are True, run_subprocess() prints all responses to the current value of stdout:

>>> from hydpy import run_subprocess
>>> import platform
>>> esc = "" if "windows" in platform.platform().lower() else "\\"
>>> result = run_subprocess(f"python -c print{esc}(1+1{esc})")
2

With verbose being False, run_subprocess() does never print out anything:

>>> result = run_subprocess(f"python -c print{esc}(1+1{esc})", verbose=False)

>>> process = run_subprocess("python", blocking=False, verbose=False)
>>> process.kill()
>>> _ = process.communicate()

When verbose is True and blocking is False, run_subprocess() prints all responses to the console (“invisible” for doctests):

>>> process = run_subprocess("python", blocking=False)
>>> process.kill()
>>> _ = process.communicate()
hydpy.exe.commandtools.exec_commands(commands: str, **parameters: Any) → None[source]

Execute the given Python commands.

Function exec_commands() is thought for testing purposes only (see the main documentation on module hyd). Separate individual commands by semicolons and replaced whitespaces with underscores:

>>> from hydpy.exe.commandtools import exec_commands
>>> import sys
>>> exec_commands("x_=_1+1;print(x)")
Start to execute the commands ['x_=_1+1', 'print(x)'] for testing purposes.
2

exec_commands() interprets double underscores as a single underscores:

>>> exec_commands("x_=_1;print(x.____class____)")
Start to execute the commands ['x_=_1', 'print(x.____class____)'] for testing purposes.
<class 'int'>

exec_commands() evaluates additional keyword arguments before it executes the given commands:

>>> exec_commands("e=x==y;print(e)", x=1, y=2)
Start to execute the commands ['e=x==y', 'print(e)'] for testing purposes.
False
hydpy.exe.commandtools.test_everything() → int[source]

Execute script test_everything.py from remote.

Whenever in doubt about the functioning of your HydPy installation, call script function test_everything(). It executes all tests employed before your actual HydPy version was released, which allows you to check for possible incompatibilities with the site-packages or the configuration of your specific system.

Actually, test_everything() only makes the following system call and returns its exit code:

>>> from hydpy import test_everything, repr_
>>> from unittest import mock
>>> with mock.patch("os.system", return_value=1) as system:
...     test_everything()
1
>>> repr_(system.call_args[0][0])   
'.../python... .../hydpy/tests/test_everything.py forcecompiling=False'
hydpy.exe.commandtools.exec_script(filepath: str) → None[source]

Execute an arbitrary Python script.

Function run_simulation() allows you to execute a predefined HydPy workflow. You can configure many details of this workflow but not change its general structure. Use function exec_script() when you want to execute HydPy remotely but strive for more flexibility. As its name suggests, function exec_script() executes any valid Python code relying on the standard library and the available site-packages.

Function exec_script() requires the name of the script to be executed as a single argument:

>>> from hydpy import print_latest_logfile, Node, TestIO, run_subprocess
>>> TestIO.clear()
>>> with TestIO():
...     result = run_subprocess('hyd.py logfile="default" exec_script temp.py')
...     print_latest_logfile()    
Invoking hyd.py with arguments `logfile=default, exec_script, temp.py` resulted in the following error:
File `...temp.py` does not exist.
...

Function exec_script() can use all HydPy features. As a simple example, we write a Python script that initialises a Node object and prints its string representation (into the log file):

>>> with TestIO():
...     with open("temp.py", "w") as file_:
...         _ = file_.write("from hydpy import Node\n")
...         _ = file_.write('print(repr(Node("valid_name")))\n')
...     result = run_subprocess('hyd.py logfile="default" exec_script temp.py')
...     print_latest_logfile()
Node("valid_name", variable="Q")

Errors are reported as usual:

>>> with TestIO():
...     with open("temp.py", "w") as file_:
...         _ = file_.write("from hydpy import Node\n")
...         _ = file_.write('print(repr(Node("invalid name")))\n')
...     result = run_subprocess('hyd.py logfile="default" exec_script temp.py')
...     print_latest_logfile()    
Invoking hyd.py with arguments `logfile=default, exec_script, temp.py` resulted in the following error:
While trying to initialize a `Node` object with value `invalid name` of type `str`, the following error occurred: The given name string `invalid name` does not define a valid variable identifier.  Valid identifiers do not contain characters like `-` or empty spaces, do not start with numbers, cannot be mistaken with Python built-ins like `for`...)
...
hydpy.exe.commandtools.start_shell(filepath: str = '') → None[source]

Open an interactive Python shell.

Writing “hyd.py start_shell” into your command line tool opens an interactive Python console with the most relevant HydPy features being imported already. In our first example, we directly prepare an Element object (without needing to import class Element first) and print its string representation:

>>> import subprocess
>>> from hydpy import TestIO
>>> TestIO.clear()
>>> with TestIO():
...     with subprocess.Popen(
...             "hyd.py start_shell",
...             stdin=subprocess.PIPE,
...             stdout=subprocess.PIPE,
...             stderr=subprocess.PIPE,
...             encoding="utf-8",
...             shell=True) as process:
...         response = process.communicate(
...             'print(repr(Element("e1", outlets="n1")))')
...         print(response[0])
Element("e1",
        outlets="n1")

You can pass the name of a Python file as an additional argument, which enables to interact with the results of the file. For demonstration purposes, we create the example file test.py simply defining a Nodes object handling two individual nodes:

>>> with TestIO():
...     with open("test.py", "w") as file_:
...         _ = file_.write("from hydpy import Nodes\n")
...         _ = file_.write('nodes = Nodes("n1", "n2")\n')

Now we can, execute this file and, for example, query the names of the defined nodes interactively:

>>> with TestIO():
...     with subprocess.Popen(
...             "hyd.py start_shell test.py",
...             stdin=subprocess.PIPE,
...             stdout=subprocess.PIPE,
...             stderr=subprocess.PIPE,
...             encoding="utf-8",
...             shell=True) as process:
...         response = process.communicate(
...             "print(nodes.names)")
...         print(response[0])
('n1', 'n2')
hydpy.exe.commandtools.print_latest_logfile(dirpath: str = '.', wait: float = 0.0) → None[source]

Print the latest log file in the current or the given working directory.

When executing processes in parallel, print_latest_logfile() may be called before any log file exists. Then pass an appropriate number of seconds to the argument wait. print_latest_logfile() then prints the contents of the latest log file, as soon as it finds one. Function print_latest_logfile() works only for “default” logfile names, as described in the documentation on function prepare_logfile().

>>> from hydpy import TestIO, print_latest_logfile, run_subprocess
>>> TestIO.clear()
>>> with TestIO():
...     result = run_subprocess("hyd.py")
...     print_latest_logfile(wait=0.5)    
Traceback (most recent call last):
...
FileNotFoundError: Cannot find a default HydPy log file in directory ...iotesting.

>>> with TestIO():
...     result1 = run_subprocess('hyd.py logfile="default" test=1')
...     result2 = run_subprocess('hyd.py logfile="default" test=2')
...     print_latest_logfile(wait=0.5)    
Invoking hyd.py with arguments `logfile=default, test=2` resulted in the following error:
...
hydpy.exe.commandtools.prepare_logfile(filename: str) → str[source]

Prepare an empty log file eventually and return its absolute path.

When passing the “filename” stdout, prepare_logfile() does not prepare any file and just returns stdout:

>>> from hydpy.exe.commandtools import prepare_logfile
>>> prepare_logfile("stdout")
'stdout'

When passing the “filename” default, prepare_logfile() generates a filename containing the actual date and time, prepares an empty file on disk, and returns its path:

>>> from hydpy import repr_, TestIO
>>> from hydpy.core.testtools import mock_datetime_now
>>> from datetime import datetime
>>> with TestIO():
...     with mock_datetime_now(datetime(2000, 1, 1, 12, 30, 0)):
...         filepath = prepare_logfile("default")
>>> import os
>>> os.path.exists(filepath)
True
>>> repr_(filepath)    
'...hydpy/tests/iotesting/hydpy_2000-01-01_12-30-00.log'

For all other strings, prepare_logfile() does not add any date or time information to the filename:

>>> with TestIO():
...     with mock_datetime_now(datetime(2000, 1, 1, 12, 30, 0)):
...         filepath = prepare_logfile("my_log_file.txt")
>>> os.path.exists(filepath)
True
>>> repr_(filepath)    
'...hydpy/tests/iotesting/my_log_file.txt'
hydpy.exe.commandtools.execute_scriptfunction() → Optional[int][source]

Execute a HydPy script function.

Function execute_scriptfunction() is indirectly applied and explained in the documentation on module hyd.

class hydpy.exe.commandtools.LogFileInterface(logfile: TextIO, logstyle: str, infotype: typing_extensions.Literal[info, warning, exception])[source]

Bases: object

Wraps a usual file object, exposing all its methods while modifying only the write method.

At the moment, class LogFileInterface only supports only two log styles, as explained in the documentation on module hyd. The following example shows its basic usage:

>>> from hydpy import TestIO
>>> from hydpy.exe.commandtools import LogFileInterface
>>> with TestIO():
...     logfile = open("test.log", "w")
>>> lfi = LogFileInterface(
...     logfile, logstyle="prefixed", infotype="exception")
>>> lfi.write("a message\n")
>>> lfi.write("another message\n")
>>> lfi.close()
>>> with TestIO():
...     with open("test.log", "r") as logfile:
...         print(logfile.read())
error: a message
error: another message

The class member style2infotype2string defines the currently available log styles.

style2infotype2string = {'plain': {'exception': '', 'info': '', 'warning': ''}, 'prefixed': {'exception': 'error: ', 'info': 'info: ', 'warning': 'warning: '}}
infotype: typing_extensions.Literal[info, warning, exception]
logfile: TextIO
write(string: str) → None[source]

Write the given string as explained in the main documentation on class LogFileInterface.

hydpy.exe.commandtools.parse_argument(string: str) → Union[str, Tuple[str, str]][source]

Return a single value for a string understood as a positional argument or a tuple containing a keyword and its value for a string understood as a keyword argument.

parse_argument() is intended to be used as a helper function for function execute_scriptfunction() only. See the following examples to see which types of keyword arguments execute_scriptfunction() covers:

>>> from hydpy.exe.commandtools import parse_argument
>>> parse_argument("x=3")
('x', '3')
>>> parse_argument('"x=3"')
'"x=3"'
>>> parse_argument("'x=3'")
"'x=3'"
>>> parse_argument('x="3==3"')
('x', '"3==3"')
>>> parse_argument("x='3==3'")
('x', "'3==3'")
hydpy.exe.commandtools.print_textandtime(text: str) → None[source]

Print the given string and the current date and time with high precision for logging purposes.

>>> from hydpy.exe.commandtools import print_textandtime
>>> from hydpy.core.testtools import mock_datetime_now
>>> from datetime import datetime
>>> with mock_datetime_now(datetime(2000, 1, 1, 12, 30, 0, 123456)):
...     print_textandtime("something happens")
something happens (2000-01-01 12:30:00.123456).