API documentation

The Utility Objects

convenience objects

class calim_pipelines.objects.ContainerDescriptor(image: str | None = None, volumes: dict | None = None, working_dir: str | None = None, detached: bool = False)[source]

Class for keeping track of the properties of a container.

static argument_parser(parser: ArgumentParser) → ArgumentParser[source]

Populate a parser with the configuration required to build an instance

Parameters:: parser (ArgumentParser) – Current Parser
Returns:: Parser with the configuration additions
Return type:: ArgumentParser

classmethod from_parser(parser: ArgumentParser)[source]

Instantiate a ContainerDescritor from an ArgumentParser

Parameters:: parser (ArgumentParser) – parser with an image, working_dir and volumes
Returns:: ContainerDescriptor

class calim_pipelines.objects.SourceDescriptor(r_a: float, dec: float, flux: float)[source]: Class that holds the position and flux of a source. Other parameters may be added later

The Pipelines

Simple flow to run a command

calim_pipelines.pipelines.run_command.main()[source]: Simple application to run any command from a container

Simple flow to image using cimager

calim_pipelines.pipelines.run_cimager.main()[source]: Simple application to run cimager

Simple flow to image using cimager

calim_pipelines.pipelines.run_csim.main()[source]: Application to configure and run the csimulator application from an

Simple flow to get the stats

calim_pipelines.pipelines.run_imgstat.main()[source]: Simple method to return the simple image details from a CASA image

Simple flow to image using cimager

calim_pipelines.pipelines.run_point_source_test.main()[source]: Simple application to run cimager and imgstat to interrogate a measurement set to see if it contains a point source

The Tasks

Prefect2 tasks This module contains the Prefect Tasks that are used in the Prefect Flows.

calim_pipelines.tasks.create_csimulator_antennas_definition(instrument: str, stations: list, outfile: str | None = None)

Creates a definition file. THis is needed to take our layouts and to turn them into definitions for the purposes of running the simultator

Parameters:

instrument (str) – ska1_low or ska1_mid
stations (list) – list of station names you want to take from the layout
outfile (str, optional) – Array description file. Defaults to None.

calim_pipelines.tasks.get_config_filenames(config_file: str, keyword: str = 'dataset')

Get the filenames from the config

Parameters:: config_file (str) – The configuration file
Raises:: ValueError – If the dataset key is not in the configuration raise an execption
Returns:: a string containing all the filenames
Return type:: _type_

calim_pipelines.tasks.get_config_value(config_file: str, key: str)

Simply return the value given the config file and the key

Parameters:

config_file (str) – The configuration file
key (str) – The Key

Raises:

ValueError – if the key is not present

Returns:

the value

Return type:

str

calim_pipelines.tasks.get_definition_file(config_file: str)

Get the array definition/layout file from the config. Very ASKAP specific.

Parameters:: config_file (str) – The configuration file
Raises:: ValueError – If the dataset key is not in the configuration raise an execption
Returns:: a string containing all the filenames
Return type:: str

calim_pipelines.tasks.get_ms_channels(ms_file: str)

Get the number of channels from a measurement set

Parameters:: ms_file (str) – the measurement set

Returns the number of channels

calim_pipelines.tasks.get_telescope_layout(instrument: str)

Gets the telescope layout using TMData

Parameters:: instrument (str) – ska1_low or ska1_nid
Raises:: ValueError – Raised if the instrument is not ska1_low or ska1_mid
Returns:: the antenna/station layout
Return type:: dictionary

calim_pipelines.tasks.hello_task(): Prefect task: Simple Hello World

calim_pipelines.tasks.pull_image(image: str)

Prefect task: Pull a docker image using the current environment

Parameters:: image (str) – The name of the image to pull

calim_pipelines.tasks.run_container(image: str, detach=False)

Simple tasks that uses the docker API to launch a container using the given image.

The stdout from the task when the container completes is then sent the the logger

Parameters:: image (str) – image to start container with

calim_pipelines.tasks.run_container_cmd(command: str, desc: ContainerDescriptor)

Run the given command in a container with the given image

Parameters:

command (str) – the command to run
desc (ContainerDescriptor) – Description of the container to run.

calim_pipelines.tasks.set_config_filenames(config_file: str, valuestr: str, keyword: str = 'dataset')

Sets the filenames in the config

Parameters:: config_file (str) – The configuration file
Raises:: ValueError – If the dataset key is not in the configuration raise an execption
Returns:: a string containing all the filenames
Return type:: _type_

The Workflows

Prefect 2 Workflows

calim_pipelines.flows.build_cimager_command_line(config: str) → str

Read the parset and build the command line.

We need to get: * the number of channels in the input measurements set (nchan) * the number of channels per core (nchanpercore) * the number of worker groups. * whether there is a solver per core or a dedicated solver.

The calculation == (nchan/nchanpercore) * nworkergroupa

We add one if an extra solver is required :param config: the parset :type config: str

calim_pipelines.flows.build_csimulator_command_line(config: str)

Simple flow to build the imgstat command line

Parameters:: file (str) – the file to interrogate
Returns:: The command line to run

calim_pipelines.flows.build_imgstat_command_line(file: str)

Simple flow to build the imgstat command line

Parameters:: file (str) – the file to interrogate
Returns:: The command line to run

calim_pipelines.flows.docker_hello_world(): Workflow that runs two tasks to pull an image and run a container.

calim_pipelines.flows.docker_run_command(command: str, desc: ContainerDescriptor)

Simple flow that can execute arbitrary commands in a docker container

Parameters:

command (str) – The command to run
desc (ContainerDescriptor) – Container description

calim_pipelines.flows.get_peak_value(file: str, desc: ContainerDescriptor)

Find the peak pixel in an image

Parameters:

config (str) – the ASKAP configuration file
image (str, optional) – image containing the command Defaults to None.
volumes (dict, optional) – volumes to mount. Defaults to None.
working_dir (str, optional) – directory you want to work in. Defaults to None.

calim_pipelines.flows.hello_flow(): Simple hello world workflow - just uses a single task

calim_pipelines.flows.point_source_test(config: str, expected: SourceDescriptor, desc: ContainerDescriptor | None = None)

Simple test that runs cimager with the configuration. Then runs get_peak_value to get the source position and flux. It then compares the found flux and position with the expected one

Parameters:

config (str) – config file
desc (ContainerDescriptor) – image to use

calim_pipelines.flows.run_cimager(config: str, desc: ContainerDescriptor)

Flow that just runs the imager.

Parameters:

config (str) – cimager config file
image (str) – Optional: image to run the imager in

calim_pipelines.flows.run_csimulator(config: str, instrument: str, stations: list, desc: ContainerDescriptor | None = None)

Flow that just runs the simulator.

Parameters:

config (str) – csimulator config file
instrument (str) – Instrument to simulate (used ti generate layout)
stations (list) – list of stations to use (used to generate layout)
descriptor (ContainerDescriptor) – Optional: image to run the imager in