Composite Service

Introduction

This document describes the backend service for Orikami's composites: entities that use data from two or more distinct experiment types. The IMWG Frailty Score is an example, combining the results of three distinct questionnaires and the patient's age to form a 'composite' result.

Interface

The Composite Service is implemented in Python and runs stand-alone in a Python interpreter. It communicates with the rest of the Krane framework using Apache pulsar messages. It uses PULSAR_TENANT/tasks/TENANT-events messages to register the start, completion or expiration of a new composite measurement and PULSAR_TENANT/experiments/TENANT-events messages to obtain result data from composing/underlying experiments. When all result data is available, a composite result is generated.

Since the time between scheduling the composite and receiving all of its composing experiment results may be quite long (days, maybe weeks?), it is possible that the Composite Service is stopped and restarted halfway processing a composite. Storage of a local state is prevented by only acknowledging task and experiment pulsar messages after completion or expiration of active composites. Pulsar messages related to experiments that do not belong to any composite are acknowledged as soon as related task and experiment messages are received.

Processed composites, containing score, interpretation and an error message if applicable, are published to the PULSAR_TENANT/experiments/TENANT-events topic.

Input

Expected format for json-formatted task messages:

Composite scheduling/creation:

    {
            'name': 'FutureTaskCreated',
            'version': 'v2',
            'tenant': 'orikami',
            'data': {
                'experimentType': 'composite',
                'featureId': 'IMWGFrailty',
                'resultId': None,
                'patientId': '6527a81258139c1e1494c4bc',
                'projectId': None,
                'scheduleTemplateId': None,
                'scheduleType': 'oneoff',
                'firstOccurrenceAt': '2023-10-13T08:51:55.206Z',
                'occurrenceId': 1,
                'startAt': '2023-10-13T08:51:55.206Z',
                'endAt': '2023-10-13T08:51:55.206Z'
            }
    }

Scheduling of a composing/underlying experiment:

    {
        'name': 'FutureTaskCreated',
        'version': 'v2',
        'tenant': 'orikami',
        'data': {
            'experimentType': 'questionnaire',
            'featureId': 'Katz-ADL',
            'resultId': None,
            'patientId': '6527a81258139c1e1494c4bc',
            'projectId': None,
            'scheduleTemplateId': None,
            'scheduleType': 'oneoff',
            'firstOccurrenceAt': '2023-10-13T08:51:55.206Z',
            'occurrenceId': 2,
            'startAt': '2023-10-13T08:51:55.206Z',
            'endAt': '2023-10-13T08:51:55.206Z'
        }
    }

Completion of a composing/underlying experiment, relating the experiment's occurrenceId to the result data's resultId:

    {
        'name': 'TaskCompleted',
        'version': 'v2',
        'tenant': 'orikami',
        'data': {
            'occurrenceId': 2,
            'resultId': '6529052bed80223630597ae0'
        }
    }

Expiration of an experiment:

    {
        'name': 'TaskNextOccurrenceExpired',
        'version': 'v2',
        'tenant': 'orikami',
        'data': {
            'occurrenceId': 1,
            'expiredAt': '2023-10-13T08:51:55.206Z'
    }

Experiment result data:

    {
        'name': 'QuestionnaireScoreGenerated',
        'version': 'v2',
        'tenant': 'orikami',
        'data':
        {
            'resultId': '6529052bed80223630597ae0',
            'score': 5,
            'interpretation': 'F'
        }
    }

Output

The algorithm's result is a dictionary in json format. It is published using Apache pulsar as described above, and has the following structure:

    {
        'name': 'CompositeScoreGenerated',
        'version': 'v2',
        'tenant': 'orikami',
        'data':
        {
            'resultId': '6529052bed80223630597ae0',
            'score': 5,
            'interpretation': 'F'
        }
    }

Use

The backend starts a pulsar client and keeps on listening and processing until it is terminated:

python3 -m src.main

NOTE that the pulsar client currently does NOT work in a Windows environment, so on a Windows machine WSL should be used. The code also requires a number of environment variables to be set in a .env file. An example .env file is included in this repository, containing the following variables:

    PULSAR_TOKEN=<token to be obtained from Orikami Group BV>
    PULSAR_URL=pulsar+ssl://pulsar.orikami.tech:6651
    PULSAR_TENANT=krane-staging

The token needs to obtained from Orikami Group BV.

Composite processing

Composites may perform a wide array of calculations on composing data. The src/composites folder contains configuration data and algorithms for composite result calculation. One Python file is used for each specific composite, and no other files need to be updated when adding, removing or renaming a composite file.

Testing

Pytest will run module tests (currently only a good weather scenario is tested):

python3 -m pytest

Note that these tests do not require pulsar running, since they use a mockup of a pulsar client and of the Experiment Service.

Sequence diagram