bluemira.codes.utilities
Utility functions for interacting with external codes
Classes
Base Model Enum |
|
Capture logs for subprocesses |
Functions
|
Read json file or raise CodesError |
|
Dynamically import code interface |
|
Check if a code is available |
|
Import guard for a code at runtime. |
|
Creates mappings for external codes |
|
Run a subprocess terminal command piping the output into bluemira's |
Module Contents
- class bluemira.codes.utilities.Model(*args, **kwds)
Bases:
enum.EnumBase Model Enum
- classmethod info()
Show Model options
- classmethod _missing_(value)
- bluemira.codes.utilities.read_mock_json_or_raise(file_path: str, name: str) dict[str, float]
Read json file or raise CodesError
- Raises:
CodesError – Cannot open mock file
- Parameters:
file_path (str)
name (str)
- Return type:
dict[str, float]
- bluemira.codes.utilities.get_code_interface(module: str) types.ModuleType
Dynamically import code interface
- Parameters:
module (str) – module to import
- Returns:
Code module
- Return type:
types.ModuleType
- bluemira.codes.utilities.is_code_available(code_module: str) bool
Check if a code is available
- Parameters:
code_module (str) – The name of the code module to check.
- Returns:
True if the code is available, False otherwise.
- Return type:
bool
- bluemira.codes.utilities.code_guard(code_module: str, add_message: str = '')
Import guard for a code at runtime.
- Parameters:
code_module (str) – The name of the code module to check.
add_message (str)
- Raises:
ImportError – If the code is not available (installed).
- bluemira.codes.utilities.create_mapping(in_mappings=None, out_mappings=None, io_mappings=None, none_mappings=None) dict[str, Any]
Creates mappings for external codes
- Returns:
A mapping from bluemira names to an external code ParameterMapping
- Return type:
dict[str, Any]
- class bluemira.codes.utilities.LogPipe(loglevel: str, flush_callable: collections.abc.Callable[[str], bool] = lambda line: ..., flush_printer: collections.abc.Callable[[str], None] | None = None)
Bases:
threading.ThreadCapture logs for subprocesses
- Parameters:
loglevel (str) – print or error flush printing
flush_callable (collections.abc.Callable[[str], bool])
flush_printer (collections.abc.Callable[[str], None] | None)
- logfunc
- logfunc_flush
- flush_callable
- pipe
- fileno()
- Returns:
The write file descriptor of the pipe
- run()
Run the thread and pipe it all into the logger.
- close()
Close the write end of the pipe.
- bluemira.codes.utilities.run_subprocess(command: list[str], run_directory: str = '.', flush_callable: collections.abc.Callable[[str], bool] = lambda line: ..., flush_printer: collections.abc.Callable[[str], None] | None = None, **kwargs) int
Run a subprocess terminal command piping the output into bluemira’s logs.
- Parameters:
command (list[str]) – The arguments of the command to run.
run_directory (str) – The directory to run the command in. Default is current working directory.
**kwargs – Arguments passed directly to subprocess.Popen.
flush_callable (collections.abc.Callable[[str], bool])
flush_printer (collections.abc.Callable[[str], None] | None)
- Returns:
return_code – The return code of the subprocess.
- Return type:
int