Utilities

Command Line

parse_python_assigns(assign_str) Parses a string, containing assign statements into a dictionary.
africanus.util.cmdline.parse_python_assigns(assign_str)[source]

Parses a string, containing assign statements into a dictionary.

data = parse_python_assigns("beta=5.6; l=[2,3], s='hello, world'")

assert data == {
    'beta': 5.6,
    'l': [2, 3],
    's': 'hello, world'
}
Parameters:
assign_str: str

Assignment string. Should only contain assignment statements assigning python literals or builtin function calls, to variable names. Multiple assignment statements should be separated by semi-colons.

Returns:
dict

Dictionary { name: value } containing assignment results.

Requirements Handling

requires_optional(*requirements) Decorator returning either the original function, or a dummy function raising a MissingPackageException when called, depending on whether the supplied requirements are present.
africanus.util.requirements.requires_optional(*requirements)[source]

Decorator returning either the original function, or a dummy function raising a MissingPackageException when called, depending on whether the supplied requirements are present.

If packages are missing and called within a test, the dummy function will call pytest.skip().

Used in the following way:

try:
    from scipy import interpolate
except ImportError as e:
    # https://stackoverflow.com/a/29268974/1611416, pep 3110 and 344
    scipy_import_error = e
else:
    scipy_import_error = None

@requires_optional('scipy', scipy_import_error)
def function(*args, **kwargs):
    return interpolate(...)
Parameters:
requirements : iterable of string, None or ImportError

Sequence of package names required by the decorated function. ImportError exceptions (or None, indicating their absence) may also be supplied and will be immediately re-raised within the decorator. This is useful for tracking down problems in user import logic.

Returns:
callable

Either the original function if all requirements are available or a dummy function that throws a MissingPackageException or skips a pytest.

Shapes

aggregate_chunks(chunks, max_chunks) Aggregate dask chunks together into chunks no larger than max_chunks.
corr_shape(ncorr, corr_shape) Returns the shape of the correlations, given ncorr and the type of correlation shape requested
africanus.util.shapes.aggregate_chunks(chunks, max_chunks)[source]

Aggregate dask chunks together into chunks no larger than max_chunks.

chunks, max_c = ((3,4,6,3,6,7),(1,1,1,1,1,1)), (10,3)
expected = ((7,9,6,7), (2,2,1,1))
assert aggregate_chunks(chunks, max_c) == expected
Parameters:
chunks : sequence of tuples or tuple
max_chunks : sequence of ints or int
Returns:
sequence of tuples or tuple
africanus.util.shapes.corr_shape(ncorr, corr_shape)[source]

Returns the shape of the correlations, given ncorr and the type of correlation shape requested

Parameters:
ncorr : integer

Number of correlations

corr_shape : {‘flat’, ‘matrix’}

Shape of output correlations

Returns:
tuple

Shape tuple describing the correlation dimensions

  • If flat returns (ncorr,)

  • If matrix returns

    • (1,) if ncorr == 1
    • (2,) if ncorr == 2
    • (2,2) if ncorr == 4

Beams

beam_filenames(filename_schema, …) Returns a dictionary of beam filename pairs, keyed on correlation,from the cartesian product of correlations and real, imaginary pairs
beam_grids(header) Extracts the FITS indices and grids for the beam dimensions in the supplied FITS header.
africanus.util.beams.beam_filenames(filename_schema, polarisation_type)[source]

Returns a dictionary of beam filename pairs, keyed on correlation,from the cartesian product of correlations and real, imaginary pairs

Given beam_$(corr)_$(reim).fits returns:

{
  'xx' : ('beam_xx_re.fits', 'beam_xx_im.fits'),
  'xy' : ('beam_xy_re.fits', 'beam_xy_im.fits'),
  ...
  'yy' : ('beam_yy_re.fits', 'beam_yy_im.fits'),
}

Given beam_$(CORR)_$(REIM).fits returns:

{
  'xx' : ('beam_XX_RE.fits', 'beam_XX_IM.fits'),
  'xy' : ('beam_XY_RE.fits', 'beam_XY_IM.fits'),
  ...
  'yy' : ('beam_YY_RE.fits', 'beam_YY_IM.fits'),
}
Parameters:
filename_schema : str

String containing the filename schema.

polarisation_type : {‘linear’, ‘circular’}

String defining the type of polarisation.

Returns:
dict

Dictionary of schema {correlation : (refile, imfile)} mapping correlations to real and imaginary filename pairs

africanus.util.beams.beam_grids(header)[source]

Extracts the FITS indices and grids for the beam dimensions in the supplied FITS header. Specifically the axes specified by

  1. L or X CTYPE
  2. M or Y CTYPE
  3. FREQ CTYPE

If the first two axes have a negative sign, such as -L, the grid will be inverted.

Any grids corresponding to axes with a CUNIT type of DEG will be converted to radians.

Parameters:
header : Header or dict

FITS header object.

Returns:
tuple

Returns ((l_axis, l_grid), (m_axis, m_grid), (freq_axis, freq_grid)) where the axis is the FORTRAN indexed FITS axis (1-indexed) and grid contains the values at each pixel along the axis.

Code

format_code(code) Formats some code with line numbers
memoize_on_key(key_fn) Memoize based on a key function supplied by the user.
africanus.util.code.format_code(code)[source]

Formats some code with line numbers

Parameters:
code : str

Code

Returns:
str

Code prefixed with line numbers

class africanus.util.code.memoize_on_key(key_fn)[source]

Memoize based on a key function supplied by the user. The key function should return a custom key for memoizing the decorated function, based on the arguments passed to it.

In the following example, the arguments required to generate the _generate_phase_delay_kernel function are the types of the lm, uvw and frequency arrays, as well as the number of correlations, ncorr.

The supplied key_fn produces a unique key based on these types and the number of correlations, which is used to cache the generated function.

def key_fn(lm, uvw, frequency, ncorrs=4):
    '''
    Produce a unique key for the arguments of
     _generate_phase_delay_kernel
    '''
    return (lm.dtype, uvw.dtype, frequency.dtype, ncorrs)

_code_template = jinja2.Template('''
#define ncorrs {{ncorrs}}

__global__ void phase_delay(
    const {{lm_type}} * lm,
    const {{uvw_type}} * uvw,
    const {{freq_type}} * frequency,
    {{out_type}} * out)
{
    ...
}
''')

_type_map = {
    np.float32: 'float',
    np.float64: 'double'
}

@memoize_on_key(key_fn)
def _generate_phase_delay_kernel(lm, uvw, frequency, ncorrs=4):
    ''' Generate the phase delay kernel '''
    out_dtype = np.result_type(lm.dtype, uvw.dtype, frequency.dtype)
    code = _code_template.render(lm_type=_type_map[lm.dtype],
                                 uvw_type=_type_map[uvw.dtype],
                                 freq_type=_type_map[frequency.dtype],
                                 ncorrs=ncorrs)
    return cp.RawKernel(code, "phase_delay")

Methods

__call__  

CUDA

grids(dims, blocks) Determine the grid size, given space dimensions sizes and blocks
africanus.util.cuda.grids(dims, blocks)[source]

Determine the grid size, given space dimensions sizes and blocks

Parameters:
dims : tuple of ints

(x, y, z) tuple

Returns:
tuple

(x, y, z) grid size tuple