Interdisciplinary Workflow

Interdisciplinary Workflow Documentation sub-package for MaRDMO.

Provides everything needed to document interdisciplinary research workflows within RDMO:

• :mod:~MaRDMO.workflow.models — Dataclasses for ProcessStep, Method, Software, Hardware, and DataSet entities from MaRDI Portal / Wikidata, as well as Variables and Parameters.
• :mod:~MaRDMO.workflow.handlers — Signal handlers that populate the questionnaire when a software, hardware, instrument, data set, method, or process-step ID is saved.
• :mod:~MaRDMO.workflow.providers — RDMO optionset providers for searching software, hardware, instruments, data sets, methods, process steps, disciplines, and workflow tasks in external knowledge graphs.
• :mod:~MaRDMO.workflow.worker — Background worker for generating workflow previews and exporting workflow entries to the MaRDI Portal.
• :mod:~MaRDMO.workflow.utils — Helpers for partitioning disciplines, extracting data-set size, reference, and archive metadata.
• :mod:~MaRDMO.workflow.constants — URI-prefix maps, property lists, and reproducibility vocabulary for the workflow catalog.

Handlers

Module containing Handlers for the Interdisciplinary Workflow Documentation.

Information inherits _entry, _collect_existing_ids, _hydrate_relatants, and _fill from BaseInformation (MaRDMO/handler_base.py).

Information

Bases: BaseInformation

Handlers for the Workflow Documentation questionnaire.

Source code in MaRDMO/workflow/handlers.py

class Information(BaseInformation):
    '''Handlers for the Workflow Documentation questionnaire.'''

    _ENTITY_KEYS = ('Workflow', 'Algorithm', 'Software', 'Hardware', 'Data Set', 'Process Step')

    def __init__(self):
        '''Load workflow questions, base URI, RDMO options, and MathAlgoDB registry.'''
        self.questions  = get_questions('workflow') | get_questions('publication')
        self.base       = BASE_URI
        self.options    = get_options()
        self.mathalgodb = get_mathalgodb()
        self.mathmoddb  = get_mathmoddb()

    # ------------------------------------------------------------------ #
    #  Public entry points (called by router via post_save signal)         #
    # ------------------------------------------------------------------ #

    def workflow(self, instance):
        '''Handle Workflow ID save: hydrate basics and SPARQL data.

        Args:
            instance: RDMO :class:`~rdmo.projects.models.Value` that was just saved.
        '''
        self._entry(instance, 'Workflow', self._fill_workflow_batch)

    def software(self, instance):
        '''Handle Software ID save: hydrate basics and SPARQL data.

        Args:
            instance: RDMO :class:`~rdmo.projects.models.Value` that was just saved.
        '''
        self._entry(instance, 'Software', self._fill_software_batch)

    def hardware(self, instance):
        '''Handle Hardware ID save: hydrate basics and SPARQL data.

        Args:
            instance: RDMO :class:`~rdmo.projects.models.Value` that was just saved.
        '''
        self._entry(instance, 'Hardware', self._fill_hardware_batch)

    def processor_cores(self, instance):
        '''Handle CPU ID save: query and write number of processor cores.

        Args:
            instance: RDMO :class:`~rdmo.projects.models.Value` that was just saved.
        '''
        if not instance.external_id:
            return

        data_by_id = _fetch_by_source(
            [(instance.text, instance.external_id, instance.set_index)],
            'workflow/queries/cpu_mardi.sparql',
            'workflow/queries/cpu_wikidata.sparql',
            Cpu,
        )
        data = data_by_id.get(instance.external_id)
        if not data or not data.cores:
            return

        value_editor(
            project=instance.project,
            uri=f'{self.base}{self.questions["Hardware"]["Cores"]["uri"]}',
            info={'text': data.cores,
                  'set_prefix': instance.set_prefix,
                  'set_index': instance.set_index})

    def data_set(self, instance):
        '''Handle Data Set ID save: hydrate basics and SPARQL data.

        Args:
            instance: RDMO :class:`~rdmo.projects.models.Value` that was just saved.
        '''
        self._entry(instance, 'Data Set', self._fill_data_set_batch)

    def algorithm(self, instance):
        '''Handle Algorithm ID save: hydrate basics and SPARQL data.

        Args:
            instance: RDMO :class:`~rdmo.projects.models.Value` that was just saved.
        '''
        self._entry(instance, 'Algorithm', self._fill_algorithm_batch)

    def process_step(self, instance):
        '''Handle Process Step ID save: hydrate basics and cascade to all related entities.

        Args:
            instance: RDMO :class:`~rdmo.projects.models.Value` that was just saved.
        '''
        self._entry(instance, 'Process Step', self._fill_process_step_batch)

    def model(self, instance):
        '''Handle Workflow Model ID save: add linked Tasks from the MaRDI Portal.

        Queries the MaRDI Portal for computational tasks linked to the selected model
        and adds any not yet present as Task values. Already-present tasks are left
        untouched so tasks from previously selected models are preserved.

        Args:
            instance: RDMO :class:`~rdmo.projects.models.Value` that was just saved.
        '''

        if not instance.external_id:
            return

        tasks = self._fetch_tasks_for_model(instance.external_id)

        if not tasks:
            return

        for idx, (identifier, label, description) in enumerate(tasks):
            value_editor(
                project = instance.project,
                uri = f'{self.base}{self.questions["Workflow"]["Task"]["uri"]}',
                info = {
                    'external_id': identifier,
                    'text': f'{label} ({description}) [mardi]',
                    'collection_index': idx,
                    'set_prefix': instance.set_prefix,
                    'set_index': instance.set_index
                },
            )

    # ------------------------------------------------------------------ #
    #  Private helpers                                                     #
    # ------------------------------------------------------------------ #

    def _fetch_tasks_for_model(self, model_id):
        '''Run ``task_mardi.sparql`` for *id_value* and return parsed task tuples.

        Args:
            id_value: Raw Wikibase QID string (e.g. ``"Q1234"``), without prefix.

        Returns:
            List of ``(identifier, label, description)`` tuples, or empty list when
            the portal returns no results or the query fails.
        '''
        _, id_value = model_id.split(':')

        query = get_sparql_query('workflow/queries/task_mardi.sparql').format(
            id_value,
            **get_items(),
            **get_properties(),
        )

        results = query_sparql(query, get_url('mardi', 'sparql'))

        if not results or not results[0].get('usedBy', {}).get('value'):
            return []

        tasks = []
        for raw in results[0]['usedBy']['value'].split(' <|> '):
            try:
                identifier, label, description = raw.split(' || ')
                tasks.append((identifier, label, description))
            except ValueError:
                logger.warning('Unexpected task format, skipping: %r', raw)

        return tasks

    # ------------------------------------------------------------------ #
    #  Batch _fill_* methods (one SPARQL query for N entities)            #
    # ------------------------------------------------------------------ #

    def _fill_workflow_batch(self, project, items, catalog='', visited=None):
        '''Hydrate multiple Workflow pages with a single SPARQL query per source.

        Writes basics, research objective, procedure (long description),
        mathematical models with task qualifiers, process-step relatant pointers,
        reproducibility statements (Yes + optional comment), and transferability
        comments.  Then cascades into the Process Step section via
        :meth:`_hydrate_relatants`.

        Args:
            project:  RDMO project instance.
            items:    List of ``(text, external_id, set_index)`` tuples to process.
            catalog:  Active catalog URI suffix (default ``""``).
            visited:  Set of external IDs already processed (mutated to avoid cycles).
        '''
        if not items:
            return

        workflow_q = self.questions['Workflow']
        data_by_id = _fetch_by_source(
            items,
            'workflow/queries/workflow_mardi.sparql',
            'workflow/queries/workflow_wikidata.sparql',
            Workflow,
        )
        section_indices = {}
        for text, external_id, set_index in items:
            data = data_by_id.get(external_id)
            if not data:
                continue

            add_basics(project=project, text=text, questions=self.questions,
                       item_type='Workflow', index=(0, set_index))

            if data.research_objective:
                value_editor(
                    project=project,
                    uri=f'{self.base}{workflow_q["Objective"]["uri"]}',
                    info={'text': ' | '.join(data.research_objective),
                          'set_prefix': set_index})

            for i, proc in enumerate(data.procedure):
                value_editor(
                    project=project,
                    uri=f'{self.base}{workflow_q["Long Description"]["uri"]}',
                    info={'text': proc, 'set_prefix': set_index, 'collection_index': i})

            model_order = []
            model_tasks = {}
            for raw in data.uses_model:
                parts = raw.split(' || ')
                while len(parts) < 6:
                    parts.append('')
                model_id, model_label, model_desc, task_id, task_label, task_desc = parts[:6]
                if not model_id:
                    continue
                if model_id not in model_tasks:
                    model_order.append((model_id, model_label, model_desc))
                    model_tasks[model_id] = []
                if task_id:
                    model_tasks[model_id].append((task_id, task_label, task_desc))

            for model_idx, (model_id, model_label, model_desc) in enumerate(model_order):
                source = model_id.split(':')[0]
                value_editor(
                    project=project,
                    uri=f'{self.base}{workflow_q["Model"]["uri"]}',
                    info={'text': f'{model_label} ({model_desc}) [{source}]',
                          'external_id': model_id,
                          'set_prefix': f"{set_index}|0", 'set_index': model_idx})
                for task_idx, (task_id, task_label, task_desc) in enumerate(model_tasks[model_id]):
                    source = task_id.split(':')[0]
                    value_editor(
                        project=project,
                        uri=f'{self.base}{workflow_q["Task"]["uri"]}',
                        info={'text': f'{task_label} ({task_desc}) [{source}]',
                              'external_id': task_id,
                              'set_prefix': f"{set_index}|0", 'set_index': model_idx,
                              'collection_index': task_idx})

            for i, ps in enumerate(data.contains_process_step):
                if ps.id:
                    source = ps.id.split(':')[0]
                    value_editor(
                        project=project,
                        uri=f'{self.base}{workflow_q["PSRelatant"]["uri"]}',
                        info={'text': f'{ps.label} ({ps.description}) [{source}]',
                              'external_id': ps.id,
                              'set_prefix': set_index, 'collection_index': i})

            for value, comment, q_key in [
                (data.mathematical,     data.mathematical_comment,    'Mathematical'),
                (data.runtime,          data.runtime_comment,         'Runtime'),
                (data.result,           data.result_comment,          'Result'),
                (data.originalplatform, data.originalplatform_comment, 'Original Platform'),
                (data.otherplatform,    data.otherplatform_comment,   'Other Platform'),
            ]:
                if value == 'Yes':
                    value_editor(
                        project=project,
                        uri=f'{self.base}{workflow_q[q_key]["uri"]}',
                        info={'option': self.options['YesLargeText'],
                              'text': comment or '',
                              'set_prefix': f"{set_index}|0"})
                elif value == 'No':
                    value_editor(
                        project=project,
                        uri=f'{self.base}{workflow_q[q_key]["uri"]}',
                        info={'option': self.options['No'],
                              'set_prefix': f"{set_index}|0"})

            if data.transferable == 'Yes':
                for i, comment in enumerate(data.transferable_comment):
                    value_editor(
                        project=project,
                        uri=f'{self.base}{workflow_q["Transferability"]["uri"]}',
                        info={'text': comment,
                              'set_prefix': set_index,
                              'set_index': 0, 'collection_index': i})

            add_relations_flexible(
                project=project, data=data,
                props={'keys': PROPS['IW2IW'], 'mapping': self.mathmoddb},
                index={'set_prefix': set_index},
                statement={
                    'relation': f'{self.base}{workflow_q["IntraClassRelation"]["uri"]}',
                    'relatant': f'{self.base}{workflow_q["IntraClassElement"]["uri"]}',
                })

            self._hydrate_publications(project, data.publications, catalog, visited)

            self._hydrate_relatants(
                project=project, data=data, prop_keys=['contains_process_step'],
                spec=_RelatantSpec(
                    question_id_uri=f'{self.base}{self.questions["Process Step"]["ID"]["uri"]}',
                    question_set_uri=f'{self.base}{self.questions["Process Step"]["uri"]}',
                    prefix='PS',
                    fill_method=partial(self._fill, item_type='Process Step',
                                        batch_fill_method=self._fill_process_step_batch),
                    catalog=catalog, visited=visited,
                    batch_fill_method=self._fill_process_step_batch,
                    section_indices=section_indices,
                ))

    def _fill_hardware_batch(self, project, items, catalog='', visited=None):
        '''Hydrate multiple Hardware pages with a single SPARQL query per source.

        Args:
            project:  RDMO project instance.
            items:    List of ``(text, external_id, set_index)`` tuples to process.
            catalog:  Active catalog URI suffix (default ``""``).
            visited:  Set of external IDs already processed (mutated to avoid cycles).
        '''
        if not items:
            return
        if visited is None:
            visited = set()

        hardware   = self.questions['Hardware']
        data_by_id = _fetch_by_source(
            items,
            'workflow/queries/hardware_mardi.sparql',
            'workflow/queries/hardware_wikidata.sparql',
            Hardware,
        )

        for text, external_id, set_index in items:
            data = data_by_id.get(external_id)
            if not data:
                continue

            add_basics(project=project, text=text, questions=self.questions,
                       item_type='Hardware', index=(0, set_index))

            if data.nodes:
                value_editor(
                    project=project,
                    uri=f'{self.base}{hardware["Nodes"]["uri"]}',
                    info={'text': data.nodes, 'set_prefix': set_index})

            for i, cpu in enumerate(data.cpu):
                source, _ = cpu.id.split(':')
                cpu_value, _ = value_editor(
                    project=project,
                    uri=f'{self.base}{hardware["CPU"]["uri"]}',
                    info={'text': f'{cpu.label} ({cpu.description}) [{source}]',
                          'external_id': cpu.id,
                          'set_prefix': f"{set_index}|0", 'set_index': i})
                if cpu.count:
                    value_editor(
                        project=project,
                        uri=f'{self.base}{hardware["Number of CPU"]["uri"]}',
                        info={'text': cpu.count,
                              'set_prefix': f"{set_index}|0", 'set_index': i})
                self.processor_cores(cpu_value)

            self._hydrate_publications(project, data.publications, catalog, visited)

    def _fill_data_set_batch(self, project, items, catalog='', visited=None):
        '''Hydrate multiple Data Set pages with a single SPARQL query per source.

        Writes basics, size, file format, binary/text type, proprietary flag,
        and publication/archival statements.

        Args:
            project:  RDMO project instance.
            items:    List of ``(text, external_id, set_index)`` tuples to process.
            catalog:  Active catalog URI suffix (default ``""``).
            visited:  Set of external IDs already processed (mutated to avoid cycles).
        '''
        if not items:
            return
        if visited is None:
            visited = set()

        data_set_q = self.questions['Data Set']
        data_by_id = _fetch_by_source(
            items,
            'workflow/queries/data_set_mardi.sparql',
            'workflow/queries/data_set_wikidata.sparql',
            DataSet,
        )

        for text, external_id, set_index in items:
            data = data_by_id.get(external_id)
            if not data:
                continue

            add_basics(project=project, text=text, questions=self.questions,
                       item_type='Data Set', index=(0, set_index))

            if data.size:
                value_editor(
                    project=project,
                    uri=f'{self.base}{data_set_q["Size"]["uri"]}',
                    info={'text': data.size[1], 'option': data.size[0],
                          'set_prefix': set_index})

            if data.file_format:
                value_editor(
                    project=project,
                    uri=f'{self.base}{data_set_q["File Format"]["uri"]}',
                    info={'text': data.file_format, 'set_prefix': set_index})

            if data.binary_or_text:
                value_editor(
                    project=project,
                    uri=f'{self.base}{data_set_q["Binary or Text"]["uri"]}',
                    info={'option': data.binary_or_text, 'set_prefix': set_index})

            if data.proprietary:
                value_editor(
                    project=project,
                    uri=f'{self.base}{data_set_q["Proprietary"]["uri"]}',
                    info={'option': data.proprietary, 'set_prefix': set_index})

            if data.to_publish:
                value_editor(
                    project=project,
                    uri=f'{self.base}{data_set_q["To Publish"]["uri"]}',
                    info={'text': data.to_publish[1], 'option': data.to_publish[0],
                          'set_prefix': set_index})

            if data.to_archive:
                value_editor(
                    project=project,
                    uri=f'{self.base}{data_set_q["To Archive"]["uri"]}',
                    info={'text': data.to_archive[1][:4], 'option': data.to_archive[0],
                          'set_prefix': set_index})

            self._hydrate_publications(project, data.publications, catalog, visited)

    def _fill_process_step_batch(self, project, items, catalog='', visited=None):
        '''Hydrate multiple Process Step pages with a single SPARQL query per source.

        Writes basics and all relation fields (input/output data sets, algorithms
        with software/hardware qualifiers, experimental methods, fields of work).
        Cascades into Data Set, Algorithm, Software, and Hardware sections via
        :meth:`_hydrate_relatants` instead of relying on signal-driven cascades.

        Args:
            project:  RDMO project instance.
            items:    List of ``(text, external_id, set_index)`` tuples to process.
            catalog:  Active catalog URI suffix (default ``""``).
            visited:  Set of external IDs already processed (mutated to avoid cycles).
        '''
        if not items:
            return
        if visited is None:
            visited = set()

        process_step = self.questions['Process Step']
        data_by_id   = _fetch_by_source(
            items,
            'workflow/queries/process_step_mardi.sparql',
            'workflow/queries/process_step_wikidata.sparql',
            ProcessStep,
        )
        section_indices = {}
        for text, external_id, set_index in items:
            data = data_by_id.get(external_id)
            if not data:
                continue

            add_basics(project=project, text=text, questions=self.questions,
                       item_type='Process Step', index=(0, set_index))

            add_relations_static(
                project=project, data=data,
                props={'keys': PROPS['PS2IDS']},
                index={'set_prefix': set_index},
                statement={'relatant': f'{self.base}{process_step["Input"]["uri"]}'})

            self._hydrate_relatants(
                project=project, data=data, prop_keys=PROPS['PS2IDS'],
                spec=_RelatantSpec(
                    question_id_uri=f'{self.base}{self.questions["Data Set"]["ID"]["uri"]}',
                    question_set_uri=f'{self.base}{self.questions["Data Set"]["uri"]}',
                    prefix='DS',
                    fill_method=partial(self._fill, item_type='Data Set',
                                        batch_fill_method=self._fill_data_set_batch),
                    catalog=catalog, visited=visited,
                    batch_fill_method=self._fill_data_set_batch,
                    section_indices=section_indices,
                ))

            add_relations_static(
                project=project, data=data,
                props={'keys': PROPS['PS2ODS']},
                index={'set_prefix': set_index},
                statement={'relatant': f'{self.base}{process_step["Output"]["uri"]}'})

            self._hydrate_relatants(
                project=project, data=data, prop_keys=PROPS['PS2ODS'],
                spec=_RelatantSpec(
                    question_id_uri=f'{self.base}{self.questions["Data Set"]["ID"]["uri"]}',
                    question_set_uri=f'{self.base}{self.questions["Data Set"]["uri"]}',
                    prefix='DS',
                    fill_method=partial(self._fill, item_type='Data Set',
                                        batch_fill_method=self._fill_data_set_batch),
                    catalog=catalog, visited=visited,
                    batch_fill_method=self._fill_data_set_batch,
                    section_indices=section_indices,
                ))

            add_relations_static(
                project=project, data=data,
                props={'keys': PROPS['PS2A']},
                index={'set_prefix': f'{set_index}|0'},
                statement={
                    'relatant':      f'{self.base}{process_step["Algorithm"]["uri"]}',
                    'platform':      f'{self.base}{process_step["Software"]["uri"]}',
                    'hardware':      f'{self.base}{process_step["Hardware"]["uri"]}',
                    'documentation': (
                        f'{self.base}{process_step["Software-Documentation"]["uri"]}'
                    ),
                    'parameter':     f'{self.base}{process_step["Algorithm-Parameter"]["uri"]}',
                })

            self._hydrate_relatants(
                project=project, data=data, prop_keys=PROPS['PS2A'],
                spec=_RelatantSpec(
                    question_id_uri=f'{self.base}{self.questions["Algorithm"]["ID"]["uri"]}',
                    question_set_uri=f'{self.base}{self.questions["Algorithm"]["uri"]}',
                    prefix='A',
                    fill_method=partial(self._fill, item_type='Algorithm',
                                        batch_fill_method=self._fill_algorithm_batch),
                    catalog=catalog, visited=visited,
                    batch_fill_method=self._fill_algorithm_batch,
                    section_indices=section_indices,
                ))

            self._hydrate_qualifier_entities(
                project=project, data=data, prop_keys=PROPS['PS2A'],
                spec=_RelatantSpec(
                    question_id_uri=f'{self.base}{self.questions["Software"]["ID"]["uri"]}',
                    question_set_uri=f'{self.base}{self.questions["Software"]["uri"]}',
                    prefix='S',
                    fill_method=partial(self._fill, item_type='Software',
                                        batch_fill_method=self._fill_software_batch),
                    catalog=catalog, visited=visited,
                    batch_fill_method=self._fill_software_batch,
                    section_indices=section_indices,
                ))

            self._hydrate_qualifier_entities(
                project=project, data=data,
                prop_keys=PROPS['PS2A'] + PROPS['PS2M'],
                spec=_RelatantSpec(
                    question_id_uri=f'{self.base}{self.questions["Hardware"]["ID"]["uri"]}',
                    question_set_uri=f'{self.base}{self.questions["Hardware"]["uri"]}',
                    prefix='HW',
                    fill_method=partial(self._fill, item_type='Hardware',
                                        batch_fill_method=self._fill_hardware_batch),
                    catalog=catalog, visited=visited,
                    batch_fill_method=self._fill_hardware_batch,
                    section_indices=section_indices,
                ),
                attr='hardware')

            add_relations_static(
                project=project, data=data,
                props={'keys': PROPS['PS2M']},
                index={'set_prefix': f'{set_index}|0'},
                statement={
                    'relatant':      f'{self.base}{process_step["Method"]["uri"]}',
                    'platform':      f'{self.base}{process_step["Instrument"]["uri"]}',
                    'hardware':      f'{self.base}{process_step["Hardware"]["uri"]}',
                    'documentation': (
                        f'{self.base}{process_step["Method-Documentation"]["uri"]}'
                    ),
                    'parameter':     f'{self.base}{process_step["Method-Parameter"]["uri"]}',
                })

            add_relations_static(
                project=project, data=data,
                props={'keys': PROPS['PS2F']},
                index={'set_prefix': set_index},
                statement={'relatant': f'{self.base}{process_step["RFRelatant"]["uri"]}'})

            self._hydrate_publications(project, data.publications, catalog, visited)

__init__()

Load workflow questions, base URI, RDMO options, and MathAlgoDB registry.

Source code in MaRDMO/workflow/handlers.py

def __init__(self):
    '''Load workflow questions, base URI, RDMO options, and MathAlgoDB registry.'''
    self.questions  = get_questions('workflow') | get_questions('publication')
    self.base       = BASE_URI
    self.options    = get_options()
    self.mathalgodb = get_mathalgodb()
    self.mathmoddb  = get_mathmoddb()

algorithm(instance)

Handle Algorithm ID save: hydrate basics and SPARQL data.

Parameters:

Name	Type	Description	Default
instance		RDMO :class:~rdmo.projects.models.Value that was just saved.	required

Source code in MaRDMO/workflow/handlers.py

def algorithm(self, instance):
    '''Handle Algorithm ID save: hydrate basics and SPARQL data.

    Args:
        instance: RDMO :class:`~rdmo.projects.models.Value` that was just saved.
    '''
    self._entry(instance, 'Algorithm', self._fill_algorithm_batch)

data_set(instance)

Handle Data Set ID save: hydrate basics and SPARQL data.

Parameters:

Name	Type	Description	Default
instance		RDMO :class:~rdmo.projects.models.Value that was just saved.	required

Source code in MaRDMO/workflow/handlers.py

def data_set(self, instance):
    '''Handle Data Set ID save: hydrate basics and SPARQL data.

    Args:
        instance: RDMO :class:`~rdmo.projects.models.Value` that was just saved.
    '''
    self._entry(instance, 'Data Set', self._fill_data_set_batch)

hardware(instance)

Handle Hardware ID save: hydrate basics and SPARQL data.

Parameters:

Name	Type	Description	Default
instance		RDMO :class:~rdmo.projects.models.Value that was just saved.	required

Source code in MaRDMO/workflow/handlers.py

def hardware(self, instance):
    '''Handle Hardware ID save: hydrate basics and SPARQL data.

    Args:
        instance: RDMO :class:`~rdmo.projects.models.Value` that was just saved.
    '''
    self._entry(instance, 'Hardware', self._fill_hardware_batch)

model(instance)

Handle Workflow Model ID save: add linked Tasks from the MaRDI Portal.

Queries the MaRDI Portal for computational tasks linked to the selected model and adds any not yet present as Task values. Already-present tasks are left untouched so tasks from previously selected models are preserved.

Parameters:

Name	Type	Description	Default
instance		RDMO :class:~rdmo.projects.models.Value that was just saved.	required

Source code in MaRDMO/workflow/handlers.py

def model(self, instance):
    '''Handle Workflow Model ID save: add linked Tasks from the MaRDI Portal.

    Queries the MaRDI Portal for computational tasks linked to the selected model
    and adds any not yet present as Task values. Already-present tasks are left
    untouched so tasks from previously selected models are preserved.

    Args:
        instance: RDMO :class:`~rdmo.projects.models.Value` that was just saved.
    '''

    if not instance.external_id:
        return

    tasks = self._fetch_tasks_for_model(instance.external_id)

    if not tasks:
        return

    for idx, (identifier, label, description) in enumerate(tasks):
        value_editor(
            project = instance.project,
            uri = f'{self.base}{self.questions["Workflow"]["Task"]["uri"]}',
            info = {
                'external_id': identifier,
                'text': f'{label} ({description}) [mardi]',
                'collection_index': idx,
                'set_prefix': instance.set_prefix,
                'set_index': instance.set_index
            },
        )

process_step(instance)

Handle Process Step ID save: hydrate basics and cascade to all related entities.

Parameters:

Name	Type	Description	Default
instance		RDMO :class:~rdmo.projects.models.Value that was just saved.	required

Source code in MaRDMO/workflow/handlers.py

def process_step(self, instance):
    '''Handle Process Step ID save: hydrate basics and cascade to all related entities.

    Args:
        instance: RDMO :class:`~rdmo.projects.models.Value` that was just saved.
    '''
    self._entry(instance, 'Process Step', self._fill_process_step_batch)

processor_cores(instance)

Handle CPU ID save: query and write number of processor cores.

Parameters:

Name	Type	Description	Default
instance		RDMO :class:~rdmo.projects.models.Value that was just saved.	required

Source code in MaRDMO/workflow/handlers.py

def processor_cores(self, instance):
    '''Handle CPU ID save: query and write number of processor cores.

    Args:
        instance: RDMO :class:`~rdmo.projects.models.Value` that was just saved.
    '''
    if not instance.external_id:
        return

    data_by_id = _fetch_by_source(
        [(instance.text, instance.external_id, instance.set_index)],
        'workflow/queries/cpu_mardi.sparql',
        'workflow/queries/cpu_wikidata.sparql',
        Cpu,
    )
    data = data_by_id.get(instance.external_id)
    if not data or not data.cores:
        return

    value_editor(
        project=instance.project,
        uri=f'{self.base}{self.questions["Hardware"]["Cores"]["uri"]}',
        info={'text': data.cores,
              'set_prefix': instance.set_prefix,
              'set_index': instance.set_index})

software(instance)

Handle Software ID save: hydrate basics and SPARQL data.

Parameters:

Name	Type	Description	Default
instance		RDMO :class:~rdmo.projects.models.Value that was just saved.	required

Source code in MaRDMO/workflow/handlers.py

def software(self, instance):
    '''Handle Software ID save: hydrate basics and SPARQL data.

    Args:
        instance: RDMO :class:`~rdmo.projects.models.Value` that was just saved.
    '''
    self._entry(instance, 'Software', self._fill_software_batch)

workflow(instance)

Handle Workflow ID save: hydrate basics and SPARQL data.

Parameters:

Name	Type	Description	Default
instance		RDMO :class:~rdmo.projects.models.Value that was just saved.	required

Source code in MaRDMO/workflow/handlers.py

def workflow(self, instance):
    '''Handle Workflow ID save: hydrate basics and SPARQL data.

    Args:
        instance: RDMO :class:`~rdmo.projects.models.Value` that was just saved.
    '''
    self._entry(instance, 'Workflow', self._fill_workflow_batch)

Models

Dataclasses that represent entities in the Interdisciplinary Workflow documentation catalog.

Each class maps to one entity type collected from MaRDI Portal and Wikidata during workflow documentation. Instances are populated from SPARQL query results and carry the fields needed to render questionnaire answers and export entries.

Provides:

• :class:ProcessStepUsage — algorithm/method usage within a process step
• :class:ProcessStep — process steps associated with a workflow
• :class:Software — software associated with a workflow
• :class:Hardware — hardware associated with a workflow
• :class:DataSet — data sets associated with a workflow

Cpu dataclass

Number of processor cores for a CPU entity.

Source code in MaRDMO/workflow/models.py

@dataclass
class Cpu:
    '''Number of processor cores for a CPU entity.'''
    cores: Optional[str] = None

    @classmethod
    def from_query_batch(cls, raw_data: list) -> 'dict[str, Cpu]':
        '''Parse a batch SPARQL result into {external_id: instance} dict.'''
        return {
            row['qid']['value']: cls(
                cores=row.get('cores', {}).get('value') or None
            )
            for row in raw_data
            if row.get('qid', {}).get('value')
        }

from_query_batch(raw_data) classmethod

Parse a batch SPARQL result into {external_id: instance} dict.

Source code in MaRDMO/workflow/models.py

@classmethod
def from_query_batch(cls, raw_data: list) -> 'dict[str, Cpu]':
    '''Parse a batch SPARQL result into {external_id: instance} dict.'''
    return {
        row['qid']['value']: cls(
            cores=row.get('cores', {}).get('value') or None
        )
        for row in raw_data
        if row.get('qid', {}).get('value')
    }

CpuEntry dataclass

One CPU type on a hardware entity, with count information.

Source code in MaRDMO/workflow/models.py

@dataclass
class CpuEntry:
    '''One CPU type on a hardware entity, with count information.'''
    id: Optional[str]
    label: Optional[str]
    description: Optional[str]
    count: Optional[str]

    @classmethod
    def from_query(cls, raw: str) -> 'CpuEntry':
        '''Parse a 4-field ``||``-delimited string into a CpuEntry instance.'''
        parts = raw.split(' || ')
        while len(parts) < 4:
            parts.append('')
        return cls(
            id=parts[0] or None,
            label=parts[1] or None,
            description=parts[2] or None,
            count=parts[3] or None,
        )

from_query(raw) classmethod

Parse a 4-field ||-delimited string into a CpuEntry instance.

Source code in MaRDMO/workflow/models.py

@classmethod
def from_query(cls, raw: str) -> 'CpuEntry':
    '''Parse a 4-field ``||``-delimited string into a CpuEntry instance.'''
    parts = raw.split(' || ')
    while len(parts) < 4:
        parts.append('')
    return cls(
        id=parts[0] or None,
        label=parts[1] or None,
        description=parts[2] or None,
        count=parts[3] or None,
    )

DataSet dataclass

Metadata and relations of a Data Set entity from MaRDI/Wikidata.

Source code in MaRDMO/workflow/models.py

@dataclass
class DataSet:
    '''Metadata and relations of a Data Set entity from MaRDI/Wikidata.'''

    size: list = field(default_factory=list)
    file_format: Optional[str] = None
    binary_or_text: Optional[str] = None
    proprietary: Optional[str] = None
    to_publish: list = field(default_factory=list)
    to_archive: list = field(default_factory=list)
    publications: list[Relatant] = field(default_factory=list)

    @classmethod
    def from_query(cls, raw_data: dict) -> 'DataSet':
        '''Parse a single-item SPARQL result (backward-compatible).'''
        return cls.from_query_single(raw_data[0])

    @classmethod
    def from_query_batch(cls, raw_data: list) -> 'dict[str, DataSet]':
        '''Parse a batch SPARQL result into {external_id: instance} dict.'''
        return {
            row['qid']['value']: cls.from_query_single(row)
            for row in raw_data
            if row.get('qid', {}).get('value')
        }

    @classmethod
    def from_query_single(cls, data: dict) -> 'DataSet':
        '''Parse one SPARQL result row into a DataSet instance.'''
        options = get_options()

        return cls(
            size=get_size(data, options),
            file_format=data.get('file_format', {}).get('value'),
            binary_or_text=(
                options[data['binary_or_text']['value']]
                if data.get('binary_or_text', {}).get('value') else ''
            ),
            proprietary=(
                options[data['proprietary']['value']]
                if data.get('proprietary', {}).get('value') else ''
            ),
            to_publish=get_option_text_pair(data, options, 'publish', 'DOI', 'URL'),
            to_archive=get_option_text_pair(data, options, 'archive', 'end_time'),
            publications=split_value(
                data=data, key='publication', transform=Relatant.from_query
            ),
        )

from_query(raw_data) classmethod

Parse a single-item SPARQL result (backward-compatible).

Source code in MaRDMO/workflow/models.py

@classmethod
def from_query(cls, raw_data: dict) -> 'DataSet':
    '''Parse a single-item SPARQL result (backward-compatible).'''
    return cls.from_query_single(raw_data[0])

from_query_batch(raw_data) classmethod

Parse a batch SPARQL result into {external_id: instance} dict.

Source code in MaRDMO/workflow/models.py

@classmethod
def from_query_batch(cls, raw_data: list) -> 'dict[str, DataSet]':
    '''Parse a batch SPARQL result into {external_id: instance} dict.'''
    return {
        row['qid']['value']: cls.from_query_single(row)
        for row in raw_data
        if row.get('qid', {}).get('value')
    }

from_query_single(data) classmethod

Parse one SPARQL result row into a DataSet instance.

Source code in MaRDMO/workflow/models.py

@classmethod
def from_query_single(cls, data: dict) -> 'DataSet':
    '''Parse one SPARQL result row into a DataSet instance.'''
    options = get_options()

    return cls(
        size=get_size(data, options),
        file_format=data.get('file_format', {}).get('value'),
        binary_or_text=(
            options[data['binary_or_text']['value']]
            if data.get('binary_or_text', {}).get('value') else ''
        ),
        proprietary=(
            options[data['proprietary']['value']]
            if data.get('proprietary', {}).get('value') else ''
        ),
        to_publish=get_option_text_pair(data, options, 'publish', 'DOI', 'URL'),
        to_archive=get_option_text_pair(data, options, 'archive', 'end_time'),
        publications=split_value(
            data=data, key='publication', transform=Relatant.from_query
        ),
    )

Hardware dataclass

Node/core counts and CPU relations of a Hardware entity.

Source code in MaRDMO/workflow/models.py

@dataclass
class Hardware:
    '''Node/core counts and CPU relations of a Hardware entity.'''

    nodes: Optional[str] = None
    cpu: list[CpuEntry] = field(default_factory=list)
    publications: list[Relatant] = field(default_factory=list)

    @classmethod
    def from_query(cls, raw_data: dict) -> 'Hardware':
        '''Parse a single-item SPARQL result (backward-compatible).'''
        return cls.from_query_single(raw_data[0])

    @classmethod
    def from_query_batch(cls, raw_data: list) -> 'dict[str, Hardware]':
        '''Parse a batch SPARQL result into {external_id: instance} dict.'''
        return {
            row['qid']['value']: cls.from_query_single(row)
            for row in raw_data
            if row.get('qid', {}).get('value')
        }

    @classmethod
    def from_query_single(cls, data: dict) -> 'Hardware':
        '''Parse one SPARQL result row into a Hardware instance.'''
        return cls(
            nodes=data.get('number_of_nodes', {}).get('value') or None,
            cpu=split_value(data=data, key='cpu', transform=CpuEntry.from_query),
            publications=split_value(
                data=data, key='publication', transform=Relatant.from_query
            ),
        )

from_query(raw_data) classmethod

Parse a single-item SPARQL result (backward-compatible).

Source code in MaRDMO/workflow/models.py

@classmethod
def from_query(cls, raw_data: dict) -> 'Hardware':
    '''Parse a single-item SPARQL result (backward-compatible).'''
    return cls.from_query_single(raw_data[0])

from_query_batch(raw_data) classmethod

Parse a batch SPARQL result into {external_id: instance} dict.

Source code in MaRDMO/workflow/models.py

@classmethod
def from_query_batch(cls, raw_data: list) -> 'dict[str, Hardware]':
    '''Parse a batch SPARQL result into {external_id: instance} dict.'''
    return {
        row['qid']['value']: cls.from_query_single(row)
        for row in raw_data
        if row.get('qid', {}).get('value')
    }

from_query_single(data) classmethod

Parse one SPARQL result row into a Hardware instance.

Source code in MaRDMO/workflow/models.py

@classmethod
def from_query_single(cls, data: dict) -> 'Hardware':
    '''Parse one SPARQL result row into a Hardware instance.'''
    return cls(
        nodes=data.get('number_of_nodes', {}).get('value') or None,
        cpu=split_value(data=data, key='cpu', transform=CpuEntry.from_query),
        publications=split_value(
            data=data, key='publication', transform=Relatant.from_query
        ),
    )

ProcessStep dataclass

Relations of a Process Step entity from MaRDI/Wikidata.

Source code in MaRDMO/workflow/models.py

@dataclass
class ProcessStep:
    '''Relations of a Process Step entity from MaRDI/Wikidata.'''

    input_data_set: list[Relatant] = field(default_factory=list)
    output_data_set: list[Relatant] = field(default_factory=list)
    uses_algorithm: list[ProcessStepUsage] = field(default_factory=list)
    uses_method: list[ProcessStepUsage] = field(default_factory=list)
    field_of_work: list[Relatant] = field(default_factory=list)
    publications: list[Relatant] = field(default_factory=list)

    @classmethod
    def from_query(cls, raw_data: list) -> 'ProcessStep':
        '''Parse a single-item SPARQL result (backward-compatible).'''
        return cls.from_query_single(raw_data[0])

    @classmethod
    def from_query_batch(cls, raw_data: list) -> 'dict[str, ProcessStep]':
        '''Parse a batch SPARQL result into {external_id: instance} dict.'''
        return {
            row['qid']['value']: cls.from_query_single(row)
            for row in raw_data
            if row.get('qid', {}).get('value')
        }

    @classmethod
    def from_query_single(cls, data: dict) -> 'ProcessStep':
        '''Parse one SPARQL result row into a ProcessStep instance.'''
        return cls(
            input_data_set=split_value(
                data=data, key='input_data_set', transform=Relatant.from_query
            ),
            output_data_set=split_value(
                data=data, key='output_data_set', transform=Relatant.from_query
            ),
            uses_algorithm=split_value(
                data=data, key='uses_algorithm', transform=ProcessStepUsage.from_query
            ),
            uses_method=split_value(
                data=data, key='uses_method', transform=ProcessStepUsage.from_query
            ),
            field_of_work=split_value(
                data=data, key='field_of_work', transform=Relatant.from_query
            ),
            publications=split_value(
                data=data, key='publication', transform=Relatant.from_query
            ),
        )

from_query(raw_data) classmethod

Parse a single-item SPARQL result (backward-compatible).

Source code in MaRDMO/workflow/models.py

@classmethod
def from_query(cls, raw_data: list) -> 'ProcessStep':
    '''Parse a single-item SPARQL result (backward-compatible).'''
    return cls.from_query_single(raw_data[0])

from_query_batch(raw_data) classmethod

Parse a batch SPARQL result into {external_id: instance} dict.

Source code in MaRDMO/workflow/models.py

@classmethod
def from_query_batch(cls, raw_data: list) -> 'dict[str, ProcessStep]':
    '''Parse a batch SPARQL result into {external_id: instance} dict.'''
    return {
        row['qid']['value']: cls.from_query_single(row)
        for row in raw_data
        if row.get('qid', {}).get('value')
    }

from_query_single(data) classmethod

Parse one SPARQL result row into a ProcessStep instance.

Source code in MaRDMO/workflow/models.py

@classmethod
def from_query_single(cls, data: dict) -> 'ProcessStep':
    '''Parse one SPARQL result row into a ProcessStep instance.'''
    return cls(
        input_data_set=split_value(
            data=data, key='input_data_set', transform=Relatant.from_query
        ),
        output_data_set=split_value(
            data=data, key='output_data_set', transform=Relatant.from_query
        ),
        uses_algorithm=split_value(
            data=data, key='uses_algorithm', transform=ProcessStepUsage.from_query
        ),
        uses_method=split_value(
            data=data, key='uses_method', transform=ProcessStepUsage.from_query
        ),
        field_of_work=split_value(
            data=data, key='field_of_work', transform=Relatant.from_query
        ),
        publications=split_value(
            data=data, key='publication', transform=Relatant.from_query
        ),
    )

ProcessStepUsage dataclass

Usage of an algorithm or method in a process step.

Covers both algorithm-in-process-step (qualifier=software, hardware=hardware) and method-in-process-step (qualifier=instrument, hardware always empty). Parsed from a fixed-position 11-field ||-delimited main block followed by an optional >|<-separated parameters section.

Source code in MaRDMO/workflow/models.py

@dataclass
class ProcessStepUsage:
    '''Usage of an algorithm or method in a process step.

    Covers both algorithm-in-process-step (qualifier=software, hardware=hardware)
    and method-in-process-step (qualifier=instrument, hardware always empty).
    Parsed from a fixed-position 11-field ``||``-delimited main block followed
    by an optional ``>|<``-separated parameters section.
    '''
    id: Optional[str]
    label: Optional[str]
    description: Optional[str]
    qualifier: Optional[str]
    qualifier_label: Optional[str]
    qualifier_description: Optional[str]
    hardware: Optional[str]
    hardware_label: Optional[str]
    hardware_description: Optional[str]
    parameters: Optional[str]
    doi: list
    url: list

    @classmethod
    def from_query(cls, raw: str) -> 'ProcessStepUsage':
        '''Parse ``id||label||desc||q||ql||qd||hw||hwl||hwd||doi||url >|< params``.'''
        options = get_options()
        if ' >|< ' in raw:
            main, parameters = raw.split(' >|< ', 1)
        else:
            main, parameters = raw, None
        parts = main.split(' || ')
        while len(parts) < 11:
            parts.append('')
        return cls(
            id=parts[0] or None,
            label=parts[1] or None,
            description=parts[2] or None,
            qualifier=parts[3] or None,
            qualifier_label=parts[4] or None,
            qualifier_description=parts[5] or None,
            hardware=parts[6] or None,
            hardware_label=parts[7] or None,
            hardware_description=parts[8] or None,
            doi=[options['DOI'], parts[9]] if parts[9] else None,
            url=[options['URL'], parts[10]] if parts[10] else None,
            parameters=parameters or None,
        )

from_query(raw) classmethod

Parse id||label||desc||q||ql||qd||hw||hwl||hwd||doi||url >|< params.

Source code in MaRDMO/workflow/models.py

@classmethod
def from_query(cls, raw: str) -> 'ProcessStepUsage':
    '''Parse ``id||label||desc||q||ql||qd||hw||hwl||hwd||doi||url >|< params``.'''
    options = get_options()
    if ' >|< ' in raw:
        main, parameters = raw.split(' >|< ', 1)
    else:
        main, parameters = raw, None
    parts = main.split(' || ')
    while len(parts) < 11:
        parts.append('')
    return cls(
        id=parts[0] or None,
        label=parts[1] or None,
        description=parts[2] or None,
        qualifier=parts[3] or None,
        qualifier_label=parts[4] or None,
        qualifier_description=parts[5] or None,
        hardware=parts[6] or None,
        hardware_label=parts[7] or None,
        hardware_description=parts[8] or None,
        doi=[options['DOI'], parts[9]] if parts[9] else None,
        url=[options['URL'], parts[10]] if parts[10] else None,
        parameters=parameters or None,
    )

Workflow dataclass

Top-level metadata of an Interdisciplinary Workflow entity from MaRDI Portal.

Source code in MaRDMO/workflow/models.py

@dataclass
class Workflow:
    '''Top-level metadata of an Interdisciplinary Workflow entity from MaRDI Portal.'''

    research_objective: list[str] = field(default_factory=list)
    procedure: list[str] = field(default_factory=list)
    mathematical: Optional[str] = None
    mathematical_comment: Optional[str] = None
    runtime: Optional[str] = None
    runtime_comment: Optional[str] = None
    result: Optional[str] = None
    result_comment: Optional[str] = None
    originalplatform: Optional[str] = None
    originalplatform_comment: Optional[str] = None
    otherplatform: Optional[str] = None
    otherplatform_comment: Optional[str] = None
    transferable: Optional[str] = None
    transferable_comment: list[str] = field(default_factory=list)
    uses_model: list[str] = field(default_factory=list)
    contains_process_step: list[Relatant] = field(default_factory=list)
    contains_workflow: list[Relatant] = field(default_factory=list)
    contained_in_workflow: list[Relatant] = field(default_factory=list)
    publications: list[Relatant] = field(default_factory=list)

    @classmethod
    def from_query_batch(cls, raw_data: list) -> 'dict[str, Workflow]':
        '''Parse a batch SPARQL result into {external_id: instance} dict.'''
        return {
            row['qid']['value']: cls.from_query_single(row)
            for row in raw_data
            if row.get('qid', {}).get('value')
        }

    @classmethod
    def from_query_single(cls, data: dict) -> 'Workflow':
        '''Parse one SPARQL result row into a Workflow instance.'''
        def _split(key):
            raw = data.get(key, {}).get('value', '')
            return [v.strip() for v in raw.split(' <|> ') if v.strip()] if raw else []

        def _val(key):
            return data.get(key, {}).get('value') or None

        return cls(
            research_objective=_split('research_objective'),
            procedure=_split('procedure'),
            mathematical=_val('mathematical'),
            mathematical_comment=_val('mathematical_comment'),
            runtime=_val('runtime'),
            runtime_comment=_val('runtime_comment'),
            result=_val('result'),
            result_comment=_val('result_comment'),
            originalplatform=_val('originalplatform'),
            originalplatform_comment=_val('originalplatform_comment'),
            otherplatform=_val('otherplatform'),
            otherplatform_comment=_val('otherplatform_comment'),
            transferable=_val('transferable'),
            transferable_comment=_split('transferable_comment'),
            uses_model=_split('uses_model'),
            contains_process_step=split_value(
                data=data, key='contains_process_step', transform=Relatant.from_query
            ),
            contains_workflow=split_value(
                data=data, key='contains_workflow', transform=Relatant.from_query
            ),
            contained_in_workflow=split_value(
                data=data, key='contained_in_workflow', transform=Relatant.from_query
            ),
            publications=split_value(
                data=data, key='publication', transform=Relatant.from_query
            ),
        )

from_query_batch(raw_data) classmethod

Parse a batch SPARQL result into {external_id: instance} dict.

Source code in MaRDMO/workflow/models.py

@classmethod
def from_query_batch(cls, raw_data: list) -> 'dict[str, Workflow]':
    '''Parse a batch SPARQL result into {external_id: instance} dict.'''
    return {
        row['qid']['value']: cls.from_query_single(row)
        for row in raw_data
        if row.get('qid', {}).get('value')
    }

from_query_single(data) classmethod

Parse one SPARQL result row into a Workflow instance.

Source code in MaRDMO/workflow/models.py

@classmethod
def from_query_single(cls, data: dict) -> 'Workflow':
    '''Parse one SPARQL result row into a Workflow instance.'''
    def _split(key):
        raw = data.get(key, {}).get('value', '')
        return [v.strip() for v in raw.split(' <|> ') if v.strip()] if raw else []

    def _val(key):
        return data.get(key, {}).get('value') or None

    return cls(
        research_objective=_split('research_objective'),
        procedure=_split('procedure'),
        mathematical=_val('mathematical'),
        mathematical_comment=_val('mathematical_comment'),
        runtime=_val('runtime'),
        runtime_comment=_val('runtime_comment'),
        result=_val('result'),
        result_comment=_val('result_comment'),
        originalplatform=_val('originalplatform'),
        originalplatform_comment=_val('originalplatform_comment'),
        otherplatform=_val('otherplatform'),
        otherplatform_comment=_val('otherplatform_comment'),
        transferable=_val('transferable'),
        transferable_comment=_split('transferable_comment'),
        uses_model=_split('uses_model'),
        contains_process_step=split_value(
            data=data, key='contains_process_step', transform=Relatant.from_query
        ),
        contains_workflow=split_value(
            data=data, key='contains_workflow', transform=Relatant.from_query
        ),
        contained_in_workflow=split_value(
            data=data, key='contained_in_workflow', transform=Relatant.from_query
        ),
        publications=split_value(
            data=data, key='publication', transform=Relatant.from_query
        ),
    )

Providers

RDMO optionset providers for the Workflow documentation catalog.

Each provider class implements get_options and is referenced from the workflow catalog configuration. Providers query MaRDI Portal and Wikidata for entity suggestions covering research disciplines, methods, tasks, hardware, instruments, data sets, process steps, and related mathematical models.

Provides:

• :class:MainMathematicalModel — look up the main mathematical model for a workflow
• :class:Method — numerical/analytical method lookup; refresh on select
• :class:RelatedMethod — method lookup with optional user creation
• :class:WorkflowTask — task lookup; refresh on save (depends on selected model)
• :class:Hardware — hardware platform lookup; refresh on select
• :class:Instrument — scientific instrument lookup; refresh on select
• :class:DataSet — data set lookup; refresh on select
• :class:RelatedInstrument — instrument lookup with optional user creation
• :class:RelatedDataSet — data-set lookup with optional user creation
• :class:ProcessStep — workflow process-step lookup; refresh on select
• :class:Discipline — research discipline lookup sourced from questionnaire answers

DataSet

Bases: Provider

Data Set Provider (MaRDI Portal / Wikidata), No User Creation, Refresh Upon Selection

Source code in MaRDMO/workflow/providers.py

class DataSet(Provider):
    '''Data Set Provider (MaRDI Portal / Wikidata),
       No User Creation, Refresh Upon Selection
    '''

    search = True
    refresh = True

    def get_options(self, project, search=None, user=None, site=None):
        '''Query external knowledge-graph source(s) and return matching options.

        Args:
            project: RDMO project instance (used for user-entry lookups when applicable).
            search:  Search string entered by the user; returns empty list when
                     fewer than 3 characters.
            user:    Requesting user (unused).
            site:    Current site (unused).

        Returns:
            List of ``{"id": …, "text": …}`` option dicts sorted by relevance.
        '''
        if not search or len(search) < 3:
            return []

        return query_sources(
            search = search,
            item_class = _ITEMS['data set'],
        )

get_options(project, search=None, user=None, site=None)

Query external knowledge-graph source(s) and return matching options.

Parameters:

Name	Type	Description	Default
project		RDMO project instance (used for user-entry lookups when applicable).	required
search		Search string entered by the user; returns empty list when fewer than 3 characters.	None
user		Requesting user (unused).	None
site		Current site (unused).	None

Returns:

Type	Description
	List of {"id": …, "text": …} option dicts sorted by relevance.

Source code in MaRDMO/workflow/providers.py

def get_options(self, project, search=None, user=None, site=None):
    '''Query external knowledge-graph source(s) and return matching options.

    Args:
        project: RDMO project instance (used for user-entry lookups when applicable).
        search:  Search string entered by the user; returns empty list when
                 fewer than 3 characters.
        user:    Requesting user (unused).
        site:    Current site (unused).

    Returns:
        List of ``{"id": …, "text": …}`` option dicts sorted by relevance.
    '''
    if not search or len(search) < 3:
        return []

    return query_sources(
        search = search,
        item_class = _ITEMS['data set'],
    )

Hardware

Bases: Provider

Hardware Provider (MaRDI Portal / Wikidata), No User Creation, Refresh Upon Selection

Source code in MaRDMO/workflow/providers.py

class Hardware(Provider):
    '''Hardware Provider (MaRDI Portal / Wikidata),
       No User Creation, Refresh Upon Selection
    '''

    search = True
    refresh = True

    def get_options(self, project, search=None, user=None, site=None):
        '''Query external knowledge-graph source(s) and return matching options.

        Args:
            project: RDMO project instance (used for user-entry lookups when applicable).
            search:  Search string entered by the user; returns empty list when
                     fewer than 3 characters.
            user:    Requesting user (unused).
            site:    Current site (unused).

        Returns:
            List of ``{"id": …, "text": …}`` option dicts sorted by relevance.
        '''
        if not search or len(search) < 3:
            return []

        return query_sources(
            search = search,
            item_class = _ITEMS['computer hardware'],
        )

get_options(project, search=None, user=None, site=None)

Query external knowledge-graph source(s) and return matching options.

Parameters:

Name	Type	Description	Default
project		RDMO project instance (used for user-entry lookups when applicable).	required
search		Search string entered by the user; returns empty list when fewer than 3 characters.	None
user		Requesting user (unused).	None
site		Current site (unused).	None

Returns:

Type	Description
	List of {"id": …, "text": …} option dicts sorted by relevance.

Source code in MaRDMO/workflow/providers.py

def get_options(self, project, search=None, user=None, site=None):
    '''Query external knowledge-graph source(s) and return matching options.

    Args:
        project: RDMO project instance (used for user-entry lookups when applicable).
        search:  Search string entered by the user; returns empty list when
                 fewer than 3 characters.
        user:    Requesting user (unused).
        site:    Current site (unused).

    Returns:
        List of ``{"id": …, "text": …}`` option dicts sorted by relevance.
    '''
    if not search or len(search) < 3:
        return []

    return query_sources(
        search = search,
        item_class = _ITEMS['computer hardware'],
    )

Instrument

Bases: Provider

Instrument Provider (MaRDI Portal / Wikidata), No User Creation, Refresh Upon Selection

Source code in MaRDMO/workflow/providers.py

class Instrument(Provider):
    '''Instrument Provider (MaRDI Portal / Wikidata),
       No User Creation, Refresh Upon Selection
    '''

    search = True
    refresh = True

    def get_options(self, project, search=None, user=None, site=None):
        '''Query external knowledge-graph source(s) and return matching options.

        Args:
            project: RDMO project instance (used for user-entry lookups when applicable).
            search:  Search string entered by the user; returns empty list when
                     fewer than 3 characters.
            user:    Requesting user (unused).
            site:    Current site (unused).

        Returns:
            List of ``{"id": …, "text": …}`` option dicts sorted by relevance.
        '''
        if not search or len(search) < 3:
            return []

        return query_sources(
            search = search,
            item_class = _ITEMS['research tool'],
        )

get_options(project, search=None, user=None, site=None)

Query external knowledge-graph source(s) and return matching options.

Parameters:

Name	Type	Description	Default
project		RDMO project instance (used for user-entry lookups when applicable).	required
search		Search string entered by the user; returns empty list when fewer than 3 characters.	None
user		Requesting user (unused).	None
site		Current site (unused).	None

Returns:

Type	Description
	List of {"id": …, "text": …} option dicts sorted by relevance.

Source code in MaRDMO/workflow/providers.py

def get_options(self, project, search=None, user=None, site=None):
    '''Query external knowledge-graph source(s) and return matching options.

    Args:
        project: RDMO project instance (used for user-entry lookups when applicable).
        search:  Search string entered by the user; returns empty list when
                 fewer than 3 characters.
        user:    Requesting user (unused).
        site:    Current site (unused).

    Returns:
        List of ``{"id": …, "text": …}`` option dicts sorted by relevance.
    '''
    if not search or len(search) < 3:
        return []

    return query_sources(
        search = search,
        item_class = _ITEMS['research tool'],
    )

MathematicalModel

Bases: Provider

Main Mathematical Model Provider (MaRDI Portal), No User Creation, Refresh Upon Selection

Source code in MaRDMO/workflow/providers.py

class MathematicalModel(Provider):
    '''Main Mathematical Model Provider (MaRDI Portal),
       No User Creation, Refresh Upon Selection
    '''

    search = True
    refresh = True

    def get_options(self, project, search=None, user=None, site=None):
        '''Query external knowledge-graph source(s) and return matching options.

        Args:
            project: RDMO project instance (used for user-entry lookups when applicable).
            search:  Search string entered by the user; returns empty list when
                     fewer than 3 characters.
            user:    Requesting user (unused).
            site:    Current site (unused).

        Returns:
            List of ``{"id": …, "text": …}`` option dicts sorted by relevance.
        '''
        if not search or len(search) < 3:
            return []

        return query_sources(
            search = search,
            item_class = _ITEMS['mathematical model'],
            sources = ['mardi'],
            not_found = False
        )

get_options(project, search=None, user=None, site=None)

Query external knowledge-graph source(s) and return matching options.

Parameters:

Name	Type	Description	Default
project		RDMO project instance (used for user-entry lookups when applicable).	required
search		Search string entered by the user; returns empty list when fewer than 3 characters.	None
user		Requesting user (unused).	None
site		Current site (unused).	None

Returns:

Type	Description
	List of {"id": …, "text": …} option dicts sorted by relevance.

Source code in MaRDMO/workflow/providers.py

def get_options(self, project, search=None, user=None, site=None):
    '''Query external knowledge-graph source(s) and return matching options.

    Args:
        project: RDMO project instance (used for user-entry lookups when applicable).
        search:  Search string entered by the user; returns empty list when
                 fewer than 3 characters.
        user:    Requesting user (unused).
        site:    Current site (unused).

    Returns:
        List of ``{"id": …, "text": …}`` option dicts sorted by relevance.
    '''
    if not search or len(search) < 3:
        return []

    return query_sources(
        search = search,
        item_class = _ITEMS['mathematical model'],
        sources = ['mardi'],
        not_found = False
    )

ProcessStep

Bases: Provider

Process Step Provider (MaRDI Portal / Wikidata), No User Creation, Refresh Upon Selection

Source code in MaRDMO/workflow/providers.py

class ProcessStep(Provider):
    '''Process Step Provider (MaRDI Portal / Wikidata),
       No User Creation, Refresh Upon Selection
    '''

    search = True
    refresh = True

    def get_options(self, project, search=None, user=None, site=None):
        '''Query external knowledge-graph source(s) and return matching options.

        Args:
            project: RDMO project instance (used for user-entry lookups when applicable).
            search:  Search string entered by the user; returns empty list when
                     fewer than 3 characters.
            user:    Requesting user (unused).
            site:    Current site (unused).

        Returns:
            List of ``{"id": …, "text": …}`` option dicts sorted by relevance.
        '''
        if not search or len(search) < 3:
            return []

        return query_sources(
            search = search,
            item_class = _ITEMS['process step'],
        )

get_options(project, search=None, user=None, site=None)

Query external knowledge-graph source(s) and return matching options.

Parameters:

Name	Type	Description	Default
project		RDMO project instance (used for user-entry lookups when applicable).	required
search		Search string entered by the user; returns empty list when fewer than 3 characters.	None
user		Requesting user (unused).	None
site		Current site (unused).	None

Returns:

Type	Description
	List of {"id": …, "text": …} option dicts sorted by relevance.

Source code in MaRDMO/workflow/providers.py

def get_options(self, project, search=None, user=None, site=None):
    '''Query external knowledge-graph source(s) and return matching options.

    Args:
        project: RDMO project instance (used for user-entry lookups when applicable).
        search:  Search string entered by the user; returns empty list when
                 fewer than 3 characters.
        user:    Requesting user (unused).
        site:    Current site (unused).

    Returns:
        List of ``{"id": …, "text": …}`` option dicts sorted by relevance.
    '''
    if not search or len(search) < 3:
        return []

    return query_sources(
        search = search,
        item_class = _ITEMS['process step'],
    )

RelatedAlgorithmWithCreation

Bases: Provider

Algorithm Provider (MaRDI Portal / Wikidata), No User Creation, No Refresh Upon Selection

Source code in MaRDMO/workflow/providers.py

class RelatedAlgorithmWithCreation(Provider):
    '''Algorithm Provider (MaRDI Portal / Wikidata),
       No User Creation, No Refresh Upon Selection
    '''

    search = True

    def get_options(self, project, search=None, user=None, site=None):
        '''Query external knowledge-graph source(s) and return matching options.

        Args:
            project: RDMO project instance (used for user-entry lookups when applicable).
            search:  Search string entered by the user; returns empty list when
                     fewer than 3 characters.
            user:    Requesting user (unused).
            site:    Current site (unused).

        Returns:
            List of ``{"id": …, "text": …}`` option dicts sorted by relevance.
        '''
        if not search:
            return []

        setup = define_setup(
            query_attributes = ['algorithm'],
            creation = True,
            item_class = [
                _ITEMS['algorithm']
            ]
        )

        return query_sources_with_user_additions(
            search = search,
            project = project,
            setup = setup
        )

get_options(project, search=None, user=None, site=None)

Query external knowledge-graph source(s) and return matching options.

Parameters:

Name	Type	Description	Default
project		RDMO project instance (used for user-entry lookups when applicable).	required
search		Search string entered by the user; returns empty list when fewer than 3 characters.	None
user		Requesting user (unused).	None
site		Current site (unused).	None

Returns:

Type	Description
	List of {"id": …, "text": …} option dicts sorted by relevance.

Source code in MaRDMO/workflow/providers.py

def get_options(self, project, search=None, user=None, site=None):
    '''Query external knowledge-graph source(s) and return matching options.

    Args:
        project: RDMO project instance (used for user-entry lookups when applicable).
        search:  Search string entered by the user; returns empty list when
                 fewer than 3 characters.
        user:    Requesting user (unused).
        site:    Current site (unused).

    Returns:
        List of ``{"id": …, "text": …}`` option dicts sorted by relevance.
    '''
    if not search:
        return []

    setup = define_setup(
        query_attributes = ['algorithm'],
        creation = True,
        item_class = [
            _ITEMS['algorithm']
        ]
    )

    return query_sources_with_user_additions(
        search = search,
        project = project,
        setup = setup
    )

RelatedCPUModelWithCreation

Bases: Provider

CPU Model Provider (MaRDI Portal / Wikidata), User Creation, No Refresh Upon Selection

Source code in MaRDMO/workflow/providers.py

class RelatedCPUModelWithCreation(Provider):
    '''CPU Model Provider (MaRDI Portal / Wikidata),
       User Creation, No Refresh Upon Selection
    '''

    search = True
    refresh = True

    def get_options(self, project, search=None, user=None, site=None):
        '''Query external knowledge-graph source(s) and return matching options.

        Args:
            project: RDMO project instance (used for user-entry lookups when applicable).
            search:  Search string entered by the user; returns empty list when
                     fewer than 3 characters.
            user:    Requesting user (unused).
            site:    Current site (unused).

        Returns:
            List of ``{"id": …, "text": …}`` option dicts sorted by relevance.
        '''
        if not search:
            return []

        setup = define_setup(
            creation = True,
            query_attributes = None,
            item_class = [
                _ITEMS['CPU model'],
            ]
        )

        return query_sources_with_user_additions(
            search = search,
            project = project,
            setup = setup
        )

get_options(project, search=None, user=None, site=None)

Query external knowledge-graph source(s) and return matching options.

Parameters:

Name	Type	Description	Default
project		RDMO project instance (used for user-entry lookups when applicable).	required
search		Search string entered by the user; returns empty list when fewer than 3 characters.	None
user		Requesting user (unused).	None
site		Current site (unused).	None

Returns:

Type	Description
	List of {"id": …, "text": …} option dicts sorted by relevance.

Source code in MaRDMO/workflow/providers.py

def get_options(self, project, search=None, user=None, site=None):
    '''Query external knowledge-graph source(s) and return matching options.

    Args:
        project: RDMO project instance (used for user-entry lookups when applicable).
        search:  Search string entered by the user; returns empty list when
                 fewer than 3 characters.
        user:    Requesting user (unused).
        site:    Current site (unused).

    Returns:
        List of ``{"id": …, "text": …}`` option dicts sorted by relevance.
    '''
    if not search:
        return []

    setup = define_setup(
        creation = True,
        query_attributes = None,
        item_class = [
            _ITEMS['CPU model'],
        ]
    )

    return query_sources_with_user_additions(
        search = search,
        project = project,
        setup = setup
    )

RelatedDataSetWithCreation

Bases: Provider

Data Set Provider (MaRDI Portal / Wikidata), User Creation, No Refresh Upon Selection

Source code in MaRDMO/workflow/providers.py

class RelatedDataSetWithCreation(Provider):
    '''Data Set Provider (MaRDI Portal / Wikidata),
       User Creation, No Refresh Upon Selection
    '''

    search = True

    def get_options(self, project, search=None, user=None, site=None):
        '''Query external knowledge-graph source(s) and return matching options.

        Args:
            project: RDMO project instance (used for user-entry lookups when applicable).
            search:  Search string entered by the user; returns empty list when
                     fewer than 3 characters.
            user:    Requesting user (unused).
            site:    Current site (unused).

        Returns:
            List of ``{"id": …, "text": …}`` option dicts sorted by relevance.
        '''
        if not search:
            return []

        setup = define_setup(
            creation = True,
            query_attributes = ['dataset'],
            item_class = _ITEMS['data set'],
        )

        return query_sources_with_user_additions(
            search = search,
            project = project,
            setup = setup
        )

get_options(project, search=None, user=None, site=None)

Query external knowledge-graph source(s) and return matching options.

Parameters:

Name	Type	Description	Default
project		RDMO project instance (used for user-entry lookups when applicable).	required
search		Search string entered by the user; returns empty list when fewer than 3 characters.	None
user		Requesting user (unused).	None
site		Current site (unused).	None

Returns:

Type	Description
	List of {"id": …, "text": …} option dicts sorted by relevance.

Source code in MaRDMO/workflow/providers.py

def get_options(self, project, search=None, user=None, site=None):
    '''Query external knowledge-graph source(s) and return matching options.

    Args:
        project: RDMO project instance (used for user-entry lookups when applicable).
        search:  Search string entered by the user; returns empty list when
                 fewer than 3 characters.
        user:    Requesting user (unused).
        site:    Current site (unused).

    Returns:
        List of ``{"id": …, "text": …}`` option dicts sorted by relevance.
    '''
    if not search:
        return []

    setup = define_setup(
        creation = True,
        query_attributes = ['dataset'],
        item_class = _ITEMS['data set'],
    )

    return query_sources_with_user_additions(
        search = search,
        project = project,
        setup = setup
    )

RelatedHardwareOrSoftwareWithoutCreation

Bases: Provider

Hardware, Software Provider (MaRDI Portal), No User Creation, No Refresh Upon Selection

Source code in MaRDMO/workflow/providers.py

class RelatedHardwareOrSoftwareWithoutCreation(Provider):
    '''Hardware, Software Provider (MaRDI Portal),
       No User Creation, No Refresh Upon Selection
    '''

    search = True

    def get_options(self, project, search=None, user=None, site=None):
        '''Query external knowledge-graph source(s) and return matching options.

        Args:
            project: RDMO project instance (used for user-entry lookups when applicable).
            search:  Search string entered by the user; returns empty list when
                     fewer than 3 characters.
            user:    Requesting user (unused).
            site:    Current site (unused).

        Returns:
            List of ``{"id": …, "text": …}`` option dicts sorted by relevance.
        '''
        if not search or len(search) < 3:
            return []

        setup = define_setup(
            query_attributes = ['hardware', 'software'],
            sources = ['mardi'],
            item_class = [
                _ITEMS['computer hardware'],
                _ITEMS['software']
            ]
        )

        return query_sources_with_user_additions(
            search = search,
            project = project,
            setup = setup
        )

get_options(project, search=None, user=None, site=None)

Query external knowledge-graph source(s) and return matching options.

Parameters:

Name	Type	Description	Default
project		RDMO project instance (used for user-entry lookups when applicable).	required
search		Search string entered by the user; returns empty list when fewer than 3 characters.	None
user		Requesting user (unused).	None
site		Current site (unused).	None

Returns:

Type	Description
	List of {"id": …, "text": …} option dicts sorted by relevance.

Source code in MaRDMO/workflow/providers.py

def get_options(self, project, search=None, user=None, site=None):
    '''Query external knowledge-graph source(s) and return matching options.

    Args:
        project: RDMO project instance (used for user-entry lookups when applicable).
        search:  Search string entered by the user; returns empty list when
                 fewer than 3 characters.
        user:    Requesting user (unused).
        site:    Current site (unused).

    Returns:
        List of ``{"id": …, "text": …}`` option dicts sorted by relevance.
    '''
    if not search or len(search) < 3:
        return []

    setup = define_setup(
        query_attributes = ['hardware', 'software'],
        sources = ['mardi'],
        item_class = [
            _ITEMS['computer hardware'],
            _ITEMS['software']
        ]
    )

    return query_sources_with_user_additions(
        search = search,
        project = project,
        setup = setup
    )

RelatedHardwareWithCreation

Bases: Provider

Hardware Provider (MaRDI Portal / Wikidata), User Creation, No Refresh Upon Selection

Source code in MaRDMO/workflow/providers.py

class RelatedHardwareWithCreation(Provider):
    '''Hardware Provider (MaRDI Portal / Wikidata),
       User Creation, No Refresh Upon Selection
    '''

    search = True

    def get_options(self, project, search=None, user=None, site=None):
        '''Query external knowledge-graph source(s) and return matching options.

        Args:
            project: RDMO project instance (used for user-entry lookups when applicable).
            search:  Search string entered by the user; returns empty list when
                     fewer than 3 characters.
            user:    Requesting user (unused).
            site:    Current site (unused).

        Returns:
            List of ``{"id": …, "text": …}`` option dicts sorted by relevance.
        '''
        if not search or len(search) < 3:
            return []

        setup = define_setup(
            query_attributes = ['hardware'],
            creation = True,
            item_class = [
                _ITEMS['computer hardware']
            ]
        )

        return query_sources_with_user_additions(
            search = search,
            project = project,
            setup = setup
        )

get_options(project, search=None, user=None, site=None)

Query external knowledge-graph source(s) and return matching options.

Parameters:

Name	Type	Description	Default
project		RDMO project instance (used for user-entry lookups when applicable).	required
search		Search string entered by the user; returns empty list when fewer than 3 characters.	None
user		Requesting user (unused).	None
site		Current site (unused).	None

Returns:

Type	Description
	List of {"id": …, "text": …} option dicts sorted by relevance.

Source code in MaRDMO/workflow/providers.py

def get_options(self, project, search=None, user=None, site=None):
    '''Query external knowledge-graph source(s) and return matching options.

    Args:
        project: RDMO project instance (used for user-entry lookups when applicable).
        search:  Search string entered by the user; returns empty list when
                 fewer than 3 characters.
        user:    Requesting user (unused).
        site:    Current site (unused).

    Returns:
        List of ``{"id": …, "text": …}`` option dicts sorted by relevance.
    '''
    if not search or len(search) < 3:
        return []

    setup = define_setup(
        query_attributes = ['hardware'],
        creation = True,
        item_class = [
            _ITEMS['computer hardware']
        ]
    )

    return query_sources_with_user_additions(
        search = search,
        project = project,
        setup = setup
    )

RelatedInstrumentWithCreation

Bases: Provider

Instrument Provider (MaRDI Portal / Wikidata), User Creation, No Refresh Upon Selection

Source code in MaRDMO/workflow/providers.py

class RelatedInstrumentWithCreation(Provider):
    '''Instrument Provider (MaRDI Portal / Wikidata),
       User Creation, No Refresh Upon Selection
    '''

    search = True

    def get_options(self, project, search=None, user=None, site=None):
        '''Query external knowledge-graph source(s) and return matching options.

        Args:
            project: RDMO project instance (used for user-entry lookups when applicable).
            search:  Search string entered by the user; returns empty list when
                     fewer than 3 characters.
            user:    Requesting user (unused).
            site:    Current site (unused).

        Returns:
            List of ``{"id": …, "text": …}`` option dicts sorted by relevance.
        '''
        if not search:
            return []

        setup = define_setup(
            creation = True,
            query_attributes = None,
            item_class = _ITEMS['research tool'],
        )

        return query_sources_with_user_additions(
            search = search,
            project = project,
            setup = setup
        )

get_options(project, search=None, user=None, site=None)

Query external knowledge-graph source(s) and return matching options.

Parameters:

Name	Type	Description	Default
project		RDMO project instance (used for user-entry lookups when applicable).	required
search		Search string entered by the user; returns empty list when fewer than 3 characters.	None
user		Requesting user (unused).	None
site		Current site (unused).	None

Returns:

Type	Description
	List of {"id": …, "text": …} option dicts sorted by relevance.

Source code in MaRDMO/workflow/providers.py

def get_options(self, project, search=None, user=None, site=None):
    '''Query external knowledge-graph source(s) and return matching options.

    Args:
        project: RDMO project instance (used for user-entry lookups when applicable).
        search:  Search string entered by the user; returns empty list when
                 fewer than 3 characters.
        user:    Requesting user (unused).
        site:    Current site (unused).

    Returns:
        List of ``{"id": …, "text": …}`` option dicts sorted by relevance.
    '''
    if not search:
        return []

    setup = define_setup(
        creation = True,
        query_attributes = None,
        item_class = _ITEMS['research tool'],
    )

    return query_sources_with_user_additions(
        search = search,
        project = project,
        setup = setup
    )

RelatedMethodWithCreation

Bases: Provider

Method Provider (MaRDI Portal / Wikidata), User Creation, No Refresh Upon Selection

Source code in MaRDMO/workflow/providers.py

class RelatedMethodWithCreation(Provider):
    '''Method Provider (MaRDI Portal / Wikidata),
       User Creation, No Refresh Upon Selection
    '''

    search = True

    def get_options(self, project, search=None, user=None, site=None):
        '''Query external knowledge-graph source(s) and return matching options.

        Args:
            project: RDMO project instance (used for user-entry lookups when applicable).
            search:  Search string entered by the user; returns empty list when
                     fewer than 3 characters.
            user:    Requesting user (unused).
            site:    Current site (unused).

        Returns:
            List of ``{"id": …, "text": …}`` option dicts sorted by relevance.
        '''
        if not search:
            return []

        setup = define_setup(
            creation = True,
            query_attributes = None,
            item_class = [
                _ITEMS['method'],
            ]
        )

        return query_sources_with_user_additions(
            search = search,
            project = project,
            setup = setup
        )

get_options(project, search=None, user=None, site=None)

Query external knowledge-graph source(s) and return matching options.

Parameters:

Name	Type	Description	Default
project		RDMO project instance (used for user-entry lookups when applicable).	required
search		Search string entered by the user; returns empty list when fewer than 3 characters.	None
user		Requesting user (unused).	None
site		Current site (unused).	None

Returns:

Type	Description
	List of {"id": …, "text": …} option dicts sorted by relevance.

Source code in MaRDMO/workflow/providers.py

def get_options(self, project, search=None, user=None, site=None):
    '''Query external knowledge-graph source(s) and return matching options.

    Args:
        project: RDMO project instance (used for user-entry lookups when applicable).
        search:  Search string entered by the user; returns empty list when
                 fewer than 3 characters.
        user:    Requesting user (unused).
        site:    Current site (unused).

    Returns:
        List of ``{"id": …, "text": …}`` option dicts sorted by relevance.
    '''
    if not search:
        return []

    setup = define_setup(
        creation = True,
        query_attributes = None,
        item_class = [
            _ITEMS['method'],
        ]
    )

    return query_sources_with_user_additions(
        search = search,
        project = project,
        setup = setup
    )

RelatedProgrammingLanguageWithCreation

Bases: Provider

Method Provider (MaRDI Portal / Wikidata), User Creation, No Refresh Upon Selection

Source code in MaRDMO/workflow/providers.py

class RelatedProgrammingLanguageWithCreation(Provider):
    '''Method Provider (MaRDI Portal / Wikidata),
       User Creation, No Refresh Upon Selection
    '''

    search = True

    def get_options(self, project, search=None, user=None, site=None):
        '''Query external knowledge-graph source(s) and return matching options.

        Args:
            project: RDMO project instance (used for user-entry lookups when applicable).
            search:  Search string entered by the user; returns empty list when
                     fewer than 3 characters.
            user:    Requesting user (unused).
            site:    Current site (unused).

        Returns:
            List of ``{"id": …, "text": …}`` option dicts sorted by relevance.
        '''
        if not search:
            return []

        setup = define_setup(
            creation = True,
            query_attributes = None,
            item_class = [
                _ITEMS['programming language'],
            ]
        )

        return query_sources_with_user_additions(
            search = search,
            project = project,
            setup = setup
        )

get_options(project, search=None, user=None, site=None)

Query external knowledge-graph source(s) and return matching options.

Parameters:

Name	Type	Description	Default
project		RDMO project instance (used for user-entry lookups when applicable).	required
search		Search string entered by the user; returns empty list when fewer than 3 characters.	None
user		Requesting user (unused).	None
site		Current site (unused).	None

Returns:

Type	Description
	List of {"id": …, "text": …} option dicts sorted by relevance.

Source code in MaRDMO/workflow/providers.py

def get_options(self, project, search=None, user=None, site=None):
    '''Query external knowledge-graph source(s) and return matching options.

    Args:
        project: RDMO project instance (used for user-entry lookups when applicable).
        search:  Search string entered by the user; returns empty list when
                 fewer than 3 characters.
        user:    Requesting user (unused).
        site:    Current site (unused).

    Returns:
        List of ``{"id": …, "text": …}`` option dicts sorted by relevance.
    '''
    if not search:
        return []

    setup = define_setup(
        creation = True,
        query_attributes = None,
        item_class = [
            _ITEMS['programming language'],
        ]
    )

    return query_sources_with_user_additions(
        search = search,
        project = project,
        setup = setup
    )

RelatedStepWithCreation

Bases: Provider

Step Provider (MaRDI Portal / Wikidata), User Creation, No Refresh Upon Selection

Source code in MaRDMO/workflow/providers.py

class RelatedStepWithCreation(Provider):
    '''Step Provider (MaRDI Portal / Wikidata),
       User Creation, No Refresh Upon Selection
    '''

    search = True

    def get_options(self, project, search=None, user=None, site=None):
        '''Query external knowledge-graph source(s) and return matching options.

        Args:
            project: RDMO project instance (used for user-entry lookups when applicable).
            search:  Search string entered by the user; returns empty list when
                     fewer than 3 characters.
            user:    Requesting user (unused).
            site:    Current site (unused).

        Returns:
            List of ``{"id": …, "text": …}`` option dicts sorted by relevance.
        '''
        if not search or len(search) < 3:
            return []

        setup = define_setup(
            query_attributes = ['processstep'],
            creation = True,
            item_class = [
                _ITEMS['process step']
            ]
        )

        return query_sources_with_user_additions(
            search = search,
            project = project,
            setup = setup
        )

get_options(project, search=None, user=None, site=None)

Query external knowledge-graph source(s) and return matching options.

Parameters:

Name	Type	Description	Default
project		RDMO project instance (used for user-entry lookups when applicable).	required
search		Search string entered by the user; returns empty list when fewer than 3 characters.	None
user		Requesting user (unused).	None
site		Current site (unused).	None

Returns:

Type	Description
	List of {"id": …, "text": …} option dicts sorted by relevance.

Source code in MaRDMO/workflow/providers.py

def get_options(self, project, search=None, user=None, site=None):
    '''Query external knowledge-graph source(s) and return matching options.

    Args:
        project: RDMO project instance (used for user-entry lookups when applicable).
        search:  Search string entered by the user; returns empty list when
                 fewer than 3 characters.
        user:    Requesting user (unused).
        site:    Current site (unused).

    Returns:
        List of ``{"id": …, "text": …}`` option dicts sorted by relevance.
    '''
    if not search or len(search) < 3:
        return []

    setup = define_setup(
        query_attributes = ['processstep'],
        creation = True,
        item_class = [
            _ITEMS['process step']
        ]
    )

    return query_sources_with_user_additions(
        search = search,
        project = project,
        setup = setup
    )

RelatedWorkflowEntityWithoutCreation

Bases: Provider

Interdisciplinary Workflow, Process Step, and Data Set Provider (MaRDI Portal / Wikidata), No User Creation, No Refresh Upon Selection

Source code in MaRDMO/workflow/providers.py

class RelatedWorkflowEntityWithoutCreation(Provider):
    '''Interdisciplinary Workflow, Process Step, and Data Set  Provider 
       (MaRDI Portal / Wikidata), No User Creation, No Refresh Upon Selection
    '''

    search = True

    def get_options(self, project, search=None, user=None, site=None):
        '''Query the MathModDB ontology and return matching options.

        Args:
            project: RDMO project instance (used for user-entry lookups when applicable).
            search:  Search string entered by the user; returns empty list when
                     fewer than 3 characters.
            user:    Requesting user (unused).
            site:    Current site (unused).

        Returns:
            List of ``{"id": …, "text": …}`` option dicts sorted by relevance.
        '''
        if not search or len(search) < 3:
            return []


        # Define the query_setup
        setup = define_setup(
            query_attributes = ['workflow', 'processstep', 'dataset'],
            sources = ['mardi'],
            item_class = [
                _ITEMS['research workflow'],
                _ITEMS['process step'],
                _ITEMS['data set'],
            ]
        )

        return query_sources_with_user_additions(
            search = search,
            project = project,
            setup = setup
        )

get_options(project, search=None, user=None, site=None)

Query the MathModDB ontology and return matching options.

Parameters:

Name	Type	Description	Default
project		RDMO project instance (used for user-entry lookups when applicable).	required
search		Search string entered by the user; returns empty list when fewer than 3 characters.	None
user		Requesting user (unused).	None
site		Current site (unused).	None

Returns:

Type	Description
	List of {"id": …, "text": …} option dicts sorted by relevance.

Source code in MaRDMO/workflow/providers.py

def get_options(self, project, search=None, user=None, site=None):
    '''Query the MathModDB ontology and return matching options.

    Args:
        project: RDMO project instance (used for user-entry lookups when applicable).
        search:  Search string entered by the user; returns empty list when
                 fewer than 3 characters.
        user:    Requesting user (unused).
        site:    Current site (unused).

    Returns:
        List of ``{"id": …, "text": …}`` option dicts sorted by relevance.
    '''
    if not search or len(search) < 3:
        return []


    # Define the query_setup
    setup = define_setup(
        query_attributes = ['workflow', 'processstep', 'dataset'],
        sources = ['mardi'],
        item_class = [
            _ITEMS['research workflow'],
            _ITEMS['process step'],
            _ITEMS['data set'],
        ]
    )

    return query_sources_with_user_additions(
        search = search,
        project = project,
        setup = setup
    )

RelatedWorkflowWithoutCreation

Bases: Provider

Workflow Provider (MaRDI Portal), No User Creation, No Refresh Upon Selection

Source code in MaRDMO/workflow/providers.py

class RelatedWorkflowWithoutCreation(Provider):
    '''Workflow Provider (MaRDI Portal),
       No User Creation, No Refresh Upon Selection
    '''

    search = True

    def get_options(self, project, search=None, user=None, site=None):
        '''Query external knowledge-graph source(s) and return matching options.

        Args:
            project: RDMO project instance (used for user-entry lookups when applicable).
            search:  Search string entered by the user; returns empty list when
                     fewer than 3 characters.
            user:    Requesting user (unused).
            site:    Current site (unused).

        Returns:
            List of ``{"id": …, "text": …}`` option dicts sorted by relevance.
        '''
        if not search or len(search) < 3:
            return []

        setup = define_setup(
            query_attributes = ['workflow'],
            sources = ['mardi'],
            item_class = _ITEMS['research workflow']
        )

        return query_sources_with_user_additions(
            search = search,
            project = project,
            setup = setup
        )

get_options(project, search=None, user=None, site=None)

Query external knowledge-graph source(s) and return matching options.

Parameters:

Name	Type	Description	Default
project		RDMO project instance (used for user-entry lookups when applicable).	required
search		Search string entered by the user; returns empty list when fewer than 3 characters.	None
user		Requesting user (unused).	None
site		Current site (unused).	None

Returns:

Type	Description
	List of {"id": …, "text": …} option dicts sorted by relevance.

Source code in MaRDMO/workflow/providers.py

def get_options(self, project, search=None, user=None, site=None):
    '''Query external knowledge-graph source(s) and return matching options.

    Args:
        project: RDMO project instance (used for user-entry lookups when applicable).
        search:  Search string entered by the user; returns empty list when
                 fewer than 3 characters.
        user:    Requesting user (unused).
        site:    Current site (unused).

    Returns:
        List of ``{"id": …, "text": …}`` option dicts sorted by relevance.
    '''
    if not search or len(search) < 3:
        return []

    setup = define_setup(
        query_attributes = ['workflow'],
        sources = ['mardi'],
        item_class = _ITEMS['research workflow']
    )

    return query_sources_with_user_additions(
        search = search,
        project = project,
        setup = setup
    )

Task

Bases: Provider

Computational Task Provider (MaRDI Portal), No User Creation, No Refresh Upon Selection

Source code in MaRDMO/workflow/providers.py

class Task(Provider):
    '''Computational Task Provider (MaRDI Portal),
       No User Creation, No Refresh Upon Selection
    '''

    search = True

    def get_options(self, project, search=None, user=None, site=None):
        '''Query external knowledge-graph source(s) and return matching options.

        Args:
            project: RDMO project instance (used for user-entry lookups when applicable).
            search:  Search string entered by the user; returns empty list when
                     fewer than 3 characters.
            user:    Requesting user (unused).
            site:    Current site (unused).

        Returns:
            List of ``{"id": …, "text": …}`` option dicts sorted by relevance.
        '''
        if not search or len(search) < 3:
            return []

        return query_sources(
            search = search,
            item_class = _ITEMS['computational task'],
            sources = ['mardi'],
            not_found = False
        )

get_options(project, search=None, user=None, site=None)

Query external knowledge-graph source(s) and return matching options.

Parameters:

Name	Type	Description	Default
project		RDMO project instance (used for user-entry lookups when applicable).	required
search		Search string entered by the user; returns empty list when fewer than 3 characters.	None
user		Requesting user (unused).	None
site		Current site (unused).	None

Returns:

Type	Description
	List of {"id": …, "text": …} option dicts sorted by relevance.

Source code in MaRDMO/workflow/providers.py

def get_options(self, project, search=None, user=None, site=None):
    '''Query external knowledge-graph source(s) and return matching options.

    Args:
        project: RDMO project instance (used for user-entry lookups when applicable).
        search:  Search string entered by the user; returns empty list when
                 fewer than 3 characters.
        user:    Requesting user (unused).
        site:    Current site (unused).

    Returns:
        List of ``{"id": …, "text": …}`` option dicts sorted by relevance.
    '''
    if not search or len(search) < 3:
        return []

    return query_sources(
        search = search,
        item_class = _ITEMS['computational task'],
        sources = ['mardi'],
        not_found = False
    )

Workflow

Bases: Provider

Workflow Provider (MaRDI Portal / Wikidata), No User Creation, Refresh Upon Selection

Source code in MaRDMO/workflow/providers.py

class Workflow(Provider):
    '''Workflow Provider (MaRDI Portal / Wikidata),
       No User Creation, Refresh Upon Selection
    '''

    search = True
    refresh = True

    def get_options(self, project, search=None, user=None, site=None):
        '''Query external knowledge-graph source(s) and return matching options.

        Args:
            project: RDMO project instance (used for user-entry lookups when applicable).
            search:  Search string entered by the user; returns empty list when
                     fewer than 3 characters.
            user:    Requesting user (unused).
            site:    Current site (unused).

        Returns:
            List of ``{"id": …, "text": …}`` option dicts sorted by relevance.
        '''
        if not search or len(search) < 3:
            return []

        return query_sources(
            search = search,
            item_class = _ITEMS['research workflow'],
        )

get_options(project, search=None, user=None, site=None)

Query external knowledge-graph source(s) and return matching options.

Parameters:

Name	Type	Description	Default
project		RDMO project instance (used for user-entry lookups when applicable).	required
search		Search string entered by the user; returns empty list when fewer than 3 characters.	None
user		Requesting user (unused).	None
site		Current site (unused).	None

Returns:

Type	Description
	List of {"id": …, "text": …} option dicts sorted by relevance.

Source code in MaRDMO/workflow/providers.py

def get_options(self, project, search=None, user=None, site=None):
    '''Query external knowledge-graph source(s) and return matching options.

    Args:
        project: RDMO project instance (used for user-entry lookups when applicable).
        search:  Search string entered by the user; returns empty list when
                 fewer than 3 characters.
        user:    Requesting user (unused).
        site:    Current site (unused).

    Returns:
        List of ``{"id": …, "text": …}`` option dicts sorted by relevance.
    '''
    if not search or len(search) < 3:
        return []

    return query_sources(
        search = search,
        item_class = _ITEMS['research workflow'],
    )

Workers

Worker module for Interdisciplinary Workflow preview and export.

Provides :class:PrepareWorkflow, which assembles RDMO questionnaire answers into a :class:~MaRDMO.payload.GeneratePayload ready for submission to the MaRDI Portal Wikibase instance.

PrepareWorkflow

Bases: PublicationExport

Prepare interdisciplinary workflow answers for preview and export.

Inherits publication export helpers from :class:~MaRDMO.publication.worker.PublicationExport and extends them with workflow-specific relation mapping and Wikibase payload generation.

Source code in MaRDMO/workflow/worker.py

class PrepareWorkflow(PublicationExport):
    '''Prepare interdisciplinary workflow answers for preview and export.

    Inherits publication export helpers from
    :class:`~MaRDMO.publication.worker.PublicationExport` and extends them
    with workflow-specific relation mapping and Wikibase payload generation.
    '''

    def __init__(self):
        '''Initialise with Wikibase vocabulary and ontology registries.'''
        super().__init__()
        self.mathmoddb           = get_mathmoddb()
        self.mathalgodb          = get_mathalgodb()
        self.publication_mapping = get_publication_mapping()

    def preview(self, answers):
        '''Enrich the answers dict with derived display structures for the preview template.

        First iterates over :data:`~MaRDMO.workflow.constants.preview_relations` and calls
        :func:`~MaRDMO.helpers.entity_relations` for each entry to resolve cross-entity
        relation labels (e.g. Workflow → Process Step) into the ``answers`` dict — the same
        pattern used by the model and algorithm preview workers.

        Then builds ``model_task_pairs`` — an ordered list of dicts with keys ``model``,
        ``tasks``, ``has_model``, and ``has_tasks`` — for every workflow in the project
        and stores it directly on each workflow's sub-dict
        (``answers['workflow'][i]['model_task_pairs']``) so the template can access it as
        ``values.model_task_pairs`` inside the ``{% for values in answers.workflow.values %}``
        loop without integer-key lookups.  Indices from both models *and* tasks are unioned
        so that a task with no matching model (or vice versa) still produces a row with the
        appropriate validation flags set.

        Similarly builds ``algorithm_software_pairs`` and ``method_instrument_pairs`` —
        ordered lists of dicts with generic keys ``primary``, ``qualifier``,
        ``has_primary``, and ``has_qualifier`` — for every process step.  Index *i*
        pairs algorithm *i* with software *i* (and method *i* with instrument *i*);
        a missing partner raises a validation flag rendered as a red warning via the
        ``_relation_block_single_with_qualifier.html`` include.

        Args:
            answers: Top-level answers dict produced by ``get_post_data``.

        Returns:
            The same *answers* dict with relation labels and ``model_task_pairs`` added.
        '''

        # Prepare Relations for Preview
        for relation in preview_relations:
            fn = entity_relations_grouped if relation.get('grouped') else entity_relations
            fn(
                data = answers,
                idx = {
                    'from': relation['from_idx'],
                    'to': relation['to_idx']
                },
                entity = {
                    'relation': relation['relation'],
                    'old_name': relation['old_name'],
                    'new_name': relation['new_name'],
                    'encryption': relation['encryption']
                },
                order = {
                    'formulation': False,
                    'task': False
                },
                assumption = False,
                mapping = (
                    self.mathmoddb           if relation.get('mapping') == 'mathmoddb'
                    else self.publication_mapping if relation.get('mapping') == 'publication'
                    else self.mathalgodb
                )
            )

        for wf_data in answers.get('workflow', {}).values():
            models = wf_data.get('model', {})
            tasks  = wf_data.get('task', {})
            all_indices = sorted(set(models) | set(tasks))
            wf_data['model_task_pairs'] = [
                {
                    'model':     models.get(idx),
                    'tasks':     list(tasks.get(idx, {}).values()),
                    'has_model': bool(models.get(idx)),
                    'has_tasks': bool(tasks.get(idx)),
                }
                for idx in all_indices
            ]

        for ps_data in answers.get('processstep', {}).values():
            # RelationA keys are 'set_index|collection_index' strings;
            # group all algorithms sharing the same set_index into one list.
            algo_by_set = {}
            for key, val in ps_data.get('RelationA', {}).items():
                set_idx = int(str(key).split('|')[0])
                algo_by_set.setdefault(set_idx, []).append(val)
            software = ps_data.get('RelationS', {})
            hardware = ps_data.get('RelationH', {})
            software_doc = ps_data.get('software-documentation', {})
            algo_params = ps_data.get('algorithm-parameter', {})
            all_indices = sorted(
                set(algo_by_set) | set(software) | set(hardware) | set(software_doc)
            )
            ps_data['algorithm_software_pairs'] = [
                {
                    'primary':       algo_by_set.get(idx, []),
                    'qualifier':     software.get(idx),
                    'has_primary':   bool(algo_by_set.get(idx)),
                    'has_qualifier': bool(software.get(idx)),
                    'parameters':    ', '.join(
                        v[1] for v in sorted(algo_params.get(idx, {}).items())
                        if v[1]
                    ),
                    'software_doc':  software_doc.get(idx, {}),
                    'hardware':      hardware.get(idx),
                }
                for idx in all_indices
            ]

            # Use raw MRelatant/IRelatant dicts so the template can display
            # Name (Description) for inline items that have no dedicated section.
            meth_by_set = {
                set_idx: [raw]
                for set_idx, raw in ps_data.get('MRelatant', {}).items()
            }
            instruments = ps_data.get('IRelatant', {})
            method_doc  = ps_data.get('method-documentation', {})
            meth_params = ps_data.get('method-parameter', {})
            all_indices = sorted(set(meth_by_set) | set(instruments) | set(method_doc))
            ps_data['method_instrument_pairs'] = [
                {
                    'primary':       meth_by_set.get(idx, []),
                    'qualifier':     instruments.get(idx),
                    'has_primary':   bool(meth_by_set.get(idx)),
                    'has_qualifier': bool(instruments.get(idx)),
                    'parameters':    ', '.join(
                        v[1] for v in sorted(meth_params.get(idx, {}).items())
                        if v[1]
                    ),
                    'method_doc':    method_doc.get(idx, {}),
                }
                for idx in all_indices
            ]

        for hw_data in answers.get('hardware', {}).values():
            cpu_dict   = hw_data.get('cpu', {})
            count_dict = hw_data.get('number-of-cpu', {})
            cores_dict = hw_data.get('cores', {})
            all_indices = sorted(set(cpu_dict) | set(count_dict) | set(cores_dict))
            hw_data['cpu_entries'] = [
                {
                    'id':          cpu_dict.get(idx, {}).get('ID'),
                    'name':        cpu_dict.get(idx, {}).get('Name'),
                    'description': cpu_dict.get(idx, {}).get('Description'),
                    'count':       count_dict.get(idx),
                    'count_valid': (
                        str(count_dict.get(idx, '')).strip().isdigit()
                        if count_dict.get(idx) else False
                    ),
                    'cores':       cores_dict.get(idx),
                    'cores_valid': (
                        str(cores_dict.get(idx, '')).strip().isdigit()
                        if cores_dict.get(idx) else False
                    ),
                }
                for idx in all_indices
            ]
            nodes = hw_data.get('nodes')
            hw_data['nodes_valid'] = str(nodes).strip().isdigit() if nodes else False

        answers['cpu'] = collect_items_without_section(answers, 'hardware', 'cpu')
        answers['programminglanguage'] = collect_items_without_section(
            answers, 'software', 'programminglanguage'
        )
        answers['algorithmictask'] = collect_items_without_section(
            answers, 'algorithm', 'PRelatant'
        )
        answers['academicdiscipline'] = collect_items_without_section(
            answers, 'processstep', 'RFRelatant'
        )
        answers['method'] = collect_items_without_section(
            answers, 'processstep', 'MRelatant', nested=True
        )
        answers['instrument'] = collect_items_without_section(
            answers, 'processstep', 'IRelatant'
        )

        options = get_options()

        for ds_data in answers.get('dataset', {}).values():
            size = ds_data.get('Size')
            if size and size[0] and size[1]:
                val = str(size[1]).strip()
                if size[0] == options['items']:
                    ds_data['size_value_valid'] = val.isdigit()
                else:
                    try:
                        float(val)
                        ds_data['size_value_valid'] = True
                    except (ValueError, TypeError):
                        ds_data['size_value_valid'] = False
            archive = ds_data.get('ToArchive')
            if archive and archive[0] and archive[1]:
                val = str(archive[1]).strip()
                ds_data['archive_year_valid'] = len(val) == 4 and val.isdigit()
            publish = ds_data.get('ToPublish')
            if publish and len(publish) > 1 and publish[1]:
                val = str(publish[1]).strip()
                if not val.startswith('10.'):
                    ds_data['publish_url_valid'] = is_valid_url(val)

        for sw_data in answers.get('software', {}).values():
            ref = sw_data.get('reference', {})
            if ref.get(2) and ref[2][1]:
                sw_data['ref_desc_url_valid'] = is_valid_url(ref[2][1])
            if ref.get(3) and ref[3][1]:
                sw_data['ref_repo_url_valid'] = is_valid_url(ref[3][1])

        for ps_data in answers.get('processstep', {}).values():
            for pair in ps_data.get('algorithm_software_pairs', []):
                pair['software_doc_url_invalid'] = {
                    idx
                    for idx, entry in pair.get('software_doc', {}).items()
                    if entry and entry[0] == options['URL'] and entry[1]
                    and not is_valid_url(entry[1])
                }
            for pair in ps_data.get('method_instrument_pairs', []):
                pair['method_doc_url_invalid'] = {
                    idx
                    for idx, entry in pair.get('method_doc', {}).items()
                    if entry and entry[0] == options['URL'] and entry[1]
                    and not is_valid_url(entry[1])
                }

        return answers

    def export(self, data, url):
        '''Assemble and return the complete Wikibase payload for a Workflow documentation export.

        Args:
            data: Top-level workflow answers dict produced by ``get_post_data``.
            url:  Target Wikibase API URL for the upload.

        Returns:
            Tuple ``(payload_dict, dependency_order)`` ready for
            :meth:`~MaRDMO.oauth2.OauthProviderMixin.post`.
        '''

        items, dependency = unique_items(data)

        payload = GeneratePayload(
            dependency = dependency,
            user_items = items,
            url = url,
            wikibase = {
                'items':      get_items(),
                'properties': get_properties(),
                'relations':  get_relations(),
            }
        )

        payload.process_items()

        self._export_workflows(
            payload = payload,
            data    = data,
        )
        self._export_processsteps(
            payload      = payload,
            processsteps = data.get('processstep', {}),
        )
        self._export_algorithms(
            payload    = payload,
            algorithms = data.get('algorithm', {}),
        )
        algo_software_keys = {
            (item.get('ID', ''), item.get('Name', ''), item.get('Description', ''))
            for algo in data.get('algorithm', {}).values()
            for item in algo.get('SRelatant', {}).values()
        }
        self._export_softwares(
            payload           = payload,
            softwares         = data.get('software', {}),
            algo_software_keys = algo_software_keys,
        )
        self._export_hardwares(
            payload   = payload,
            hardwares = data.get('hardware', {}),
        )
        self._export_datasets(
            payload  = payload,
            datasets = data.get('dataset', {}),
        )
        self._export_cpus(
            payload = payload,
            cpus    = data.get('cpu', {}),
        )
        self._export_programming_languages(
            payload              = payload,
            programminglanguages = data.get('programminglanguage', {}),
        )
        self._export_methods(
            payload = payload,
            methods = data.get('method', {}),
        )
        self._export_instruments(
            payload     = payload,
            instruments = data.get('instrument', {}),
        )
        self._export_algorithmic_tasks(
            payload           = payload,
            algorithmic_tasks = data.get('algorithmictask', {}),
        )
        self._export_academic_disciplines(
            payload              = payload,
            academic_disciplines = data.get('academicdiscipline', {}),
        )
        self._export_authors(
            payload      = payload,
            publications = data.get('publication', {}),
        )
        self._export_journals(
            payload      = payload,
            publications = data.get('publication', {}),
        )
        self._export_publications(
            payload      = payload,
            publications = data.get('publication', {}),
            relations    = [
                ('P2A', 'ARelatant'), ('P2BS', 'HSRelatant'), ('P2E', 'EntityRelatant'),
            ],
        )

        payload.add_item_payload()

        if any(key.startswith('RELATION') for key in payload.get_dictionary()):
            query = payload.build_relation_check_query()

            check = None
            for attempt in range(2):
                try:
                    check = query_sparql(
                        query           = query,
                        sparql_endpoint = get_url('mardi', 'sparql'),
                    )
                    break
                except Exception as e:
                    logging.warning("SPARQL query attempt %s failed: %s", attempt + 1, e)
                    if attempt == 0:
                        time.sleep(1)
            if not check:
                check = [{}]

            payload.add_check_results(check=check)

        return payload.get_dictionary(), payload.dependency

    # ------------------------------------------------------------------
    # Entity export helpers
    # ------------------------------------------------------------------

    def _export_workflows(self, payload, data: dict):
        '''Export the top-level research workflow item.

        Creates the workflow item (instance of: research workflow), adds a
        long description, all reproducibility aspects (from REPRODUCIBILITY
        constant) with optional comment qualifiers, and transferability with
        comment qualifiers.  Then links the workflow to its mathematical model
        with computational-task (used-by) qualifiers, contains process steps,
        and workflow-to-workflow intra-class relations.
        '''
        options = get_options()

        for entry in data.get('workflow', {}).values():
            if not entry.get('ID'):
                continue

            payload.get_item_key(value=entry)
            payload.set_class('Workflow')

            self._add_common_metadata(
                payload      = payload,
                qclass       = self.items["research workflow"],
                profile_type = "MaRDI workflow profile",
                description_long = True,
            )

            # Research objective
            if entry.get('objective'):
                payload.add_answer(
                    verb            = self.properties["problem statement"],
                    object_and_type = [entry['objective'], "string"],
                )

            # Reproducibility aspects
            for key in REPRODUCIBILITY:
                for val in entry.get(key, {}).values():
                    if not val:
                        continue
                    if val[0] == options['YesLargeText']:
                        qualifiers = []
                        if val[1]:
                            qualifiers += payload.add_qualifier(
                                self.properties["comment"], "string", val[1]
                            )
                        payload.add_answer(
                            verb            = self.properties["instance of"],
                            object_and_type = [self.items[REPRODUCIBILITY[key]], "wikibase-item"],
                            qualifier       = qualifiers,
                        )
                    elif val[0] == options['No']:
                        payload.add_answer(
                            verb            = self.properties["instance of"],
                            object_and_type = [self.items[IRREPRODUCIBILITY[key]], "wikibase-item"],
                        )

            # Transferability
            transferability = entry.get('transferability', {})
            if transferability:
                qualifiers = []
                for coll_dict in transferability.values():
                    for val in coll_dict.values():
                        if val:
                            qualifiers += payload.add_qualifier(
                                self.properties["comment"], "string", val
                            )
                payload.add_answer(
                    verb            = self.properties["instance of"],
                    object_and_type = [
                        self.items["transferable research workflow"], "wikibase-item",
                    ],
                    qualifier       = qualifiers,
                )

            # Mathematical model → uses, with computational task qualifiers
            for set_idx, model_item in entry.get('model', {}).items():
                if not model_item.get('ID'):
                    continue
                model_key = payload.get_item_key(model_item, 'object')
                if model_key not in payload.get_dictionary():
                    continue
                qualifiers = []
                for task_item in entry.get('task', {}).get(set_idx, {}).values():
                    if not task_item.get('ID'):
                        continue
                    task_key = payload.get_item_key(task_item, 'object')
                    if task_key not in payload.get_dictionary():
                        continue
                    qualifiers += payload.add_qualifier(
                        self.properties["used by"], "wikibase-item", task_key
                    )
                payload.add_answer(
                    verb            = self.properties["uses"],
                    object_and_type = [model_key, "wikibase-item"],
                    qualifier       = qualifiers,
                )

            # Process steps
            payload.add_single_relation(
                statement = {'relation': self.properties["contains"], 'relatant': "PSRelatant"}
            )

            # Workflow–Workflow intra-class relations
            payload.add_multiple_relation(
                statement = {'relation': "IntraClassRelation", 'relatant': "IntraClassElement"}
            )

    def _export_processsteps(self, payload, processsteps: dict):
        '''Export each process step item.

        Sets instance of: process step, then adds input/output data-set
        relations, applied methods with parameter qualifiers, software
        environment (platform relation with object-has-role: software
        qualifier) and instrument environment (platform relation with
        object-has-role: research tool qualifier), and field-of-work using
        a Wikibase item for each discipline.
        '''
        options = get_options()

        for entry in processsteps.values():
            if not entry.get("ID"):
                continue

            payload.get_item_key(value=entry)
            payload.set_class('Process Step')

            payload.add_answer(
                verb = self.properties["instance of"],
                object_and_type = [self.items["process step"], "wikibase-item"],
            )

            payload.add_single_relation(
                statement = {
                    'relation': self.properties["input data set"], 'relatant': "IDSRelatant",
                }
            )
            payload.add_single_relation(
                statement = {
                    'relation': self.properties["output data set"], 'relatant': "ODSRelatant",
                }
            )

            # Algorithms: one 'uses' statement per (set_idx, collection_idx)
            for set_idx, algo_items in entry.get('ARelatant', {}).items():
                qualifiers = []

                sw = entry.get('SRelatant', {}).get(set_idx)
                if sw and sw.get('ID'):
                    sw_key = payload.get_item_key(sw, 'object')
                    if sw_key in payload.get_dictionary():
                        qualifiers += payload.add_qualifier(
                            self.properties["platform"], "wikibase-item", sw_key
                        )

                hw = entry.get('HRelatant', {}).get(set_idx)
                if hw and hw.get('ID'):
                    hw_key = payload.get_item_key(hw, 'object')
                    if hw_key in payload.get_dictionary():
                        qualifiers += payload.add_qualifier(
                            self.properties["platform"], "wikibase-item", hw_key
                        )

                for param in entry.get('algorithm-parameter', {}).get(set_idx, {}).values():
                    if param:
                        qualifiers += payload.add_qualifier(
                            self.properties["comment"], "string", param
                        )

                for doc in entry.get('software-documentation', {}).get(set_idx, {}).values():
                    if doc and len(doc) > 1 and doc[1]:
                        if doc[0] == options['DOI']:
                            qualifiers += payload.add_qualifier(
                                self.properties["DOI"], "external-id", doc[1]
                            )
                        elif doc[0] == options['URL']:
                            qualifiers += payload.add_qualifier(
                                self.properties["URL"], "url", doc[1]
                            )

                for algo_item in algo_items.values():
                    if not algo_item.get('ID'):
                        continue
                    algo_key = payload.get_item_key(algo_item, 'object')
                    if algo_key not in payload.get_dictionary():
                        continue
                    payload.add_answer(
                        verb = self.properties["uses"],
                        object_and_type = [algo_key, "wikibase-item"],
                        qualifier = qualifiers,
                    )

            # Methods: one 'uses' statement per (set_idx, collection_idx)
            for set_idx, meth_items in entry.get('MRelatant', {}).items():
                qualifiers = []

                instr = entry.get('IRelatant', {}).get(set_idx)
                if instr and instr.get('ID'):
                    instr_key = payload.get_item_key(instr, 'object')
                    if instr_key in payload.get_dictionary():
                        qualifiers += payload.add_qualifier(
                            self.properties["platform"], "wikibase-item", instr_key
                        )

                for param in entry.get('method-parameter', {}).get(set_idx, {}).values():
                    if param:
                        qualifiers += payload.add_qualifier(
                            self.properties["comment"], "string", param
                        )

                for doc in entry.get('method-documentation', {}).get(set_idx, {}).values():
                    if doc and len(doc) > 1 and doc[1]:
                        if doc[0] == options['DOI']:
                            qualifiers += payload.add_qualifier(
                                self.properties["DOI"], "external-id", doc[1]
                            )
                        elif doc[0] == options['URL']:
                            qualifiers += payload.add_qualifier(
                                self.properties["URL"], "url", doc[1]
                            )

                for meth_item in meth_items.values():
                    if not meth_item.get('ID'):
                        continue
                    meth_key = payload.get_item_key(meth_item, 'object')
                    if meth_key not in payload.get_dictionary():
                        continue
                    payload.add_answer(
                        verb = self.properties["uses"],
                        object_and_type = [meth_key, "wikibase-item"],
                        qualifier = qualifiers,
                    )

            payload.add_single_relation(
                statement = {'relation': self.properties["field of work"], 'relatant': "RFRelatant"}
            )

    def _export_algorithms(self, payload, algorithms: dict):
        '''Export each algorithm item.

        Sets instance of: algorithm (community: MathAlgoDB, profile: MaRDI
        algorithm profile), links to its algorithmic tasks via solved-by
        (reverse) and to implementing software via implemented-by.
        No algorithm-to-algorithm intra-class relations are exported for the
        workflow catalog.
        '''
        for entry in algorithms.values():
            if not entry.get("ID"):
                continue

            payload.get_item_key(value=entry)
            payload.set_class('Algorithm')

            self._add_common_metadata(
                payload      = payload,
                community    = self.items["MathAlgoDB"],
                qclass       = self.items["algorithm"],
                profile_type = "MaRDI algorithm profile",
            )

            payload.add_single_relation(
                statement = {
                    'relation': self.properties["solved by"],
                    'relatant': "PRelatant"
                },
                reverse = True
            )

            payload.add_single_relation(
                statement = {
                    'relation': self.properties["implemented by"],
                    'relatant': "SRelatant"
                }
            )

    def _export_softwares(self, payload, softwares: dict, algo_software_keys: set):
        '''Export each software item.

        Software linked to an algorithm via "implemented by" receives community
        MathAlgoDB and the MaRDI software profile; all other software receives
        only the profile.  No benchmark (tested-by) relation is exported for
        the workflow catalog.
        '''
        for entry in softwares.values():
            if not entry.get("ID"):
                continue

            payload.get_item_key(value=entry)
            payload.set_class('Software')

            entry_key = (entry.get('ID', ''), entry.get('Name', ''), entry.get('Description', ''))
            if entry_key in algo_software_keys:
                self._add_common_metadata(
                    payload      = payload,
                    community    = self.items["MathAlgoDB"],
                    qclass       = self.items["software"],
                    profile_type = "MaRDI software profile",
                )
            else:
                self._add_common_metadata(
                    payload      = payload,
                    qclass       = self.items["software"],
                    profile_type = "MaRDI software profile",
                )

            payload.add_single_relation(
                statement = {
                    'relation': self.properties["programmed in"],
                    'relatant': "programminglanguage"
                }
            )

            payload.add_single_relation(
                statement = {
                    'relation': self.properties["depends on software"],
                    'relatant': "dependency"
                }
            )

            if entry.get("reference"):
                if entry['reference'].get(0):
                    payload.add_answer(
                        verb = self.properties["DOI"],
                        object_and_type = [entry["reference"][0][1], "external-id"],
                    )
                if entry['reference'].get(1):
                    payload.add_answer(
                        verb = self.properties["swMath work ID"],
                        object_and_type = [entry["reference"][1][1], "external-id"],
                    )
                if entry['reference'].get(2):
                    payload.add_answer(
                        verb = self.properties["source code repository URL"],
                        object_and_type = [entry["reference"][2][1], "URL"],
                    )
                if entry['reference'].get(3):
                    payload.add_answer(
                        verb = self.properties["described at URL"],
                        object_and_type = [entry["reference"][3][1], "URL"],
                    )

    def _export_hardwares(self, payload, hardwares: dict):
        '''Export each hardware item.

        Sets instance of: computer hardware, adds CPU relation, number-of-
        compute-nodes (has-part: compute node with quantity qualifier), and
        number-of-processor-cores (quantity).
        '''
        for entry in hardwares.values():
            if not entry.get("ID"):
                continue

            payload.get_item_key(value=entry)
            payload.set_class('Hardware')

            self._add_common_metadata(
                payload = payload,
                qclass  = self.items["computer hardware"],
            )

            nodes = str(entry.get('nodes', '')).strip()
            if nodes.isdigit():
                payload.add_answer(
                    verb = self.properties["number of compute nodes"],
                    object_and_type = [
                        {"amount": f"+{nodes}", "unit": "1"},
                        "quantity",
                    ],
                )

            for idx, cpu_entry in entry.get('cpu', {}).items():
                if not cpu_entry.get('ID'):
                    continue

                # Add number of processor cores as a statement on the CPU item
                cores = str(entry.get('cores', {}).get(idx, '')).strip()
                if cores.isdigit():
                    payload.get_item_key(cpu_entry)
                    payload.add_answer(
                        verb = self.properties["number of processor cores"],
                        object_and_type = [
                            {"amount": f"+{cores}", "unit": "1"},
                            "quantity",
                        ],
                    )
                    payload.get_item_key(value=entry)

                # Add CPU relation with count qualifier on the hardware item
                cpu_item = payload.get_item_key(cpu_entry, 'object')
                if cpu_item not in payload.get_dictionary():
                    continue
                qualifiers = []
                count = str(entry.get('number-of-cpu', {}).get(idx, '')).strip()
                if count.isdigit():
                    qualifiers += payload.add_qualifier(
                        self.properties["quantity_property"],
                        "quantity",
                        {"amount": f"+{count}", "unit": "1"},
                    )
                payload.add_answer(
                    verb = self.properties["CPU"],
                    object_and_type = [cpu_item, "wikibase-item"],
                    qualifier = qualifiers,
                )

    def _export_datasets(self, payload, datasets: dict):
        '''Export each data set item.

        Sets instance of: data set, adds data-size (with unit from
        kilobyte/megabyte/gigabyte/terabyte options) or number-of-records,
        file-extension, binary/text classification, proprietary/open-data
        classification, data-publishing mandate (with optional DOI and URL),
        and research-data-archiving mandate (with optional end-time qualifier).
        '''
        options = get_options()

        for entry in datasets.values():
            if not entry.get("ID"):
                continue

            payload.get_item_key(value=entry)
            payload.set_class('Dataset')

            self._add_common_metadata(
                payload      = payload,
                qclass       = self.items["data set"],
                profile_type = "MaRDI data set profile",
            )

            if entry.get("Size"):
                unit_map = {
                    options["kilobyte"]: self.items["kilobyte"],
                    options["megabyte"]: self.items["megabyte"],
                    options["gigabyte"]: self.items["gigabyte"],
                    options["terabyte"]: self.items["terabyte"],
                }
                size_unit = entry["Size"][0]
                size_val  = entry["Size"][1]
                if size_unit in unit_map:
                    payload.add_answer(
                        verb = self.properties["data size"],
                        object_and_type = [
                            {
                                "amount": f"+{size_val}",
                                "unit": (
                                    f"{get_url('mardi', 'entity_uri')}"
                                    f"/entity/{unit_map[size_unit]}"
                                ),
                            },
                            "quantity",
                        ],
                    )
                elif size_unit == options["items"]:
                    payload.add_answer(
                        verb = self.properties["number of records"],
                        object_and_type = [
                            {"amount": f"+{size_val}", "unit": "1"},
                            "quantity",
                        ],
                    )

            if entry.get("FileFormat"):
                payload.add_answer(
                    verb = self.properties["file extension"],
                    object_and_type = [entry["FileFormat"], "string"],
                )

            if entry.get("BinaryText"):
                if entry["BinaryText"] == options["binary"]:
                    payload.add_answer(
                        verb = self.properties["instance of"],
                        object_and_type = [self.items["binary data"], "wikibase-item"],
                    )
                elif entry["BinaryText"] == options["text"]:
                    payload.add_answer(
                        verb = self.properties["instance of"],
                        object_and_type = [self.items["text data"], "wikibase-item"],
                    )

            if entry.get("Proprietary"):
                if entry["Proprietary"] == options["Yes"]:
                    payload.add_answer(
                        verb = self.properties["instance of"],
                        object_and_type = [self.items["proprietary information"], "wikibase-item"],
                    )
                elif entry["Proprietary"] == options["No"]:
                    payload.add_answer(
                        verb = self.properties["instance of"],
                        object_and_type = [self.items["open data"], "wikibase-item"],
                    )

            if entry.get("ToPublish"):
                if entry["ToPublish"][0] == options["YesText"]:
                    payload.add_answer(
                        verb = self.properties["mandates"],
                        object_and_type = [self.items["data publishing"], "wikibase-item"],
                    )
                    val = entry["ToPublish"][1] if len(entry["ToPublish"]) > 1 else ""
                    if val.startswith("10."):
                        payload.add_answer(
                            verb = self.properties["DOI"],
                            object_and_type = [val, "external-id"],
                        )
                    elif val.startswith(("http://", "https://")):
                        payload.add_answer(
                            verb = self.properties["URL"],
                            object_and_type = [val, "url"],
                        )

            if entry.get("ToArchive"):
                if entry["ToArchive"][0] == options["YesText"]:
                    qualifier = []
                    if entry["ToArchive"][1]:
                        qualifier = payload.add_qualifier(
                            self.properties["end time"],
                            "time",
                            {
                                "time": f"+{entry['ToArchive'][1]}-00-00T00:00:00Z",
                                "precision": 9,
                                "calendarmodel": "http://www.wikidata.org/entity/Q1985727",
                            },
                        )
                    payload.add_answer(
                        verb = self.properties["mandates"],
                        object_and_type = [self.items["research data archiving"], "wikibase-item"],
                        qualifier = qualifier,
                    )

    def _export_cpus(self, payload, cpus: dict):
        '''Export each CPU item collected from hardware sections.'''
        for entry in cpus.values():
            if not entry.get("ID"):
                continue

            payload.get_item_key(
                value = entry
            )
            payload.set_class('CPU Model')

            payload.add_answer(
                verb = self.properties["instance of"],
                object_and_type = [self.items["CPU model"], "wikibase-item"],
            )

    def _export_programming_languages(self, payload, programminglanguages: dict):
        '''Export each programming language item collected from software sections.'''
        for entry in programminglanguages.values():
            if not entry.get("ID"):
                continue

            payload.get_item_key(
                value = entry
            )
            payload.set_class('Programming Language')

            payload.add_answer(
                verb = self.properties["instance of"],
                object_and_type = [self.items["programming language"], "wikibase-item"],
            )

    def _export_methods(self, payload, methods: dict):
        '''Export each experimental method item collected from process-step sections.'''
        for entry in methods.values():
            if not entry.get("ID"):
                continue

            payload.get_item_key(
                value = entry
            )
            payload.set_class('Experimental Method')

            payload.add_answer(
                verb = self.properties["instance of"],
                object_and_type = [self.items["method"], "wikibase-item"],
            )

    def _export_instruments(self, payload, instruments: dict):
        '''Export each instrument item collected from process-step sections.'''
        for entry in instruments.values():
            if not entry.get("ID"):
                continue

            payload.get_item_key(
                value = entry
            )
            payload.set_class('Experimental Equipment')

            payload.add_answer(
                verb = self.properties["instance of"],
                object_and_type = [self.items["research tool"], "wikibase-item"],
            )

    def _export_algorithmic_tasks(self, payload, algorithmic_tasks: dict):
        '''Export each algorithmic task item collected from algorithm sections.'''
        for entry in algorithmic_tasks.values():
            if not entry.get("ID"):
                continue

            payload.get_item_key(
                value = entry
            )
            payload.set_class('Algorithmic Task')

            self._add_common_metadata(
                payload = payload,
                community = self.items["MathAlgoDB"],
                qclass =self.items["algorithmic task"],
                profile_type = "MaRDI task profile",
            )

    def _export_academic_disciplines(self, payload, academic_disciplines: dict):
        '''Export each academic discipline item collected from process-step sections.'''
        for entry in academic_disciplines.values():
            if not entry.get("ID"):
                continue

            payload.get_item_key(
                value = entry
            )
            payload.set_class('Academic Discipline')

            payload.add_answer(
                verb = self.properties["instance of"],
                object_and_type = [self.items["academic discipline"], "wikibase-item"],
            )

__init__()

Initialise with Wikibase vocabulary and ontology registries.

Source code in MaRDMO/workflow/worker.py

def __init__(self):
    '''Initialise with Wikibase vocabulary and ontology registries.'''
    super().__init__()
    self.mathmoddb           = get_mathmoddb()
    self.mathalgodb          = get_mathalgodb()
    self.publication_mapping = get_publication_mapping()

export(data, url)

Assemble and return the complete Wikibase payload for a Workflow documentation export.

Parameters:

Name	Type	Description	Default
data		Top-level workflow answers dict produced by get_post_data.	required
url		Target Wikibase API URL for the upload.	required

Returns:

Type	Description
	Tuple (payload_dict, dependency_order) ready for
	meth:~MaRDMO.oauth2.OauthProviderMixin.post.

Source code in MaRDMO/workflow/worker.py

def export(self, data, url):
    '''Assemble and return the complete Wikibase payload for a Workflow documentation export.

    Args:
        data: Top-level workflow answers dict produced by ``get_post_data``.
        url:  Target Wikibase API URL for the upload.

    Returns:
        Tuple ``(payload_dict, dependency_order)`` ready for
        :meth:`~MaRDMO.oauth2.OauthProviderMixin.post`.
    '''

    items, dependency = unique_items(data)

    payload = GeneratePayload(
        dependency = dependency,
        user_items = items,
        url = url,
        wikibase = {
            'items':      get_items(),
            'properties': get_properties(),
            'relations':  get_relations(),
        }
    )

    payload.process_items()

    self._export_workflows(
        payload = payload,
        data    = data,
    )
    self._export_processsteps(
        payload      = payload,
        processsteps = data.get('processstep', {}),
    )
    self._export_algorithms(
        payload    = payload,
        algorithms = data.get('algorithm', {}),
    )
    algo_software_keys = {
        (item.get('ID', ''), item.get('Name', ''), item.get('Description', ''))
        for algo in data.get('algorithm', {}).values()
        for item in algo.get('SRelatant', {}).values()
    }
    self._export_softwares(
        payload           = payload,
        softwares         = data.get('software', {}),
        algo_software_keys = algo_software_keys,
    )
    self._export_hardwares(
        payload   = payload,
        hardwares = data.get('hardware', {}),
    )
    self._export_datasets(
        payload  = payload,
        datasets = data.get('dataset', {}),
    )
    self._export_cpus(
        payload = payload,
        cpus    = data.get('cpu', {}),
    )
    self._export_programming_languages(
        payload              = payload,
        programminglanguages = data.get('programminglanguage', {}),
    )
    self._export_methods(
        payload = payload,
        methods = data.get('method', {}),
    )
    self._export_instruments(
        payload     = payload,
        instruments = data.get('instrument', {}),
    )
    self._export_algorithmic_tasks(
        payload           = payload,
        algorithmic_tasks = data.get('algorithmictask', {}),
    )
    self._export_academic_disciplines(
        payload              = payload,
        academic_disciplines = data.get('academicdiscipline', {}),
    )
    self._export_authors(
        payload      = payload,
        publications = data.get('publication', {}),
    )
    self._export_journals(
        payload      = payload,
        publications = data.get('publication', {}),
    )
    self._export_publications(
        payload      = payload,
        publications = data.get('publication', {}),
        relations    = [
            ('P2A', 'ARelatant'), ('P2BS', 'HSRelatant'), ('P2E', 'EntityRelatant'),
        ],
    )

    payload.add_item_payload()

    if any(key.startswith('RELATION') for key in payload.get_dictionary()):
        query = payload.build_relation_check_query()

        check = None
        for attempt in range(2):
            try:
                check = query_sparql(
                    query           = query,
                    sparql_endpoint = get_url('mardi', 'sparql'),
                )
                break
            except Exception as e:
                logging.warning("SPARQL query attempt %s failed: %s", attempt + 1, e)
                if attempt == 0:
                    time.sleep(1)
        if not check:
            check = [{}]

        payload.add_check_results(check=check)

    return payload.get_dictionary(), payload.dependency

preview(answers)

Enrich the answers dict with derived display structures for the preview template.

First iterates over :data:~MaRDMO.workflow.constants.preview_relations and calls :func:~MaRDMO.helpers.entity_relations for each entry to resolve cross-entity relation labels (e.g. Workflow → Process Step) into the answers dict — the same pattern used by the model and algorithm preview workers.

Then builds model_task_pairs — an ordered list of dicts with keys model, tasks, has_model, and has_tasks — for every workflow in the project and stores it directly on each workflow's sub-dict (answers['workflow'][i]['model_task_pairs']) so the template can access it as values.model_task_pairs inside the {% for values in answers.workflow.values %} loop without integer-key lookups. Indices from both models and tasks are unioned so that a task with no matching model (or vice versa) still produces a row with the appropriate validation flags set.

Similarly builds algorithm_software_pairs and method_instrument_pairs — ordered lists of dicts with generic keys primary, qualifier, has_primary, and has_qualifier — for every process step. Index i pairs algorithm i with software i (and method i with instrument i); a missing partner raises a validation flag rendered as a red warning via the _relation_block_single_with_qualifier.html include.

Parameters:

Name	Type	Description	Default
answers		Top-level answers dict produced by get_post_data.	required

Returns:

Type	Description
	The same answers dict with relation labels and model_task_pairs added.

Source code in MaRDMO/workflow/worker.py

def preview(self, answers):
    '''Enrich the answers dict with derived display structures for the preview template.

    First iterates over :data:`~MaRDMO.workflow.constants.preview_relations` and calls
    :func:`~MaRDMO.helpers.entity_relations` for each entry to resolve cross-entity
    relation labels (e.g. Workflow → Process Step) into the ``answers`` dict — the same
    pattern used by the model and algorithm preview workers.

    Then builds ``model_task_pairs`` — an ordered list of dicts with keys ``model``,
    ``tasks``, ``has_model``, and ``has_tasks`` — for every workflow in the project
    and stores it directly on each workflow's sub-dict
    (``answers['workflow'][i]['model_task_pairs']``) so the template can access it as
    ``values.model_task_pairs`` inside the ``{% for values in answers.workflow.values %}``
    loop without integer-key lookups.  Indices from both models *and* tasks are unioned
    so that a task with no matching model (or vice versa) still produces a row with the
    appropriate validation flags set.

    Similarly builds ``algorithm_software_pairs`` and ``method_instrument_pairs`` —
    ordered lists of dicts with generic keys ``primary``, ``qualifier``,
    ``has_primary``, and ``has_qualifier`` — for every process step.  Index *i*
    pairs algorithm *i* with software *i* (and method *i* with instrument *i*);
    a missing partner raises a validation flag rendered as a red warning via the
    ``_relation_block_single_with_qualifier.html`` include.

    Args:
        answers: Top-level answers dict produced by ``get_post_data``.

    Returns:
        The same *answers* dict with relation labels and ``model_task_pairs`` added.
    '''

    # Prepare Relations for Preview
    for relation in preview_relations:
        fn = entity_relations_grouped if relation.get('grouped') else entity_relations
        fn(
            data = answers,
            idx = {
                'from': relation['from_idx'],
                'to': relation['to_idx']
            },
            entity = {
                'relation': relation['relation'],
                'old_name': relation['old_name'],
                'new_name': relation['new_name'],
                'encryption': relation['encryption']
            },
            order = {
                'formulation': False,
                'task': False
            },
            assumption = False,
            mapping = (
                self.mathmoddb           if relation.get('mapping') == 'mathmoddb'
                else self.publication_mapping if relation.get('mapping') == 'publication'
                else self.mathalgodb
            )
        )

    for wf_data in answers.get('workflow', {}).values():
        models = wf_data.get('model', {})
        tasks  = wf_data.get('task', {})
        all_indices = sorted(set(models) | set(tasks))
        wf_data['model_task_pairs'] = [
            {
                'model':     models.get(idx),
                'tasks':     list(tasks.get(idx, {}).values()),
                'has_model': bool(models.get(idx)),
                'has_tasks': bool(tasks.get(idx)),
            }
            for idx in all_indices
        ]

    for ps_data in answers.get('processstep', {}).values():
        # RelationA keys are 'set_index|collection_index' strings;
        # group all algorithms sharing the same set_index into one list.
        algo_by_set = {}
        for key, val in ps_data.get('RelationA', {}).items():
            set_idx = int(str(key).split('|')[0])
            algo_by_set.setdefault(set_idx, []).append(val)
        software = ps_data.get('RelationS', {})
        hardware = ps_data.get('RelationH', {})
        software_doc = ps_data.get('software-documentation', {})
        algo_params = ps_data.get('algorithm-parameter', {})
        all_indices = sorted(
            set(algo_by_set) | set(software) | set(hardware) | set(software_doc)
        )
        ps_data['algorithm_software_pairs'] = [
            {
                'primary':       algo_by_set.get(idx, []),
                'qualifier':     software.get(idx),
                'has_primary':   bool(algo_by_set.get(idx)),
                'has_qualifier': bool(software.get(idx)),
                'parameters':    ', '.join(
                    v[1] for v in sorted(algo_params.get(idx, {}).items())
                    if v[1]
                ),
                'software_doc':  software_doc.get(idx, {}),
                'hardware':      hardware.get(idx),
            }
            for idx in all_indices
        ]

        # Use raw MRelatant/IRelatant dicts so the template can display
        # Name (Description) for inline items that have no dedicated section.
        meth_by_set = {
            set_idx: [raw]
            for set_idx, raw in ps_data.get('MRelatant', {}).items()
        }
        instruments = ps_data.get('IRelatant', {})
        method_doc  = ps_data.get('method-documentation', {})
        meth_params = ps_data.get('method-parameter', {})
        all_indices = sorted(set(meth_by_set) | set(instruments) | set(method_doc))
        ps_data['method_instrument_pairs'] = [
            {
                'primary':       meth_by_set.get(idx, []),
                'qualifier':     instruments.get(idx),
                'has_primary':   bool(meth_by_set.get(idx)),
                'has_qualifier': bool(instruments.get(idx)),
                'parameters':    ', '.join(
                    v[1] for v in sorted(meth_params.get(idx, {}).items())
                    if v[1]
                ),
                'method_doc':    method_doc.get(idx, {}),
            }
            for idx in all_indices
        ]

    for hw_data in answers.get('hardware', {}).values():
        cpu_dict   = hw_data.get('cpu', {})
        count_dict = hw_data.get('number-of-cpu', {})
        cores_dict = hw_data.get('cores', {})
        all_indices = sorted(set(cpu_dict) | set(count_dict) | set(cores_dict))
        hw_data['cpu_entries'] = [
            {
                'id':          cpu_dict.get(idx, {}).get('ID'),
                'name':        cpu_dict.get(idx, {}).get('Name'),
                'description': cpu_dict.get(idx, {}).get('Description'),
                'count':       count_dict.get(idx),
                'count_valid': (
                    str(count_dict.get(idx, '')).strip().isdigit()
                    if count_dict.get(idx) else False
                ),
                'cores':       cores_dict.get(idx),
                'cores_valid': (
                    str(cores_dict.get(idx, '')).strip().isdigit()
                    if cores_dict.get(idx) else False
                ),
            }
            for idx in all_indices
        ]
        nodes = hw_data.get('nodes')
        hw_data['nodes_valid'] = str(nodes).strip().isdigit() if nodes else False

    answers['cpu'] = collect_items_without_section(answers, 'hardware', 'cpu')
    answers['programminglanguage'] = collect_items_without_section(
        answers, 'software', 'programminglanguage'
    )
    answers['algorithmictask'] = collect_items_without_section(
        answers, 'algorithm', 'PRelatant'
    )
    answers['academicdiscipline'] = collect_items_without_section(
        answers, 'processstep', 'RFRelatant'
    )
    answers['method'] = collect_items_without_section(
        answers, 'processstep', 'MRelatant', nested=True
    )
    answers['instrument'] = collect_items_without_section(
        answers, 'processstep', 'IRelatant'
    )

    options = get_options()

    for ds_data in answers.get('dataset', {}).values():
        size = ds_data.get('Size')
        if size and size[0] and size[1]:
            val = str(size[1]).strip()
            if size[0] == options['items']:
                ds_data['size_value_valid'] = val.isdigit()
            else:
                try:
                    float(val)
                    ds_data['size_value_valid'] = True
                except (ValueError, TypeError):
                    ds_data['size_value_valid'] = False
        archive = ds_data.get('ToArchive')
        if archive and archive[0] and archive[1]:
            val = str(archive[1]).strip()
            ds_data['archive_year_valid'] = len(val) == 4 and val.isdigit()
        publish = ds_data.get('ToPublish')
        if publish and len(publish) > 1 and publish[1]:
            val = str(publish[1]).strip()
            if not val.startswith('10.'):
                ds_data['publish_url_valid'] = is_valid_url(val)

    for sw_data in answers.get('software', {}).values():
        ref = sw_data.get('reference', {})
        if ref.get(2) and ref[2][1]:
            sw_data['ref_desc_url_valid'] = is_valid_url(ref[2][1])
        if ref.get(3) and ref[3][1]:
            sw_data['ref_repo_url_valid'] = is_valid_url(ref[3][1])

    for ps_data in answers.get('processstep', {}).values():
        for pair in ps_data.get('algorithm_software_pairs', []):
            pair['software_doc_url_invalid'] = {
                idx
                for idx, entry in pair.get('software_doc', {}).items()
                if entry and entry[0] == options['URL'] and entry[1]
                and not is_valid_url(entry[1])
            }
        for pair in ps_data.get('method_instrument_pairs', []):
            pair['method_doc_url_invalid'] = {
                idx
                for idx, entry in pair.get('method_doc', {}).items()
                if entry and entry[0] == options['URL'] and entry[1]
                and not is_valid_url(entry[1])
            }

    return answers

Constants

Compile-time constants and configuration builders for the Workflow catalog.

Centralises the property URIs, question/item mappings, and publication-order definitions used by the workflow documentation sub-package at runtime. Values are loaded once from the RDMO database via the getters helpers and then referenced by handlers, providers, and workers.

Provides:

• get_uri_prefix_map() — returns the {prefix: URI} map used to expand compact IDs
• order_to_publish() — returns the ordered list of workflow attributes for portal export
• Module-level constants built from the above

get_relations()

Return the workflow relation mapping.

Extends the shared publication-role mapping with the workflow intra-class relations (contains / contained in) from MathModDB.

Source code in MaRDMO/workflow/constants.py

def get_relations():
    '''Return the workflow relation mapping.

    Extends the shared publication-role mapping with the workflow intra-class
    relations (contains / contained in) from MathModDB.
    '''
    mathmoddb  = get_mathmoddb()
    properties = get_properties()
    return {
        **get_publication_relations(),
        mathmoddb.get(key='contains_workflow')['url']: [
            properties['contains'],
            'forward'
        ],
        mathmoddb.get(key='contained_in_workflow')['url']: [
            properties['contains'],
            'backward'
        ],
    }

get_uri_prefix_map()

Build the attribute-URI → section config mapping for the Workflow Documentation.

Maps each relation attribute URI to the corresponding questionnaire section metadata needed to add or hydrate the related entity.

Returns:

Type	Description
	Dict mapping full RDMO attribute URI strings to dicts with keys
	question_set, question_id, and prefix.

Source code in MaRDMO/workflow/constants.py

def get_uri_prefix_map():
    '''Build the attribute-URI → section config mapping for the Workflow Documentation.

    Maps each relation attribute URI to the corresponding questionnaire section
    metadata needed to add or hydrate the related entity.

    Returns:
        Dict mapping full RDMO attribute URI strings to dicts with keys
        ``question_set``, ``question_id``, and ``prefix``.
    '''
    questions = get_questions('workflow')
    uri_prefix_map = {
        f'{BASE_URI}{questions["Process Step"]["Algorithm"]["uri"]}': {
            "question_set": f'{BASE_URI}{questions["Algorithm"]["uri"]}',
            "question_id": f'{BASE_URI}{questions["Algorithm"]["ID"]["uri"]}',
            "prefix": "A"
        },
        f'{BASE_URI}{questions["Process Step"]["Hardware"]["uri"]}': {
            "question_set": f'{BASE_URI}{questions["Hardware"]["uri"]}',
            "question_id": f'{BASE_URI}{questions["Hardware"]["ID"]["uri"]}',
            "prefix": "HW"
        },
        f'{BASE_URI}{questions["Process Step"]["Input"]["uri"]}': {
            "question_set": f'{BASE_URI}{questions["Data Set"]["uri"]}',
            "question_id": f'{BASE_URI}{questions["Data Set"]["ID"]["uri"]}',
            "prefix": "DS"
        },
        f'{BASE_URI}{questions["Process Step"]["Output"]["uri"]}': {
            "question_set": f'{BASE_URI}{questions["Data Set"]["uri"]}',
            "question_id": f'{BASE_URI}{questions["Data Set"]["ID"]["uri"]}',
            "prefix": "DS"
        },
        f'{BASE_URI}{questions["Workflow"]["PSRelatant"]["uri"]}': {
            "question_set": f'{BASE_URI}{questions["Process Step"]["uri"]}',
            "question_id": f'{BASE_URI}{questions["Process Step"]["ID"]["uri"]}',
            "prefix": "PS"
        },
        f'{BASE_URI}{questions["Process Step"]["Software"]["uri"]}': {
            "question_set": f'{BASE_URI}{questions["Software"]["uri"]}',
            "question_id": f'{BASE_URI}{questions["Software"]["ID"]["uri"]}',
            "prefix": "S"
        },
        f'{BASE_URI}{questions["Algorithm"]["SRelatant"]["uri"]}': {
            "question_set": f'{BASE_URI}{questions["Software"]["uri"]}',
            "question_id": f'{BASE_URI}{questions["Software"]["ID"]["uri"]}',
            "prefix": "S"
        },
        f'{BASE_URI}{questions["Software"]["Dependency"]["uri"]}': {
            "question_set": f'{BASE_URI}{questions["Software"]["uri"]}',
            "question_id": f'{BASE_URI}{questions["Software"]["ID"]["uri"]}',
            "prefix": "S"
        },
    }
    return uri_prefix_map

order_to_publish()

Build an ordered mapping of publishing option keys to (rank, option_value) tuples.

Returns:

Type	Description
	Dict {"Yes": (0, …), "doi": (1, …), "url": (2, …), "No": (3, …)}
	where values are resolved RDMO option strings.

Source code in MaRDMO/workflow/constants.py

def order_to_publish():
    '''Build an ordered mapping of publishing option keys to ``(rank, option_value)`` tuples.

    Returns:
        Dict ``{"Yes": (0, …), "doi": (1, …), "url": (2, …), "No": (3, …)}``
        where values are resolved RDMO option strings.
    '''
    options = get_options()
    order = {
        'Yes': (0, options['Yes']),
        'doi': (1, options['DOI']),
        'url': (2, options['URL']),
        'No': (3, options['No'])
        }
    return order

Utils

Utility functions for extracting and normalising workflow-related data.

Provides helpers that read structured answers from the RDMO questionnaire and derive derived values (data-set sizes, references, archive URLs) needed by the workflow worker and handler layers.

Provides:

• get_size — extract the data-set size value from answer options
• get_option_text_pair — resolve an option label and paired text value from a SPARQL result

get_option_text_pair(data, options, opt_key, *value_keys)

Resolve an option label and the first available text value from a SPARQL result dict.

Parameters:

Name	Type	Description	Default
data		SPARQL result row dict with nested {"value": ...} entries.	required
options		Options dict mapping RDMO option URIs to string values.	required
opt_key		Key whose resolved option URI is looked up in options.	required
*value_keys		One or more keys tried left-to-right for the text value; the first non-empty one is used.	()

Returns:

Type	Description
	[option_label, text] if the option is set; [] otherwise.

Source code in MaRDMO/workflow/utils.py

def get_option_text_pair(data, options, opt_key, *value_keys):
    '''Resolve an option label and the first available text value from a SPARQL result dict.

    Args:
        data:       SPARQL result row dict with nested ``{"value": ...}`` entries.
        options:    Options dict mapping RDMO option URIs to string values.
        opt_key: Key whose resolved option URI is looked up in *options*.
        *value_keys: One or more keys tried left-to-right for the text value;
                    the first non-empty one is used.

    Returns:
        ``[option_label, text]`` if the option is set; ``[]`` otherwise.
    '''
    option_label = options[data[opt_key]['value']] if data.get(opt_key, {}).get('value') else ''
    text = next(
        (data.get(k, {}).get('value', '') for k in value_keys if data.get(k, {}).get('value')),
        ''
    )
    return [option_label, text] if option_label else []

get_size(data, options)

Extract the size unit and value from a data-set answer dict.

Resolves the unit from either size_unit (RDMO option URI) or falls back to items when a record count is given.

Parameters:

Name	Type	Description	Default
data		Answer sub-dict for the data-set size questions.	required
options		Options dict mapping RDMO option URIs to string values.	required

Returns:

Type	Description
	[unit, value] if both are present; [] otherwise.

Source code in MaRDMO/workflow/utils.py

def get_size(data, options):
    '''Extract the size unit and value from a data-set answer dict.

    Resolves the unit from either ``size_unit`` (RDMO option URI) or falls
    back to ``items`` when a record count is given.

    Args:
        data:    Answer sub-dict for the data-set size questions.
        options: Options dict mapping RDMO option URIs to string values.

    Returns:
        ``[unit, value]`` if both are present; ``[]`` otherwise.
    '''
    size_unit = data.get('size_unit', {}).get('value', '')
    size_value = data.get('size_value', {}).get('value', '')
    size_record = data.get('size_record', {}).get('value', '')

    unit = options.get(size_unit) if size_unit else (options['items'] if size_record else '')
    value = size_value or size_record

    return [unit, value] if unit and value else []