openstb.simulator.controller.simple_points

Simple simulation with idealised point targets.

Classes:

Name	Description
ChunkSettings	Container for simulation settings specific to a single chunk.
CommonSettings	Container for simulation settings that are common to all chunks.
SimplePointConfig	Specification for the SimplePointSimulation configuration dictionary.
SimplePointSimulation	Controller plugin to simulate with idealised point targets.

Functions:

Name	Description
point_simulation_chunk	Perform a single chunk of the simulation.
point_simulation_store	Store the result of a piece of the simulation.

ChunkSettings dataclass

ChunkSettings(ping_time, rx_position, rx_ori, rx_distortion, max_t)

Container for simulation settings specific to a single chunk.

These settings vary per ping or per receiver, and so have to be set for each chunk.

Attributes:

Name	Type	Description
max_t	float	Maximum sample time in seconds relative to the start of transmission.
ping_time	float	Time the ping starts in seconds relative to the trajectory start.
rx_distortion	list[Distortion]	Receiver distortions to apply to the received signal.
rx_ori	ndarray	Orientation of the receiver in system coordinates.
rx_position	ndarray	Position of the receiver in system coordinates.

max_t instance-attribute

max_t

Maximum sample time in seconds relative to the start of transmission.

ping_time instance-attribute

ping_time

Time the ping starts in seconds relative to the trajectory start.

rx_distortion instance-attribute

rx_distortion

Receiver distortions to apply to the received signal.

rx_ori instance-attribute

rx_ori

Orientation of the receiver in system coordinates.

rx_position instance-attribute

rx_position

Position of the receiver in system coordinates.

CommonSettings dataclass

CommonSettings(f, S, signal_frequency_bounds, baseband_frequency, travel_time, trajectory, environment, tx_position, tx_ori, emitted_distortion, echo_distortion)

Container for simulation settings that are common to all chunks.

These settings are broadcast to all workers before starting the simulation as they do not change during the simulation.

Attributes:

Name	Type	Description
S	ndarray	Baseband spectrum of the transmitted signal.
baseband_frequency	float	Frequency used for basebanding.
echo_distortion	list[Distortion]	Distortions to apply to the echo before it reaches the receiver.
emitted_distortion	list[Distortion]	Distortions to apply to the signal before it hits a target.
environment	Environment	Environmental properties to simulate.
f	ndarray	Passband simulation frequencies.
signal_frequency_bounds	tuple[float, float]	Passband frequency bounds of the transmitted signal.
trajectory	Trajectory	Trajectory followed by the system.
travel_time	TravelTime	Travel time calculator to use.
tx_ori	ndarray	Orientation of the transmitter in system coordinates.
tx_position	ndarray	Position of the transmitter in system coordinates.

S instance-attribute

S

Baseband spectrum of the transmitted signal.

baseband_frequency instance-attribute

baseband_frequency

Frequency used for basebanding.

echo_distortion instance-attribute

echo_distortion

Distortions to apply to the echo before it reaches the receiver.

emitted_distortion instance-attribute

emitted_distortion

Distortions to apply to the signal before it hits a target.

This should include any distortions from the transmitter.

environment instance-attribute

environment

Environmental properties to simulate.

f instance-attribute

f

Passband simulation frequencies.

signal_frequency_bounds instance-attribute

signal_frequency_bounds

Passband frequency bounds of the transmitted signal.

trajectory instance-attribute

trajectory

Trajectory followed by the system.

travel_time instance-attribute

travel_time

Travel time calculator to use.

tx_ori instance-attribute

tx_ori

Orientation of the transmitter in system coordinates.

tx_position instance-attribute

tx_position

Position of the transmitter in system coordinates.

SimplePointConfig

Bases: TypedDict

Specification for the SimplePointSimulation configuration dictionary.

Attributes:

Name	Type	Description
dask_cluster	DaskCluster	Dask cluster to run the simulation on.
echo_distortion	NotRequired[list[Distortion]]	Plugins to apply distortion to the echoed signal.
emitted_distortion	NotRequired[list[Distortion]]	Plugins to apply distortion to the emitted signal.
environment	Environment	Plugin to provide details about the simulated environment.
ping_times	PingTimes	Plugin to calculate ping start times.
result_converter	NotRequired[ResultConverter]	Plugin to convert the simulation output into a desired format.
system	System	Details about the system to simulate.
targets	list[PointTargets]	A list of plugins giving the point targets to simulate.
trajectory	Trajectory	Plugin specifying the trajectory followed by the system.
travel_time	TravelTime	Plugin to calculate the travel times to and from each target.

dask_cluster instance-attribute

dask_cluster

Dask cluster to run the simulation on.

echo_distortion instance-attribute

echo_distortion

Plugins to apply distortion to the echoed signal.

These are applied to the signal after it has been scattered by a target and before it reaches the receiver. They are applied in the order they are given here.

emitted_distortion instance-attribute

emitted_distortion

Plugins to apply distortion to the emitted signal.

These are applied to the signal after it has been transmitted and before it reaches the target. They are applied in the order they are given here.

environment instance-attribute

environment

Plugin to provide details about the simulated environment.

ping_times instance-attribute

ping_times

Plugin to calculate ping start times.

result_converter instance-attribute

result_converter

Plugin to convert the simulation output into a desired format.

system instance-attribute

system

Details about the system to simulate.

targets instance-attribute

targets

A list of plugins giving the point targets to simulate.

trajectory instance-attribute

trajectory

Plugin specifying the trajectory followed by the system.

travel_time instance-attribute

travel_time

Plugin to calculate the travel times to and from each target.

SimplePointSimulation

SimplePointSimulation(result_filename, points_per_chunk, sample_rate, baseband_frequency, max_samples=None, task_lower_threshold=2.0, task_upper_threshold=3.0, reduction_node_count=4, reduction_levels=3)

Bases: LoopController[tuple[int, int, ndarray, ndarray], tuple[CommonSettings, ChunkSettings], ndarray, SimplePointConfig]

Controller plugin to simulate with idealised point targets.

The echo from each point target is summed to get the final result. Occlusions are not modelled (the targets are infinitesimally small and so cannot cast shadows). The scattering strength of each target is a fixed value independent of aspect. The simulation is performed in the temporal frequency domain and the results are basebanded.

The targets are divided into chunks and submitted to the cluster for simulation. To combine the results from multiple chunks, a reduction tree is used. This sums the results from a small number of chunks, allowing the initial results to be freed from memory. Groups of these summed results are themselves recursively summed in the same manner until a single combined result remains. For eight chunks and a reduction node count of two, this means instead of computing the result as

result = r1 + r2 + r3 + r4 + r5 + r6 + f7 + r8

we might compute it as

result = ((r1 + r2) + (r3 + r4)) + ((r5 + r6) + (r7 + r8))

which has the same number of operations but allows memory to be freed earlier. Note that, since floating-point operations are generally not associative, these two results will differ by some small amount proportional to the machine precision and the number of operations.

This reduction will be performed for a few levels, after which the combined set of results will be written to disk. The number of chunks in each write is given by the reduction_node_count parameter raised to the power of the reduction_levels parameter.

Parameters:

Name	Type	Description	Default
result_filename	PathLike[str] | str	Filename to store the results under. If this already exists, an exception will be raised.	required
points_per_chunk	int	The maximum number of point targets to simulate in each chunk.	required
sample_rate	float	Sampling rate in Hertz of the results.	required
baseband_frequency	float	Frequency used for downconversion during basebanding (carrier frequency).	required
max_samples	int | None	The maximum number of samples each receiver will capture per ping. If not given, this is calculated from the maximum interval between pings and the sampling rate. The maximum length of the trace in seconds can be found by dividing max_samples by sample_rate.	None
task_lower_threshold	float	When the number of simulation tasks per worker in the scheduler drops below this value, add more tasks.	2.0
task_upper_threshold	float	When submitting simulation tasks to the scheduler, add enough to ensure there are at least this many per worker.	3.0
reduction_node_count	int	How many results to sum at each level of the reduction tree.	4
reduction_levels	int	How many levels to use in the reduction tree before writing the combined results to disk.	3

Source code in openstb/simulator/controller/simple_points.py

def __init__(
    self,
    result_filename: os.PathLike[str] | str,
    points_per_chunk: int,
    sample_rate: float,
    baseband_frequency: float,
    max_samples: int | None = None,
    task_lower_threshold: float = 2.0,
    task_upper_threshold: float = 3.0,
    reduction_node_count: int = 4,
    reduction_levels: int = 3,
):
    """
    Parameters
    ----------
    result_filename
        Filename to store the results under. If this already exists, an exception
        will be raised.
    points_per_chunk
        The maximum number of point targets to simulate in each chunk.
    sample_rate
        Sampling rate in Hertz of the results.
    baseband_frequency
        Frequency used for downconversion during basebanding (carrier frequency).
    max_samples
        The maximum number of samples each receiver will capture per ping. If not
        given, this is calculated from the maximum interval between pings and the
        sampling rate. The maximum length of the trace in seconds can be found by
        dividing `max_samples` by `sample_rate`.
    task_lower_threshold
        When the number of simulation tasks per worker in the scheduler drops below
        this value, add more tasks.
    task_upper_threshold
        When submitting simulation tasks to the scheduler, add enough to ensure
        there are at least this many per worker.
    reduction_node_count
        How many results to sum at each level of the reduction tree.
    reduction_levels
        How many levels to use in the reduction tree before writing the combined
        results to disk.

    """
    super().__init__()

    # Do not overwrite existing results.
    self.result_filename = Path(result_filename)
    if self.result_filename.exists():
        raise ValueError(_("specified output path already exists"))

    # Basic parameter checks.
    if points_per_chunk < 1:
        raise ValueError(_("points per chunk must be at least one"))
    if sample_rate < 1:
        raise ValueError(_("sample rate must be at least one"))
    if baseband_frequency < 0:
        raise ValueError(_("baseband frequency cannot be negative"))
    if reduction_node_count < 2:
        raise ValueError(_("reduction node count must be at least two"))
    if reduction_levels < 1:
        raise ValueError(_("reduction levels must be at least one"))

    # If given, max samples must be given.
    if max_samples is not None and max_samples < 1:
        raise ValueError(_("max samples must be at least one"))

    # Ensure thresholds are valid.
    if task_lower_threshold < 1:
        raise ValueError(_("task lower threshold must be at least one"))
    if task_upper_threshold < task_lower_threshold:
        raise ValueError(
            _("task upper threshold cannot be less than lower threshold")
        )

    self.points_per_chunk = points_per_chunk
    self.sample_rate = sample_rate
    self.baseband_frequency = baseband_frequency
    self.max_samples = max_samples
    self.loop_lower_threshold = task_lower_threshold
    self.loop_upper_threshold = task_upper_threshold
    self.reduction_node_count = reduction_node_count
    self.reduction_levels = reduction_levels

point_simulation_chunk

point_simulation_chunk(chunk, params)

Perform a single chunk of the simulation.

Parameters:

Name	Type	Description	Default
chunk	tuple[int, int, ndarray, ndarray]	The chunk of targets to simulate. This corresponds to one item of the iterator return by openstb.simulator.target.points.target_chunk_iterator.	required
params	tuple[CommonSettings, ChunkSettings]	A tuple of the common and chunk-specific settings.	required

Returns:

Type	Description
echo_fdomain	The frequeny-domain echo for this chunk of the simulation.

Source code in openstb/simulator/controller/simple_points.py

def point_simulation_chunk(
    chunk: tuple[int, int, np.ndarray, np.ndarray],
    params: tuple[CommonSettings, ChunkSettings],
) -> np.ndarray:
    """Perform a single chunk of the simulation.

    Parameters
    ----------
    chunk
        The chunk of targets to simulate. This corresponds to one item of the iterator
        return by [openstb.simulator.target.points.target_chunk_iterator][].
    params
        A tuple of the common and chunk-specific settings.

    Returns
    -------
    echo_fdomain
        The frequeny-domain echo for this chunk of the simulation.

    """
    # Unpack the inputs.
    _, _, position, reflectivity = chunk
    common, settings = params

    # Calculate the travel times.
    tt_result = common.travel_time.calculate(
        common.trajectory,
        settings.ping_time,
        common.environment,
        common.tx_position,
        common.tx_ori,
        settings.rx_position.reshape(1, 3),
        settings.rx_ori.reshape(1, 4),
        position,
    )

    # Start with the transmitted spectrum.
    Schunk = common.S[np.newaxis, :, np.newaxis]

    # Apply any distortions from the transmitter or during its travel to the target.
    for distortion in common.emitted_distortion:
        Schunk = distortion.apply(
            settings.ping_time,
            common.f,
            Schunk,
            common.baseband_frequency,
            common.environment,
            common.signal_frequency_bounds,
            tt_result,
        )

    # Apply the phase shift corresponding to each travel time.
    Schunk = Schunk * np.exp(
        -2j * np.pi * common.f[:, np.newaxis] * tt_result.travel_time[:, np.newaxis, :]
    )
    Schunk *= (tt_result.travel_time <= settings.max_t)[:, np.newaxis, :]

    # Scale by the reflectivity of the target.
    Schunk *= reflectivity

    # Apply any distortions from its travel during return or by the receiver.
    for distortion in common.echo_distortion + settings.rx_distortion:
        Schunk = distortion.apply(
            settings.ping_time,
            common.f,
            Schunk,
            common.baseband_frequency,
            common.environment,
            common.signal_frequency_bounds,
            tt_result,
        )

    # Sum over the targets.
    return Schunk.sum(axis=-1).squeeze()

point_simulation_store

point_simulation_store(storage, ping, receiver, echo_fdomain)

Store the result of a piece of the simulation.

Parameters:

Name	Type	Description	Default
storage	Array	The zarr array to store the result in.	required
ping	int	The ping index of the result.	required
receiver	int	The receiver of the result.	required
echo_fdomain	ndarray	The Fourier domain echoes to add to the storage. This is returned to the time domain and added to the current result in the storage.	required

Source code in openstb/simulator/controller/simple_points.py

def point_simulation_store(
    storage: zarr.Array, ping: int, receiver: int, echo_fdomain: np.ndarray
) -> None:
    """Store the result of a piece of the simulation.

    Parameters
    ----------
    storage
        The zarr array to store the result in.
    ping
        The ping index of the result.
    receiver
        The receiver of the result.
    echo_fdomain
        The Fourier domain echoes to add to the storage. This is returned to the time
        domain and added to the current result in the storage.

    """
    # Return to the time domain.
    result = np.fft.ifft(np.fft.ifftshift(echo_fdomain, axes=-1), norm="forward")

    # Remove any guard band.
    Nt = storage.shape[-1]
    if Nt < result.shape[-1]:
        result = result[..., :Nt]

    if Nt > result.shape[-1]:
        raise ValueError(_("result has fewer samples than expected"))

    with distributed.Lock("write-pressure"):
        storage[ping, receiver, :] += result  # type:ignore[operator]