Coverage for src/mafw/processor/base.py: 98%
718 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-26 09:13 +0000
1# Copyright 2025–2026 European Union
2# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
3# SPDX-License-Identifier: EUPL-1.2
4"""Core Processor class for the MAFw pipeline.
6This module defines the :class:`~mafw.processor.Processor` base class — the fundamental building
7block of every MAFw analysis pipeline. A Processor encapsulates a single
8processing step (reading data, applying corrections, fitting models, etc.) and
9can be orchestrated inside a :class:`~mafw.processor.ProcessorList`.
11Scientists subclass :class:`~mafw.processor.Processor` and override a small set of methods
12(:meth:`~mafw.processor.Processor.get_items`, :meth:`~mafw.processor.Processor.process`,
13:meth:`~mafw.processor.Processor.accept_item`) to define their analysis logic, while the
14framework manages looping, resource acquisition, parameter configuration, and
15progress reporting.
17.. versionadded:: 2.3
18 Extracted from the monolithic ``processor.py`` module for improved
19 maintainability and focused testing.
20"""
22from __future__ import annotations
24import contextlib
25import inspect
26import logging
27import os
28import queue
29import threading
30import time
31import warnings
32from collections import OrderedDict
33from collections.abc import Callable, Collection, Iterator
34from concurrent.futures import FIRST_COMPLETED, Future, ThreadPoolExecutor, wait
35from copy import copy, deepcopy
36from functools import wraps
37from itertools import count
38from typing import (
39 TYPE_CHECKING,
40 Any,
41 Union,
42 cast,
43 get_args,
44 get_origin,
45)
47import peewee
48from peewee import Database
50# noinspection PyUnresolvedReferences
51from playhouse.db_url import connect
53import mafw.db.db_filter
54from mafw.active import Active
55from mafw.db.db_connection import build_connection_parameters
56from mafw.db.db_model import MAFwBaseModel, database_proxy, mafw_model_register
57from mafw.enumerators import LoopingStatus, LoopType, ProcessorExitStatus, ProcessorStatus
58from mafw.mafw_errors import (
59 MissingDatabase,
60 MissingOverloadedMethod,
61 MissingSuperCall,
62 ProcessorParameterError,
63)
64from mafw.models.filter_schema import FilterSchema
65from mafw.models.loop_payloads import LoopItem, LoopResult
66from mafw.models.parameter_schema import ParameterSchema
67from mafw.models.processor_schema import ProcessorSchema
68from mafw.processor.meta import ProcessorMeta
69from mafw.processor.parameters import (
70 ParameterType,
71 PassiveParameter,
72 _ensure_parameter_definitions,
73 ensure_parameter_registration,
74)
75from mafw.processor.utils import validate_database_conf
76from mafw.timer import Timer, pretty_format_duration
77from mafw.tools.generics import deep_update
78from mafw.tools.parallel import is_free_threading
79from mafw.ui.abstract_user_interface import UserInterfaceBase
80from mafw.ui.console_user_interface import ConsoleInterface
82log = logging.getLogger(__name__)
85# noinspection PyProtectedMember
86class Processor(metaclass=ProcessorMeta):
87 """
88 The basic processor.
90 A very comprehensive description of what a Processor does and how it works is available at :ref:`doc_processor`.
92 """
94 processor_status = Active(ProcessorStatus.Unknown)
95 """Processor execution status"""
97 progress_message: str = f'{__qualname__} is working'
98 """Message displayed to show the progress.
100 It can be customized with information about the current item in the loop by overloading the
101 :meth:`format_progress_message`."""
103 #: List of methods that should invoke their super implementation when overridden.
104 _methods_to_be_checked_for_super: tuple[str, ...] = ('start', 'finish')
106 @classmethod
107 def parameter_schema(cls) -> list[ParameterSchema]:
108 """
109 Return the ordered static schema for the processor parameters defined on the class.
111 The schema is derived without instantiating the processor, keeping toolchains free from side effects.
112 """
113 definitions = getattr(cls, '_parameter_definitions', None)
114 if definitions is None:
115 definitions = _ensure_parameter_definitions(cls)
116 return [param.to_schema() for param in definitions.values()]
118 @classmethod
119 def filter_schema(cls) -> FilterSchema | None:
120 """
121 Optional metadata describing the models available for filtering.
122 """
123 return getattr(cls, '_filter_schema', None)
125 @classmethod
126 def processor_schema(cls) -> ProcessorSchema:
127 """
128 Return the static schema for the processor.
129 """
130 return ProcessorSchema(parameters=cls.parameter_schema(), filter=cls.filter_schema())
132 _ids = count(0)
133 """A counter for all processor instances"""
135 new_defaults: dict[str, Any] = {}
136 """
137 A dictionary containing defaults value for the parameters to be overridden
139 .. versionadded:: v2.0
140 """
142 new_only_flag = 'new_only'
144 def __init__(
145 self,
146 name: str | None = None,
147 description: str | None = None,
148 config: dict[str, Any] | None = None,
149 looper: LoopType | str = LoopType.ForLoop,
150 user_interface: UserInterfaceBase | None = None,
151 timer: Timer | None = None,
152 timer_params: dict[str, Any] | None = None,
153 database: Database | None = None,
154 database_conf: dict[str, Any] | None = None,
155 remove_orphan_files: bool = True,
156 replica_id: str | None = None,
157 create_standard_tables: bool = True,
158 max_workers: int | None = None,
159 queue_size: int | None = None,
160 queue_batch_size: int | None = None,
161 *args: Any,
162 **kwargs: Any,
163 ) -> None:
164 """
165 Constructor parameters
167 :param name: The name of the processor. If None is provided, the class name is used instead. Defaults to None.
168 :type name: str, Optional
169 :param description: A short description of the processor task. Defaults to the processor name.
170 :type description: str, Optional
171 :param config: A configuration dictionary for this processor. Defaults to None.
172 :type config: dict, Optional
173 :param looper: Enumerator to define the looping type. Defaults to LoopType.ForLoop
174 :type looper: LoopType, Optional
175 :param user_interface: A user interface instance to be used by the processor to interact with the user.
176 :type user_interface: UserInterfaceBase, Optional
177 :param timer: A timer object to measure process duration.
178 :type timer: Timer, Optional
179 :param timer_params: Parameters for the timer object.
180 :type timer_params: dict, Optional
181 :param database: A database instance. Defaults to None.
182 :type database: Database, Optional
183 :param database_conf: Configuration for the database. Default to None.
184 :type database_conf: dict, Optional
185 :param remove_orphan_files: Boolean flag to remove files on disc without a reference to the database.
186 See :ref:`std_tables` and :meth:`~mafw.processor.Processor._remove_orphan_files`. Defaults to True
187 :type remove_orphan_files: bool, Optional
188 :param replica_id: The replica identifier for the current processor.
189 :type replica_id: str, Optional
190 :param create_standard_tables: Boolean flag to create std tables on disk. Defaults to True
191 When a nested steering configuration is loaded, this value can be overridden by the
192 global create_standard_tables entry. Flat processor configurations keep the constructor value.
193 :type create_standard_tables: bool, Optional
194 :param max_workers: Number of worker threads for parallel loops.
195 :type max_workers: int, Optional
196 :param queue_size: Maximum size of the internal queue for the queue-based parallel loop.
197 :type queue_size: int, Optional
198 :param queue_batch_size: Number of items processed per worker task in the queue-based parallel loop.
199 :type queue_batch_size: int, Optional
200 :param kwargs: Keyword arguments that can be used to set processor parameters.
201 """
203 self.name = name or self.__class__.__name__
204 """The name of the processor."""
206 self.unique_id = next(self._ids)
207 """A unique identifier representing how many instances of Processor has been created."""
209 self.replica_id = replica_id
210 """
211 The replica identifier specified in the constructor
213 .. versionadded:: v2.0.0
214 """
216 self.description = description or self.name
217 """A short description of the processor task."""
219 self._item: Any = None
220 """The current item of the loop."""
222 self._looping_status: LoopingStatus = LoopingStatus.Continue
223 """The looping status for the main thread."""
225 self._thread_local = threading.local()
226 """Thread-local storage for loop attributes in parallel execution.
228 .. versionadded:: v2.1.0
229 """
231 self._wall_clock_start: float | None = None
232 """Timestamp when the looping execution began."""
234 self.processor_exit_status = ProcessorExitStatus.Successful
235 """Processor exit status"""
237 self.loop_type: LoopType = LoopType(looper)
238 """
239 The loop type.
241 The value of this parameter can also be changed by the :func:`~mafw.decorators.execution_workflow` decorator
242 factory.
244 See :class:`~mafw.enumerators.LoopType` for more details.
245 """
247 self.create_standard_tables = create_standard_tables
248 """The boolean flag to proceed or skip with standard table creation and initialisation"""
250 self.max_workers = max_workers if max_workers is not None else self._compute_default_max_workers()
251 """Maximum number of worker threads used in parallel loops."""
253 computed_queue_size = max(1, self.max_workers * 2)
254 self.queue_size = queue_size if queue_size is not None else computed_queue_size
255 """Maximum size of the queue used by :class:`~mafw.enumerators.LoopType.ParallelForLoopWithQueue`."""
257 self.queue_batch_size = max(1, queue_batch_size or 1)
258 """Number of items processed per worker task in :class:`~mafw.enumerators.LoopType.ParallelForLoopWithQueue`."""
260 # private attributes
261 self._config: dict[str, Any] = {}
262 """
263 A dictionary containing the processor configuration object.
265 This dictionary is populated with configuration parameter (always type 2) during the
266 :meth:`._load_parameter_configuration` method.
268 The original value of the configuration dictionary that is passed to the constructor is stored in
269 :attr:`._orig_config`.
271 .. versionchanged:: v2.0.0
272 Now it is an empty dictionary until the :meth:`._load_parameter_configuration` is called.
274 """
276 self._orig_config = deepcopy(config) if config is not None else {}
277 """
278 A copy of the original configuration dictionary.
280 .. versionadded:: v2.0.0
281 """
283 self._processor_parameters: OrderedDict[str, PassiveParameter[Any]] = OrderedDict()
284 """
285 A dictionary to store all the processor parameter instances.
287 The name of the parameter is used as a key, while for the value an instance of the
288 :class:`.processor.PassiveParameter` is used.
289 """
290 self._parameter_registered = False
291 """A boolean flag to confirm successful parameter registration."""
292 self._kwargs = kwargs
294 # loops attributes
295 self._i_item: int = -1
296 self._n_item: int | None = -1
297 self._process_durations: list[float] = []
298 self._super_call_flags: dict[str, bool] = {}
299 """Tracks super-call usage for methods that require it."""
301 # resource stack
302 self._resource_stack: contextlib.ExitStack
303 self._resource_acquisition: bool = True
305 # processor timer
306 self.timer: Timer | None = timer
307 self._timer_parameters: dict[str, Any] = timer_params or {}
309 # user interface
310 if user_interface is None:
311 self._user_interface: UserInterfaceBase = ConsoleInterface()
312 else:
313 self._user_interface = user_interface
315 # database stuff
316 self._database: peewee.Database | None = database
317 self._database_conf: dict[str, Any] | None = validate_database_conf(database_conf)
318 self.filter_register: mafw.db.db_filter.ProcessorFilter = mafw.db.db_filter.ProcessorFilter()
319 """The DB filter register of the Processor."""
320 self.remove_orphan_files: bool = remove_orphan_files
321 """The flag to remove or protect the orphan files. Defaults to True"""
323 self.initialise_parameters()
325 def initialise_parameters(self) -> None:
326 """
327 Initialises processor parameters by registering them and applying various configuration sources.
329 This method orchestrates the parameter initialisation process by performing the following steps in order:
331 #. Registers processor parameters defined as :class:`.processor.ActiveParameter` instances
332 #. Overrides default parameter values with any configured overrides
333 #. Loads parameter configuration from the processor's configuration dictionary
334 #. Applies keyword arguments as parameter overrides
336 The method ensures that all processor parameters are properly configured before the processor
337 execution begins. It is automatically called during processor initialisation and should not
338 typically be called directly by users.
340 .. seealso::
341 :meth:`_register_parameters`, :meth:`_override_defaults`,
342 :meth:`_load_parameter_configuration`, :meth:`_overrule_kws_parameters`
344 .. versionadded:: v2.0.0
345 """
346 self._register_parameters()
347 self._override_defaults()
348 self._load_parameter_configuration()
349 self._overrule_kws_parameters()
351 def __post_init__(self) -> None:
352 """
353 Performs post-initialisation tasks for the processor.
355 This method is automatically called after the processor initialisation is complete.
356 It performs validation checks on overloaded methods and sets the initial processor status.
358 .. seealso::
359 :meth:`validate_configuration`, :meth:`_check_method_overload`,
360 :attr:`~mafw.processor.Processor.processor_status`
362 .. versionchanged:: v2.0.0
363 Moved the parameter initialisation to :meth:`initialise_parameters` and now executed as last step of the
364 init method.
366 Added the validate configuration check. This method should silently check that configuration provided
367 with the processor parameters is valid. If not, a :exc:`.ProcessorParameterError` is raised.
368 """
369 self.validate_configuration()
370 self._check_method_overload()
371 self.processor_status = ProcessorStatus.Init
373 def _register_parameters(self) -> None:
374 """
375 Register processor parameters defined as ActiveParameter instances in the class.
377 This private method scans the class definition for any :class:`.processor.ActiveParameter` instances and creates
378 corresponding :class:`.processor.PassiveParameter` instances to store the actual parameter values and metadata.
379 It ensures that all processor parameters are properly initialised and available for configuration
380 through the processor's configuration system.
382 The method checks for duplicate parameter names and raises a :exc:`.ProcessorParameterError` if duplicates
383 are detected. It also sets the internal flag :attr:`.processor.Processor._parameter_registered` to True once
384 registration is complete.
386 .. note::
387 This method is automatically called during processor initialisation and should not be called directly
388 by users.
390 .. seealso::
391 :class:`.processor.Processor`, :meth:`.processor.Processor._override_defaults`,
392 :meth:`.processor.Processor._load_parameter_configuration`, :meth:`.processor.Processor._overrule_kws_parameters`
394 .. versionchanged:: v2.0.0
395 Only :class:`.processor.ActiveParameter` are not registered. The use of
396 :class:`.processor.PassiveParameter` is only meant to store the value and metadata of the active
397 counterpart.
398 """
399 if self._parameter_registered:
400 return
402 definitions = getattr(self.__class__, '_parameter_definitions', None)
403 if definitions is None:
404 definitions = _ensure_parameter_definitions(self.__class__)
406 for attr in definitions.values():
407 ext_name = attr._external_name
408 if ext_name in self._processor_parameters:
409 raise ProcessorParameterError(f'Duplicated parameter name ({ext_name}).')
410 self._processor_parameters[ext_name] = PassiveParameter(
411 ext_name, attr._value, attr._default, attr._help_doc
412 )
414 self._parameter_registered = True
416 def _override_defaults(self) -> None:
417 """
418 Override default parameter values with values from :attr:`new_defaults`.
420 This private method iterates through the :attr:`new_defaults` dictionary and updates
421 the corresponding processor parameters with new values. Only parameters that exist
422 in both :attr:`new_defaults` and :attr:`_processor_parameters` are updated.
424 .. versionadded:: v2.0.0
425 """
426 for key, value in self.new_defaults.items():
427 if key in self._processor_parameters:
428 self._processor_parameters[key].value = value
430 def _reset_parameters(self) -> None:
431 """
432 Reset processor parameters to their initial state.
434 This method clears all currently registered processor parameters and triggers
435 a fresh registration process. It's useful when parameter configurations need
436 to be reinitialized or when parameters have been modified and need to be reset.
438 .. seealso::
439 :meth:`_register_parameters`, :meth:`_register_parameters`
440 """
441 self._processor_parameters = OrderedDict()
442 self._parameter_registered = False
443 self._register_parameters()
445 @ensure_parameter_registration
446 def _load_parameter_configuration(self) -> None:
447 """
448 Load processor parameter configuration from the internal configuration dictionary.
450 This method processes the processor's configuration dictionary to set parameter values.
451 It handles two configuration formats:
453 1. Nested format: ``{'ProcessorName': {'param1': value1, ...}}``
454 2. Flat format: ``{'param1': value1, ...}``
456 The method also handles filter configurations by collecting filter table names
457 and deferring their initialisation until after the global filter has been processed.
459 .. versionchanged:: v2.0.0
460 For option 1 combining configuration from name and name_replica
462 .. versionchanged:: v2.1.2
463 When a nested steering configuration is loaded, the processor-level
464 create_standard_tables value is overridden from the global steering-file setting.
465 Flat configurations keep the constructor value untouched.
467 :raises ProcessorParameterError: If a parameter in the configuration is not registered.
469 .. seealso::
470 :meth:`mafw.db.db_filter.ModelFilter.from_conf`
471 """
472 original_config = copy(self._orig_config)
473 flt_list = []
475 # by default the flag new_only is set to true
476 # unless the user specify differently in the general section of the steering file
477 self.filter_register.new_only = original_config.get(self.new_only_flag, True)
479 # we need to check if the configuration object is of type 1 or type 2
480 if any([name for name in [self.name, self.replica_name] if name in original_config]):
481 # one of the two names (the base or the replica) must be present in case of option 1
482 # we start from the base name. If not there, then take an empty dict
483 option1_config_base = original_config.get(self.name, {})
484 if self.name != self.replica_name:
485 # if there is the replica name, then update the base configuration with the replica value
486 # we get the replica configuration
487 option1_config_replica = original_config.get(self.replica_name, {})
489 # let's check if the user wants to have inheritance default
490 # by default is True
491 inheritance = option1_config_replica.get('__inheritance__', True)
492 if inheritance:
493 # we update the base with the replica without changing the base
494 option1_config_update = deep_update(option1_config_base, option1_config_replica, copy_first=True)
495 else:
496 # we do not use the base with the replica specific, we pass the replica as the updated
497 option1_config_update = option1_config_replica
499 # we modify the type 1 original so that the table for the replica has the updated configuration
500 # this is used for the filter configuration at the end.
501 original_config[self.replica_name] = option1_config_update
502 else:
503 # there is not replica, so the update is equal to the base.
504 option1_config_update = option1_config_base
506 self._config = option1_config_update
507 if 'create_standard_tables' in original_config:
508 self.create_standard_tables = bool(original_config['create_standard_tables'])
509 else:
510 # for type 2 we are already good to go
511 self._config = original_config
513 filter_config = deepcopy(original_config)
515 def _sanitize_filter_config(processor_name: str) -> None:
516 processor_config = filter_config.get(processor_name)
517 if not isinstance(processor_config, dict):
518 return
519 filter_table = processor_config.get('__filter__')
520 if not isinstance(filter_table, dict):
521 return
523 sanitized_table: dict[str, Any] = {}
524 for model_name, model_config in filter_table.items():
525 if not isinstance(model_config, dict):
526 sanitized_table[model_name] = model_config
527 continue
529 model_config_copy = deepcopy(model_config)
530 if not bool(model_config_copy.pop('__enable__', True)):
531 continue
533 for field_name, field_value in list(model_config_copy.items()):
534 if (
535 isinstance(field_value, dict)
536 and not ('op' in field_value and 'value' in field_value)
537 and '__enable__' in field_value
538 ):
539 field_enabled = bool(field_value.pop('__enable__', True))
540 if not field_enabled: 540 ↛ 533line 540 didn't jump to line 533 because the condition on line 540 was always true
541 model_config_copy.pop(field_name, None)
543 conditionals = model_config_copy.get('__conditional__')
544 if isinstance(conditionals, list):
545 filtered_conditionals: list[Any] = []
546 for conditional in conditionals:
547 if isinstance(conditional, dict):
548 conditional_enabled = bool(conditional.pop('__enable__', True))
549 if not conditional_enabled:
550 continue
551 filtered_conditionals.append(conditional)
552 model_config_copy['__conditional__'] = filtered_conditionals
554 sanitized_table[model_name] = model_config_copy
556 processor_config['__filter__'] = sanitized_table
558 _sanitize_filter_config(self.replica_name)
560 for key, value in self._config.items():
561 if key in self._processor_parameters:
562 type_: ParameterType = type(self.get_parameter(key).value) # type: ignore[valid-type] # TypeVar used at runtime for dynamic type inference
563 self.set_parameter_value(key, type_(value)) # type: ignore[misc] # runtime call on dynamically-inferred type
564 elif key == '__filter__':
565 # we got a filter table!
566 # it should contain one table for each model
567 # we add all the names to a list for deferred initialisation
568 flt_table = self._config[key]
569 if isinstance(flt_table, dict):
570 for model_name, model_config in flt_table.items():
571 if isinstance(model_config, dict) and not bool(model_config.get('__enable__', True)):
572 continue
573 flt_list.append(f'{self.replica_name}.__filter__.{model_name}')
574 elif key == '__logic__':
575 # we got a filter logic string
576 # we store it in the filter register directly
577 self.filter_register._logic = self._config[key]
578 elif key == '__new_only__':
579 # we got a new only boolean, we store it in the filter register
580 self.filter_register.new_only = self._config[key]
582 # only now, after the configuration file has been totally read, we can do the real filter initialisation.
583 # This is to be sure that if there were a GlobalFilter table, this has been read.
584 # The global filter region will be used as a starting point for the construction of a new filter (default
585 # parameter in the from_conf class method).
586 for flt_name in flt_list:
587 model_name = flt_name.split('.')[-1]
588 self.filter_register[model_name] = mafw.db.db_filter.ModelFilter.from_conf(flt_name, filter_config)
590 @ensure_parameter_registration
591 def _overrule_kws_parameters(self) -> None:
592 """
593 Override processor parameters with values from keyword arguments.
595 This method applies parameter values passed as keyword arguments during processor
596 initialisation. It ensures that the parameter types match the expected types
597 before setting the values.
599 .. seealso::
600 :meth:`_register_parameters`, :meth:`_load_parameter_configuration`,
601 :meth:`set_parameter_value`
602 """
603 for key, value in self._kwargs.items():
604 if key in self._processor_parameters:
605 type_: ParameterType = type(self.get_parameter(key).value) # type: ignore[valid-type] # TypeVar used at runtime for dynamic type inference
606 self.set_parameter_value(key, type_(value)) # type: ignore[misc] # runtime call on dynamically-inferred type
608 def validate_configuration(self) -> None:
609 """
610 Validate the configuration provided via the processor parameters.
612 Method to be implemented by subclasses if a configuration validation is needed.
614 The method should silently check for the proper configuration, if this is not obtained,
615 then the :exc:`.InvalidConfigurationError` must be raised.
617 .. versionadded:: v2.0.0
618 """
619 pass
621 def _check_method_overload(self) -> None:
622 """
623 Check if the user overloaded the required methods.
625 Depending on the loop type, the user must overload different methods.
626 This method is doing the check and if the required methods are not overloaded a warning is emitted.
627 """
628 methods_dict: dict[LoopType, list[str]] = {
629 LoopType.WhileLoop: ['while_condition'],
630 LoopType.ForLoop: ['get_items'],
631 LoopType.ParallelForLoop: ['get_items'],
632 LoopType.ParallelForLoopWithQueue: ['get_items'],
633 }
634 required_methods: list[str] = methods_dict.get(self.loop_type, [])
635 for method in required_methods:
636 if getattr(type(self), method) == getattr(Processor, method):
637 warnings.warn(
638 MissingOverloadedMethod(
639 '%s was not overloaded. The process execution workflow might not work.' % method
640 )
641 )
643 @classmethod
644 def _apply_super_call_wrappers(cls) -> None:
645 """
646 Wraps overridden methods so the class can detect whether they called `super()`.
648 This method runs immediately after the class is created (see :class:`.processor.ProcessorMeta`). For every
649 method that we expect scientists to extend (start, finish, etc.) we replace their implementation with a
650 wrapper. The wrapper resets a per-instance flag, invokes the real override, and only after that method returns
651 it checks whether the real `super()` was ever reached; if not, it emits a :class:`~mafw.mafw_errors.MissingSuperCall`
652 warning. In other words, the wiring happens while the subclass is defined, and the actual smoke test executes
653 each time the method runs.
654 """
655 methods = getattr(cls, '_methods_to_be_checked_for_super', ())
656 for method in methods:
657 if method not in cls.__dict__:
658 continue
659 if not any(hasattr(base, method) for base in cls.__mro__[1:]):
660 continue
661 original: Callable[..., Any] = getattr(cls, method)
662 # we are adding an attribute to the method, it looks strange, but it is possible
663 # in this way we avoid wrapping the same method more than once.
664 if getattr(original, '_mafw_super_check_wrapped', False):
665 continue
667 def _make_wrapper(
668 __orig: Callable[..., Any],
669 __method: str, # pragma: no cover
670 ) -> Callable[..., Any]:
671 @wraps(__orig)
672 def _wrapper(self: Any, *args: Any, **kwargs: Any) -> Any:
673 # self is the processor instance.
674 # _reset_super_call_flag is resetting the call status
675 # for method as False
676 self._reset_super_call_flag(__method)
677 # in the base method, the super call flag is set to True
678 result = __orig(self, *args, **kwargs)
679 # if the super call flag is not True, then it is because the base
680 # method was not called.
681 # emit the warning and return the original method return value
682 if not self._did_call_super(__method):
683 warnings.warn(
684 MissingSuperCall(
685 'The overloaded %s is not invoking its super method. The processor might not work.'
686 % __method
687 )
688 )
689 return result
691 return _wrapper
693 # we create a wrapped method from the original
694 wrapper = _make_wrapper(original, method)
695 # we set a flag for the method to avoid multiple wrapping
696 wrapper._mafw_super_check_wrapped = True # type: ignore[attr-defined]
697 setattr(cls, method, wrapper)
699 def _reset_super_call_flag(self, method: str) -> None:
700 """
701 Reset the super-call flag for a method.
702 """
703 self._super_call_flags[method] = False
705 def _mark_super_call(self, method: str) -> None:
706 """
707 Mark a method as having called its super implementation.
708 """
709 self._super_call_flags[method] = True
711 def _did_call_super(self, method: str) -> bool:
712 """
713 Check whether a method called its super implementation.
714 """
715 return self._super_call_flags.get(method, False)
717 @ensure_parameter_registration
718 def dump_parameter_configuration(self, option: int = 1) -> dict[str, Any]:
719 """
720 Dumps the processor parameter values in a dictionary.
722 The snippet below explains the meaning of `option`.
724 .. code-block:: python
726 # option 1
727 conf_dict1 = {
728 'Processor': {'param1': 5, 'input_table': 'my_table'}
729 }
731 # option 2
732 conf_dict2 = {'param1': 5, 'input_table': 'my_table'}
734 In the case of option 1, the replica aware name (:meth:`.replica_name`) will be used as a key for the
735 configuration dictionary.
737 .. versionchanged:: v2.0.0
738 With option 1, using :meth:`.replica_name` instead of :attr:`~.processor.Processor.name` as key of the configuration
739 dictionary.
741 :param option: Select the dictionary style. Defaults to 1.
742 :type option: int, Optional
743 :return: A parameter configuration dictionary.
744 :rtype: dict
745 """
746 inner_dict = {}
747 for key, value in self._processor_parameters.items():
748 inner_dict[key] = value.value
750 if option == 1:
751 outer_dict = {self.replica_name: inner_dict}
752 elif option == 2:
753 outer_dict = inner_dict
754 else:
755 log.warning('Unknown option %s. Using option 2' % option)
756 outer_dict = inner_dict
757 return outer_dict
759 @ensure_parameter_registration
760 def get_parameter(self, name: str) -> PassiveParameter[Any]:
761 """
762 Gets the processor parameter named name.
764 :param name: The name of the parameter.
765 :type name: str
766 :return: The processor parameter
767 :rtype: processor.PassiveParameter
768 :raises ProcessorParameterError: If a parameter with `name` is not registered.
769 """
770 if name in self._processor_parameters:
771 return self._processor_parameters[name]
772 raise ProcessorParameterError(f'No parameter ({name}) found for {self.name}')
774 @ensure_parameter_registration
775 def get_parameters(self) -> dict[str, PassiveParameter[Any]]:
776 """
777 Returns the full dictionary of registered parameters for this processor.
779 Useful when dumping the parameter specification in a configuration file, for example.
781 :return: The dictionary with the registered parameters.
782 :rtype: dict[str, processor.PassiveParameter[ParameterType]
783 """
784 return self._processor_parameters
786 @ensure_parameter_registration
787 def delete_parameter(self, name: str) -> None:
788 """
789 Deletes a processor parameter.
791 :param name: The name of the parameter to be deleted.
792 :type name: str
793 :raises ProcessorParameterError: If a parameter with `name` is not registered.
794 """
795 if name in self._processor_parameters:
796 del self._processor_parameters[name]
797 else:
798 raise ProcessorParameterError(f'No parameter ({name}) found for {self.name}')
800 @ensure_parameter_registration
801 def set_parameter_value(self, name: str, value: ParameterType) -> None:
802 """
803 Sets the value of a processor parameter.
805 :param name: The name of the parameter to be deleted.
806 :type name: str
807 :param value: The value to be assigned to the parameter.
808 :type value: ParameterType
809 :raises ProcessorParameterError: If a parameter with `name` is not registered.
810 """
811 if name in self._processor_parameters:
812 self._processor_parameters[name].value = value
813 else:
814 raise ProcessorParameterError(f'No parameter ({name}) found for {self.name}')
816 def get_filter(self, model_name: str) -> mafw.db.db_filter.ModelFilter:
817 """
818 Returns a registered :class:`~mafw.db.db_filter.ModelFilter` via the model name.
820 If a filter for the provided model_name does not exist, a KeyError is raised.
822 :param model_name: The model name for which the filter will be returned.
823 :type model_name: str
824 :return: The registered filter
825 :rtype: mafw.db.db_filter.ModelFilter
826 :raises: KeyError is a filter with the give name is not found.
827 """
828 return self.filter_register[model_name]
830 def on_processor_status_change(self, old_status: ProcessorStatus, new_status: ProcessorStatus) -> None:
831 """
832 Callback invoked when the processor status is changed.
834 :param old_status: The old processor status.
835 :type old_status: ProcessorStatus
836 :param new_status: The new processor status.
837 :type new_status: ProcessorStatus
838 """
839 self._user_interface.change_of_processor_status(self.name, old_status, new_status)
841 def on_looping_status_set(self, status: LoopingStatus) -> None:
842 """
843 Call back invoked when the looping status is set.
845 The user can overload this method according to the needs.
847 :param status: The set looping status.
848 :type status: LoopingStatus
849 """
850 if status == LoopingStatus.Skip:
851 log.warning('Skipping item %s' % self.i_item)
852 elif status == LoopingStatus.Abort:
853 log.error('Looping has been aborted')
854 elif status == LoopingStatus.Quit:
855 log.warning('Looping has been quit')
857 def format_progress_message(self) -> None:
858 """Customizes the progress message with information about the current item.
860 The user can overload this method in order to modify the message being displayed during the process loop with
861 information about the current item.
863 The user can access the current value, its position in the looping cycle and the total number of items using
864 :attr:`.processor.Processor.item`, :obj:`.processor.Processor.i_item` and :obj:`.processor.Processor.n_item`.
865 """
866 pass
868 @contextlib.contextmanager
869 def _thread_loop_context(self, i_item: int, n_item: int, item: Any) -> Iterator[None]:
870 """
871 Context manager to set thread-local loop attributes for parallel execution.
873 :param i_item: Item index for the loop.
874 :type i_item: int
875 :param n_item: Total number of items in the loop.
876 :type n_item: int
877 :param item: Item payload.
878 :type item: Any
879 """
880 self._thread_local.in_worker = True
881 self._thread_local.i_item = i_item
882 self._thread_local.n_item = n_item
883 self._thread_local.item = item
884 self._thread_local.looping_status = LoopingStatus.Continue
885 try:
886 yield
887 finally:
888 for name in ('i_item', 'n_item', 'item', 'looping_status', 'in_worker'):
889 if hasattr(self._thread_local, name): 889 ↛ 888line 889 didn't jump to line 888 because the condition on line 889 was always true
890 delattr(self._thread_local, name)
892 def _in_thread_context(self) -> bool:
893 """Return True when running inside a parallel worker thread."""
894 return bool(getattr(self._thread_local, 'in_worker', False))
896 @property
897 def item(self) -> Any:
898 """The current item of the loop."""
899 if self._in_thread_context() and hasattr(self._thread_local, 'item'):
900 return self._thread_local.item
901 return self._item
903 @item.setter
904 def item(self, value: Any) -> None:
905 if self._in_thread_context():
906 self._thread_local.item = value
907 else:
908 self._item = value
910 @property
911 def i_item(self) -> int:
912 """The enumeration of the current item being processed."""
913 if self._in_thread_context() and hasattr(self._thread_local, 'i_item'):
914 return cast(int, self._thread_local.i_item)
915 return self._i_item
917 @i_item.setter
918 def i_item(self, value: int) -> None:
919 if self._in_thread_context():
920 self._thread_local.i_item = value
921 else:
922 self._i_item = value
924 @property
925 def n_item(self) -> int | None:
926 """The total number of items to be processed or None for an undefined loop"""
927 if self._in_thread_context() and hasattr(self._thread_local, 'n_item'):
928 return cast(int | None, self._thread_local.n_item)
929 return self._n_item
931 @n_item.setter
932 def n_item(self, value: int | None) -> None:
933 if self._in_thread_context():
934 self._thread_local.n_item = value
935 else:
936 self._n_item = value
938 @property
939 def looping_status(self) -> LoopingStatus:
940 """The looping status for the current thread context."""
941 if self._in_thread_context():
942 value = getattr(self._thread_local, 'looping_status', LoopingStatus.Continue)
943 else:
944 value = self._looping_status
945 if hasattr(self, 'on_looping_status_get'):
946 self.on_looping_status_get(value)
947 return value
949 @looping_status.setter
950 def looping_status(self, value: LoopingStatus) -> None:
951 if self._in_thread_context():
952 current = getattr(self._thread_local, 'looping_status', LoopingStatus.Continue)
953 self._thread_local.looping_status = value
954 else:
955 current = self._looping_status
956 self._looping_status = value
957 if current != value:
958 if hasattr(self, 'on_looping_status_change'):
959 self.on_looping_status_change(current, value)
960 else:
961 if hasattr(self, 'on_looping_status_set'): 961 ↛ exitline 961 didn't return from function 'looping_status' because the condition on line 961 was always true
962 self.on_looping_status_set(value)
964 @property
965 def unique_name(self) -> str:
966 """Returns the unique name for the processor."""
967 return f'{self.name}_{self.unique_id}'
969 @property
970 def replica_name(self) -> str:
971 """
972 Returns the replica aware name of the processor.
974 If no replica_id is specified, then return the pure name, otherwise join the two string using the '#' symbol.
976 .. versionadded:: v2.0.0
978 :return: The replica aware name of the processor.
979 :rtype: str
980 """
981 if self.replica_id is None:
982 return self.name
983 else:
984 return self.name + '#' + self.replica_id
986 @property
987 def local_resource_acquisition(self) -> bool:
988 """
989 Checks if resources should be acquired locally.
991 When the processor is executed in stand-alone mode, it is responsible to acquire and release its own external
992 resources, but when it is executed from a ProcessorList, then is a good practice to share and distribute
993 resources among the whole processor list. In this case, resources should not be acquired locally by the
994 single processor, but from the parent execution context.
996 :return: True if resources are to be acquired locally by the processor. False, otherwise.
997 :rtype: bool
998 """
999 return self._resource_acquisition
1001 @local_resource_acquisition.setter
1002 def local_resource_acquisition(self, flag: bool) -> None:
1003 self._resource_acquisition = flag
1005 @property
1006 def database(self) -> peewee.Database:
1007 """
1008 Returns the database instance
1010 :return: A database object.
1011 :raises MissingDatabase: If the database connection has not been established.
1012 """
1013 if self._database is None:
1014 raise MissingDatabase('Database connection not initialized')
1015 return self._database
1017 def execute(self) -> None:
1018 """Execute the processor tasks.
1020 This method works as a dispatcher, reassigning the call to a more specific execution implementation depending
1021 on the :attr:`~mafw.processor.Processor.loop_type`.
1022 """
1023 dispatcher: dict[LoopType, Callable[[], None]] = {
1024 LoopType.SingleLoop: self._execute_single,
1025 LoopType.ForLoop: self._execute_for_loop,
1026 LoopType.ParallelForLoop: self._execute_for_loop,
1027 LoopType.ParallelForLoopWithQueue: self._execute_for_loop,
1028 LoopType.WhileLoop: self._execute_while_loop,
1029 }
1030 dispatcher[self.loop_type]()
1032 @staticmethod
1033 def _compute_default_max_workers() -> int:
1034 """
1035 Helper to compute the default number of workers for parallel execution.
1037 Returns min(32, cpu_count + 4).
1038 """
1039 return min(32, (os.cpu_count() or 1) + 4)
1041 def _execute_single(self) -> None:
1042 """Execute the processor in single mode.
1044 **Private method**. Do not overload nor invoke it directly. The :meth:`execute` method will call the
1045 appropriate implementation depending on the processor LoopType.
1046 """
1047 with contextlib.ExitStack() as self._resource_stack:
1048 self.acquire_resources()
1049 self._wall_clock_start = time.perf_counter()
1050 self.start()
1051 self.processor_status = ProcessorStatus.Run
1052 self.process()
1053 self.finish()
1055 def _execute_for_loop(self) -> None:
1056 """Executes the processor within a for loop.
1058 **Private method**. Do not overload nor invoke it directly. The :meth:`execute` method will call the
1059 appropriate implementation depending on the processor LoopType.
1060 """
1062 with contextlib.ExitStack() as self._resource_stack:
1063 self.acquire_resources()
1064 # we cannot use a Timer context here to measure the whole duration because it spans
1065 # over different methods. Instead we directly use a performance clock.
1066 self._wall_clock_start = time.perf_counter()
1067 self.start()
1069 # get the input item list and filter it
1070 item_list = self.get_items()
1072 # get the total number of items.
1073 self.n_item = len(item_list)
1075 # turn the processor status to run
1076 self.processor_status = ProcessorStatus.Run
1078 # create a new task in the progress bar interface
1079 self._user_interface.create_task(self.replica_name, self.description, completed=0, total=self.n_item)
1081 # verify if we can use parallel for loop. If not, switch back to a serial for loop.
1082 if (
1083 self.loop_type in (LoopType.ParallelForLoop, LoopType.ParallelForLoopWithQueue)
1084 and not is_free_threading()
1085 ):
1086 warnings.warn(
1087 'Parallel for-loop requires free-threading; falling back to serial for loop.',
1088 stacklevel=2,
1089 )
1090 self.loop_type = LoopType.ForLoop
1092 if self.loop_type == LoopType.ParallelForLoopWithQueue:
1093 self._process_parallel_for_loop_with_queue(item_list)
1094 elif self.loop_type == LoopType.ParallelForLoop:
1095 self._process_parallel_for_loop(item_list)
1096 else:
1097 self._process_for_loop(item_list)
1099 self._user_interface.update_task(self.replica_name, completed=self.n_item, total=self.n_item)
1101 self.finish()
1103 def _build_loop_item(self) -> LoopItem:
1104 """
1105 Build a LoopItem payload for the current loop context.
1107 :return: The LoopItem payload.
1108 :rtype: mafw.models.LoopItem
1109 """
1110 return LoopItem(self.i_item, int(self.n_item or 0), self.item)
1112 def _build_loop_result(self, payload: Any, duration: float) -> LoopResult:
1113 """
1114 Build a LoopResult payload for the current loop context.
1116 :param payload: The optional payload returned by process.
1117 :type payload: Any
1118 :param duration: Wall-clock duration of the item processing.
1119 :type duration: float
1120 :return: The LoopResult payload.
1121 :rtype: mafw.models.LoopResult
1122 """
1123 return LoopResult(self.i_item, int(self.n_item or 0), self.looping_status, payload, duration)
1125 def _payload_annotation_matches(self, annotation: Any) -> bool:
1126 """
1127 Check whether an annotation matches LoopItem or LoopResult.
1129 Handles both ``typing.Union[X, Y]`` and the built-in ``X | Y`` union syntax.
1131 :param annotation: The annotation to inspect.
1132 :type annotation: Any
1133 :return: True if the annotation matches LoopItem or LoopResult.
1134 :rtype: bool
1135 """
1136 import types
1138 if annotation in (LoopItem, LoopResult):
1139 return True
1140 origin = get_origin(annotation)
1141 if origin is Union or isinstance(annotation, types.UnionType):
1142 return any(arg in (LoopItem, LoopResult) for arg in get_args(annotation))
1143 return False
1145 def _call_with_optional_payload(self, func: Callable[..., Any], payload: Any) -> Any:
1146 """
1147 Invoke a processor hook with an optional payload if the signature allows it.
1149 :param func: The callable to invoke.
1150 :type func: Callable
1151 :param payload: The payload to pass if supported.
1152 :type payload: Any
1153 :return: The callable return value.
1154 :rtype: Any
1155 """
1156 signature = inspect.signature(func)
1157 parameters = list(signature.parameters.values())
1158 if parameters and parameters[0].name == 'self':
1159 parameters = parameters[1:]
1160 if not parameters:
1161 return func()
1162 if any(param.kind in (param.VAR_POSITIONAL, param.VAR_KEYWORD) for param in parameters):
1163 return func(payload)
1164 first = parameters[0]
1165 if first.kind in (first.POSITIONAL_ONLY, first.POSITIONAL_OR_KEYWORD): 1165 ↛ 1170line 1165 didn't jump to line 1170 because the condition on line 1165 was always true
1166 if first.name in ('item', 'loop_item', 'result', 'loop_result', 'payload'):
1167 return func(payload)
1168 if self._payload_annotation_matches(first.annotation): 1168 ↛ 1169line 1168 didn't jump to line 1169 because the condition on line 1168 was never true
1169 return func(payload)
1170 return func()
1172 def _process_for_loop(self, item_list: Collection[Any]) -> None:
1173 """
1174 Process items with the standard serial for-loop.
1176 :param item_list: The list of items to process.
1177 :type item_list: Collection[Any]
1178 """
1179 for self.i_item, self.item in enumerate(item_list):
1180 self.looping_status = LoopingStatus.Continue
1182 self.format_progress_message()
1183 self._user_interface.display_progress_message(self.progress_message, self.i_item, self.n_item, 0.1)
1185 loop_item = self._build_loop_item()
1186 with Timer(suppress_message=True) as timer:
1187 payload = self._call_with_optional_payload(self.process, loop_item)
1188 self._process_durations.append(timer.duration)
1190 loop_result = self._build_loop_result(payload, timer.duration)
1192 if self.looping_status == LoopingStatus.Continue:
1193 self._call_with_optional_payload(self.accept_item, loop_result)
1194 elif self.looping_status == LoopingStatus.Skip:
1195 self._call_with_optional_payload(self.skip_item, loop_result)
1196 else: # Abort or Quit
1197 break
1199 self._user_interface.update_task(self.replica_name, increment=1)
1201 def _process_parallel_for_loop(self, item_list: Collection[Any]) -> None:
1202 """
1203 Process items in parallel using a thread pool.
1205 :param item_list: The list of items to process.
1206 :type item_list: Collection[Any]
1207 """
1208 max_workers = self.max_workers
1209 abort_event = threading.Event()
1210 abort_lock = threading.Lock()
1211 abort_status: LoopingStatus | None = None
1213 def _set_abort(status: LoopingStatus) -> None:
1214 nonlocal abort_status
1215 with abort_lock:
1216 if abort_status == LoopingStatus.Abort:
1217 return
1218 if status == LoopingStatus.Abort:
1219 abort_status = LoopingStatus.Abort
1220 elif abort_status is None: 1220 ↛ 1222line 1220 didn't jump to line 1222
1221 abort_status = LoopingStatus.Quit
1222 abort_event.set()
1224 def _worker(i_item: int, item: Any) -> tuple[int, Any, LoopingStatus, Any, float]:
1225 with self._thread_loop_context(i_item, int(self.n_item or 0), item):
1226 self.looping_status = LoopingStatus.Continue
1227 loop_item = self._build_loop_item()
1228 with Timer(suppress_message=True) as timer:
1229 payload = self._call_with_optional_payload(self.process, loop_item)
1230 status = self.looping_status
1232 if status in (LoopingStatus.Abort, LoopingStatus.Quit):
1233 _set_abort(status)
1234 return i_item, item, status, payload, timer.duration
1236 if abort_event.is_set():
1237 return i_item, item, status, payload, timer.duration
1239 loop_result = self._build_loop_result(payload, timer.duration)
1240 if status == LoopingStatus.Continue:
1241 self._call_with_optional_payload(self.accept_item, loop_result)
1242 elif status == LoopingStatus.Skip: 1242 ↛ 1245line 1242 didn't jump to line 1245 because the condition on line 1242 was always true
1243 self._call_with_optional_payload(self.skip_item, loop_result)
1245 return i_item, item, status, payload, timer.duration
1247 pending: set[Future[tuple[int, Any, LoopingStatus, Any, float]]] = set()
1248 items_iter = iter(enumerate(item_list))
1250 def _submit_next() -> bool:
1251 try:
1252 idx, itm = next(items_iter)
1253 except StopIteration:
1254 return False
1255 fut: Future[tuple[int, Any, LoopingStatus, Any, float]] = executor.submit(_worker, idx, itm)
1256 pending.add(fut)
1257 return True
1259 with ThreadPoolExecutor(max_workers=max_workers) as executor:
1260 while len(pending) < max_workers and _submit_next():
1261 pass
1263 while pending:
1264 done, _ = wait(pending, return_when=FIRST_COMPLETED)
1265 for fut in done:
1266 pending.remove(fut)
1267 i_item, item, status, _payload, duration = fut.result()
1268 self._process_durations.append(duration)
1269 self.item = item
1270 self.i_item = i_item
1271 self.format_progress_message()
1272 self._user_interface.display_progress_message(self.progress_message, i_item, self.n_item, 0.1)
1273 self._user_interface.update_task(self.replica_name, increment=1)
1275 if abort_event.is_set():
1276 continue
1278 while len(pending) < max_workers and _submit_next():
1279 pass
1281 if abort_status is not None:
1282 self.looping_status = abort_status
1284 def _process_parallel_for_loop_with_queue(self, item_list: Collection[Any]) -> None:
1285 """
1286 Process items in parallel using a producer/consumer queue.
1288 Items are processed in worker threads, while a dedicated consumer thread handles the post-processing hooks.
1290 :param item_list: The list of items to process.
1291 :type item_list: Collection[Any]
1292 """
1293 # the idea is the following, we have a single consumer thread that is constantly trying to pull
1294 # items out of a shared queue and we have a pool of producer threads that are putting items in the queue as
1295 # long as there is space into it (back-pressure).
1296 # it the queue is full, then the producer threads are set to sleep for a short interval of time and waken up
1297 # again after some times for another attempt to write the output in the queue.
1298 # when there are no more items to execute, the main thread is pushing a sentinel object into the queue
1299 # marking the end of the processing, so that the consumer thread can be gracefully terminated.
1300 max_workers = self.max_workers
1301 result_queue: queue.Queue[object] = queue.Queue(maxsize=self.queue_size)
1302 abort_event = threading.Event()
1303 abort_lock = threading.Lock()
1304 abort_status: LoopingStatus | None = None
1305 sentinel = object()
1307 def _set_abort(status: LoopingStatus) -> None:
1308 nonlocal abort_status
1309 with abort_lock:
1310 if abort_status == LoopingStatus.Abort: 1310 ↛ 1311line 1310 didn't jump to line 1311 because the condition on line 1310 was never true
1311 return
1312 if status == LoopingStatus.Abort: 1312 ↛ 1314line 1312 didn't jump to line 1314 because the condition on line 1312 was always true
1313 abort_status = LoopingStatus.Abort
1314 elif abort_status is None:
1315 abort_status = LoopingStatus.Quit
1316 abort_event.set()
1318 def _get_abort_status() -> LoopingStatus | None:
1319 with abort_lock:
1320 return abort_status
1322 def _worker(batch_items: list[tuple[int, Any]]) -> None:
1323 batch_results: list[tuple[LoopItem, LoopResult]] = []
1324 for i_item, item in batch_items:
1325 if abort_event.is_set(): 1325 ↛ 1326line 1325 didn't jump to line 1326 because the condition on line 1325 was never true
1326 break
1327 with self._thread_loop_context(i_item, int(self.n_item or 0), item):
1328 self.looping_status = LoopingStatus.Continue
1329 loop_item = self._build_loop_item()
1330 with Timer(suppress_message=True) as timer:
1331 payload = self._call_with_optional_payload(self.process, loop_item)
1332 status = self.looping_status
1333 if status in (LoopingStatus.Abort, LoopingStatus.Quit):
1334 _set_abort(status)
1335 loop_result = self._build_loop_result(payload, timer.duration)
1336 batch_results.append((loop_item, loop_result))
1337 if status in (LoopingStatus.Abort, LoopingStatus.Quit):
1338 break
1339 if batch_results: 1339 ↛ exitline 1339 didn't return from function '_worker' because the condition on line 1339 was always true
1340 result_queue.put(batch_results)
1342 def _consumer() -> None:
1343 self.consumer_start()
1344 try:
1345 while True:
1346 queued = result_queue.get()
1347 if queued is sentinel:
1348 break
1349 batch_results = cast(list[tuple[LoopItem, LoopResult]], queued)
1350 for loop_item, loop_result in batch_results:
1351 with self._thread_loop_context(loop_item.i_item, loop_item.n_item, loop_item.payload):
1352 self.looping_status = loop_result.looping_status
1353 self.format_progress_message()
1354 self._user_interface.display_progress_message(
1355 self.progress_message, loop_item.i_item, self.n_item, 0.1
1356 )
1357 self._process_durations.append(loop_result.duration)
1358 self._user_interface.update_task(self.replica_name, increment=1)
1360 if _get_abort_status() is None and loop_result.looping_status in (
1361 LoopingStatus.Continue,
1362 LoopingStatus.Skip,
1363 ):
1364 self._call_with_optional_payload(self.consumer_process, loop_result)
1365 finally:
1366 self.consumer_finish()
1368 pending: set[Future[None]] = set()
1369 items_iter = iter(enumerate(item_list))
1371 def _submit_next() -> bool:
1372 if abort_event.is_set():
1373 return False
1374 batch_items: list[tuple[int, Any]] = []
1375 for _ in range(self.queue_batch_size):
1376 try:
1377 idx, itm = next(items_iter)
1378 except StopIteration:
1379 break
1380 batch_items.append((idx, itm))
1381 if not batch_items:
1382 return False
1383 fut: Future[None] = executor.submit(_worker, batch_items)
1384 pending.add(fut)
1385 return True
1387 consumer_thread = threading.Thread(target=_consumer, name=f'{self.name}-consumer')
1388 consumer_thread.start()
1390 try:
1391 with ThreadPoolExecutor(max_workers=max_workers) as executor:
1392 while len(pending) < max_workers and _submit_next():
1393 pass
1395 while pending:
1396 done, _ = wait(pending, return_when=FIRST_COMPLETED)
1397 for fut in done:
1398 pending.remove(fut)
1399 fut.result()
1401 if abort_event.is_set():
1402 continue
1404 while len(pending) < max_workers and _submit_next(): 1404 ↛ 1405line 1404 didn't jump to line 1405 because the condition on line 1404 was never true
1405 pass
1406 finally:
1407 result_queue.put(sentinel)
1408 consumer_thread.join()
1410 if abort_status is not None:
1411 self.looping_status = abort_status
1413 def _execute_while_loop(self) -> None:
1414 """Executes the processor within a while loop.
1416 **Private method**. Do not overload nor invoke it directly. The :meth:`execute` method will call the
1417 appropriate implementation depending on the processor LoopType.
1418 """
1419 # it is a while loop, so a priori we don't know how many iterations we will have, nevertheless, we
1420 # can have a progress bar with 'total' set to None, so that it goes in the so-called indeterminate
1421 # progress. See https://rich.readthedocs.io/en/stable/progress.html#indeterminate-progress
1422 # we initialise n_item outside the loop, because it is possible that the user has a way to define n_item
1423 # and he can do it within the loop.
1424 self.n_item = None
1425 with contextlib.ExitStack() as self._resource_stack:
1426 self.acquire_resources()
1427 self._wall_clock_start = time.perf_counter()
1428 self.start()
1430 # turn the processor status to run
1431 self.processor_status = ProcessorStatus.Run
1433 self._user_interface.create_task(self.replica_name, self.description, completed=0, total=self.n_item)
1435 # we are ready to start the looping. For statistics, we can count the iterations.
1436 self.i_item = 0
1437 while self.while_condition():
1438 # set the looping status to Continue. The user may want to change it in the process method.
1439 self.looping_status = LoopingStatus.Continue
1441 # send a message to the user interface
1442 self.format_progress_message()
1443 self._user_interface.display_progress_message(
1444 self.progress_message, self.i_item, self.n_item, frequency=0.1
1445 )
1447 # wrap the execution in a timer to measure how long it too for statistical reasons.
1448 with Timer(suppress_message=True) as timer:
1449 self.process()
1450 self._process_durations.append(timer.duration)
1452 # modify the loop depending on the looping status
1453 if self.looping_status == LoopingStatus.Continue:
1454 self.accept_item()
1455 elif self.looping_status == LoopingStatus.Skip:
1456 self.skip_item()
1457 else: # equiv to if self.looping_status in [LoopingStatus.Abort, LoopingStatus.Quit]:
1458 break
1460 # update the progress bar. if self.n_item is still None, then the progress bar will show indeterminate
1461 # progress.
1462 self._user_interface.update_task(self.replica_name, self.i_item + 1, 1, self.n_item)
1464 # now that the loop is finished, we know how many elements we processed
1465 if self.n_item is None:
1466 self.n_item = self.i_item
1467 self._user_interface.update_task(self.replica_name, completed=self.n_item, total=self.n_item)
1469 self.finish()
1471 def acquire_resources(self) -> None:
1472 """
1473 Acquires resources and add them to the resource stack.
1475 The whole body of the :meth:`execute` method is within a context structure. The idea is that if any part of
1476 the code inside should throw an exception that breaking the execution, we want to be sure that all stateful
1477 resources are properly closed.
1479 Since the number of resources may vary, the variable number of nested `with` statements has been replaced by
1480 an `ExitStack <https://docs.python.org/3/library/contextlib.html#contextlib.ExitStack>`_. Resources,
1481 like open files, timers, db connections, need to be added to the resource stacks in this method.
1483 In the case a processor is being executed within a :class:`~mafw.processor.ProcessorList`, then some resources might be shared, and
1484 for this reason they are not added to the stack. This selection can be done via the private
1485 :attr:`local_resource_acquisition`. This is normally True, meaning that the processor will handle its resources
1486 independently, but when the processor is executed from a :class:`~mafw.processor.ProcessorList`, this flag is automatically turned to
1487 False.
1489 If the user wants to add additional resources, he has to overload this method calling the super to preserve
1490 the original resources. If he wants to have shared resources among different processors executed from inside
1491 a processor list, he has to overload the :class:`~mafw.processor.ProcessorList` class as well.
1492 """
1493 # Both the timer and the user interface will be added to the processor resource stack only if the processor is
1494 # set to acquire its own resources.
1495 # The timer and the user interface have in-built enter and exit method.
1496 if self._resource_acquisition:
1497 self.timer = self._resource_stack.enter_context(Timer(**self._timer_parameters))
1498 self._resource_stack.enter_context(self._user_interface)
1500 # For the database it is a bit different.
1501 if self._database is None and self._database_conf is None:
1502 # no database, nor configuration.
1503 # we cannot do anything
1504 pass
1505 elif self._database is None and self._database_conf is not None:
1506 # no db, but we got a configuration.
1507 # we can make a db.
1508 # This processor will try to make a valid connection, and in case it succeeds, it will add the database to
1509 # the resource stack.
1510 # The database has an enter method, but it is to generate transaction.
1511 # We will add the database.close via the callback method.
1512 if 'DBConfiguration' in self._database_conf:
1513 conf = self._database_conf['DBConfiguration'] # type1
1514 else:
1515 conf = self._database_conf # type2
1517 db_url, connection_parameters = build_connection_parameters(conf)
1519 self._database = connect(db_url, **connection_parameters) # type: ignore[no-untyped-call] # playhouse.db_url.connect lacks type stubs
1520 self._resource_stack.callback(self._database.close)
1521 try:
1522 self._database.connect()
1523 except peewee.OperationalError as e:
1524 log.critical('Unable to connect to %s', db_url)
1525 raise e
1526 database_proxy.initialize(self._database)
1527 if self.create_standard_tables:
1528 standard_tables = mafw_model_register.get_standard_tables()
1529 self.database.create_tables(standard_tables)
1530 for table in standard_tables:
1531 table.init()
1533 else: # equivalent to: if self._database is not None:
1534 # we got a database, so very likely we are inside a processor list
1535 # the connection has been already set and the initialisation as well.
1536 # nothing else to do here.
1537 # do not put the database in the exit stack. who create it has also to close it.
1538 pass
1540 def start(self) -> None:
1541 """
1542 Start method.
1544 The user can overload this method, including all steps that should be performed at the beginning of the
1545 operation.
1547 If the user decides to overload it, it should include a call to the super method.
1548 """
1549 self._mark_super_call('start')
1550 self.processor_status = ProcessorStatus.Start
1551 self._remove_orphan_files()
1553 def get_items(self) -> Collection[Any]:
1554 """
1555 Returns the item collections for the processor loop.
1557 This method must be overloaded for the processor to work. Generally, this is getting a list of rows from the
1558 database, or a list of files from the disk to be processed.
1560 :return: A collection of items for the loop
1561 :rtype: Collection[Any]
1562 """
1563 return []
1565 def while_condition(self) -> bool:
1566 """
1567 Return the while condition
1569 :return: True if the while loop has to continue, false otherwise.
1570 :rtype: bool
1571 """
1572 return False
1574 def process(self) -> None:
1575 """
1576 Processes the current item.
1578 This is the core of the Processor, where the user has to define the calculations required.
1580 In parallel for loops, the method can optionally accept a :class:`~mafw.models.loop_payloads.LoopItem`
1581 parameter if the user prefers not to rely on thread-local access to :attr:`.processor.Processor.item`, :attr:`.processor.Processor.i_item` and
1582 :attr:`.processor.Processor.n_item`.
1583 """
1584 pass
1586 def accept_item(self) -> None:
1587 """
1588 Does post process actions on a successfully processed item.
1590 Within the :meth:`process`, the user left the looping status to Continue, so it means that everything looks
1591 good and this is the right place to perform database updates or file savings.
1593 .. seealso:
1594 Have a look at :meth:`skip_item` for what to do in case something went wrong.
1596 In parallel for loops, the method can optionally accept a :class:`~mafw.models.loop_payloads.LoopResult`
1597 parameter for direct access to the processed payload and looping status.
1598 """
1599 pass
1601 def skip_item(self) -> None:
1602 """
1603 Does post process actions on a *NOT* successfully processed item.
1605 Within the :meth:`process`, the user set the looping status to Skip, so it means that something went wrong
1606 and here corrective actions can be taken if needed.
1608 .. seealso:
1609 Have a look at :meth:`accept_item` for what to do in case everything was OK.
1611 In parallel for loops, the method can optionally accept a :class:`~mafw.models.loop_payloads.LoopResult`
1612 parameter for direct access to the processed payload and looping status.
1613 """
1614 pass
1616 def consumer_start(self) -> None:
1617 """
1618 Executes once in the consumer thread for the parallel queue loop.
1620 This hook mirrors :meth:`start` but is only used with the queue-based parallel loop.
1621 """
1622 pass
1624 def consumer_process(self, loop_result: LoopResult | None = None) -> None:
1625 """
1626 Handle processed items in the consumer thread for the parallel queue loop.
1628 By default this dispatches to :meth:`accept_item` or :meth:`skip_item` based on the looping status. The method
1629 can optionally accept a :class:`~mafw.models.loop_payloads.LoopResult` payload.
1630 """
1631 if loop_result is None:
1632 loop_result = self._build_loop_result(None, 0.0)
1633 if self.looping_status == LoopingStatus.Continue:
1634 self._call_with_optional_payload(self.accept_item, loop_result)
1635 elif self.looping_status == LoopingStatus.Skip: 1635 ↛ exitline 1635 didn't return from function 'consumer_process' because the condition on line 1635 was always true
1636 self._call_with_optional_payload(self.skip_item, loop_result)
1638 def consumer_finish(self) -> None:
1639 """
1640 Executes once in the consumer thread after the queue has been drained.
1641 """
1642 pass
1644 def finish(self) -> None:
1645 """
1646 Concludes the execution.
1648 The user can reimplement this method if there are some conclusive tasks that must be achieved.
1649 Always include a call to super().
1650 """
1651 self._mark_super_call('finish')
1652 self.processor_status = ProcessorStatus.Finish
1653 if self.looping_status == LoopingStatus.Abort:
1654 self.processor_exit_status = ProcessorExitStatus.Aborted
1655 self.print_process_statistics()
1657 def print_process_statistics(self) -> None:
1658 """
1659 Print the process statistics.
1661 A utility method to display the fastest, the slowest and the average timing required to process on a single
1662 item. This is particularly useful when the looping processor is part of a ProcessorList.
1663 """
1664 if len(self._process_durations):
1665 log.info('[cyan] Processed %s items.' % len(self._process_durations))
1666 log.info(
1667 '[cyan] Fastest item process duration: %s '
1668 % pretty_format_duration(min(self._process_durations), n_digits=3)
1669 )
1670 log.info(
1671 '[cyan] Slowest item process duration: %s '
1672 % pretty_format_duration(max(self._process_durations), n_digits=3)
1673 )
1674 log.info(
1675 '[cyan] Average item process duration: %s '
1676 % pretty_format_duration((sum(self._process_durations) / len(self._process_durations)), n_digits=3)
1677 )
1678 if self._wall_clock_start is not None:
1679 total_duration = time.perf_counter() - self._wall_clock_start
1680 else:
1681 total_duration = sum(self._process_durations)
1682 log.info('[cyan] Total process duration: %s' % pretty_format_duration(total_duration, n_digits=3))
1684 def _remove_orphan_files(self) -> None:
1685 """
1686 Remove orphan files.
1688 If a connection to the database is available, then the OrphanFile standard table is queried for all its entries,
1689 and all the files are then removed.
1691 The user can turn off this behaviour by switching the :attr:`~mafw.processor.Processor.remove_orphan_files` to False.
1693 """
1694 if self._database is None or self.remove_orphan_files is False:
1695 # no database connection or no wish to remove orphan files, it does not make sense to continue
1696 return
1698 try:
1699 OrphanFile = cast(MAFwBaseModel, mafw_model_register.get_model('OrphanFile'))
1700 except KeyError:
1701 log.warning('OrphanFile table not found in DB. Please verify database integrity')
1702 return
1704 if TYPE_CHECKING:
1705 assert hasattr(OrphanFile, '_meta')
1707 orphan_files = OrphanFile.select().execute()
1708 if len(orphan_files) != 0:
1709 msg = f'[yellow]Pruning orphan files ({sum(len(f.filenames) for f in orphan_files)})...'
1710 log.info(msg)
1711 for orphan in orphan_files:
1712 # filenames is a list of files:
1713 for f in orphan.filenames:
1714 f.unlink(missing_ok=True)
1716 OrphanFile.delete().execute()