Tasks

Available tasks

List of supported tasks.

GAS tasks

class app.workspace.tasks.GAS.GASMeasureAll(*args, **kwargs)

Bases: BaseTask

Measures all measurable values and saves them to database.

Extra parameters:

‘device_id’: str - ID of target device, ‘sleep_period’: float - measurement period

end()

End the task.

start()

Start the task.

GMS tasks

class app.workspace.tasks.GMS.GMSMeasureAll(*args, **kwargs)

Bases: BaseTask

Measures all measurable values and saves them to database.

Extra parameters:

  • ‘device_id’: str - ID of target device

  • ‘sleep_period’: float - measurement period

  • ‘task_id’: str - ID of target task

end()

End the task.

start()

Start the task.

PBR tasks

class app.workspace.tasks.PBR.PBRGeneralPump(*args, **kwargs)

Bases: BaseTask, Observer

Turbidostat task based on optical density (OD) values.

Turns on/off (by specifying command for that) when OD reaches max_od/min_od value.

Extra parameters:

  • ‘min_od’: int - lowed OD bound,

  • ‘max_od’: int - upper OD bound,

  • ‘device_id’: str - ID of target device,

  • ‘measure_all_task_id’: str - associated measurement task,

  • ‘pump_on_command’: dict- command to turn on pump,

  • ‘pump_off_command’: dict - command to turn off pump

change_pump_state(state: bool)

Switch state of the pump.

Parameters

state – target state

end()

End the task.

get_pump_command(state: bool) → Command

Create command to change pump state.

Parameters

state – desired pump state

Returns

create Command

is_od_value_too_high(od)
is_od_value_too_low(od)
stabilize(od)

Update state of pump if OD is out of allowed bounds

Parameters

od – current OD

start()

Start the task.

turn_pump_off()
turn_pump_on()
update(observable: Observable)
class app.workspace.tasks.PBR.PBRMeasureAll(*args, **kwargs)

Bases: BaseTask

Measures all measurable values and saves them to database.

Automatically detects outliers for optical density.

Extra parameters:

  • ‘device_id’: str - ID of target device,

  • ‘sleep_period’: float - measurement period,

  • ‘max_outliers’: int - maximum number of outliers on row,

  • ‘pump_id’: int - control pump ID,

  • ‘lower_tol’: int - lower outlier tolerance,

  • ‘upper_tol’: int - upper outlier tolerance,

  • ‘od_attribute’: int - OD channel to be used

calculate_average()

Helper method which calculates the average of a list while removing the elements from the objects deque.

Returns

The average of the deque

end()

End the task.

get_od_for_init()

Measure sample value of OD.

Returns

obtained value

handle_outlier(measured_od) → bool

Decides whether the measured OD value is an outlier or not.

Parameters

measured_od – optical density value

Returns

True if it is an outlier, False otherwise

measure_initial_od_average()

Collect the OD value from 5 measurements and calculate the average OD from the measured data.

Returns

measured value

start()

Start the task.

class app.workspace.tasks.PBR.ePBRMeasureAll(*args, **kwargs)

Bases: PBRMeasureAll

Measures all measurable values and saves them to database.

Extra parameters: ‘device_id’: str - ID of target device, ‘sleep_period’: float - measurement period

SICS tasks

class app.workspace.tasks.SICS.MeasureWeight(*args, **kwargs)

Bases: BaseTask

Measures weight and saves it to database.

Extra parameters:

  • ‘device_id’: str - ID of target device,

  • ‘sleep_period’: float - measurement period

end()

End the task.

start()

Start the task.

TH_IB2 tasks

class app.workspace.tasks.TH_IB2.MeasureAll(*args, **kwargs)

Bases: BaseTask

Measures all measurable values and saves them to database.

Extra parameters:

  • ‘device_id’: str - ID of target device,

  • ‘sleep_period’: float - measurement period

end()

End the task.

start()

Start the task.

U1W_TVSL tasks

class app.workspace.tasks.U1W_TVSL.MeasureAll(*args, **kwargs)

Bases: BaseTask

Measures all measurable values and saves them to database.

Extra parameters:

  • ‘device_id’: str - ID of target device,

  • ‘sleep_period’: float - measurement period

end()

End the task.

start()

Start the task.

General tasks

class app.workspace.tasks.general.MeasureAllDesync(*args, **kwargs)

Bases: BaseTask

Asynchronous periodical measurement of chosen variables.

It is necessary to provide a dictionary, where for each time period particular variables are given:

{time_period_1: {"variable_X": command_X, "variable_Y": command_Y, ...},
 time_period_2: {"variable_Z": command_Z, "variable_W": command_W, ...},
 ...}
end()

End the task.

execute_commands(commands)

Execute given command.

Parameters

commands – given Command

Returns

response

find_gcd_counters(frequency_to_commands)

Find greatest common divisor to find optimal frequency to execute periodical commands.

Parameters

frequency_to_commands – mapping of required frequencies

Returns

optimal frequency

start()

Start the task.

class app.workspace.tasks.general.PeriodicRegime(*args, **kwargs)

Bases: BaseTask

Periodically call specified commands to change a regime.

Fully customisable by two parameters:

  1. intervals - specifies periods when next set of commands should be called e.g.

'intervals': [8, 16],  # change regime after 8 hours and then after 16 hours

  1. commands - list of commands for each regime change which should be executed, e.g.

'commands': [[{'id': '10', 'args': [0, 20]}, {'id': '10', 'args': [1, 20]}],   # night
            [{'id': '10', 'args': [0, 200]}, {'id': '10', 'args': [1, 200]}]]  # day
end()

End the task.

execute_commands(commands)

Execute given command.

Parameters

commands – given Command

Returns

response

start()

Start the task.

Abstract task

Serves as a template for tasks.

class app.src.utils.abstract_task.BaseTask(*args, **kwargs)

Bases: object

An interface which every implemented Task must implement. For additional information, see Wiki.

abstract end()

Terminate the task.

abstract start()

Start the task.

task_id()

ID of the task.

validate_attributes(required, class_name)