Spaces:
Sleeping
Sleeping
File size: 4,895 Bytes
63ceb34 f3b0e4a 63ceb34 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 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 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 | """``ProjectorRegistry`` β Sprint A14-S13.
Container instanciΓ© explicitement qui mappe ``projector_name``
vers une instance ``Projector``. SymΓ©trique du ``MetricRegistry``
(S5) : pas de singleton global, pas de side-effect d'import.
Pattern d'utilisation
---------------------
.. code-block:: python
from picarones.evaluation.projectors import (
ProjectorRegistry, AltoToText,
)
from picarones.formats.alto import AltoToText as _AltoToText
registry = ProjectorRegistry()
registry.register(_AltoToText())
registry.register(PageToText())
projector = registry.get("alto_to_text")
target_artifact, payload, report = projector.project(source_artifact, {})
Au S20, ce registre sera construit par
``app/services/registry_service.py`` au dΓ©marrage de l'application.
Pour S13-S18, chaque test ou consommateur l'instancie explicitement.
Anti-sur-ingΓ©nierie
-------------------
Pas de versioning de projecteur, pas de namespace, pas de recherche
par tag. Ces extras viendront quand un caller en aura concrètement
besoin (probablement avec les projecteurs contribuΓ©s par des modules
tiers, post-livraison).
"""
from __future__ import annotations
from picarones.domain.errors import PicaronesError
from picarones.evaluation.projectors.base import Projector
class ProjectorRegistrationError(PicaronesError):
"""Tentative d'enregistrement invalide d'un projecteur."""
class ProjectorNotFoundError(PicaronesError):
"""Le projecteur demandΓ© n'est pas enregistrΓ©."""
class ProjectorRegistry:
"""Container mutable de projecteurs indexΓ©s par ``name``.
Thread-safe en lecture après initialisation ; la séquence
d'enregistrement attendue est : un seul service, au dΓ©marrage,
enregistre tous les projecteurs en une fois, puis l'instance
est figΓ©e par convention.
"""
def __init__(self) -> None:
self._projectors: dict[str, Projector] = {}
# ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
# Enregistrement
# ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
def register(self, projector: Projector) -> None:
"""Enregistre un projecteur.
Raises
------
ProjectorRegistrationError
Si un projecteur du mΓͺme nom est dΓ©jΓ enregistrΓ© (sauf
re-enregistrement strict du mΓͺme objet, tolΓ©rΓ© pour les
tests qui re-instancient).
"""
if not hasattr(projector, "name"):
raise ProjectorRegistrationError(
"register : l'objet n'expose pas d'attribut ``name``."
)
if not isinstance(projector, Projector):
raise ProjectorRegistrationError(
f"register : {projector!r} ne satisfait pas le protocole "
"Projector (attributs ``name``, ``source_type``, "
"``target_type``, mΓ©thode ``project``)."
)
existing = self._projectors.get(projector.name)
if existing is not None:
if existing is projector:
return # idempotent
raise ProjectorRegistrationError(
f"Projecteur {projector.name!r} dΓ©jΓ enregistrΓ© avec "
"une autre instance."
)
self._projectors[projector.name] = projector
# ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
# Lecture
# ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
def __contains__(self, name: str) -> bool:
return name in self._projectors
def __len__(self) -> int:
return len(self._projectors)
def names(self) -> list[str]:
"""Liste des noms enregistrΓ©s (ordre d'enregistrement)."""
return list(self._projectors.keys())
def get(self, name: str) -> Projector:
"""Récupère le projecteur par son ``name``.
Raises
------
ProjectorNotFoundError
Si le nom n'est pas enregistrΓ©.
"""
if name not in self._projectors:
raise ProjectorNotFoundError(
f"Projecteur {name!r} non enregistrΓ©. "
f"Disponibles : {sorted(self._projectors)}."
)
return self._projectors[name]
__all__ = [
"ProjectorRegistry",
"ProjectorRegistrationError",
"ProjectorNotFoundError",
]
|