Skip to content

Validation Manager

Manager for data validation processes.

ValidationManager 

ValidationManager(parent=None)

Bases: QObject

Manages the validation process including starting, cancelling or getting the status of running validation tasks.

Source code in src/cplus_plugin/lib/validation/manager.py 
30
31
32
33
34
35
36
37
38
39
40
def __init__(self, parent=None):
    super().__init__(parent)

    # Data validator (value) indexed by task id (key)
    self._validation_tasks = {}

    # Validation results (value) indexed by task id (key)
    self._validation_results = {}

    self.task_manager = QgsApplication.instance().taskManager()
    self.task_manager.statusChanged.connect(self.on_validation_status_changed)

cancel 

cancel(result)

Cancels a validation process matching the result of the submission.

Parameters:

Name Type Description Default
result SubmitResult

Result of the validation submission.

 required

Returns:

Type Description
bool

True if the validation task was successfully cancelled else False if the submit results status was not successful or if the validation task was not found.
Source code in src/cplus_plugin/lib/validation/manager.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
def cancel(self, result: SubmitResult) -> bool:
    """Cancels a validation process matching the result of the submission.

    :param result: Result of the validation submission.
    :type result: SubmitResult

    :returns: True if the validation task was successfully cancelled
    else False if the submit results status was not successful or if
    the validation task was not found.
    :rtype: bool
    """
    if not result.success:
        return False

    if result.identifier not in self._validation_tasks:
        return False

    validator = self._validation_tasks[result.identifier]
    validator.cancel()

    del self._validation_tasks[result.identifier]

    return True

cancel_ncs_validation 

cancel_ncs_validation()

Cancel all validation processes of NCS pathway datasets.

Source code in src/cplus_plugin/lib/validation/manager.py 
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def cancel_ncs_validation(self):
    """Cancel all validation processes of NCS pathway datasets."""
    for task_id in list(self._validation_tasks):
        ncs_validator = self._validation_tasks[task_id]
        # In some OSs, the underlying C++ task object is de-referenced
        # unexpectedly hence the need to check if it exists.
        if sip.isdeleted(ncs_validator):
            continue

        status = ncs_validator.status()
        if (
            status == QgsTask.TaskStatus.Running
            or status == QgsTask.TaskStatus.Queued
            or status == QgsTask.TaskStatus.OnHold
        ):
            ncs_validator.cancel()
            del self._validation_tasks[task_id]

is_validation_complete 

is_validation_complete(result)

Checks whether the validation process, based on the given submission result, is complete.

Parameters:

Name Type Description Default
result SubmitResult

Result of the request for validation.

 required

Returns:

Type Description
bool

True if the validation process is complete, else False if it is still in progress or has failed.
Source code in src/cplus_plugin/lib/validation/manager.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def is_validation_complete(self, result: SubmitResult) -> bool:
    """Checks whether the validation process, based on the given
    submission result, is complete.

    :param result: Result of the request for validation.
    :type result: SubmitResult

    :returns: True if the validation process is complete, else
    False if it is still in progress or has failed.
    :rtype: bool
    """
    if not result.success:
        return False

    if result.identifier in self._validation_results:
        return True

    return False

last_ncs_result 

last_ncs_result()

Gets the result of the last successful validation of NCS pathways.

Returns:

Type Description
ValidationResult

Result of the last successful NCS pathway validation.
Source code in src/cplus_plugin/lib/validation/manager.py 
222
223
224
225
226
227
228
229
230
231
232
def last_ncs_result(self) -> typing.Union[ValidationResult, None]:
    """Gets the result of the last successful validation of NCS pathways.

    :returns: Result of the last successful NCS pathway validation.
    :rtype: ValidationResult
    """
    ncs_results = self.ncs_results()
    if len(ncs_results) == 0:
        return None

    return ncs_results[len(ncs_results) - 1]

ncs_results 

ncs_results()

Gets the validation results for NCS pathways.

Returns:

Type Description
list

A list containing the validation results for NCS pathways or an empty list if no results are found.
Source code in src/cplus_plugin/lib/validation/manager.py 
213
214
215
216
217
218
219
220
def ncs_results(self) -> typing.List[ValidationResult]:
    """Gets the validation results for NCS pathways.

    :returns: A list containing the validation results for NCS
    pathways or an empty list if no results are found.
    :rtype: list
    """
    return self.results_by_component_type(ModelComponentType.NCS_PATHWAY)

on_validation_status_changed 

on_validation_status_changed(task_id, status)

Slot raised when the status of a validation task has changed.

This function will emit when the validation task has started, when it has completed successfully or terminated due to an error.

Parameters:

Name Type Description Default
task_id int

ID of the task.

 required
status TaskStatus

New task status.

 required
Source code in src/cplus_plugin/lib/validation/manager.py 
 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
def on_validation_status_changed(self, task_id: int, status: QgsTask.TaskStatus):
    """Slot raised when the status of a validation task has changed.

    This function will emit when the validation task has started, when it
    has completed successfully or terminated due to an error.

    :param task_id: ID of the task.
    :type task_id: int

    :param status: New task status.
    :type status: QgsTask.TaskStatus
    """
    if status == QgsTask.TaskStatus.Running:
        self.validation_started.emit(str(task_id))

    elif status == QgsTask.TaskStatus.Complete:
        if str(task_id) not in self._validation_tasks:
            return

        # Get result
        validator = self.task_manager.task(task_id)
        result = validator.result
        if result is not None:
            self._validation_results[str(task_id)] = result

        # Remove task
        if str(task_id) in self._validation_tasks:
            del self._validation_tasks[str(task_id)]

        self.validation_completed.emit(str(task_id))

results_by_component_type 

results_by_component_type(component_type)

Gets the validation results based on the model component type of the validator.

Parameters:

Name Type Description Default
component_type ModelComponentType

Model component type e.g. NCS pathway.

 required

Returns:

Type Description
list

A list containing the validation results for the given model component type or an empty list if no results are found.
Source code in src/cplus_plugin/lib/validation/manager.py 
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
def results_by_component_type(
    self, component_type: ModelComponentType
) -> typing.List[ValidationResult]:
    """Gets the validation results based on the model component
    type of the validator.

    :param component_type: Model component type e.g. NCS pathway.
    :type component_type: ModelComponentType

    :returns: A list containing the validation results for the given
    model component type or an empty list if no results are found.
    :rtype: list
    """
    if component_type == ModelComponentType.UNKNOWN:
        return []

    return [
        result
        for result in self._validation_results.values()
        if result.component_type == component_type
    ]

validate_ncs_pathways 

validate_ncs_pathways(pathways, cancel_running=True)

Validates a set of NcsPathway datasets and returns the status of the submission.

Parameters:

Name Type Description Default
pathways List[NcsPathway]

A list of NcsPathway objects to be validated. At least one NcsPathway is required for the validation process to be executed.

 required
cancel_running bool

True if any running validation processes are to be cancelled else False if multiple concurrent validation processes are to be executed.

 True

Returns:

Type Description
SubmitResult

Result object containing the task id and status of the submission.
Source code in src/cplus_plugin/lib/validation/manager.py 
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
def validate_ncs_pathways(
    self, pathways: typing.List[NcsPathway], cancel_running=True
) -> SubmitResult:
    """Validates a set of NcsPathway datasets and returns the status
    of the submission.

    :param pathways: A list of NcsPathway objects to be validated. At least one
    NcsPathway is required for the validation process to be executed.
    :type pathways: list

    :param cancel_running: True if any running validation processes are to be cancelled
    else False if multiple concurrent validation processes are to be executed.
    :type cancel_running: bool

    :returns: Result object containing the task id and status of the submission.
    :rtype: SubmitResult
    """
    if len(pathways) == 0:
        return SubmitResult("", False, None)

    if cancel_running:
        self.cancel_ncs_validation()

    # Clone the pathways
    cloned_pathways = list(map(clone_ncs_pathway, pathways))

    ncs_validator = NcsDataValidator()
    ncs_validator.model_components = cloned_pathways
    task_id = self.task_manager.addTask(ncs_validator)

    if task_id == 0:
        return SubmitResult("", False, None)

    self._validation_tasks[str(task_id)] = ncs_validator

    return SubmitResult(str(task_id), True, ncs_validator.feedback)

validation_result 

validation_result(result)

Gets the validation result based on the given submission result.

Parameters:

Name Type Description Default
result SubmitResult

Result of the request for validation.

 required

Returns:

Type Description
ValidationResult

Validation result if the process was complete else None if the submission or actual validation process failed.
Source code in src/cplus_plugin/lib/validation/manager.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def validation_result(
    self, result: SubmitResult
) -> typing.Union[None, ValidationResult]:
    """Gets the validation result based on the given submission result.

    :param result: Result of the request for validation.
    :type result: SubmitResult

    :returns: Validation result if the process was complete else None
    if the submission or actual validation process failed.
    :rtype: ValidationResult
    """
    if not result.success:
        return None

    if result.identifier not in self._validation_results:
        return None

    return self._validation_results[result.identifier]
Last update: October 2, 2024
Back to top