Task ID

Methods to handle task id’s

class momotor.options.task_id.StepTaskId(step_id, task_number)

A step-id and task-number pair

step_id: str
task_number: Optional[Tuple[int, ...]]
momotor.options.task_id.apply_task_number(depend_id, task_id)

Replace $ references in dependency strings with their value from the task_id parameter, e.g. $0 in depend_id will be replaced with task_id.task_number[0]

Simple arithmetic on the values can be done, available operators are +, -, *, / and %, for the usual operations add, subtract, multiply, integer division and modulo. Arithmetic operations are evaluated from left to right, there is no operator precedence.

When subtraction results in a negative value or division in infinity, this will not directly throw an exception, but instead will generate an invalid task-id containing #NEG or #INF.

Special value $@ will be replaced with the full task number.

Raises InvalidDependencies if depend_id contains invalid references or is syntactically incorrect.

Examples:

>>> tid = StepTaskId('step', (0, 1, 2))
>>> apply_task_number('test', tid)
'test'

>>> apply_task_number('test.$0', tid)
'test.0'

>>> apply_task_number('test.$0.$1', tid)
'test.0.1'

>>> apply_task_number('test.$1.$2', tid)
'test.1.2'

>>> apply_task_number('test.$0-1.$1-1.$2-1', tid)
'test.#NEG.0.1'

>>> apply_task_number('test.$0+1.$1+1.$2+1', tid)
'test.1.2.3'

>>> apply_task_number('test.$0*2.$1*2.$2*2', tid)
'test.0.2.4'

>>> apply_task_number('test.$0/2.$1/2.$2/2', tid)
'test.0.0.1'

>>> apply_task_number('test.$0%2.$1%2.$2%2', tid)
'test.0.1.0'

>>> apply_task_number('test.$0*2+1.$1*2+1.$2*2+1', tid)
'test.1.3.5'

>>> apply_task_number('test.$0+1*2.$1+1*2.$2+1*2', tid)
'test.2.4.6'

>>> apply_task_number('test.$0+$1+$2', tid)
'test.3'

>>> apply_task_number('test.$1/0', tid)
'test.#INF'

>>> apply_task_number('test.$@', tid)
'test.0.1.2'

>>> apply_task_number('test.$X', tid)
Traceback (most recent call last):
...
momotor.bundles.exception.InvalidDependencies: Task 'step.0.1.2' has invalid dependency 'test.$X'

>>> apply_task_number('test.$1$2', tid)
Traceback (most recent call last):
...
momotor.bundles.exception.InvalidDependencies: Task 'step.0.1.2' has invalid dependency 'test.$1$2'

>>> apply_task_number('test.$1^4', tid)
Traceback (most recent call last):
...
momotor.bundles.exception.InvalidDependencies: Task 'step.0.1.2' has invalid dependency 'test.$1^4'

>>> apply_task_number('test.$9', tid)
Traceback (most recent call last):
...
momotor.bundles.exception.InvalidDependencies: Task 'step.0.1.2' has invalid dependency 'test.$9'
Return type

str

momotor.options.task_id.get_task_id_lookup(task_ids)

Convert an iterable of StepTaskId objects into a lookup table to convert a string representation of a task-id to the StepTaskId

>>> get_task_id_lookup({StepTaskId('step', (0, 0))})
{'step.0.0': StepTaskId(step_id='step', task_number=(0, 0))}
Parameters

task_ids (Iterable[StepTaskId]) – the task ids to convert

Return type

Dict[str, StepTaskId]

Returns

the lookup table

momotor.options.task_id.iter_task_ids(step_id, sub_tasks)

Generate all the task-ids for the subtasks.

>>> list(str(t) for t in iter_task_ids('step', tuple()))
['step']

>>> list(str(t) for t in iter_task_ids('step', (2,)))
['step.0', 'step.1']

>>> list(str(t) for t in iter_task_ids('step', (2, 2)))
['step.0.0', 'step.0.1', 'step.1.0', 'step.1.1']
Parameters

  • step_id (str) – the id of the step

  • sub_tasks (Tuple[int, …]) – sequence of integers with the number of sub-tasks for each level

Return type

Generator[StepTaskId, None, None]

Returns

the task numbers

momotor.options.task_id.iter_task_numbers(sub_tasks)

Generate all the task-numbers for the subtasks.

>>> list(iter_task_numbers(tuple()))
[None]

>>> list(iter_task_numbers((1,)))
[(0,)]

>>> list(iter_task_numbers((3,)))
[(0,), (1,), (2,)]

>>> list(iter_task_numbers((2, 2)))
[(0, 0), (0, 1), (1, 0), (1, 1)]

>>> list(iter_task_numbers((2, 3)))
[(0, 0), (0, 1), (0, 2), (1, 0), (1, 1), (1, 2)]
Parameters

sub_tasks (Tuple[int, …]) – sequence of integers with the number of sub-tasks for each level

Return type

Generator[Optional[Tuple[int, …]], None, None]

Returns

the task numbers

momotor.options.task_id.task_id_from_number(task_number)

Convert a task number (tuple of ints) into a task id (dotted string)

>>> task_id_from_number(None)
''

>>> task_id_from_number((1,))
'1'

>>> task_id_from_number((1, 2,))
'1.2'

>>> task_id_from_number((1, 2, 3,))
'1.2.3'
Return type

str

momotor.options.task_id.task_number_from_id(task_id)

Convert a task_id string into a task number

>>> task_number_from_id('') is None
True

>>> task_number_from_id('1')
(1,)

>>> task_number_from_id('1.2')
(1, 2)

>>> task_number_from_id('1.2.3')
(1, 2, 3)
Return type

Optional[Tuple[int, …]]