Skip to content

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 
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
class Information(BaseInformation):
    '''Handlers for the Workflow Documentation questionnaire.'''

    _ENTITY_KEYS = ('Software', 'Hardware', 'Instrument', 'Data Set', 'Method', 'Process Step')

    def __init__(self):
        '''Load workflow questions, base URI, and RDMO options.'''
        self.questions = get_questions('workflow')
        self.base      = BASE_URI
        self.options   = get_options()

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

    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 instrument(self, instance):
        '''Handle Instrument ID save: hydrate basics only (no SPARQL data available).

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

    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 method(self, instance):
        '''Handle Method ID save: hydrate basics and cascade to Software/Instrument.

        Args:
            instance: RDMO :class:`~rdmo.projects.models.Value` that was just saved.
        '''
        self._entry(instance, 'Method', self._fill_method_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)

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

    def _fill_software_batch(self, project, items, catalog='', visited=None):
        '''Hydrate multiple Software 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()

        software   = self.questions['Software']
        data_by_id = _fetch_by_source(
            items,
            'workflow/queries/software_mardi.sparql',
            'workflow/queries/software_wikidata.sparql',
            Software,
        )

        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='Software', index=(0, set_index))

            add_references(project=project, data=data,
                           uri=f'{self.base}{software["Reference"]["uri"]}',
                           set_prefix=set_index)

            add_relations_static(
                project=project, data=data,
                props={'keys': PROPS['S2PL']},
                index={'set_prefix': set_index},
                statement={
                    'relatant': f'{self.base}{software["Programming Language"]["uri"]}'
                })

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

    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))

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

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

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

    def _fill_instrument_batch(self, project, items, catalog='', visited=None):
        '''Hydrate multiple Instrument pages with basic label/description only.

        Instruments have no external SPARQL data; only :func:`~.adders.add_basics`
        is called for each item.

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

        for text, _, set_index in items:
            add_basics(project=project, text=text, questions=self.questions,
                       item_type='Instrument', index=(0, set_index))

    def _fill_data_set_batch(self, project, items, catalog='', visited=None):
        '''Hydrate multiple Data Set 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()

        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))
            self._write_data_set_fields(project, data_set_q, data, set_index)

    def _write_data_set_fields(self, project, data_set_q, data, set_index):
        '''Write size, publication, archival, and reference fields for one Data Set page.

        Args:
            project:    RDMO project instance.
            data_set_q: Questions sub-dict for the Data Set section.
            data:       Parsed data-set dataclass instance.
            set_index:  Set-index of the data set page to write into.
        '''
        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})

        add_relations_static(
            project=project, data=data,
            props={'keys': PROPS['DS2DT']},
            index={'set_prefix': set_index},
            statement={
                'relatant': f'{self.base}{data_set_q["Data Type"]["uri"]}'
            })

        add_relations_static(
            project=project, data=data,
            props={'keys': PROPS['DS2RF']},
            index={'set_prefix': set_index},
            statement={
                'relatant': f'{self.base}{data_set_q["Representation Format"]["uri"]}'
            })

        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})

        add_references(project=project, data=data,
                       uri=f'{self.base}{data_set_q["To Publish"]["uri"]}',
                       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], 'option': data.to_archive[0],
                      'set_prefix': set_index})

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

        Explicitly cascades into Software and Instrument via
        :meth:`_hydrate_relatants` rather than 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()

        method     = self.questions['Method']
        data_by_id = _fetch_by_source(
            items,
            'workflow/queries/method_mardi.sparql',
            'workflow/queries/method_wikidata.sparql',
            Method,
        )

        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='Method', index=(0, set_index))

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

            self._hydrate_relatants(
                project=project, data=data, prop_keys=PROPS['M2S'],
                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,
                ))

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

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

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

        Explicitly cascades into Data Set, Method, Software, and Instrument 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))

            self._fill_process_step_relations(
                project, process_step, data, set_index, catalog, visited,
                section_indices,
            )

    def _fill_process_step_relations(
        self, project, process_step, data, set_index, catalog, visited,
        section_indices=None,
    ):
        '''Write all relation fields and cascade hydration for one Process Step page.

        Adds Input/Output Data Set, Method, Software, and Instrument relations,
        then cascades into related entities via :meth:`_hydrate_relatants`.

        Args:
            project:         RDMO project instance.
            process_step:    Questions sub-dict for the Process Step section.
            data:            Parsed process-step dataclass instance.
            set_index:       Set-index of the process step page to write into.
            catalog:         Active catalog URI suffix.
            visited:         Set of external IDs already processed.
            section_indices: Dict mapping section names to next available set
                             indices (mutated in place for cascade writes).
        '''
        # Input Data Sets
        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,
            ))

        # Output Data Sets
        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,
            ))

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

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

        # Platform Software
        add_relations_static(
            project=project, data=data,
            props={'keys': PROPS['PS2PLS']},
            index={'set_prefix': set_index},
            statement={
                'relatant':
                    f'{self.base}{process_step["Environment-Software"]["uri"]}'
            })

        self._hydrate_relatants(
            project=project, data=data, prop_keys=PROPS['PS2PLS'],
            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,
            ))

        # Platform Instrument
        add_relations_static(
            project=project, data=data,
            props={'keys': PROPS['PS2PLI']},
            index={'set_prefix': set_index},
            statement={
                'relatant':
                    f'{self.base}{process_step["Environment-Instrument"]["uri"]}'
            })

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

        # Fields of Work (static, no cascade)
        add_relations_static(
            project=project, data=data,
            props={'keys': PROPS['PS2F']},
            index={'set_prefix': set_index},
            statement={'relatant': f'{self.base}{process_step["Discipline"]["uri"]}'})

        # MSC IDs (static, no cascade)
        add_relations_static(
            project=project, data=data,
            props={'keys': PROPS['PS2MA']},
            index={'set_prefix': set_index},
            statement={'relatant': f'{self.base}{process_step["Discipline"]["uri"]}'})

__init__()

Load workflow questions, base URI, and RDMO options.

Source code in MaRDMO/workflow/handlers.py 
27
28
29
30
31
def __init__(self):
    '''Load workflow questions, base URI, and RDMO options.'''
    self.questions = get_questions('workflow')
    self.base      = BASE_URI
    self.options   = get_options()

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 
61
62
63
64
65
66
67
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 
45
46
47
48
49
50
51
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)

instrument(instance)

Handle Instrument ID save: hydrate basics only (no SPARQL data available).

Parameters:

Name Type Description Default
instance

RDMO :class:~rdmo.projects.models.Value that was just saved.

 required
Source code in MaRDMO/workflow/handlers.py 
53
54
55
56
57
58
59
def instrument(self, instance):
    '''Handle Instrument ID save: hydrate basics only (no SPARQL data available).

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

method(instance)

Handle Method ID save: hydrate basics and cascade to Software/Instrument.

Parameters:

Name Type Description Default
instance

RDMO :class:~rdmo.projects.models.Value that was just saved.

 required
Source code in MaRDMO/workflow/handlers.py 
69
70
71
72
73
74
75
def method(self, instance):
    '''Handle Method ID save: hydrate basics and cascade to Software/Instrument.

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

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 
77
78
79
80
81
82
83
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)

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 
37
38
39
40
41
42
43
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)

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:`ModelProperties — properties of model associated with a workflow
  • :class:`Variables — variables of model associated with a workflow
  • :class:`Parameters — parameters of model associated with a workflow
  • :class:ProcessStep— process steps associated with a workflow
  • :class: Method — methods 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

DataSet dataclass

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

Source code in MaRDMO/workflow/models.py 
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
@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
    reference: list = field(default_factory=list)
    to_archive: list = field(default_factory=list)
    data_type: list[Relatant] = field(default_factory=list)
    representation_format: 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 ''
            ),
            reference=get_reference(data, options),
            to_archive=get_archive(data, options),
            data_type=split_value(
                data=data, key='data_type', transform=Relatant.from_query
            ),
            representation_format=split_value(
                data=data, key='representation_format', transform=Relatant.from_query
            ),
        )

from_query(raw_data) classmethod

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

Source code in MaRDMO/workflow/models.py 
323
324
325
326
@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 
328
329
330
331
332
333
334
335
@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 
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
@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 ''
        ),
        reference=get_reference(data, options),
        to_archive=get_archive(data, options),
        data_type=split_value(
            data=data, key='data_type', transform=Relatant.from_query
        ),
        representation_format=split_value(
            data=data, key='representation_format', transform=Relatant.from_query
        ),
    )

Hardware dataclass

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

Source code in MaRDMO/workflow/models.py 
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
@dataclass
class Hardware:
    '''Node/core counts and CPU relations of a Hardware entity.'''

    nodes: Optional[str] = None
    cores: Optional[str] = None
    cpu: 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('nodes', {}).get('value', ''),
            cores=data.get('cores', {}).get('value', ''),
            cpu=split_value(data=data, key='cpu', transform=Relatant.from_query),
        )

from_query(raw_data) classmethod

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

Source code in MaRDMO/workflow/models.py 
286
287
288
289
@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 
291
292
293
294
295
296
297
298
@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 
300
301
302
303
304
305
306
307
@classmethod
def from_query_single(cls, data: dict) -> 'Hardware':
    '''Parse one SPARQL result row into a Hardware instance.'''
    return cls(
        nodes=data.get('nodes', {}).get('value', ''),
        cores=data.get('cores', {}).get('value', ''),
        cpu=split_value(data=data, key='cpu', transform=Relatant.from_query),
    )

Method dataclass

Relations of a Method entity from MaRDI/Wikidata.

Source code in MaRDMO/workflow/models.py 
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
@dataclass
class Method:
    '''Relations of a Method entity from MaRDI/Wikidata.'''

    implemented_by_software: list[Relatant] = field(default_factory=list)
    implemented_by_instrument: list[Relatant] = field(default_factory=list)

    @classmethod
    def from_query(cls, raw_data: list) -> 'Method':
        '''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, Method]':
        '''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) -> 'Method':
        '''Parse one SPARQL result row into a Method instance.'''
        return cls(
            implemented_by_software=split_value(
                data=data, key='implemented_by_software', transform=Relatant.from_query
            ),
            implemented_by_instrument=split_value(
                data=data, key='implemented_by_instrument', transform=Relatant.from_query
            ),
        )

from_query(raw_data) classmethod

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

Source code in MaRDMO/workflow/models.py 
209
210
211
212
@classmethod
def from_query(cls, raw_data: list) -> 'Method':
    '''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 
214
215
216
217
218
219
220
221
@classmethod
def from_query_batch(cls, raw_data: list) -> 'dict[str, Method]':
    '''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 Method instance.

Source code in MaRDMO/workflow/models.py 
223
224
225
226
227
228
229
230
231
232
233
@classmethod
def from_query_single(cls, data: dict) -> 'Method':
    '''Parse one SPARQL result row into a Method instance.'''
    return cls(
        implemented_by_software=split_value(
            data=data, key='implemented_by_software', transform=Relatant.from_query
        ),
        implemented_by_instrument=split_value(
            data=data, key='implemented_by_instrument', transform=Relatant.from_query
        ),
    )

ModelProperties dataclass

Properties (time/space discretisation) of a mathematical model.

Source code in MaRDMO/workflow/models.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
@dataclass
class ModelProperties:
    '''Properties (time/space discretisation) of a mathematical model.'''

    time: Optional[str] = None
    space: Optional[str] = None

    @classmethod
    def from_query(cls, raw_data: list) -> 'ModelProperties':
        '''Parse a single SPARQL result row into a ModelProperties instance.'''
        data = raw_data[0]

        if data.get('isTimeContinuous', {}).get('value') == 'True':
            time = 'continuous'
        elif data.get('isTimeDiscrete', {}).get('value') == 'True':
            time = 'discrete'
        else:
            time = ''

        if data.get('isSpaceContinuous', {}).get('value') == 'True':
            space = 'continuous'
        elif data.get('isSpaceDiscrete', {}).get('value') == 'True':
            space = 'discrete'
        else:
            space = ''

        return cls(
            time = time,
            space = space,
        )

from_query(raw_data) classmethod

Parse a single SPARQL result row into a ModelProperties instance.

Source code in MaRDMO/workflow/models.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
@classmethod
def from_query(cls, raw_data: list) -> 'ModelProperties':
    '''Parse a single SPARQL result row into a ModelProperties instance.'''
    data = raw_data[0]

    if data.get('isTimeContinuous', {}).get('value') == 'True':
        time = 'continuous'
    elif data.get('isTimeDiscrete', {}).get('value') == 'True':
        time = 'discrete'
    else:
        time = ''

    if data.get('isSpaceContinuous', {}).get('value') == 'True':
        space = 'continuous'
    elif data.get('isSpaceDiscrete', {}).get('value') == 'True':
        space = 'discrete'
    else:
        space = ''

    return cls(
        time = time,
        space = space,
    )

Parameters dataclass

A single parameter entry from MathModDB.

Source code in MaRDMO/workflow/models.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
@dataclass
class Parameters:
    '''A single parameter entry from MathModDB.'''

    identifier: Optional[str] = None
    name: Optional[str] = None
    unit: Optional[str] = None
    symbol: Optional[str] = None
    task: Optional[str] = None

    @classmethod
    def from_query(cls, data: dict) -> 'Parameters':
        '''Parse one SPARQL result row into a Parameters instance.'''

        parameters = {
            # Identifier of Parameter
            'identifier': data.get('ID', {}).get('value'),
            # Name of Parameter
            'name': data.get('Name', {}).get('value'),
            # Unit of Parameter
            'unit': data.get('Unit', {}).get('value'),
            # Symbol of Parameter
            'symbol': re.sub(
                r'(<math\b(?![^>]*\bdisplay=)[^>]*)(>)',
                r'\1 display="inline"\2',
                data.get('Symbol', {}).get('value'),
            ),
            # Task Parameter is Part Of
            'task': data.get('label', {}).get('value'),
        }

        return cls(
            **parameters
        )

from_query(data) classmethod

Parse one SPARQL result row into a Parameters instance.

Source code in MaRDMO/workflow/models.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
@classmethod
def from_query(cls, data: dict) -> 'Parameters':
    '''Parse one SPARQL result row into a Parameters instance.'''

    parameters = {
        # Identifier of Parameter
        'identifier': data.get('ID', {}).get('value'),
        # Name of Parameter
        'name': data.get('Name', {}).get('value'),
        # Unit of Parameter
        'unit': data.get('Unit', {}).get('value'),
        # Symbol of Parameter
        'symbol': re.sub(
            r'(<math\b(?![^>]*\bdisplay=)[^>]*)(>)',
            r'\1 display="inline"\2',
            data.get('Symbol', {}).get('value'),
        ),
        # Task Parameter is Part Of
        'task': data.get('label', {}).get('value'),
    }

    return cls(
        **parameters
    )

ProcessStep dataclass

Relations of a Process Step entity from MaRDI/Wikidata.

Source code in MaRDMO/workflow/models.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
@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: list[Relatant] = field(default_factory=list)
    platform_software: list[Relatant] = field(default_factory=list)
    platform_instrument: list[Relatant] = field(default_factory=list)
    field_of_work: list[Relatant] = field(default_factory=list)
    msc_id: 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=split_value(
                data=data, key='uses', transform=Relatant.from_query
            ),
            platform_software=split_value(
                data=data, key='platform_software', transform=Relatant.from_query
            ),
            platform_instrument=split_value(
                data=data, key='platform_instrument', transform=Relatant.from_query
            ),
            field_of_work=split_value(
                data=data, key='field_of_work', transform=Relatant.from_query
            ),
            msc_id=[
                Relatant.from_msc(msc_id, *_MSC_BY_ID[msc_id])
                for msc_id in split_value(data, 'msc_id')
                if msc_id in _MSC_BY_ID
            ],
        )

from_query(raw_data) classmethod

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

Source code in MaRDMO/workflow/models.py 
158
159
160
161
@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 
163
164
165
166
167
168
169
170
@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 
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
@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=split_value(
            data=data, key='uses', transform=Relatant.from_query
        ),
        platform_software=split_value(
            data=data, key='platform_software', transform=Relatant.from_query
        ),
        platform_instrument=split_value(
            data=data, key='platform_instrument', transform=Relatant.from_query
        ),
        field_of_work=split_value(
            data=data, key='field_of_work', transform=Relatant.from_query
        ),
        msc_id=[
            Relatant.from_msc(msc_id, *_MSC_BY_ID[msc_id])
            for msc_id in split_value(data, 'msc_id')
            if msc_id in _MSC_BY_ID
        ],
    )

Software dataclass

References and relations of a Software entity from MaRDI/Wikidata.

Source code in MaRDMO/workflow/models.py 
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
@dataclass
class Software:
    '''References and relations of a Software entity from MaRDI/Wikidata.'''

    reference: dict[int, list[str]] = field(default_factory=dict)
    programmed_in: list[Relatant] = field(default_factory=list)
    depends_on_software: list[Relatant] = field(default_factory=list)

    @classmethod
    def from_query(cls, raw_data: dict) -> 'Software':
        '''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, Software]':
        '''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) -> 'Software':
        '''Parse one SPARQL result row into a Software instance.'''
        options = get_options()

        return cls(
            reference={
                idx: [options[prop], data[prop]['value']]
                for idx, prop in enumerate(software_reference_ids)
                if data.get(prop, {}).get('value')
            },
            programmed_in=split_value(
                data=data, key='programmed_in', transform=Relatant.from_query
            ),
            depends_on_software=split_value(
                data=data, key='depends_on_software', transform=Relatant.from_query
            ),
        )

from_query(raw_data) classmethod

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

Source code in MaRDMO/workflow/models.py 
244
245
246
247
@classmethod
def from_query(cls, raw_data: dict) -> 'Software':
    '''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 
249
250
251
252
253
254
255
256
@classmethod
def from_query_batch(cls, raw_data: list) -> 'dict[str, Software]':
    '''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 Software instance.

Source code in MaRDMO/workflow/models.py 
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
@classmethod
def from_query_single(cls, data: dict) -> 'Software':
    '''Parse one SPARQL result row into a Software instance.'''
    options = get_options()

    return cls(
        reference={
            idx: [options[prop], data[prop]['value']]
            for idx, prop in enumerate(software_reference_ids)
            if data.get(prop, {}).get('value')
        },
        programmed_in=split_value(
            data=data, key='programmed_in', transform=Relatant.from_query
        ),
        depends_on_software=split_value(
            data=data, key='depends_on_software', transform=Relatant.from_query
        ),
    )

Variables dataclass

A single variable entry from MathModDB.

Source code in MaRDMO/workflow/models.py 
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
@dataclass
class Variables:
    '''A single variable entry from MathModDB.'''

    identifier: Optional[str] = None
    name: Optional[str] = None
    unit: Optional[str] = None
    symbol: Optional[str] = None
    task: Optional[str] = None
    type: Optional[str] = None

    @classmethod
    def from_query(cls, data: dict) -> 'Variables':
        '''Parse one SPARQL result row into a Variables instance.'''

        variables = {
            # Identifier of Variable
            'identifier': data.get('ID', {}).get('value'),
            # Name of Variable
            'name': data.get('Name', {}).get('value'),
            # Unit of Variable
            'unit': data.get('Unit', {}).get('value'),
            # Symbol of Variable
            'symbol': re.sub(
                r'(<math\b(?![^>]*\bdisplay=)[^>]*)(>)',
                r'\1 display="inline"\2',
                data.get('Symbol', {}).get('value'),
            ),
            # Task Variable is Part Of
            'task': data.get('label', {}).get('value'),
            # Role of Variable in Task (Input/Output)
            'type': data.get('Type', {}).get('value'),
        }

        return cls(
            **variables
        )

from_query(data) classmethod

Parse one SPARQL result row into a Variables instance.

Source code in MaRDMO/workflow/models.py 
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
@classmethod
def from_query(cls, data: dict) -> 'Variables':
    '''Parse one SPARQL result row into a Variables instance.'''

    variables = {
        # Identifier of Variable
        'identifier': data.get('ID', {}).get('value'),
        # Name of Variable
        'name': data.get('Name', {}).get('value'),
        # Unit of Variable
        'unit': data.get('Unit', {}).get('value'),
        # Symbol of Variable
        'symbol': re.sub(
            r'(<math\b(?![^>]*\bdisplay=)[^>]*)(>)',
            r'\1 display="inline"\2',
            data.get('Symbol', {}).get('value'),
        ),
        # Task Variable is Part Of
        'task': data.get('label', {}).get('value'),
        # Role of Variable in Task (Input/Output)
        'type': data.get('Type', {}).get('value'),
    }

    return cls(
        **variables
    )

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:MaRDIAndWikidataSearch — generic entity search across MaRDI Portal and Wikidata
  • :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/computation step lookup; refresh on select
  • :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 
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
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 
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
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'],
    )

Discipline

Bases: Provider

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

Source code in MaRDMO/workflow/providers.py 
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
class Discipline(Provider):
    '''Discipline Provider (MaRDI Portal / Wikidata / MSC),
       No User Creation, Refresh Upon Selection
    '''

    msc = get_data('data/msc2020.json')

    search = True
    refresh = True

    def get_options(self, project, search=None, user=None, site=None):
        '''Query knowledge graphs and the MSC 2020 classification for matching disciplines.

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

        Returns:
            Combined, sorted list of up to 30 ``{"id": …, "text": …}`` option dicts
            from MaRDI Portal / Wikidata (``academic discipline`` class) and MSC 2020.
        '''
        if not search or len(search) < 3:
            return []

        # Discipline from Knowledge Graphs
        options = query_sources(
            search = search,
            item_class = _ITEMS['academic discipline'],
            not_found = False,
        )

        # Mathematical Subjects
        options.extend(
            [
                {
                    'id': f"msc:{self.msc[key]['id']}",
                    'text': f"{key} ({self.msc[key]['quote']}) [msc]"
                }
                for key in self.msc if search.lower() in key.lower()
            ]
        )

        return sorted(options, key=lambda option: option['text'].lower())[:30]

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

Query knowledge graphs and the MSC 2020 classification for matching disciplines.

Parameters:

Name Type Description Default
project

RDMO project instance (unused).

 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

Combined, sorted list of up to 30 {"id": …, "text": …} option dicts

from MaRDI Portal / Wikidata (academic discipline class) and MSC 2020.
Source code in MaRDMO/workflow/providers.py 
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
def get_options(self, project, search=None, user=None, site=None):
    '''Query knowledge graphs and the MSC 2020 classification for matching disciplines.

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

    Returns:
        Combined, sorted list of up to 30 ``{"id": …, "text": …}`` option dicts
        from MaRDI Portal / Wikidata (``academic discipline`` class) and MSC 2020.
    '''
    if not search or len(search) < 3:
        return []

    # Discipline from Knowledge Graphs
    options = query_sources(
        search = search,
        item_class = _ITEMS['academic discipline'],
        not_found = False,
    )

    # Mathematical Subjects
    options.extend(
        [
            {
                'id': f"msc:{self.msc[key]['id']}",
                'text': f"{key} ({self.msc[key]['quote']}) [msc]"
            }
            for key in self.msc if search.lower() in key.lower()
        ]
    )

    return sorted(options, key=lambda option: option['text'].lower())[:30]

Hardware

Bases: Provider

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

Source code in MaRDMO/workflow/providers.py 
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
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 
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
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 
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
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 
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
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'],
    )

MaRDIAndWikidataSearch

Bases: Provider

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

Source code in MaRDMO/workflow/providers.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
class MaRDIAndWikidataSearch(Provider):
    '''General 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 or len(search) < 3:
            return []

        return query_sources(
            search = search
        )

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 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
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
    )

MainMathematicalModel

Bases: Provider

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

Source code in MaRDMO/workflow/providers.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
class MainMathematicalModel(Provider):
    '''Main Mathematical Model 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['mathematical model'],
            sources = ['mardi'],
        )

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 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
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'],
    )

Method

Bases: Provider

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

Source code in MaRDMO/workflow/providers.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
class Method(Provider):
    '''Method 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['method'],
                _ITEMS['algorithm']
            ]
        )

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 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
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['method'],
            _ITEMS['algorithm']
        ]
    )

ProcessStep

Bases: Provider

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

Source code in MaRDMO/workflow/providers.py 
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
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 
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
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'],
    )

RelatedDataSet

Bases: Provider

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

Source code in MaRDMO/workflow/providers.py 
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
class RelatedDataSet(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 = ['data-set'],
            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 
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
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 = ['data-set'],
        item_class = _ITEMS['data set'],
    )

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

RelatedInstrument

Bases: Provider

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

Source code in MaRDMO/workflow/providers.py 
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
class RelatedInstrument(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 = ['instrument'],
            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 
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
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 = ['instrument'],
        item_class = _ITEMS['research tool'],
    )

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

RelatedMethod

Bases: Provider

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

Source code in MaRDMO/workflow/providers.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
class RelatedMethod(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(
            query_attributes = ['method'],
            creation = True,
            item_class = [
                _ITEMS['method'],
                _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 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
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 = ['method'],
        creation = True,
        item_class = [
            _ITEMS['method'],
            _ITEMS['algorithm']
        ]
    )

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

WorkflowTask

Bases: Provider

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

Source code in MaRDMO/workflow/providers.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
class WorkflowTask(Provider):
    '''Task Provider (MaRDI Portal),
       User Creation, No Refresh Upon Selection
    '''

    def get_options(self, project, search=None, user=None, site=None):
        '''Query the MaRDI Portal for Computational Tasks linked to the currently selected Model.

        Reads the Model ID from the questionnaire, fetches associated Task
        items via SPARQL, and returns them as options.

        Args:
            project: RDMO project instance used to look up the current model ID.
            search:  Unused (tasks are filtered by model, not free-text).
            user:    Requesting user (unused).
            site:    Current site (unused).

        Returns:
            List of ``{"id": …, "text": …}`` option dicts for matching tasks.
        '''

        questions = get_questions('workflow')
        options = []
        model_id = ''

        values = project.values.filter(
            snapshot = None,
            attribute = Attribute.objects.get(
                uri = f'{BASE_URI}{questions["Model"]["ID"]["uri"]}'
            )
        )

        for value in values:
            model_id = value.external_id

        if not model_id:
            return options

        _, 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:
            return options

        if results[0].get('usedBy', {}).get('value'):
            tasks = results[0]['usedBy']['value'].split(' <|> ')
            for task in tasks:
                identifier, label, description = task.split(' || ')
                options.append(
                    {
                        'id': identifier,
                        'text': f'{label} ({description}) [mardi]'
                    }
                )

        return options

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

Query the MaRDI Portal for Computational Tasks linked to the currently selected Model.

Reads the Model ID from the questionnaire, fetches associated Task items via SPARQL, and returns them as options.

Parameters:

Name Type Description Default
project

RDMO project instance used to look up the current model ID.

 required
search

Unused (tasks are filtered by model, not free-text).

 None
user

Requesting user (unused).

 None
site

Current site (unused).

 None

Returns:

Type Description

List of {"id": …, "text": …} option dicts for matching tasks.
Source code in MaRDMO/workflow/providers.py 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
def get_options(self, project, search=None, user=None, site=None):
    '''Query the MaRDI Portal for Computational Tasks linked to the currently selected Model.

    Reads the Model ID from the questionnaire, fetches associated Task
    items via SPARQL, and returns them as options.

    Args:
        project: RDMO project instance used to look up the current model ID.
        search:  Unused (tasks are filtered by model, not free-text).
        user:    Requesting user (unused).
        site:    Current site (unused).

    Returns:
        List of ``{"id": …, "text": …}`` option dicts for matching tasks.
    '''

    questions = get_questions('workflow')
    options = []
    model_id = ''

    values = project.values.filter(
        snapshot = None,
        attribute = Attribute.objects.get(
            uri = f'{BASE_URI}{questions["Model"]["ID"]["uri"]}'
        )
    )

    for value in values:
        model_id = value.external_id

    if not model_id:
        return options

    _, 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:
        return options

    if results[0].get('usedBy', {}).get('value'):
        tasks = results[0]['usedBy']['value'].split(' <|> ')
        for task in tasks:
            identifier, label, description = task.split(' || ')
            options.append(
                {
                    'id': identifier,
                    'text': f'{label} ({description}) [mardi]'
                }
            )

    return options

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

Prepare interdisciplinary workflow answers for preview and export.

Loads Wikibase vocabulary (items and properties) on instantiation so they are available to both :meth:preview and :meth:export.

Source code in MaRDMO/workflow/worker.py 
  18
  19
  20
  21
  22
  23
  24
  25
  26
  27
  28
  29
  30
  31
  32
  33
  34
  35
  36
  37
  38
  39
  40
  41
  42
  43
  44
  45
  46
  47
  48
  49
  50
  51
  52
  53
  54
  55
  56
  57
  58
  59
  60
  61
  62
  63
  64
  65
  66
  67
  68
  69
  70
  71
  72
  73
  74
  75
  76
  77
  78
  79
  80
  81
  82
  83
  84
  85
  86
  87
  88
  89
  90
  91
  92
  93
  94
  95
  96
  97
  98
  99
 100
 101
 102
 103
 104
 105
 106
 107
 108
 109
 110
 111
 112
 113
 114
 115
 116
 117
 118
 119
 120
 121
 122
 123
 124
 125
 126
 127
 128
 129
 130
 131
 132
 133
 134
 135
 136
 137
 138
 139
 140
 141
 142
 143
 144
 145
 146
 147
 148
 149
 150
 151
 152
 153
 154
 155
 156
 157
 158
 159
 160
 161
 162
 163
 164
 165
 166
 167
 168
 169
 170
 171
 172
 173
 174
 175
 176
 177
 178
 179
 180
 181
 182
 183
 184
 185
 186
 187
 188
 189
 190
 191
 192
 193
 194
 195
 196
 197
 198
 199
 200
 201
 202
 203
 204
 205
 206
 207
 208
 209
 210
 211
 212
 213
 214
 215
 216
 217
 218
 219
 220
 221
 222
 223
 224
 225
 226
 227
 228
 229
 230
 231
 232
 233
 234
 235
 236
 237
 238
 239
 240
 241
 242
 243
 244
 245
 246
 247
 248
 249
 250
 251
 252
 253
 254
 255
 256
 257
 258
 259
 260
 261
 262
 263
 264
 265
 266
 267
 268
 269
 270
 271
 272
 273
 274
 275
 276
 277
 278
 279
 280
 281
 282
 283
 284
 285
 286
 287
 288
 289
 290
 291
 292
 293
 294
 295
 296
 297
 298
 299
 300
 301
 302
 303
 304
 305
 306
 307
 308
 309
 310
 311
 312
 313
 314
 315
 316
 317
 318
 319
 320
 321
 322
 323
 324
 325
 326
 327
 328
 329
 330
 331
 332
 333
 334
 335
 336
 337
 338
 339
 340
 341
 342
 343
 344
 345
 346
 347
 348
 349
 350
 351
 352
 353
 354
 355
 356
 357
 358
 359
 360
 361
 362
 363
 364
 365
 366
 367
 368
 369
 370
 371
 372
 373
 374
 375
 376
 377
 378
 379
 380
 381
 382
 383
 384
 385
 386
 387
 388
 389
 390
 391
 392
 393
 394
 395
 396
 397
 398
 399
 400
 401
 402
 403
 404
 405
 406
 407
 408
 409
 410
 411
 412
 413
 414
 415
 416
 417
 418
 419
 420
 421
 422
 423
 424
 425
 426
 427
 428
 429
 430
 431
 432
 433
 434
 435
 436
 437
 438
 439
 440
 441
 442
 443
 444
 445
 446
 447
 448
 449
 450
 451
 452
 453
 454
 455
 456
 457
 458
 459
 460
 461
 462
 463
 464
 465
 466
 467
 468
 469
 470
 471
 472
 473
 474
 475
 476
 477
 478
 479
 480
 481
 482
 483
 484
 485
 486
 487
 488
 489
 490
 491
 492
 493
 494
 495
 496
 497
 498
 499
 500
 501
 502
 503
 504
 505
 506
 507
 508
 509
 510
 511
 512
 513
 514
 515
 516
 517
 518
 519
 520
 521
 522
 523
 524
 525
 526
 527
 528
 529
 530
 531
 532
 533
 534
 535
 536
 537
 538
 539
 540
 541
 542
 543
 544
 545
 546
 547
 548
 549
 550
 551
 552
 553
 554
 555
 556
 557
 558
 559
 560
 561
 562
 563
 564
 565
 566
 567
 568
 569
 570
 571
 572
 573
 574
 575
 576
 577
 578
 579
 580
 581
 582
 583
 584
 585
 586
 587
 588
 589
 590
 591
 592
 593
 594
 595
 596
 597
 598
 599
 600
 601
 602
 603
 604
 605
 606
 607
 608
 609
 610
 611
 612
 613
 614
 615
 616
 617
 618
 619
 620
 621
 622
 623
 624
 625
 626
 627
 628
 629
 630
 631
 632
 633
 634
 635
 636
 637
 638
 639
 640
 641
 642
 643
 644
 645
 646
 647
 648
 649
 650
 651
 652
 653
 654
 655
 656
 657
 658
 659
 660
 661
 662
 663
 664
 665
 666
 667
 668
 669
 670
 671
 672
 673
 674
 675
 676
 677
 678
 679
 680
 681
 682
 683
 684
 685
 686
 687
 688
 689
 690
 691
 692
 693
 694
 695
 696
 697
 698
 699
 700
 701
 702
 703
 704
 705
 706
 707
 708
 709
 710
 711
 712
 713
 714
 715
 716
 717
 718
 719
 720
 721
 722
 723
 724
 725
 726
 727
 728
 729
 730
 731
 732
 733
 734
 735
 736
 737
 738
 739
 740
 741
 742
 743
 744
 745
 746
 747
 748
 749
 750
 751
 752
 753
 754
 755
 756
 757
 758
 759
 760
 761
 762
 763
 764
 765
 766
 767
 768
 769
 770
 771
 772
 773
 774
 775
 776
 777
 778
 779
 780
 781
 782
 783
 784
 785
 786
 787
 788
 789
 790
 791
 792
 793
 794
 795
 796
 797
 798
 799
 800
 801
 802
 803
 804
 805
 806
 807
 808
 809
 810
 811
 812
 813
 814
 815
 816
 817
 818
 819
 820
 821
 822
 823
 824
 825
 826
 827
 828
 829
 830
 831
 832
 833
 834
 835
 836
 837
 838
 839
 840
 841
 842
 843
 844
 845
 846
 847
 848
 849
 850
 851
 852
 853
 854
 855
 856
 857
 858
 859
 860
 861
 862
 863
 864
 865
 866
 867
 868
 869
 870
 871
 872
 873
 874
 875
 876
 877
 878
 879
 880
 881
 882
 883
 884
 885
 886
 887
 888
 889
 890
 891
 892
 893
 894
 895
 896
 897
 898
 899
 900
 901
 902
 903
 904
 905
 906
 907
 908
 909
 910
 911
 912
 913
 914
 915
 916
 917
 918
 919
 920
 921
 922
 923
 924
 925
 926
 927
 928
 929
 930
 931
 932
 933
 934
 935
 936
 937
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
class prepareWorkflow:
    '''Prepare interdisciplinary workflow answers for preview and export.

    Loads Wikibase vocabulary (items and properties) on instantiation so
    they are available to both :meth:`preview` and :meth:`export`.
    '''

    def __init__(self):
        '''Initialise with Wikibase items and properties from MaRDMOConfig.'''
        self.items = get_items()
        self.properties = get_properties()

    def preview(self, data):
        '''Enrich workflow answers with supplemental MaRDI Portal data for preview rendering.

        Fetches model properties, process-step entities, and research information
        via SPARQL and merges them into *data* so the preview template can
        render a complete picture.

        Args:
            data: Top-level workflow answers dict (mutated in place).

        Returns:
            The mutated *data* dict.
        '''
        # Update Model Properties via MathModDB
        if data.get('model',{}).get('ID'):
            _, identifier = data['model']['ID'].split(':')

            query = get_sparql_query(
                f'workflow/queries/preview_basic.sparql'
            ).format(
                identifier,
                **self.items,
                **self.properties
            )

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

            if basic:
                data.get('model', {}).update(asdict(ModelProperties.from_query(basic)))

        # Update Model Variables and Parameters via MathModDB
        if data.get('model', {}).get('task'):

            query = get_sparql_query(
                f'workflow/queries/preview_variable.sparql'
            ).format(
                ' '.join(f"wd:{value.get('ID', '').split(':')[1]}"
                for _, value in data['model']['task'].items()),
                **self.items,
                **self.properties
            )

            variables = query_sparql(query, get_url('mardi', 'sparql'))
            if variables:
                for idx, variable in enumerate(variables):
                    data.setdefault('variables', {}).update(
                        {
                            idx: asdict(Variables.from_query(variable))
                        }
                    )

            query = get_sparql_query(
                f'workflow/queries/preview_parameter.sparql'
            ).format(
                ' '.join(f"wd:{value.get('ID', '').split(':')[1]}"
                for _, value in data['model']['task'].items()),
                **self.items,
                **self.properties
            )

            parameters = query_sparql(query, get_url('mardi', 'sparql'))
            if parameters:
                for idx, parameter in enumerate(parameters):
                    data.setdefault('parameters', {}).update(
                        {
                            idx: asdict(Parameters.from_query(parameter))
                        }
                    )

        return data

    def export(self, data, title, 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``.
            title: Workflow title string used to seed item labels.
            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, title)

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

        # Load Options
        options = get_options()

        # Add / Retrieve Components of Interdisciplinary Workflow Item
        payload.process_items()

        ### Add additional Algorithms / Methods Information
        for method in data.get('method', {}).values():

            # Continue if no ID exists
            if not method.get('ID'):
                continue

            # Get Item Key
            payload.get_item_key(method)

            # Add Class
            if 'mathalgodb' in method['ID']:
                payload.add_answer(
                    verb=self.properties['instance of'],
                    object_and_type=[
                        self.items['algorithm'],
                        'wikibase-item',
                    ]
                )
            else:
                payload.add_answer(
                    verb=self.properties['instance of'],
                    object_and_type=[
                        self.items['method'],
                        'wikibase-item',
                    ]
                )

        ### Add additional Software Information
        for software in data.get('software', {}).values():

            # Continue if no ID exists
            if not software.get('ID'):
                continue

            # Get Item Key
            payload.get_item_key(software)

            # Add Class
            payload.add_answer(
                    verb=self.properties['instance of'],
                    object_and_type=[
                        self.items['software'],
                        'wikibase-item',
                    ]
                )

            # Add References of the Software
            for reference in software.get('Reference', {}).values():
                if reference[0] == options['DOI']:
                    payload.add_answer(
                        verb=self.properties['DOI'],
                        object_and_type=[
                            reference[1],
                            'external-id',
                        ]
                    )
                elif reference[0] == options['SWMATH']:
                    payload.add_answer(
                        verb=self.properties['swMath work ID'],
                        object_and_type=[
                            reference[1],
                            'external-id',
                        ]
                    )
                elif reference[0] == options['SOURCECODE_URL']:
                    payload.add_answer(
                        verb=self.properties['source code reposiory URL'],
                        object_and_type=[
                            reference[1],
                            'url',
                        ]
                    )
                elif reference[0] == options['DESCRIPTION_URL']:
                    payload.add_answer(
                        verb=self.properties['described at URL'],
                        object_and_type=[
                            reference[1],
                            'url',
                        ]
                    )

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

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

            # Add Source Code Repository
            if software.get('Published', [''])[0] == options['YesText']:
                payload.add_answer(
                    verb=self.properties['source code repository URL'],
                    object_and_type=[
                        software['Published'][1],
                        'url',
                    ]
                )

            # Add Documentation / Manual
            if software.get('Documented', [''])[0] == options['YesText']:
                payload.add_answer(
                    verb=self.properties['user manual URL'],
                    object_and_type=[
                        software['Documented'][1],
                        'url',
                    ]
                )

        ### Add additional Hardware Information
        for hardware in data.get('hardware', {}).values():

            # Continue if no ID exists
            if not hardware.get('ID'):
                continue

            # Get Item Key
            payload.get_item_key(hardware)

            # Add Class
            payload.add_answer(
                verb=self.properties['instance of'],
                object_and_type=[
                    self.items['computer hardware'],
                    'wikibase-item',
                ]
            )

            # Add CPU
            payload.add_single_relation(
                statement = {
                    'relation': self.properties['CPU'],
                    'relatant': 'cpu'
                }
            )

            # Add Number of Computing Nodes
            if hardware['Nodes']:
                payload.add_answer(
                    verb=self.properties['has part(s)'],
                    object_and_type=[
                        self.items['compute node'],
                        'wikibase-item',
                    ],
                    qualifier = [
                                    {
                                        "property": {
                                            "id": self.properties['quantity_property'],
                                        },
                                        "value": {
                                            "type": "value",
                                            "content": {
                                                "amount": f"+{hardware['Nodes']}",
                                                "unit": "1",
                                            },
                                        },
                                    }
                                ]
                )

            # Add Number of Processor Cores
            if hardware['Cores']:
                payload.add_answer(
                    verb=self.properties['number of processor cores'],
                    object_and_type=[
                        {
                            "amount": f"+{hardware['Cores']}",
                            "unit":"1"
                        },
                        'quantity'
                    ]
                )

        ### Add additional Instrument Information
        for instrument in data.get('instrument', {}).values():

            # Continue if no ID exists
            if not instrument.get('ID'):
                continue

            # Get Item Key
            payload.get_item_key(instrument)

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

            ### MORE INSTRUMENT INFORMATION TO ADD ###

        ### Add additional Dataset Information
        for dataset in data.get('dataset', {}).values():

            # Continue if no ID exists
            if not dataset.get('ID'):
                continue

            # Get Item Key
            payload.get_item_key(dataset)

            # Add Class
            payload.add_answer(
                verb=self.properties['instance of'],
                object_and_type=[
                    self.items['data set'],
                    'wikibase-item',
                ]
            )

            # Size of the data set
            if dataset.get('Size'):
                if dataset['Size'][0] == options['kilobyte']:
                    verb = self.properties['data size']
                    object = {
                        "amount": f"+{dataset['Size'][1]}",
                        "unit": f"{get_url('mardi', 'uri')}/entity/{self.items['kilobyte']}"
                    }
                elif dataset['Size'][0] == options['megabyte']:
                    verb = self.properties['data size']
                    object = {
                        "amount": f"+{dataset['Size'][1]}",
                        "unit": f"{get_url('mardi', 'uri')}/entity/{self.items['megabyte']}"
                    }
                elif dataset['Size'][0] == options['gigabyte']:
                    verb = self.properties['data size']
                    object = {
                        "amount": f"+{dataset['Size'][1]}",
                        "unit": f"{get_url('mardi', 'uri')}/entity/{self.items['gigabyte']}"
                    }
                elif dataset['Size'][0] == options['terabyte']:
                    verb = self.properties['data size']
                    object = {
                        "amount": f"+{dataset['Size'][1]}",
                        "unit": f"{get_url('mardi', 'uri')}/entity/{self.items['terabyte']}"
                    }
                elif dataset['Size'][0] == options['items']:
                    verb = self.properties['number of records']
                    object = {
                        "amount": f"+{dataset['Size'][1]}","unit":"1"
                    }

                payload.add_answer(
                    verb=verb,
                    object_and_type=[
                        object,
                        'quantity',
                    ]
                )

            # Add Data Type
            payload.add_single_relation(
                statement = {
                    'relation': self.properties['uses'], 
                    'relatant': 'datatype'
                },
                qualifier = payload.add_qualifier(
                    self.properties['object has role'],
                    'wikibase-item',
                    self.items['data type']
                )
            )

            # Add Representation Format
            payload.add_single_relation(
                statement = {
                    'relation': self.properties['uses'], 
                    'relatant': 'representationformat'
                },
                qualifier = payload.add_qualifier(
                    self.properties['object has role'],
                    'wikibase-item',
                    self.items['representation format']
                )
            )

            # Add File Format
            if dataset.get('FileFormat'):
                payload.add_answer(
                    verb=self.properties['file extension'],
                    object_and_type=[
                        dataset['FileFormat'],
                        'string',
                    ]
                )

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

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

            # Data Set to Publish
            if dataset.get('ToPublish'):
                if dataset['ToPublish'].get(0, ['',''])[0] == options['Yes']:
                    payload.add_answer(
                        verb=self.properties['mandates'],
                        object_and_type=[
                            self.items['data publishing'],
                            'wikibase-item',
                        ]
                    )
                    if dataset['ToPublish'].get(1, ['',''])[0] == options['DOI']:
                        payload.add_answer(
                            verb=self.properties['DOI'],
                            object_and_type=[
                                dataset['ToPublish'][1][1],
                                'external-id',
                            ]
                        )
                    if dataset['ToPublish'].get(2, ['',''])[0] == options['URL']:
                        payload.add_answer(
                            verb=self.properties['URL'],
                            object_and_type=[
                                dataset['ToPublish'][2][2],
                                'url',
                            ]
                        )

            # Data Set To Archive
            if dataset.get('ToArchive'):
                if dataset['ToArchive'][0] == options['YesText']:
                    qualifier = []
                    if dataset['ToArchive'][1]:
                        qualifier = payload.add_qualifier(
                            self.properties['end time'],
                            'time',
                            {
                                "time": f"+{dataset['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
                        )

        ### Add Process Step Information
        for processstep in data.get('processstep', {}).values():

            # Continue if no ID exists
            if not processstep.get('ID'):
                continue

            # Get Item Key
            payload.get_item_key(processstep)

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

            # Add Input Data Sets
            payload.add_single_relation(
                statement = {
                    'relation': self.properties['input data set'],
                    'relatant': 'input'
                }
            )

            # Add Output Data Sets
            payload.add_single_relation(
                statement = {
                    'relation': self.properties['output data set'],
                    'relatant': 'output'
                }
            )

            # Add applied Methods
            for method in processstep.get('method', {}).values():
                # Continue if no ID exists
                if not method.get('ID'):
                    continue
                # Get Entry Key
                method_item = payload.get_item_key(method, 'object')
                # Get Qualifier
                qualifier = []
                for parameter in method.get('Parameter', {}).values():
                    qualifier.extend(
                        payload.add_qualifier(
                            self.properties['comment'],
                            'string',
                            parameter
                        )
                    )
                # Add to Payload
                payload.add_answer(
                            verb=self.properties['uses'],
                            object_and_type=[
                                method_item,
                                'wikibase-item',
                            ],
                            qualifier=qualifier
                        )

            # Add Software Environment
            payload.add_single_relation(
                statement = {
                    'relation': self.properties['platform'], 
                    'relatant': 'environmentSoftware'
                },
                qualifier = payload.add_qualifier(
                    self.properties['object has role'],
                    'wikibase-item',
                    self.items['software']
                )
            )

            # Add Instrument Environment
            payload.add_single_relation(
                statement = {
                    'relation': self.properties['platform'], 
                    'relatant': 'environmentInstrument'
                },
                qualifier = payload.add_qualifier(
                    self.properties['object has role'],
                    'wikibase-item',
                    self.items['research tool']
                )
            )

            # Add Disciplines (math and non-math)
            for discipline in processstep.get('discipline', {}).values():
                # Check if new ID exists
                if 'msc:' in discipline.get('ID'):
                    _, id = discipline['ID'].split(':')
                    payload.add_answer(
                        verb=self.properties['MSC ID'],
                        object_and_type=[
                            id,
                            'external-id',
                        ]
                    )
                else:
                    # Get Discipline Key
                    discipline_item = payload.get_item_key(discipline, 'object')
                    # Add to Payload
                    payload.add_answer(
                        verb=self.properties['field of work'],
                        object_and_type=[
                            discipline_item,
                            'wikibase-item',
                        ]
                    )

        for publication in data.get('publication', {}).values():

            # Continue if no ID exists
            if not publication.get('ID'):
                continue

            # Get Item Key
            payload.get_item_key(publication)

            if 'mardi' not in publication['ID']:

                # Set and add the Class of the Publication
                if publication.get('entrytype') == 'scholarly article':
                    pclass = self.items['scholarly article']
                else:
                    pclass = self.items['publication']

                payload.add_answer(
                    verb=self.properties['instance of'],
                    object_and_type=[
                        pclass,
                        'wikibase-item',
                    ]
                )

                # Add Publication Profile
                payload.add_answer(
                        verb = self.properties["MaRDI profile type"],
                        object_and_type = [
                            self.items["MaRDI publication profile"],
                            "wikibase-item"
                        ],
                    )

                # Add the DOI of the Publication
                if publication.get('reference', {}).get(0):
                    payload.add_answer(
                        verb=self.properties['DOI'],
                        object_and_type=[
                            publication['reference'][0][1],
                            'external-id',
                        ]
                    )

                # Add the Title of the Publication
                if publication.get('title'):
                    payload.add_answer(
                        verb=self.properties['title'],
                        object_and_type=[
                            {"text": publication['title'], "language": "en"},
                            'monolingualtext',
                        ]
                    )

                # Add the Volume of the Publication
                if publication.get('volume'):
                    payload.add_answer(
                        verb=self.properties['volume'],
                        object_and_type=[
                            publication['volume'],
                            'string',
                        ]
                    )

                # Add the Issue of the Publication
                if publication.get('issue'):
                    payload.add_answer(
                        verb=self.properties['issue'],
                        object_and_type=[
                            publication['issue'],
                            'string',
                        ]
                    )

                # Add the Page(s) of the Publication
                if publication.get('page'):
                    payload.add_answer(
                        verb=self.properties['page(s)'],
                        object_and_type=[
                            publication['page'],
                            'string',
                        ]
                    )

                # Add the Date of the Publication
                if publication.get('date'):
                    payload.add_answer(
                        verb=self.properties['publication date'],
                        object_and_type=[
                            {
                                "time": f"+{publication['date']}T00:00:00Z",
                                "precision": 11,
                                "calendarmodel": "http://www.wikidata.org/entity/Q1985727"
                            },
                            'time',
                        ]
                    )

                # Add the Language of the Publication
                payload.add_single_relation(
                    statement = {
                        'relation': self.properties['language of work or name'],
                        'relatant': 'language'
                    }
                )

                # Add the Journal of the Publication
                payload.add_single_relation(
                    statement = {
                        'relation': self.properties['published in'],
                        'relatant': 'journal'
                    }
                )

                # Add the Authors of the Publication
                payload.add_single_relation(
                    statement = {
                        'relation': self.properties['author'],
                        'relatant': 'author'
                    },
                    alt_statement = {
                        'relation': self.properties['author name string'], 
                        'relatant': 'Name'
                    }
                )

        # Add Interdisciplinary Workflow Information
        workflow = {
            'ID': 'not found',
            'Name': title,
            'Description': data.get('general', {}).get('objective')
        }

        # Get Item Key
        payload.get_item_key(workflow)

        # Add Class
        payload.add_answer(
            verb=self.properties['instance of'],
            object_and_type=[
                self.items['research workflow'],
                'wikibase-item',
            ]
        )

        # Procedure Description to Workflow
        if data.get('general', {}).get('procedure'):
            payload.add_answer(
                verb=self.properties['description'],
                object_and_type=[
                    data['general']['procedure'],
                    'string',
                ]
            )

        # Add Reproducibility Aspects
        for key, value in REPRODUCIBILITY.items():
            if data.get('reproducibility', {}).get(key) == options['Yes']:
                qualifier = []
                if data['reproducibility'].get(f'{key}condition'):
                    qualifier.extend(
                        payload.add_qualifier(
                            self.properties['comment'],
                            'string',
                            data['reproducibility'][f'{key}condition']
                        )
                    )
                payload.add_answer(
                    verb=self.properties['instance of'],
                    object_and_type=[
                        self.items[value],
                        'wikibase-item',
                    ],
                    qualifier=qualifier
                )

        # Add Transferability Aspects
        if data.get('reproducibility', {}).get('transferability'):
            qualifier = []
            for value in data['reproducibility']['transferability'].values():
                qualifier.extend(
                    payload.add_qualifier(
                        self.properties['comment'],
                        'string',
                        value
                    )
                )
            payload.add_answer(
                verb=self.properties['instance of'],
                object_and_type=[
                    self.items['transferable research workflow'],
                    'wikibase-item',
                ],
                qualifier=qualifier
            )

        # Add Model and Task the Workflow Uses
        if data.get('model'):
            #Continue if ID exists
            if data['model'].get('ID'):
                # Get Item Key
                model_item = payload.get_item_key(data['model'], 'object')
                # Add Statement with Qualifier
                qualifier = []
                for task in data['model'].get('task', {}).values():
                    qualifier.extend(
                        payload.add_qualifier(
                            self.properties['used by'],
                            'wikibase-item',
                            payload.get_item_key(task, 'object')
                        )
                    )
                payload.add_answer(
                    verb=self.properties['uses'],
                    object_and_type=[
                        model_item,
                        'wikibase-item',
                    ],
                    qualifier=qualifier
                )

        # Add Methods the Workflow Uses
        for value in data.get('method', {}).values():
            # Continue if no ID exists
            if not value.get('ID'):
                continue
            # Get Item Key
            method_item = payload.get_item_key(value, 'object')
            # Add Statement with Qualifier
            qualifier = []
            for parameter in value.get('Parameter', {}).values():
                qualifier.extend(
                    payload.add_qualifier(
                        self.properties['comment'],
                        'string',
                        parameter
                    )
                )
            for software in value.get('software', {}).values():
                qualifier.extend(
                    payload.add_qualifier(
                        self.properties['implemented by'],
                        payload.get_item_key(software, 'object')
                    )
                )
            for instrument in value.get('instrument', {}).values():
                qualifier.extend(
                    payload.add_qualifier(
                        self.properties['implemented by'],
                        payload.get_item_key(instrument, 'object')
                    )
                )
            payload.add_answer(
                verb=self.properties['uses'],
                object_and_type=[
                    method_item,
                    'wikibase-item',
                ],
                qualifier=qualifier
            )

        # Add Software the Workflow uses
        for value in data.get('software', {}).values():
            # Continue if no ID exists
            if not value.get('ID'):
                continue
            # Get Item Key
            software_item = payload.get_item_key(value, 'object')
            # Add Statement with Qualifier
            qualifier = []
            for hardware in data.get('hardware', {}).values():
                for software in hardware.get('software', {}).values():
                    if (software.get('ID'), software.get('Name'), software.get('Description')) == (value['ID'], value['Name'], value['Description']):
                        hardware_item = payload.get_item_key(hardware, 'object')
                        qualifier.extend(payload.add_qualifier(self.properties['platform'], 'wikibase-item', hardware_item))
            if value.get('Version'):
                qualifier = payload.add_qualifier(
                    self.properties['software version identifier'],
                    'string',
                    value['Version']
                )
            payload.add_answer(
                verb=self.properties['uses'],
                object_and_type=[
                    software_item,
                    'wikibase-item',
                ],
                qualifier=qualifier
            )

        # Add Hardware the Workflow Uses
        for value in data.get('hardware', {}).values():
            # Continue if no ID exists
            if not value.get('ID'):
                continue
            # Get Item Key
            hardware_item = payload.get_item_key(value, 'object')
            # Add Satement with Qualifier
            qualifier = []
            for compiler in value.get('compiler', {}).values():
                qualifier.extend(
                    payload.add_qualifier(
                        self.properties['uses'],
                        'wikibase-item',
                        payload.get_item_key(compiler, 'object')
                    )
                )
            payload.add_answer(
                verb=self.properties['uses'],
                object_and_type=[
                    hardware_item,
                    'wikibase-item',
                ],
                qualifier=qualifier
            )

        # Add instruments the workflow Uses
        for value in data.get('instrument', {}).values():
            # Continue if no ID exists
            if not value.get('ID'):
                continue
            # Get Item Key
            instrument_item = payload.get_item_key(value, 'object')
            # Add Statement with Qualifer
            qualifier = []
            if value.get('Version'):
                qualifier.extend(
                    payload.add_qualifier(
                        self.properties['edition number'],
                        'string',
                        value['Version']
                    )
                )
            if value.get('SerialNumber'):
                qualifier.extend(
                    payload.add_qualifier(
                        self.properties['serial number'],
                        'string',
                        value['SerialNumber']
                    )
                )
            for location in value.get('location', {}).values():
                qualifier.extend(
                    payload.add_qualifier(
                        self.properties['location'],
                        'wikibase-item',
                        payload.get_item_key(location, 'object')
                    )
                )
            for software in value.get('software', {}).values():
                qualifier.extend(
                    payload.add_qualifier(
                        self.properties['uses'],
                        'wikibase-item',
                        payload.get_item_key(software, 'object')
                    )
                )
            payload.add_answer(
                verb=self.properties['uses'],
                object_and_type=[
                    instrument_item,
                    'wikibase-item',
                ],
                qualifier=qualifier
            )

        # Add Data Sets the Workflow Uses
        for value in data.get('dataset', {}).values():
            # Continue if no ID exists
            if not value.get('ID'):
                continue
            # Get Item Key
            dataset_item = payload.get_item_key(value, 'object')
            # Add Statement
            payload.add_answer(
                verb=self.properties['uses'],
                object_and_type=[
                    dataset_item,
                    'wikibase-item',
                ]
            )

        # Add Process Steps the Workflow Uses
        for value in data.get('processstep', {}).values():
            # Continue if no ID exists
            if not value.get('ID'):
                continue
            # Get Item Key
            processstep_item = payload.get_item_key(value, 'object')
            # Add Statement with Qualifier
            qualifier = []
            for parameter in value.get('parameter', {}).values():
                qualifier.extend(
                    payload.add_qualifier(
                        self.properties['comment'],
                        'string',
                        parameter
                    )
                )
            payload.add_answer(
                verb=self.properties['uses'],
                object_and_type=[
                    processstep_item,
                    'wikibase-item',
                ],
                qualifier=qualifier
            )

        # Add Publications related to the Workflow
        for value in data.get('publication', {}).values():
            # Continue if no ID exists
            if not value.get('ID'):
                continue
            # Get Item Key
            publication_item = payload.get_item_key(value, 'object')
            # Add Statement
            payload.add_answer(
                verb=self.properties['cites work'],
                object_and_type=[
                    publication_item,
                    'wikibase-item',
                ]
            )

        # Construct Item Payloads
        payload.add_item_payload()

        # If Relations are added, check if they exist
        if any(key.startswith('RELATION') for key in payload.get_dictionary()):

            # Generate SPARQL Check Query
            query = payload.build_relation_check_query()

            # Perform Check Query for Relations
            check = query_sparql(query, get_url('mardi', 'sparql'))

            # Add Check Results
            payload.add_check_results(check)

        return payload.get_dictionary(), payload.dependency

__init__()

Initialise with Wikibase items and properties from MaRDMOConfig.

Source code in MaRDMO/workflow/worker.py 
25
26
27
28
def __init__(self):
    '''Initialise with Wikibase items and properties from MaRDMOConfig.'''
    self.items = get_items()
    self.properties = get_properties()

export(data, title, 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
title

Workflow title string used to seed item labels.

 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 
 101
 102
 103
 104
 105
 106
 107
 108
 109
 110
 111
 112
 113
 114
 115
 116
 117
 118
 119
 120
 121
 122
 123
 124
 125
 126
 127
 128
 129
 130
 131
 132
 133
 134
 135
 136
 137
 138
 139
 140
 141
 142
 143
 144
 145
 146
 147
 148
 149
 150
 151
 152
 153
 154
 155
 156
 157
 158
 159
 160
 161
 162
 163
 164
 165
 166
 167
 168
 169
 170
 171
 172
 173
 174
 175
 176
 177
 178
 179
 180
 181
 182
 183
 184
 185
 186
 187
 188
 189
 190
 191
 192
 193
 194
 195
 196
 197
 198
 199
 200
 201
 202
 203
 204
 205
 206
 207
 208
 209
 210
 211
 212
 213
 214
 215
 216
 217
 218
 219
 220
 221
 222
 223
 224
 225
 226
 227
 228
 229
 230
 231
 232
 233
 234
 235
 236
 237
 238
 239
 240
 241
 242
 243
 244
 245
 246
 247
 248
 249
 250
 251
 252
 253
 254
 255
 256
 257
 258
 259
 260
 261
 262
 263
 264
 265
 266
 267
 268
 269
 270
 271
 272
 273
 274
 275
 276
 277
 278
 279
 280
 281
 282
 283
 284
 285
 286
 287
 288
 289
 290
 291
 292
 293
 294
 295
 296
 297
 298
 299
 300
 301
 302
 303
 304
 305
 306
 307
 308
 309
 310
 311
 312
 313
 314
 315
 316
 317
 318
 319
 320
 321
 322
 323
 324
 325
 326
 327
 328
 329
 330
 331
 332
 333
 334
 335
 336
 337
 338
 339
 340
 341
 342
 343
 344
 345
 346
 347
 348
 349
 350
 351
 352
 353
 354
 355
 356
 357
 358
 359
 360
 361
 362
 363
 364
 365
 366
 367
 368
 369
 370
 371
 372
 373
 374
 375
 376
 377
 378
 379
 380
 381
 382
 383
 384
 385
 386
 387
 388
 389
 390
 391
 392
 393
 394
 395
 396
 397
 398
 399
 400
 401
 402
 403
 404
 405
 406
 407
 408
 409
 410
 411
 412
 413
 414
 415
 416
 417
 418
 419
 420
 421
 422
 423
 424
 425
 426
 427
 428
 429
 430
 431
 432
 433
 434
 435
 436
 437
 438
 439
 440
 441
 442
 443
 444
 445
 446
 447
 448
 449
 450
 451
 452
 453
 454
 455
 456
 457
 458
 459
 460
 461
 462
 463
 464
 465
 466
 467
 468
 469
 470
 471
 472
 473
 474
 475
 476
 477
 478
 479
 480
 481
 482
 483
 484
 485
 486
 487
 488
 489
 490
 491
 492
 493
 494
 495
 496
 497
 498
 499
 500
 501
 502
 503
 504
 505
 506
 507
 508
 509
 510
 511
 512
 513
 514
 515
 516
 517
 518
 519
 520
 521
 522
 523
 524
 525
 526
 527
 528
 529
 530
 531
 532
 533
 534
 535
 536
 537
 538
 539
 540
 541
 542
 543
 544
 545
 546
 547
 548
 549
 550
 551
 552
 553
 554
 555
 556
 557
 558
 559
 560
 561
 562
 563
 564
 565
 566
 567
 568
 569
 570
 571
 572
 573
 574
 575
 576
 577
 578
 579
 580
 581
 582
 583
 584
 585
 586
 587
 588
 589
 590
 591
 592
 593
 594
 595
 596
 597
 598
 599
 600
 601
 602
 603
 604
 605
 606
 607
 608
 609
 610
 611
 612
 613
 614
 615
 616
 617
 618
 619
 620
 621
 622
 623
 624
 625
 626
 627
 628
 629
 630
 631
 632
 633
 634
 635
 636
 637
 638
 639
 640
 641
 642
 643
 644
 645
 646
 647
 648
 649
 650
 651
 652
 653
 654
 655
 656
 657
 658
 659
 660
 661
 662
 663
 664
 665
 666
 667
 668
 669
 670
 671
 672
 673
 674
 675
 676
 677
 678
 679
 680
 681
 682
 683
 684
 685
 686
 687
 688
 689
 690
 691
 692
 693
 694
 695
 696
 697
 698
 699
 700
 701
 702
 703
 704
 705
 706
 707
 708
 709
 710
 711
 712
 713
 714
 715
 716
 717
 718
 719
 720
 721
 722
 723
 724
 725
 726
 727
 728
 729
 730
 731
 732
 733
 734
 735
 736
 737
 738
 739
 740
 741
 742
 743
 744
 745
 746
 747
 748
 749
 750
 751
 752
 753
 754
 755
 756
 757
 758
 759
 760
 761
 762
 763
 764
 765
 766
 767
 768
 769
 770
 771
 772
 773
 774
 775
 776
 777
 778
 779
 780
 781
 782
 783
 784
 785
 786
 787
 788
 789
 790
 791
 792
 793
 794
 795
 796
 797
 798
 799
 800
 801
 802
 803
 804
 805
 806
 807
 808
 809
 810
 811
 812
 813
 814
 815
 816
 817
 818
 819
 820
 821
 822
 823
 824
 825
 826
 827
 828
 829
 830
 831
 832
 833
 834
 835
 836
 837
 838
 839
 840
 841
 842
 843
 844
 845
 846
 847
 848
 849
 850
 851
 852
 853
 854
 855
 856
 857
 858
 859
 860
 861
 862
 863
 864
 865
 866
 867
 868
 869
 870
 871
 872
 873
 874
 875
 876
 877
 878
 879
 880
 881
 882
 883
 884
 885
 886
 887
 888
 889
 890
 891
 892
 893
 894
 895
 896
 897
 898
 899
 900
 901
 902
 903
 904
 905
 906
 907
 908
 909
 910
 911
 912
 913
 914
 915
 916
 917
 918
 919
 920
 921
 922
 923
 924
 925
 926
 927
 928
 929
 930
 931
 932
 933
 934
 935
 936
 937
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
def export(self, data, title, 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``.
        title: Workflow title string used to seed item labels.
        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, title)

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

    # Load Options
    options = get_options()

    # Add / Retrieve Components of Interdisciplinary Workflow Item
    payload.process_items()

    ### Add additional Algorithms / Methods Information
    for method in data.get('method', {}).values():

        # Continue if no ID exists
        if not method.get('ID'):
            continue

        # Get Item Key
        payload.get_item_key(method)

        # Add Class
        if 'mathalgodb' in method['ID']:
            payload.add_answer(
                verb=self.properties['instance of'],
                object_and_type=[
                    self.items['algorithm'],
                    'wikibase-item',
                ]
            )
        else:
            payload.add_answer(
                verb=self.properties['instance of'],
                object_and_type=[
                    self.items['method'],
                    'wikibase-item',
                ]
            )

    ### Add additional Software Information
    for software in data.get('software', {}).values():

        # Continue if no ID exists
        if not software.get('ID'):
            continue

        # Get Item Key
        payload.get_item_key(software)

        # Add Class
        payload.add_answer(
                verb=self.properties['instance of'],
                object_and_type=[
                    self.items['software'],
                    'wikibase-item',
                ]
            )

        # Add References of the Software
        for reference in software.get('Reference', {}).values():
            if reference[0] == options['DOI']:
                payload.add_answer(
                    verb=self.properties['DOI'],
                    object_and_type=[
                        reference[1],
                        'external-id',
                    ]
                )
            elif reference[0] == options['SWMATH']:
                payload.add_answer(
                    verb=self.properties['swMath work ID'],
                    object_and_type=[
                        reference[1],
                        'external-id',
                    ]
                )
            elif reference[0] == options['SOURCECODE_URL']:
                payload.add_answer(
                    verb=self.properties['source code reposiory URL'],
                    object_and_type=[
                        reference[1],
                        'url',
                    ]
                )
            elif reference[0] == options['DESCRIPTION_URL']:
                payload.add_answer(
                    verb=self.properties['described at URL'],
                    object_and_type=[
                        reference[1],
                        'url',
                    ]
                )

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

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

        # Add Source Code Repository
        if software.get('Published', [''])[0] == options['YesText']:
            payload.add_answer(
                verb=self.properties['source code repository URL'],
                object_and_type=[
                    software['Published'][1],
                    'url',
                ]
            )

        # Add Documentation / Manual
        if software.get('Documented', [''])[0] == options['YesText']:
            payload.add_answer(
                verb=self.properties['user manual URL'],
                object_and_type=[
                    software['Documented'][1],
                    'url',
                ]
            )

    ### Add additional Hardware Information
    for hardware in data.get('hardware', {}).values():

        # Continue if no ID exists
        if not hardware.get('ID'):
            continue

        # Get Item Key
        payload.get_item_key(hardware)

        # Add Class
        payload.add_answer(
            verb=self.properties['instance of'],
            object_and_type=[
                self.items['computer hardware'],
                'wikibase-item',
            ]
        )

        # Add CPU
        payload.add_single_relation(
            statement = {
                'relation': self.properties['CPU'],
                'relatant': 'cpu'
            }
        )

        # Add Number of Computing Nodes
        if hardware['Nodes']:
            payload.add_answer(
                verb=self.properties['has part(s)'],
                object_and_type=[
                    self.items['compute node'],
                    'wikibase-item',
                ],
                qualifier = [
                                {
                                    "property": {
                                        "id": self.properties['quantity_property'],
                                    },
                                    "value": {
                                        "type": "value",
                                        "content": {
                                            "amount": f"+{hardware['Nodes']}",
                                            "unit": "1",
                                        },
                                    },
                                }
                            ]
            )

        # Add Number of Processor Cores
        if hardware['Cores']:
            payload.add_answer(
                verb=self.properties['number of processor cores'],
                object_and_type=[
                    {
                        "amount": f"+{hardware['Cores']}",
                        "unit":"1"
                    },
                    'quantity'
                ]
            )

    ### Add additional Instrument Information
    for instrument in data.get('instrument', {}).values():

        # Continue if no ID exists
        if not instrument.get('ID'):
            continue

        # Get Item Key
        payload.get_item_key(instrument)

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

        ### MORE INSTRUMENT INFORMATION TO ADD ###

    ### Add additional Dataset Information
    for dataset in data.get('dataset', {}).values():

        # Continue if no ID exists
        if not dataset.get('ID'):
            continue

        # Get Item Key
        payload.get_item_key(dataset)

        # Add Class
        payload.add_answer(
            verb=self.properties['instance of'],
            object_and_type=[
                self.items['data set'],
                'wikibase-item',
            ]
        )

        # Size of the data set
        if dataset.get('Size'):
            if dataset['Size'][0] == options['kilobyte']:
                verb = self.properties['data size']
                object = {
                    "amount": f"+{dataset['Size'][1]}",
                    "unit": f"{get_url('mardi', 'uri')}/entity/{self.items['kilobyte']}"
                }
            elif dataset['Size'][0] == options['megabyte']:
                verb = self.properties['data size']
                object = {
                    "amount": f"+{dataset['Size'][1]}",
                    "unit": f"{get_url('mardi', 'uri')}/entity/{self.items['megabyte']}"
                }
            elif dataset['Size'][0] == options['gigabyte']:
                verb = self.properties['data size']
                object = {
                    "amount": f"+{dataset['Size'][1]}",
                    "unit": f"{get_url('mardi', 'uri')}/entity/{self.items['gigabyte']}"
                }
            elif dataset['Size'][0] == options['terabyte']:
                verb = self.properties['data size']
                object = {
                    "amount": f"+{dataset['Size'][1]}",
                    "unit": f"{get_url('mardi', 'uri')}/entity/{self.items['terabyte']}"
                }
            elif dataset['Size'][0] == options['items']:
                verb = self.properties['number of records']
                object = {
                    "amount": f"+{dataset['Size'][1]}","unit":"1"
                }

            payload.add_answer(
                verb=verb,
                object_and_type=[
                    object,
                    'quantity',
                ]
            )

        # Add Data Type
        payload.add_single_relation(
            statement = {
                'relation': self.properties['uses'], 
                'relatant': 'datatype'
            },
            qualifier = payload.add_qualifier(
                self.properties['object has role'],
                'wikibase-item',
                self.items['data type']
            )
        )

        # Add Representation Format
        payload.add_single_relation(
            statement = {
                'relation': self.properties['uses'], 
                'relatant': 'representationformat'
            },
            qualifier = payload.add_qualifier(
                self.properties['object has role'],
                'wikibase-item',
                self.items['representation format']
            )
        )

        # Add File Format
        if dataset.get('FileFormat'):
            payload.add_answer(
                verb=self.properties['file extension'],
                object_and_type=[
                    dataset['FileFormat'],
                    'string',
                ]
            )

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

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

        # Data Set to Publish
        if dataset.get('ToPublish'):
            if dataset['ToPublish'].get(0, ['',''])[0] == options['Yes']:
                payload.add_answer(
                    verb=self.properties['mandates'],
                    object_and_type=[
                        self.items['data publishing'],
                        'wikibase-item',
                    ]
                )
                if dataset['ToPublish'].get(1, ['',''])[0] == options['DOI']:
                    payload.add_answer(
                        verb=self.properties['DOI'],
                        object_and_type=[
                            dataset['ToPublish'][1][1],
                            'external-id',
                        ]
                    )
                if dataset['ToPublish'].get(2, ['',''])[0] == options['URL']:
                    payload.add_answer(
                        verb=self.properties['URL'],
                        object_and_type=[
                            dataset['ToPublish'][2][2],
                            'url',
                        ]
                    )

        # Data Set To Archive
        if dataset.get('ToArchive'):
            if dataset['ToArchive'][0] == options['YesText']:
                qualifier = []
                if dataset['ToArchive'][1]:
                    qualifier = payload.add_qualifier(
                        self.properties['end time'],
                        'time',
                        {
                            "time": f"+{dataset['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
                    )

    ### Add Process Step Information
    for processstep in data.get('processstep', {}).values():

        # Continue if no ID exists
        if not processstep.get('ID'):
            continue

        # Get Item Key
        payload.get_item_key(processstep)

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

        # Add Input Data Sets
        payload.add_single_relation(
            statement = {
                'relation': self.properties['input data set'],
                'relatant': 'input'
            }
        )

        # Add Output Data Sets
        payload.add_single_relation(
            statement = {
                'relation': self.properties['output data set'],
                'relatant': 'output'
            }
        )

        # Add applied Methods
        for method in processstep.get('method', {}).values():
            # Continue if no ID exists
            if not method.get('ID'):
                continue
            # Get Entry Key
            method_item = payload.get_item_key(method, 'object')
            # Get Qualifier
            qualifier = []
            for parameter in method.get('Parameter', {}).values():
                qualifier.extend(
                    payload.add_qualifier(
                        self.properties['comment'],
                        'string',
                        parameter
                    )
                )
            # Add to Payload
            payload.add_answer(
                        verb=self.properties['uses'],
                        object_and_type=[
                            method_item,
                            'wikibase-item',
                        ],
                        qualifier=qualifier
                    )

        # Add Software Environment
        payload.add_single_relation(
            statement = {
                'relation': self.properties['platform'], 
                'relatant': 'environmentSoftware'
            },
            qualifier = payload.add_qualifier(
                self.properties['object has role'],
                'wikibase-item',
                self.items['software']
            )
        )

        # Add Instrument Environment
        payload.add_single_relation(
            statement = {
                'relation': self.properties['platform'], 
                'relatant': 'environmentInstrument'
            },
            qualifier = payload.add_qualifier(
                self.properties['object has role'],
                'wikibase-item',
                self.items['research tool']
            )
        )

        # Add Disciplines (math and non-math)
        for discipline in processstep.get('discipline', {}).values():
            # Check if new ID exists
            if 'msc:' in discipline.get('ID'):
                _, id = discipline['ID'].split(':')
                payload.add_answer(
                    verb=self.properties['MSC ID'],
                    object_and_type=[
                        id,
                        'external-id',
                    ]
                )
            else:
                # Get Discipline Key
                discipline_item = payload.get_item_key(discipline, 'object')
                # Add to Payload
                payload.add_answer(
                    verb=self.properties['field of work'],
                    object_and_type=[
                        discipline_item,
                        'wikibase-item',
                    ]
                )

    for publication in data.get('publication', {}).values():

        # Continue if no ID exists
        if not publication.get('ID'):
            continue

        # Get Item Key
        payload.get_item_key(publication)

        if 'mardi' not in publication['ID']:

            # Set and add the Class of the Publication
            if publication.get('entrytype') == 'scholarly article':
                pclass = self.items['scholarly article']
            else:
                pclass = self.items['publication']

            payload.add_answer(
                verb=self.properties['instance of'],
                object_and_type=[
                    pclass,
                    'wikibase-item',
                ]
            )

            # Add Publication Profile
            payload.add_answer(
                    verb = self.properties["MaRDI profile type"],
                    object_and_type = [
                        self.items["MaRDI publication profile"],
                        "wikibase-item"
                    ],
                )

            # Add the DOI of the Publication
            if publication.get('reference', {}).get(0):
                payload.add_answer(
                    verb=self.properties['DOI'],
                    object_and_type=[
                        publication['reference'][0][1],
                        'external-id',
                    ]
                )

            # Add the Title of the Publication
            if publication.get('title'):
                payload.add_answer(
                    verb=self.properties['title'],
                    object_and_type=[
                        {"text": publication['title'], "language": "en"},
                        'monolingualtext',
                    ]
                )

            # Add the Volume of the Publication
            if publication.get('volume'):
                payload.add_answer(
                    verb=self.properties['volume'],
                    object_and_type=[
                        publication['volume'],
                        'string',
                    ]
                )

            # Add the Issue of the Publication
            if publication.get('issue'):
                payload.add_answer(
                    verb=self.properties['issue'],
                    object_and_type=[
                        publication['issue'],
                        'string',
                    ]
                )

            # Add the Page(s) of the Publication
            if publication.get('page'):
                payload.add_answer(
                    verb=self.properties['page(s)'],
                    object_and_type=[
                        publication['page'],
                        'string',
                    ]
                )

            # Add the Date of the Publication
            if publication.get('date'):
                payload.add_answer(
                    verb=self.properties['publication date'],
                    object_and_type=[
                        {
                            "time": f"+{publication['date']}T00:00:00Z",
                            "precision": 11,
                            "calendarmodel": "http://www.wikidata.org/entity/Q1985727"
                        },
                        'time',
                    ]
                )

            # Add the Language of the Publication
            payload.add_single_relation(
                statement = {
                    'relation': self.properties['language of work or name'],
                    'relatant': 'language'
                }
            )

            # Add the Journal of the Publication
            payload.add_single_relation(
                statement = {
                    'relation': self.properties['published in'],
                    'relatant': 'journal'
                }
            )

            # Add the Authors of the Publication
            payload.add_single_relation(
                statement = {
                    'relation': self.properties['author'],
                    'relatant': 'author'
                },
                alt_statement = {
                    'relation': self.properties['author name string'], 
                    'relatant': 'Name'
                }
            )

    # Add Interdisciplinary Workflow Information
    workflow = {
        'ID': 'not found',
        'Name': title,
        'Description': data.get('general', {}).get('objective')
    }

    # Get Item Key
    payload.get_item_key(workflow)

    # Add Class
    payload.add_answer(
        verb=self.properties['instance of'],
        object_and_type=[
            self.items['research workflow'],
            'wikibase-item',
        ]
    )

    # Procedure Description to Workflow
    if data.get('general', {}).get('procedure'):
        payload.add_answer(
            verb=self.properties['description'],
            object_and_type=[
                data['general']['procedure'],
                'string',
            ]
        )

    # Add Reproducibility Aspects
    for key, value in REPRODUCIBILITY.items():
        if data.get('reproducibility', {}).get(key) == options['Yes']:
            qualifier = []
            if data['reproducibility'].get(f'{key}condition'):
                qualifier.extend(
                    payload.add_qualifier(
                        self.properties['comment'],
                        'string',
                        data['reproducibility'][f'{key}condition']
                    )
                )
            payload.add_answer(
                verb=self.properties['instance of'],
                object_and_type=[
                    self.items[value],
                    'wikibase-item',
                ],
                qualifier=qualifier
            )

    # Add Transferability Aspects
    if data.get('reproducibility', {}).get('transferability'):
        qualifier = []
        for value in data['reproducibility']['transferability'].values():
            qualifier.extend(
                payload.add_qualifier(
                    self.properties['comment'],
                    'string',
                    value
                )
            )
        payload.add_answer(
            verb=self.properties['instance of'],
            object_and_type=[
                self.items['transferable research workflow'],
                'wikibase-item',
            ],
            qualifier=qualifier
        )

    # Add Model and Task the Workflow Uses
    if data.get('model'):
        #Continue if ID exists
        if data['model'].get('ID'):
            # Get Item Key
            model_item = payload.get_item_key(data['model'], 'object')
            # Add Statement with Qualifier
            qualifier = []
            for task in data['model'].get('task', {}).values():
                qualifier.extend(
                    payload.add_qualifier(
                        self.properties['used by'],
                        'wikibase-item',
                        payload.get_item_key(task, 'object')
                    )
                )
            payload.add_answer(
                verb=self.properties['uses'],
                object_and_type=[
                    model_item,
                    'wikibase-item',
                ],
                qualifier=qualifier
            )

    # Add Methods the Workflow Uses
    for value in data.get('method', {}).values():
        # Continue if no ID exists
        if not value.get('ID'):
            continue
        # Get Item Key
        method_item = payload.get_item_key(value, 'object')
        # Add Statement with Qualifier
        qualifier = []
        for parameter in value.get('Parameter', {}).values():
            qualifier.extend(
                payload.add_qualifier(
                    self.properties['comment'],
                    'string',
                    parameter
                )
            )
        for software in value.get('software', {}).values():
            qualifier.extend(
                payload.add_qualifier(
                    self.properties['implemented by'],
                    payload.get_item_key(software, 'object')
                )
            )
        for instrument in value.get('instrument', {}).values():
            qualifier.extend(
                payload.add_qualifier(
                    self.properties['implemented by'],
                    payload.get_item_key(instrument, 'object')
                )
            )
        payload.add_answer(
            verb=self.properties['uses'],
            object_and_type=[
                method_item,
                'wikibase-item',
            ],
            qualifier=qualifier
        )

    # Add Software the Workflow uses
    for value in data.get('software', {}).values():
        # Continue if no ID exists
        if not value.get('ID'):
            continue
        # Get Item Key
        software_item = payload.get_item_key(value, 'object')
        # Add Statement with Qualifier
        qualifier = []
        for hardware in data.get('hardware', {}).values():
            for software in hardware.get('software', {}).values():
                if (software.get('ID'), software.get('Name'), software.get('Description')) == (value['ID'], value['Name'], value['Description']):
                    hardware_item = payload.get_item_key(hardware, 'object')
                    qualifier.extend(payload.add_qualifier(self.properties['platform'], 'wikibase-item', hardware_item))
        if value.get('Version'):
            qualifier = payload.add_qualifier(
                self.properties['software version identifier'],
                'string',
                value['Version']
            )
        payload.add_answer(
            verb=self.properties['uses'],
            object_and_type=[
                software_item,
                'wikibase-item',
            ],
            qualifier=qualifier
        )

    # Add Hardware the Workflow Uses
    for value in data.get('hardware', {}).values():
        # Continue if no ID exists
        if not value.get('ID'):
            continue
        # Get Item Key
        hardware_item = payload.get_item_key(value, 'object')
        # Add Satement with Qualifier
        qualifier = []
        for compiler in value.get('compiler', {}).values():
            qualifier.extend(
                payload.add_qualifier(
                    self.properties['uses'],
                    'wikibase-item',
                    payload.get_item_key(compiler, 'object')
                )
            )
        payload.add_answer(
            verb=self.properties['uses'],
            object_and_type=[
                hardware_item,
                'wikibase-item',
            ],
            qualifier=qualifier
        )

    # Add instruments the workflow Uses
    for value in data.get('instrument', {}).values():
        # Continue if no ID exists
        if not value.get('ID'):
            continue
        # Get Item Key
        instrument_item = payload.get_item_key(value, 'object')
        # Add Statement with Qualifer
        qualifier = []
        if value.get('Version'):
            qualifier.extend(
                payload.add_qualifier(
                    self.properties['edition number'],
                    'string',
                    value['Version']
                )
            )
        if value.get('SerialNumber'):
            qualifier.extend(
                payload.add_qualifier(
                    self.properties['serial number'],
                    'string',
                    value['SerialNumber']
                )
            )
        for location in value.get('location', {}).values():
            qualifier.extend(
                payload.add_qualifier(
                    self.properties['location'],
                    'wikibase-item',
                    payload.get_item_key(location, 'object')
                )
            )
        for software in value.get('software', {}).values():
            qualifier.extend(
                payload.add_qualifier(
                    self.properties['uses'],
                    'wikibase-item',
                    payload.get_item_key(software, 'object')
                )
            )
        payload.add_answer(
            verb=self.properties['uses'],
            object_and_type=[
                instrument_item,
                'wikibase-item',
            ],
            qualifier=qualifier
        )

    # Add Data Sets the Workflow Uses
    for value in data.get('dataset', {}).values():
        # Continue if no ID exists
        if not value.get('ID'):
            continue
        # Get Item Key
        dataset_item = payload.get_item_key(value, 'object')
        # Add Statement
        payload.add_answer(
            verb=self.properties['uses'],
            object_and_type=[
                dataset_item,
                'wikibase-item',
            ]
        )

    # Add Process Steps the Workflow Uses
    for value in data.get('processstep', {}).values():
        # Continue if no ID exists
        if not value.get('ID'):
            continue
        # Get Item Key
        processstep_item = payload.get_item_key(value, 'object')
        # Add Statement with Qualifier
        qualifier = []
        for parameter in value.get('parameter', {}).values():
            qualifier.extend(
                payload.add_qualifier(
                    self.properties['comment'],
                    'string',
                    parameter
                )
            )
        payload.add_answer(
            verb=self.properties['uses'],
            object_and_type=[
                processstep_item,
                'wikibase-item',
            ],
            qualifier=qualifier
        )

    # Add Publications related to the Workflow
    for value in data.get('publication', {}).values():
        # Continue if no ID exists
        if not value.get('ID'):
            continue
        # Get Item Key
        publication_item = payload.get_item_key(value, 'object')
        # Add Statement
        payload.add_answer(
            verb=self.properties['cites work'],
            object_and_type=[
                publication_item,
                'wikibase-item',
            ]
        )

    # Construct Item Payloads
    payload.add_item_payload()

    # If Relations are added, check if they exist
    if any(key.startswith('RELATION') for key in payload.get_dictionary()):

        # Generate SPARQL Check Query
        query = payload.build_relation_check_query()

        # Perform Check Query for Relations
        check = query_sparql(query, get_url('mardi', 'sparql'))

        # Add Check Results
        payload.add_check_results(check)

    return payload.get_dictionary(), payload.dependency

preview(data)

Enrich workflow answers with supplemental MaRDI Portal data for preview rendering.

Fetches model properties, process-step entities, and research information via SPARQL and merges them into data so the preview template can render a complete picture.

Parameters:

Name Type Description Default
data

Top-level workflow answers dict (mutated in place).

 required

Returns:

Type Description

The mutated data dict.
Source code in MaRDMO/workflow/worker.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def preview(self, data):
    '''Enrich workflow answers with supplemental MaRDI Portal data for preview rendering.

    Fetches model properties, process-step entities, and research information
    via SPARQL and merges them into *data* so the preview template can
    render a complete picture.

    Args:
        data: Top-level workflow answers dict (mutated in place).

    Returns:
        The mutated *data* dict.
    '''
    # Update Model Properties via MathModDB
    if data.get('model',{}).get('ID'):
        _, identifier = data['model']['ID'].split(':')

        query = get_sparql_query(
            f'workflow/queries/preview_basic.sparql'
        ).format(
            identifier,
            **self.items,
            **self.properties
        )

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

        if basic:
            data.get('model', {}).update(asdict(ModelProperties.from_query(basic)))

    # Update Model Variables and Parameters via MathModDB
    if data.get('model', {}).get('task'):

        query = get_sparql_query(
            f'workflow/queries/preview_variable.sparql'
        ).format(
            ' '.join(f"wd:{value.get('ID', '').split(':')[1]}"
            for _, value in data['model']['task'].items()),
            **self.items,
            **self.properties
        )

        variables = query_sparql(query, get_url('mardi', 'sparql'))
        if variables:
            for idx, variable in enumerate(variables):
                data.setdefault('variables', {}).update(
                    {
                        idx: asdict(Variables.from_query(variable))
                    }
                )

        query = get_sparql_query(
            f'workflow/queries/preview_parameter.sparql'
        ).format(
            ' '.join(f"wd:{value.get('ID', '').split(':')[1]}"
            for _, value in data['model']['task'].items()),
            **self.items,
            **self.properties
        )

        parameters = query_sparql(query, get_url('mardi', 'sparql'))
        if parameters:
            for idx, parameter in enumerate(parameters):
                data.setdefault('parameters', {}).update(
                    {
                        idx: asdict(Parameters.from_query(parameter))
                    }
                )

    return data

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_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 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
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"]["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["Process Step"]["Method"]["uri"]}': {
            "question_set": f'{BASE_URI}{questions["Method"]["uri"]}',
            "question_id": f'{BASE_URI}{questions["Method"]["ID"]["uri"]}',
            "prefix": "M"
        },
        f'{BASE_URI}{questions["Process Step"]["Environment-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["Process Step"]["Environment-Instrument"]["uri"]}': {
            "question_set": f'{BASE_URI}{questions["Instrument"]["uri"]}',
            "question_id": f'{BASE_URI}{questions["Instrument"]["ID"]["uri"]}',
            "prefix": "I"
        },
        f'{BASE_URI}{questions["Method"]["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["Method"]["Instrument"]["uri"]}': {
            "question_set": f'{BASE_URI}{questions["Instrument"]["uri"]}',
            "question_id": f'{BASE_URI}{questions["Instrument"]["ID"]["uri"]}',
            "prefix": "I"
        },
        f'{BASE_URI}{questions["Instrument"]["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["Hardware"]["Software"]["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 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
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 (discipline labels, data-set sizes, references, archive URLs) needed by the workflow worker and handler layers.

Provides:

  • get_discipline — derive the research discipline label from questionnaire answers
  • get_size — extract the data-set size value from answer options
  • get_reference — extract a data-set or instrument reference identifier
  • get_archive — extract the data-archive URL or identifier

get_archive(data, options)

Extract archival intent and optional end-year from a data-set answer dict.

Parameters:

Name Type Description Default
data

Answer sub-dict for the data-set archival questions.

 required
options

Options dict mapping RDMO option URIs to string values.

 required

Returns:

Type Description

[archive_option, year] if archival is set; [] otherwise.

year is the first four characters of the end_time value, or

an empty string if no date was given.
Source code in MaRDMO/workflow/utils.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def get_archive(data, options):
    '''Extract archival intent and optional end-year from a data-set answer dict.

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

    Returns:
        ``[archive_option, year]`` if archival is set; ``[]`` otherwise.
        *year* is the first four characters of the ``end_time`` value, or
        an empty string if no date was given.
    '''
    archive = options[data['archive']['value']] if data.get('archive', {}).get('value') else ''
    year = data.get('end_time', {}).get('value', '')[:4]
    return [archive, year] if archive else []

get_discipline(answers)

Partition process-step disciplines into non-math and MSC (math subject classification) categories.

Iterates over all processstep discipline entries in answers, deduplicates them by ID, and adds two new top-level keys to answers:

  • nonmathdiscipline – items with a mardi: or wikidata: prefix
  • mathsubject – items with an msc: prefix

Parameters:

Name Type Description Default
answers

Top-level answers dict (mutated in place).

 required

Returns:

Type Description

The mutated answers dict.
Source code in MaRDMO/workflow/utils.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def get_discipline(answers):
    '''Partition process-step disciplines into non-math and MSC (math subject classification) categories.

    Iterates over all ``processstep`` discipline entries in *answers*, deduplicates
    them by ID, and adds two new top-level keys to *answers*:

    * ``nonmathdiscipline`` – items with a ``mardi:`` or ``wikidata:`` prefix
    * ``mathsubject``       – items with an ``msc:`` prefix

    Args:
        answers: Top-level answers dict (mutated in place).

    Returns:
        The mutated *answers* dict.
    '''
    ids = []
    md = 0
    nmd = 0
    for key in answers.get('processstep', []):
        for key2 in answers['processstep'][key].get('discipline', []):
            if not answers['processstep'][key]['discipline'][key2].get('ID'):
                continue
            if answers['processstep'][key]['discipline'][key2]['ID'] in ids:
                continue
            if answers['processstep'][key]['discipline'][key2]['ID'].split(':')[0] in ('mardi', 'wikidata'):
                answers.setdefault('nonmathdiscipline', {}).update(
                    {
                        nmd:
                            {
                                'ID': answers['processstep'][key]['discipline'][key2]['ID'],
                                'Name': answers['processstep'][key]['discipline'][key2]['Name']
                            }
                    }
                )
                nmd += 1
                ids.append(answers['processstep'][key]['discipline'][key2]['ID'])
            elif answers['processstep'][key]['discipline'][key2]['ID'].split(':')[0] == 'msc':
                answers.setdefault('mathsubject', {}).update(
                    {
                        md:
                            {
                                'ID': answers['processstep'][key]['discipline'][key2]['ID'],
                                'Name': answers['processstep'][key]['discipline'][key2]['Name']
                            }
                    }
                )
                md += 1
                ids.append(answers['processstep'][key]['discipline'][key2]['ID'])
    return answers

get_reference(data, options)

Build the reference dict for a data-set answer dict.

Iterates over the predefined data_set_reference_ids keys and maps each present value to [option_uri, value] at a numeric index.

Parameters:

Name Type Description Default
data

Answer sub-dict for the data-set reference questions.

 required
options

Options dict mapping RDMO option URIs to string values.

 required

Returns:

Type Description

Dict {index: [option_uri, value]} for all present references;

empty dict if none are set.
Source code in MaRDMO/workflow/utils.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def get_reference(data, options):
    '''Build the reference dict for a data-set answer dict.

    Iterates over the predefined ``data_set_reference_ids`` keys and maps
    each present value to ``[option_uri, value]`` at a numeric index.

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

    Returns:
        Dict ``{index: [option_uri, value]}`` for all present references;
        empty dict if none are set.
    '''
    result = {}

    for idx, key in enumerate(data_set_reference_ids):
        if key in ('Yes', 'No'):
            if value := data.get('publish', {}).get('value') == key:
                result[idx] = [options[key], '']
        else:
            if value := data.get(key, {}).get('value'):
                result[idx] = [options[key], value]

    return result

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 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
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 []