Ancillary Functions

This script gathers central functions and classes for general applications

dissolve	list and tuple flattening
finder	function for finding files/folders in folders and their subdirectories
HiddenPrints
multicore	wrapper for multicore process execution
parallel_apply_along_axis	Like numpy.apply_along_axis() but using multiple threads.
parse_literal	return the smallest possible data type for a string or list of strings
run
sampler	General function to select random sample indexes from arrays.

class spatialist.ancillary.HiddenPrints

Suppress console stdout prints, i.e. redirect them to a temporary string object.

Adapted from https://stackoverflow.com/questions/8391411/suppress-calls-to-print-python

Examples

>>> with HiddenPrints():
>>>     print('foobar')
>>> print('foobar')

spatialist.ancillary.dissolve(inlist)

list and tuple flattening

Parameters:

inlist (list[Any] | tuple[Any, ...]) – the list with sub-lists or tuples to be flattened

Returns:

the flattened result

Return type:

list[Any]

Examples

>>> dissolve([[1, 2], [3, 4]])
[1, 2, 3, 4]

>>> dissolve([(1, 2, (3, 4)), [5, (6, 7)]])
[1, 2, 3, 4, 5, 6, 7]

spatialist.ancillary.finder(target, matchlist, foldermode=0, regex=False, recursive=True)

function for finding files/folders in folders and their subdirectories

Parameters:

• target (str | list[str]) – a directory, zip- or tar-archive or a list of them to be searched

• matchlist (list[str]) – a list of search patterns

• foldermode (int) –

  • 0: only files

  • 1: files and folders

  • 2: only folders

• regex (bool) – are the search patterns in matchlist regular expressions or unix shell standard (default)?

• recursive (bool) – search target recursively into all subdirectories or only in the top level? This is currently only implemented for parameter target being a directory.

Returns:

the absolute names of files/folders matching the patterns

Return type:

list[str]

spatialist.ancillary.multicore(function, cores, multiargs, pbar=False, **singleargs)

wrapper for multicore process execution

Parameters:

• function (Callable[..., Any]) – individual function to be applied to each process item

• cores (int) – the number of subprocesses started/CPUs used; this value is reduced in case the number of subprocesses is smaller

• multiargs (dict[str, Any]) – a dictionary containing sub-function argument names as keys and lists of arguments to be distributed among the processes as values

• pbar (bool) – add a progress bar? Does not yet work on Windows.

• singleargs (Any) – all remaining arguments which are invariant among the subprocesses

Returns:

the return of the function for all subprocesses

Return type:

list[Any] | None

Notes

• all multiargs value lists must be of same length, i.e. all argument keys must be explicitly defined for each subprocess

• all function arguments passed via singleargs must be provided with the full argument name and its value (i.e. argname=argval); default function args are not accepted

• if the processes return anything else than None, this function will return a list of results

• if all processes return None, this function will be of type void

Examples

>>> def add(x, y, z):
>>>     return x + y + z
>>> multicore(add, cores=2, multiargs={'x': [1, 2]}, y=5, z=9)
[15, 16]
>>> multicore(add, cores=2, multiargs={'x': [1, 2], 'y': [5, 6]}, z=9)
[15, 17]

See also

pathos.multiprocessing

spatialist.ancillary.parallel_apply_along_axis(func1d, axis, arr, cores=4, *args, **kwargs)

Like numpy.apply_along_axis() but using multiple threads. Adapted from here.

Parameters:

• func1d (Callable[..., Any]) – the function to be applied

• axis (int) – the axis along which to apply func1d

• arr (ndarray[tuple[Any, ...], dtype[Any]]) – the input array

• cores (int) – the number of parallel cores

• args (Any) – Additional arguments to func1d.

• kwargs (Any) – Additional named arguments to func1d.

Return type:

ndarray[tuple[Any, ...], dtype[Any]]

spatialist.ancillary.parse_literal(x)

return the smallest possible data type for a string or list of strings

Parameters:

x (str | bytes | list[str | bytes]) – a string to be parsed

Returns:

the parsing result

Return type:

int | float | str | bytes | list[int | float | str | bytes]

Examples

>>> isinstance(parse_literal('1.5'), float)
True

>>> isinstance(parse_literal('1'), int)
True

>>> isinstance(parse_literal('foobar'), str)
True

spatialist.ancillary.run(cmd, outdir=None, logfile=None, inlist=None, void=True, errorpass=False, env=None)

Wrapper for subprocess execution including logfile writing and command prompt piping.

This is a convenience wrapper around the subprocess module and calls its class Popen internally.

Parameters:

• cmd (list[Any]) – the command arguments

• outdir (str | None) – the directory to execute the command in

• logfile (str | None) – a file to write stdout to

• inlist (list[Any] | None) – a list of arguments passed to stdin, i.e., arguments passed to interactive input of the program

• void (bool) – return stdout and stderr?

• errorpass (bool) – if False, a subprocess.CalledProcessError is raised if the command fails

• env (dict[str, Any] | None) – the environment to be passed to the subprocess

Return type:

tuple[int, str, str] | None

Returns:

a tuple of (returncode, stdout, stderr) if void=False otherwise None

spatialist.ancillary.sampler(mask, samples=None, dim=1, replace=False, seed=42)

General function to select random sample indexes from arrays. Adapted from package S1_ARD.

Parameters:

• mask (ndarray[tuple[Any, ...], dtype[bool]]) – A 2D boolean mask to limit the sample selection.

• samples (int | None) – The number of samples to select. If None, the positions of all matching values are returned. If there are fewer values than required samples, the positions of all values are returned.

• dim (Literal[1, 2]) – The dimensions of the output array and its indexes. If 1, the returned array has one dimension and the indexes refer to the one-dimensional (i.e., flattened) representation of the input mask. If 2, the output array is of shape (2, samples) with two separate 2D arrays for y (index 0) and x respectively, which reference positions in the original 2D shape of the input array.

• replace (bool) – Draw samples with or without replacement?

• seed (int) – Seed used to initialize the pseudo-random number generator.

Returns:

The index positions of the generated random samples as 1D or 2D array.

Return type:

ndarray[tuple[Any, ...], dtype[integer[Any]]]

Examples

>>> import numpy as np
>>> from spatialist.ancillary import sampler
>>> array = np.array([[1, 2], [3, 4], [5, 6], [7, 8]])
>>> mask = array > 2
>>> s1d = sampler(mask=mask, samples=2, dim=1)
>>> s2d = sampler(mask=mask, samples=2, dim=2)
>>> print(s1d)
[2 3]
>>> print(s2d)
[[1 1]
 [0 1]]
>>> print(array.flatten()[s1d] == array[s2d[0], s2d[1]])
[ True  True]

See also

numpy.random.seed, numpy.random.choice