Ligare.programming.R

final class Ligare.programming.R.RProcessStepBuilder(log=None)[source]

Configure an object to execute an R script with Rscript.

final class RProcessExecutorStepBuilder(parent, log=None)[source]

The final step in configurating an R process executor. This class can run the command with the execute() method.

__init__(parent, log=None)[source]

execute()[source]

Execute the configured R script.

Return CompletedProcess[bytes]:: The completed process object from process.run(…)
Return type:: tuple[CompletedProcess[bytes], bytes]

static execute_R_process(args, parameter_read_fd, data)[source]

Run the process specified by args, which is passed into subprocess.run as the first argument. This method adds the METHOD_ARG_READ_FD environment variable whose value is str(parameter_read_fd). This method blocks thread execution until the process completes.

Parameters:

args (list[str|bytes|PathLike[str]|PathLike[bytes]]) – The argument list to pass to subprocess.run.
parameter_read_fd (int) – A file descriptor number from which R will read method parameters.
data (bytes | None) – Data that is written to the executed process’s STDIN.

Return subprocess.CompletedProcess[bytes]:

The completed process.

Return type:

tuple[CompletedProcess[bytes], bytes]

static handle_process_errors(proc, log=None)[source]

Decode a finished process’s STDERR. If the process’s STDOUT is empty, an error is considered to have occurred, and an exception is raised. Otherwise, STDERR is logged and the function returns.

Parameters:

proc (subprocess.CompletedProcess[bytes]) – The completed process.
log (Logger) – The Logger that is written to

Raises:

Exception – Raised only if STDERR is not empty and STDOUT is empty.

Return type:

None

property method: RProcessMethodStepBuilder

property process: RProcessStepBuilder

property script: RProcessScriptStepBuilder

with_log(log)[source]

Configure the logger.

Parameters:: log (Logger)
Return Self:
Return type:: Self

final class RProcessMethodStepBuilder(parent, log=None)[source]

Configure method parameters and input for the R script to read.

__init__(parent, log=None)[source]

property script: RProcessScriptStepBuilder

with_data(data)[source]

The input data for the R script.

Parameters:: data (bytes) – The data
Return RProcessStepBuilder.RProcessExecutorStepBuilder:: The next step for configuration
Return type:: RProcessExecutorStepBuilder

with_log(log)[source]

Configure the logger.

Parameters:: log (Logger)
Return Self:
Return type:: Self

with_method_parameters(parameters)[source]

The method parameters and their values.

Parameters:: parameters (dict[str, Any]) – The parameters
Return Self:
Return type:: Self

static write_method_parameters(parameters)[source]

Open a FIFO pipe and write the dictionary parameters as a CSV to write_fd.

Parameters:: parameters (dict[str, Any]) – The parameters to write to the pipe
Return tuple[int, int]:: (read_fd, write_fd) the input/output file descriptors of the pipe
Return type:: tuple[int, int]

final class RProcessScriptStepBuilder(parent, log=None)[source]

Configure the R script to execute with Rscript.

__init__(parent, log=None)[source]

property process: RProcessStepBuilder

with_R_script_path(rscript)[source]

The path to the R script to execute with Rscript

Parameters:: rscript (_path_like) – The script path
Return RProcessStepBuilder.RProcessMethodStepBuilder:: The next step for configuration
Return type:: RProcessMethodStepBuilder

with_args(args)[source]

with_log(log)[source]

Configure the logger.

Parameters:: log (Logger)
Return Self:
Return type:: Self

__init__(log=None)[source]

with_Rscript_binary_path(rscript_path)[source]

The path to the Rscript binary.

Parameters:: rscript_path (_path_like) – The Rscript binary path
Return RProcessStepBuilder.RProcessScriptStepBuilder:: The next step for configuration
Return type:: RProcessScriptStepBuilder

with_log(log)[source]

Configure the logger.

Parameters:: log (Logger)
Return Self:
Return type:: Self

exception Ligare.programming.R.RscriptProcessError(proc, *args)[source]

An error occurred with Rscript. Note: This exception is not used when a script executed by Rscript errors; it is only used when Rscript itself raises an error.

__init__(proc, *args)[source]

exception Ligare.programming.R.RscriptScriptError(proc, *args)[source]

An error occurred executing a script with Rscript. Note: This exception is not used when Rscript itself errors; it is only used when a script executed by Rscript raises an error.

__init__(proc, *args)[source]

Ligare.programming.R.boolean(value)[source]

Returns the R string “TRUE” or “FALSE” from the Python strings “True”, “T”, “False”, or “F” (lowercased), or boolean True or False, respectively. If value is None, “FALSE” is returned.

Parameters:: value (str) – The input string “True”, “T”, “False”, or “F”, or boolean True or False.
Return str:: The R string “TRUE” or “FALSE”.
Return type:: str

Usage

>>> boolean(True)
'TRUE'

>>> boolean("True")
'TRUE'

>>> boolean("T")
'TRUE'

>>> boolean("true")
'TRUE'

>>> boolean(False)
'FALSE'

>>> boolean("False")
'FALSE'

>>> boolean("F")
'FALSE'

>>> boolean("false")
'FALSE'

>>> boolean(1)
'FALSE'

>>> boolean(0)
'FALSE'

>>> boolean(None)
'FALSE'

Ligare.programming.R.string(value, *, comma_separated=False, vector=False)[source]

Return type:: str | None

Ligare.programming.R.string_from_csv(value)[source]

This method is a pass-through for string(value, comma_separated=True).

Remove all characters from a string that are not whitelisted in the SAFE_COMMA_SEPARATED_STRING_PATTERN regex.

The string is comma-separated and this method accepts , as valid.

Parameters:: value (str) – The string to sanitize.
Return string:: The sanitized string, or None if the string only consists of invalid characters.
Return type:: str | None

Usage

>>> string_from_csv("abc")
"'abc'"

>>> string_from_csv("a,b,c")
"'a','b','c'"

>>> string_from_csv(None) is None
True

>>> string_from_csv("\t,\t")
"'\t','\t'"

>>> string_from_csv(",") is None
True

Ligare.programming.R.vector_from_csv(value)[source]

This method is a pass-through for string(value, vector=True).

Remove all characters from a string that are not whitelisted in the SAFE_COMMA_SEPARATED_STRING_PATTERN regex.

The string is comma-separated: and this method accepts , as valid. This method returns a string value formatted as an R vector c(…) containing the values from the CSV value string.

Parameters:: value (str) – The string to sanitize.
Return str | None:: The vectorized CSV string, or None if no valid characters were found
Return type:: str | None

Usage

>>> vector_from_csv("'") is None
True

>>> vector_from_csv("''") is None
True

>>> vector_from_csv("'a'")
"c('a')"

>>> vector_from_csv("a-b-c")
"c('a-b-c')"

>>> vector_from_csv("a-b-.c")
"c('a-b-.c')"

>>> vector_from_csv("a-b-^%^$%^c")
"c('a-b-c')"

>>> vector_from_csv("")
'c()'

>>> vector_from_csv("abc")
"c('abc')"

>>> vector_from_csv("a,b,c")
"c('a','b','c')"

>>> vector_from_csv(None)
'c()'

>>> vector_from_csv("!!!") is None
True

>>> vector_from_csv("a!b!c!")
"c('abc')"

>>> vector_from_csv("a!,b!,c!")
"c('a','b','c')"

>>> vector_from_csv("a,b,c,")
"c('a','b','c')"

>>> vector_from_csv("a.b.c")
"c('a.b.c')"

Ligare.programming.R.vector_from_parts(parts, new_part_key, existing_part_keys, default='__NULL__')[source]

Add a new key to the parts dictionary named new_part_key.

The value of the new key is:

A Python string representing an R vector "c(...)" where each value comes from the parts dictionary for each key in the existing_part_keys list.
If every value from the parts dictionary is None or an empty string, the new key’s value is the values of default.

Each key in existing_part_keys is deleted from the parts dictionary.

Parameters:

parts (dict[str, Any]) – A dictionary of parameters
new_part_key (str) – The name of the new key to add to the dictionary
existing_part_keys (list[str]) – The names of the keys from which to create the value of the new key
default (Any) – The default value of the new key if all key values in parts for the keys existing_part_keys are None or an empty string, defaults to “__NULL__”

Return type:

None

Usage

Convert two keys with values into a single two-item vector.

>>> query_params = {
...     "scale.bar.coords.x": 0.5,
...     "scale.bar.coords.y": 1.0
... }
>>> vector_from_parts(
...     query_params,
...     "scale.bar.coords",
...     ["scale.bar.coords.x", "scale.bar.coords.y"]
... )
>>> query_params
{'scale.bar.coords': 'c(0.5,1.0)'}

Convert two keys with numerical and string values into a single two-item vector.

>>> query_params = {
...     "scale.bar.coords.x": 0.5,
...     "scale.bar.coords.y": '1.0'
... }
>>> vector_from_parts(
...     query_params,
...     "scale.bar.coords",
...     ["scale.bar.coords.x", "scale.bar.coords.y"]
... )
>>> query_params
{'scale.bar.coords': "c(0.5,'1.0')"}

Convert two keys into a single two-item vector where one value is `None`

>>> query_params = {
...     "scale.bar.coords.x": 0.5,
...     "scale.bar.coords.y": None
... }
>>> vector_from_parts(
...     query_params,
...     "scale.bar.coords",
...     ["scale.bar.coords.x", "scale.bar.coords.y"]
... )
>>> query_params
{'scale.bar.coords': "c(0.5,'__NULL__')"}

Convert two keys into an empty value where all values are `None`

>>> query_params = {
...     "scale.bar.coords.x": None,
...     "scale.bar.coords.y": None
... }
>>> vector_from_parts(
...     query_params,
...     "scale.bar.coords",
...     ["scale.bar.coords.x", "scale.bar.coords.y"]
... )
>>> query_params
{'scale.bar.coords': '__NULL__'}

Convert many keys of varying types into a single vector.

>>> query_params = {
...     "scale.bar.coords.x": 0.5,
...     "scale.bar.coords.y": '1.0',
...     "scale.bar.coords.z": None,
...     "scale.bar.coords.a": 123,
...     "scale.bar.coords.b": True,
...     "scale.bar.coords.c": False
... }
>>> vector_from_parts(
...     query_params,
...     "scale.bar.coords",
...     [
...         "scale.bar.coords.x",
...         "scale.bar.coords.y",
...         "scale.bar.coords.z",
...         "scale.bar.coords.a",
...         "scale.bar.coords.b",
...         "scale.bar.coords.c",
...     ]
... )
>>> query_params
{'scale.bar.coords': "c(0.5,'1.0','__NULL__',123,'TRUE','FALSE')"}

Modules

`exception`
`process`
`type_conversion`