Base Classes API

Base classes for SPECULA processing objects and data structures.

Base Data Obj

class specula.base_data_obj.BaseDataObj(target_device_idx: int = None, precision: int = None)

Bases: BaseTimeObj

Initialize the base data object.

Parameters: target_device_idx: int, optional

device to be targeted for data storage. Set to -1 for CPU, to 0 for the first GPU device, 1 for the second GPU device, etc.

precision: int, optional: if None will use the global_precision, otherwise set to 0 for double, 1 for single

Methods

`copyTo`(target_device_idx)	Duplicate a data object on another device, alllocating all CPU/GPU arrays on the new device.
`to_xp`(v[, dtype, force_copy])	Method wrapping the global to_xp function.
`transferDataTo`(destobj[, force_reallocation])	Copy CPU/GPU arrays into an existing data object: iterate over all self attributes and, if a CPU or GPU array is detected, copy data into destobj without reallocating.

init_logging
monitorMem
printMemUsage
seconds_to_t
startMemUsageCount
stopMemUsageCount
t_to_seconds

copyTo(target_device_idx): Duplicate a data object on another device, alllocating all CPU/GPU arrays on the new device.

transferDataTo(destobj, force_reallocation=False)

Copy CPU/GPU arrays into an existing data object: iterate over all self attributes and, if a CPU or GPU array is detected, copy data into destobj without reallocating.

Destination (CPU or GPU device) is inferred by destobj.target_device_idx, which must be set correctly before calling this method.

specula.base_data_obj.get_properties(cls)

Base Processing Obj

class specula.base_processing_obj.BaseProcessingObj(target_device_idx=None, precision=None)

Bases: BaseTimeObj

Initialize the base processing object.

Parameters: precision (int, optional): if None will use the global_precision, otherwise pass 0 for double, 1 for single target_device_idx (int, optional): if None will use the default_target_device_idx, otherwise pass -1 for cpu, i for GPU of index i

Methods

`checkInputTimes`()	Determine whether this processing object needs to execute the trigger method, based on the input states
`check_input_names`()	Check that all input names declared in self.input_names are present in self.inputs
`check_output_names`()	Check that all output names declared in self.output_names are present in self.outputs
`finalize`()	Override this method to perform any actions after the simulation is completed
`get_all_inputs`()	Perform get() on all inputs.
`post_trigger`()	Make sure we are using the correct device and that any previous CUDA graph has been synchronized
`prepare_trigger`(t)
`sanity_check`()	Check that all inputs and outputs have been setup correctly.
`send_outputs`([skip_delayed, delayed_only, ...])	Send all remote outputs via MPI.
`setup`()	Override this method to perform any setup just before the simulation is started.
`to_xp`(v[, dtype, force_copy])	Method wrapping the global to_xp function.
`trigger_code`()	Implementations in derived classes should run GPU operations using the xp module on arrays allocated with self.xp.

addRemoteOutput
build_stream
capture_stream
check_ready
device_stream
init_logging
monitorMem
printMemUsage
seconds_to_t
send_remote_output
startMemUsageCount
stopMemUsageCount
t_to_seconds
trigger

addRemoteOutput(name, remote_output)

build_stream(allow_parallel=True)

capture_stream()

checkInputTimes(): Determine whether this processing object needs to execute the trigger method, based on the input states

check_input_names()

Check that all input names declared in self.input_names are present in self.inputs

with the correct type (InputValue or InputList).

Supported declaration grammar for input names: - Exact names, e.g. in_ef - Placeholder patterns, e.g. in_sensor_{idx}

Notes

Placeholder segments {...} are treated as wildcards.
- In project classes, placeholder-style declarations are mandatory;
  do not use raw * declarations in input_names().
Optional declarations are identified by descriptions ending with
(optional).
Exact-key matches take precedence over pattern matches.

check_output_names()

Check that all output names declared in self.output_names are present in self.outputs

Supported declaration grammar for output names: - Exact names, e.g. out_ef

Placeholder patterns, e.g. out_modes_{sensor_idx}
(ModalrecMultirate)

Placeholder patterns, e.g. out_{source_name_}layer
(AtmoRandomPhase)

Notes

Placeholder segments {...} are treated as wildcards.
- In project classes, placeholder-style declarations are mandatory;
  do not use raw * declarations in output_names().
Exact-key matches take precedence over pattern matches.

check_ready(t)

classmethod device_stream(target_device_idx)

finalize(): Override this method to perform any actions after the simulation is completed

get_all_inputs(): Perform get() on all inputs. Remote inputs, if any, are received via MPI. Data is transferred between devices if necessary.

post_trigger(): Make sure we are using the correct device and that any previous CUDA graph has been synchronized

prepare_trigger(t)

sanity_check(): Check that all inputs and outputs have been setup correctly.

send_outputs(skip_delayed=False, delayed_only=False, first_mpi_send=True)

Send all remote outputs via MPI. If skip_delayed is True, skip sending delayed outputs. This is used during the last iteration when the simulation is ending and no one would receive delayed inputs.

If delayed_only is True, only send delayed outputs. This is used while setting up the simulation to initialize outputs that are delayed and would not be received otherwise.

send_remote_output(item, dest_rank, dest_tag, first_mpi_send=True, out_name='')

setup()

Override this method to perform any setup just before the simulation is started.

The base class implementation also checks that all non-optional inputs have been set.

trigger()

trigger_code()

Implementations in derived classes should run GPU operations using the xp module on arrays allocated with self.xp.

Avoid explicit numpy or pure-Python operations and avoid using values from variables that are reallocated by prepare_trigger() or post_trigger().

When stream capture is enabled, a CUDA graph is generated, non-GPU operations are skipped, and GPU memory addresses from the first run are reused.

class specula.base_processing_obj.InputDesc(type, desc)

Bases: tuple

Methods

`count`(value, /)	Return number of occurrences of value.
`index`(value[, start, stop])	Return first index of value.

Create new instance of InputDesc(type, desc)

Methods

`count`(value, /)	Return number of occurrences of value.
`index`(value[, start, stop])	Return first index of value.

desc: Alias for field number 1

type: Alias for field number 0

class specula.base_processing_obj.OutputDesc(type, desc)

Bases: tuple

Methods

`count`(value, /)	Return number of occurrences of value.
`index`(value[, start, stop])	Return first index of value.

Create new instance of OutputDesc(type, desc)

Methods

`count`(value, /)	Return number of occurrences of value.
`index`(value[, start, stop])	Return first index of value.

desc: Alias for field number 1

type: Alias for field number 0

Base Time Obj

class specula.base_time_obj.BaseTimeObj(target_device_idx=None, precision=None)

Bases: object

Creates a new base_time object.

Parameters: precision (int, optional): if None will use the global_precision, otherwise pass 0 for double, 1 for single target_device_idx (int, optional): if None will use the default_target_device_idx, otherwise pass -1 for cpu, i for GPU of index i

Methods

to_xp(v[, dtype, force_copy])

Method wrapping the global to_xp function.

init_logging
monitorMem
printMemUsage
seconds_to_t
startMemUsageCount
stopMemUsageCount
t_to_seconds

init_logging(level=None)

monitorMem()

printMemUsage()

seconds_to_t(seconds)

startMemUsageCount()

stopMemUsageCount()

t_to_seconds(t)

to_xp(v, dtype=None, force_copy=False): Method wrapping the global to_xp function.

Base Value

class specula.base_value.BaseValue(description='', value=None, target_device_idx=None, precision=None)

Bases: BaseDataObj

Initialize the base value object.

Parameters: description (str, optional) value (any, optional): data to store. If not set, the value is initialized to None.

Methods

`copyTo`(target_device_idx)	Duplicate a data object on another device, alllocating all CPU/GPU arrays on the new device.
`to_xp`(v[, dtype, force_copy])	Method wrapping the global to_xp function.
`transferDataTo`(destobj[, force_reallocation])	Copy CPU/GPU arrays into an existing data object: iterate over all self attributes and, if a CPU or GPU array is detected, copy data into destobj without reallocating.

array_for_display
get_fits_header
get_value
init_logging
monitorMem
printMemUsage
restore
save
seconds_to_t
set_value
startMemUsageCount
stopMemUsageCount
t_to_seconds

array_for_display()

get_fits_header()

get_value()

static restore(filename, target_device_idx=None)

save(filename, overwrite=False)

set_value(val)

Calib Manager

class specula.calib_manager.CalibManager(root_dir)

Bases: object

Initialize the calibration manager object.

Parameters: root_dir (str): Root path of the calibration tree

Methods

`filename`(subdir, name)	Construct the full file path for a calibration file.
`read_data`(name)	Read data from a FITS file in the 'data' subdirectory.
`readfits`(subdir, name)	Read data from a FITS file.
`root_subdir`(type)	Returns the full path to the subdirectory corresponding to the given type.
`write_data`(name, data)	Write data to a FITS file in the 'data' subdirectory.
`writefits`(subdir, name, data)	Write data to a FITS file.

filename(subdir, name)

Construct the full file path for a calibration file.

Parameters:

subdir (str) – The key representing the calibration data type or subdirectory.
name (str) – The base name of the file (without extension).

Returns:

The absolute path to the FITS file, ensuring the ‘.fits’ extension is present.

Return type:

str

read_data(name)

Read data from a FITS file in the ‘data’ subdirectory.

Parameters:: name (str) – The base name of the file (without extension).
Returns:: The data read from the file.
Return type:: array-like

readfits(subdir, name)

Read data from a FITS file.

Parameters:

subdir (str) – The key representing the calibration data type or subdirectory.
name (str) – The base name of the file (without extension).

Returns:

The data read from the file.

Return type:

array-like

root_subdir(type)

Returns the full path to the subdirectory corresponding to the given type.

Parameters:: type (str) – The key representing the calibration data type.
Returns:: The absolute path to the subdirectory for the specified type.
Return type:: str

write_data(name, data)

Write data to a FITS file in the ‘data’ subdirectory.

Parameters:

name (str) – The base name of the file (without extension).
data (array-like) – The data to be written to the file.

writefits(subdir, name, data)

Write data to a FITS file.

Parameters:

subdir (str) – The key representing the calibration data type or subdirectory.
name (str) – The base name of the file (without extension).
data (array-like) – The data to be written to the file.

Connections

class specula.connections.InputList(type, optional=False)

Bases: object

Wrapper for input lists exchanged by objects. All inputs and outputs are managed as lists. Singles values use the InputValue() class below, which just reduces to a list with a single value.

Each list element is a separate _InputItem instance, which is able to perform its own MPI receive if needed. This allows to mix in the same list inputs with different sources (useful e.g. in propagation)

Methods

`append`(item[, remote_rank, tag])	Append an item to the input list, optionally specifying a remote rank and tag.
`set`(values_list[, remote_rank, tag])	Set the input values for the list.

get

append(item, remote_rank=None, tag=None): Append an item to the input list, optionally specifying a remote rank and tag. If the item is a list, it will be flattened and each item will be added to the input list.

get(target_device_idx)

set(values_list, remote_rank=None, tag=None): Set the input values for the list.

class specula.connections.InputValue(type, optional=False)

Bases: InputList

Convenience class for single values: calling get() will return a single item

Methods

`append`(item[, remote_rank, tag])	Append an item to the input list, optionally specifying a remote rank and tag.
`set`(item[, remote_rank, tag])	Set a single item as the input list

get

Wrapper for input lists exchanged by objects. All inputs and outputs are managed as lists. Singles values use the InputValue() class below, which just reduces to a list with a single value.

Each list element is a separate _InputItem instance, which is able to perform its own MPI receive if needed. This allows to mix in the same list inputs with different sources (useful e.g. in propagation)

Methods

`append`(item[, remote_rank, tag])	Append an item to the input list, optionally specifying a remote rank and tag.
`set`(item[, remote_rank, tag])	Set a single item as the input list

get

get(target_device_idx)

set(item, remote_rank=None, tag=None): Set a single item as the input list

Field Analyser

class specula.field_analyser.FieldAnalyser(data_dir: str, tracking_number: str, polar_coordinates: ndarray, wavelength_nm: float = 750.0, start_time: float = 0.1, end_time: float | None = None, display: bool = False, log_level: str | None = None)

Bases: object

Class to analyze field PSF, modal analysis, and phase cubes for a given tracking number in the Specula framework. This class replicates the functionality of the previous compute_off_axis_psf, compute_off_axis_modal_analysis, and compute_off_axis_cube methods, providing a structured way to handle field sources and their analysis. .. attribute:: data_dir

Directory containing tracking number data.

type:

str

tracking_number

The tracking number for the analysis.

Type:: str

polar_coordinates

Polar coordinates of field sources.

Type:: np.ndarray

wavelength_nm

Wavelength in nanometers.

Type:: float

start_time

Start time for the analysis.

Type:: float

end_time

End time for the analysis, if applicable.

Type:: Optional[float]

log_level

Logging level if different from default specula one

Type:: str

Methods

`compute_field_psf`([psf_sampling, ...])	Calculate field PSF using SPECULA's replay system
`compute_modal_analysis`([modal_params, ...])	Calculate field modal analysis using replay system
`compute_phase_cube`([force_recompute])	Calculate field phase cubes using replay system

compute_field_psf(psf_sampling: float | None = None, psf_pixel_size_mas: float | None = None, force_recompute: bool = False) → Dict

Calculate field PSF using SPECULA’s replay system

Parameters:

psf_sampling – PSF sampling factor (alternative to psf_pixel_size_mas)
psf_pixel_size_mas – Desired PSF pixel size in milliarcseconds (alternative to psf_sampling)
force_recompute – Force recomputation even if files exist

Note

Either psf_sampling or psf_pixel_size_mas must be specified, but not both.

compute_modal_analysis(modal_params: Dict | None = None, force_recompute: bool = False) → Dict

Calculate field modal analysis using replay system

Parameters:

modal_params – Dictionary of ModalAnalysis arguments. Typical keys: ifunc_ref, ifunc_inv_ref, ifunc_object, ifunc_inv_object, type_str, nmodes/nzern, npixels, obsratio, diaratio, wavelengthInNm, dorms. If None, attempts to extract from DM configuration.
force_recompute – Force recomputation even if files exist

compute_phase_cube(force_recompute: bool = False) → Dict: Calculate field phase cubes using replay system

Template Processing Obj

class specula.template_processing_obj.ProcessingObjName(parameter1: float = 0, parameter2: Recmat = None, target_device_idx=None, precision=None)

Bases: BaseProcessingObj

Obj Name

Methods

`checkInputTimes`()	Determine whether this processing object needs to execute the trigger method, based on the input states
`check_input_names`()	Check that all input names declared in self.input_names are present in self.inputs
`check_output_names`()	Check that all output names declared in self.output_names are present in self.outputs
`finalize`()	Override this method to perform any actions after the simulation is completed
`get_all_inputs`()	Perform get() on all inputs.
`post_trigger`()	Make sure we are using the correct device and that any previous CUDA graph has been synchronized
`sanity_check`()	Check that all inputs and outputs have been setup correctly.
`send_outputs`([skip_delayed, delayed_only, ...])	Send all remote outputs via MPI.
`setup`()	Override this method to perform any setup just before the simulation is started.
`to_xp`(v[, dtype, force_copy])	Method wrapping the global to_xp function.
`trigger_code`()	Implementations in derived classes should run GPU operations using the xp module on arrays allocated with self.xp.

addRemoteOutput
build_stream
capture_stream
check_ready
device_stream
init_logging
monitorMem
prepare_trigger
printMemUsage
seconds_to_t
send_remote_output
startMemUsageCount
stopMemUsageCount
t_to_seconds
trigger

Initialize the base processing object.

Parameters: precision (int, optional): if None will use the global_precision, otherwise pass 0 for double, 1 for single target_device_idx (int, optional): if None will use the default_target_device_idx, otherwise pass -1 for cpu, i for GPU of index i

Methods

`checkInputTimes`()	Determine whether this processing object needs to execute the trigger method, based on the input states
`check_input_names`()	Check that all input names declared in self.input_names are present in self.inputs
`check_output_names`()	Check that all output names declared in self.output_names are present in self.outputs
`finalize`()	Override this method to perform any actions after the simulation is completed
`get_all_inputs`()	Perform get() on all inputs.
`post_trigger`()	Make sure we are using the correct device and that any previous CUDA graph has been synchronized
`sanity_check`()	Check that all inputs and outputs have been setup correctly.
`send_outputs`([skip_delayed, delayed_only, ...])	Send all remote outputs via MPI.
`setup`()	Override this method to perform any setup just before the simulation is started.
`to_xp`(v[, dtype, force_copy])	Method wrapping the global to_xp function.
`trigger_code`()	Implementations in derived classes should run GPU operations using the xp module on arrays allocated with self.xp.

addRemoteOutput
build_stream
capture_stream
check_ready
device_stream
init_logging
monitorMem
prepare_trigger
printMemUsage
seconds_to_t
send_remote_output
startMemUsageCount
stopMemUsageCount
t_to_seconds
trigger

post_trigger(): Make sure we are using the correct device and that any previous CUDA graph has been synchronized

prepare_trigger(t)

setup()

Override this method to perform any setup just before the simulation is started.

The base class implementation also checks that all non-optional inputs have been set.

trigger_code()

Implementations in derived classes should run GPU operations using the xp module on arrays allocated with self.xp.

Avoid explicit numpy or pure-Python operations and avoid using values from variables that are reallocated by prepare_trigger() or post_trigger().

When stream capture is enabled, a CUDA graph is generated, non-GPU operations are skipped, and GPU memory addresses from the first run are reused.