hein_control package¶

Subpackages¶

hein_control.automation module¶

DEPRECATED: use the .sequencing.automation module instead.

Module for managing automation sequences (series of Steps).

hein_control.scheduler module¶

Tools for scheduling sequences of actions to execute at prescribed times.

class hein_control.scheduler.ExecuteOnTimePoints(action: Callable, *timepoints, configuration_name: str = None, enabled: bool = True, **kwargs)¶

Bases: hein_control.action.configured.ConfiguredAction

Class for executing an action on the specified timepoints. When the action is triggered, the time point list of the instance will be checked against the currently executing timepoint name. If the condition matches, the action will be executed.

Parameters:	action – action to execute when triggered and the conditions are met timepoints – time point numbers to execute on configuration_name – name for the configuration enabled – whether the action should be enabled (use this to disable any execution) kwargs – kwarg pass through to ConfiguredAction

evaluate_condition() → bool¶: Performs the logic check to determine whether the action should be executed

regex¶: regex pattern for matching timepoint names

class hein_control.scheduler.SamplingScheduler(*actions, time_deltas: List[float] = None, sequence_name: str = None)¶

Bases: hein_control.mixin.reg.InstanceRegistry, hein_control.mixin.serialization.DumpableState

Scheduler for sequences of configured actions. Actions are stored in a configured sequence (sequence attribute).

Parameters:	time_deltas – a list of time deltas to initialize the sequence with (passed directly to insert_time_point_list) actions – callable action or the name of a callable action previously defined as a SchedulerAction sequence_name – convenience reference name for the current experiment

CYCLE_TIME = 0.5¶

UUID_PREFIX = 'sched-'¶

WAIT_IF_BUSY = True¶

apply_default_timepoint_names(prefix: str = 'Time Point #')¶

Applies default timepoint naming across the timepoint list, ensuring consistent naming order.

Parameters:	prefix – prefix to use for naming

as_dict() → dict¶: Returns a key: value representation of the instance

busy¶: whether there are any time points currently executing

completed_time_points¶: number of time points which have been completed

current_time¶: the current time since starting the scheduler

estimated_time_until_next_point¶: estimated time in seconds until next time point

classmethod get_newest_instance() → hein_control.scheduler.SamplingScheduler¶: Gets the newest instance of the sampling scheduler

get_time_point_return(key: str = None) → dict¶

Retrieves the returned values of the specified action time point.

Parameters:	key – identifier key for the desired ActionTimePoint instance. If not specified, the last triggered time point will be retrieved.
Returns:	method returns

insert_time_point(time_delta: float = None, position: int = -1, relative_to: str = 'start', shift_subsequent: bool = True, number_of_points: int = 1, timepoint_name: str = None, additional_actions: Iterable[Union[str, callable, hein_control.timepoint.ActionTimePoint]] = None, specified_actions: Iterable[Union[str, callable, hein_control.timepoint.ActionTimePoint]] = None)¶

Inserts a time point into the list of time points.

Parameters:

time_delta – time delta in seconds from the reference (relative_to)
position – position in the time point list to insert the new time point (default is the end of of the list)
relative_to – whether the specified time delta should be relative to the sequence time (‘start’) or the time point previous to the insertion point (‘previous’)
shift_subsequent – whether to shift subsequent time points in the time point list by the time delta
number_of_points – number of points with these specification to add
timepoint_name – optional name for inserted time point. If multiple points are specified, a counter will be applied.

insert_time_point_list(*time_deltas, position: int = -1, relative_to: str = 'previous', additional_actions: Iterable[Union[str, callable, hein_control.timepoint.ActionTimePoint]] = None, specified_actions: Iterable[Union[str, callable, hein_control.timepoint.ActionTimePoint]] = None)¶

Inserts a list of time points into the current list of time points.

Parameters:	time_deltas – list of time deltas (seconds) position – position in the current list to insert the incoming time points relative_to – whether the specified time delta should be relative to the sequence time (‘start’) or the time point previous to the insertion point (‘previous’; default)

insert_time_point_list_pairs(*num_delta_pairs, position: int = -1, relative_to: str = 'previous', additional_actions: Iterable[Union[str, callable, hein_control.timepoint.ActionTimePoint]] = None, specified_actions: Iterable[Union[str, callable, hein_control.timepoint.ActionTimePoint]] = None)¶

Inserts a list of time points in number-of-points/time-delta form.

e.g. 3, 10 will result in 3 time points with 10 second spacing. 3, 10, 5, 20 will results in 3 time points with 10 second spacing followed by 5 time points with 20 second spacing.

Pairs may also be provided in number/time-delta tuples. e.g. (3, 10), (5, 20)

Parameters:	num_delta_pairs – pairs of arguments in (number-of-points, time_delta) form position – position in the current list to insert the incoming time points relative_to – whether the specified time delta should be relative to the sequence time (‘start’) or the time point previous to the insertion point (‘previous’; default)

join()¶: waits for defined time points in the scheduler to finish

kill_sequence()¶: This method will kill the sequence and perform cleanup. You will not be able to restart the thread after calling this method

last_triggered_time_point¶: The last time point which was triggered

next_time_point¶: The next time point for triggering

pause_sequence()¶: Pauses the sequence (actions will not be automatically triggered until resumed)

paused¶: whether the sequence is paused

remaining_time_points¶: number of time points remaining to be sampled

start_sequence()¶: Starts the sequence loop and triggers time points as specified

started¶: whether the sequence has been started

terminated¶: whether the sequence has been terminated

time_points¶: list of time points for the sequence

total_time_points¶: total number of time points associated with the scheduler instance

trigger(additional_actions: Iterable[Union[str, callable, hein_control.timepoint.ActionTimePoint]] = None, specified_actions: Iterable[Union[str, callable, hein_control.timepoint.ActionTimePoint]] = None, wait: bool = None, timepoint_name: str = None, **action_kwargs) → hein_control.timepoint.ActionTimePoint¶

Manually triggers the sequence with the actions provided. If no actions are provided, the default actions of the instance are used. If additional actions are provided, those actions will be performed in addition to the default actions (actions will be performed in order after the default actions are complete). If specified_actions are provided, only those actions will be performed.

Parameters:

additional_actions – actions to run in addition to the default actions of the scheduler instance. These actions will be run after the default actions.
specified_actions – actions to run instead of the default actions of the instance
wait – whether to wait for completion of the actions before returning
timepoint_name – optional name for the timepoint
action_kwargs – keyword arguments for the actions. Keyword arguments will be passed through to their respective methods. See ActionTimePoint.trigger for details on kwarg handling

Returns:

created ActionTimePoint instance

triggered_time_points¶: number of triggered time points

uncompleted_time_points¶: number of time points which have not been completed

uuid¶: uuid4 for the scheduler

visualize_timepoints()¶

wait_for_time_point_completion()¶: If a time point is being executed, this method will wait for it to be complete

write_timepoints_to_csv(file_name: str = None, overwrite: bool = True)¶

Writes the time point data to the specified csv file. If no filename is passed, the time points will be saved to “YYYY-MM-DD scheduler output.csv”.

Parameters:	file_name – file name to write to overwrite – whether to overwrite the filename if it already exists

hein_control.timepoint module¶

Module for managing time points in a sequence. TimePoint simply manages a time point relative to some start, and ActionTimePoint manages context, sequences of actions, and execution of those actions when desired.

class hein_control.timepoint.ActionTimePoint(time_delta: float, configuration: Union[str, hein_control.sequencing.execution.ConfiguredSequence], name: str = None, **action_kwargs)¶

Bases: hein_control.timepoint.TimePoint, hein_control.sequencing.execution.ExecutionSequence

A class for describing and tracking an event time point relative to some other time where actions (e.g. a sampling sequence) are to be triggered. The sequence of actions is stored in a ConfiguredSequence instance.

Parameters:	time_delta – time delta for the time point (relative to some other time) name – convenience name for time point differentiation actions – action methods to call. These methods will be called in the order provided action_kwargs – action kwargs to associate with list of actions (primary keys must map to function names)

UUID_PREFIX = 'atp-'¶

as_dict() → dict¶: Returns a key: value representation of the instance

copy_configuration() → hein_control.timepoint.ActionTimePoint¶

Copies the configuration of an ActionTimePoint instance.

Returns:	ActionTimePoint instance with identical configuration

classmethod create_from_actions(time_delta: float, actions: Iterable[Union[str, ConfiguredAction]], name: str = None) → hein_control.timepoint.ActionTimePoint¶

Creates an instance from a list of configured actions. This differs from calling init by accepting an iterable argument containing actions which are used to create the ConfiguredSequence required by init.

Parameters:	time_delta – time delta for the time point (relative to some other time) actions – configured actions or named configurations name – name for the time point (optional)
Returns:	configured instance

relative_start_times¶: relative start times for actions

remove_action(action: Union[str, Callable, hein_control.action.basic.Action])¶

Removes an action from the list of actions associated with a timepoint.

Parameters:	action – action identifier
Returns:

string_time_delta¶: formatted string version of time delta

trigger(wait: bool = None, **kwargs)¶

Triggers (starts execution of) the action sequence of the ActionTimePoint. Keyword arguments will be passed through to their respective action methods by the name of the action. This method executes the built-in __call__ of ExecutionSequence and it will block until the thread of the ExecutionSequence has started.

e.g. if there is a method named “do_this_thing” which expects the keyword argument “value”, the expected syntax of kwargs would be

{

“do_this_thing”: {“value”: 42},

}

Execution Tree: 1. Execution sequence __call__ 2. ThreadedTrackedAction using operators.execute_sequence_actions action 3. Waits until the thread starts (prevents double-triggering 4. waits for completion if specified

Parameters:	wait – whether to wait for completion before returning kwargs – keyword arguments for the actions. Keyword arguments will be passed through to their respective methods.

class hein_control.timepoint.TimePoint(time_delta: float, name: str = None, description: str = None, parent_sequence_id: str = None)¶

Bases: hein_control.mixin.reg.InstanceRegistry, hein_control.mixin.serialization.DumpableState

A class for describing and tracking an event time point relative to some start time.

Parameters:	time_delta – time delta for the time point (seconds; relative to some other time) name – convenience name for time point differentiation description – extended description for differentiation parent_sequence_id – uuid of parent SamplingScheduler (if applicable)

UUID_PREFIX = 'tp-'¶

as_dict() → dict¶: Returns a key: value representation of the instance

classmethod get_instance(key: str) → hein_control.timepoint.TimePoint¶

Retreives a timepoint instance by key. The key may be a uuid, or class-uuid

Parameters:	key – key to search by
Returns:	TimePoint (or subclass) instance

sample_time_from_reference(reference_time: float)¶

Returns the sample time for the time point from the reference time (in seconds)

Parameters:	reference_time – reference time to calculate from
Returns:	time to trigger (seconds)

uuid¶: uuid for the time point