Pipeline¶

@classmethod
def from_rules(cls, language: str) -> PipelineConfig:
    """Build a complete pipeline by projecting a language's zen principles.

    Loads the [`LanguageZenPrinciples`][mcp_zen_of_languages.rules.base_models.LanguageZenPrinciples]
    for *language*, then delegates to
    ``configs_from_rules``
    to project each principle's metrics onto the matching detector configs.

    Args:
        language (str): Language key recognised by [`get_language_zen`][mcp_zen_of_languages.rules.get_language_zen]
            (e.g. ``"python"``).

    Returns:
        PipelineConfig: A fully populated ``PipelineConfig`` whose detector list reflects
        all zen principles defined for the language.

    Raises:
        ValueError: If no zen rules exist for *language*.

    Examples:
        >>> cfg = PipelineConfig.from_rules("python")
        >>> cfg.language
        'python'
    """
    lang_zen = get_language_zen(language)
    if not lang_zen:
        msg = f"No zen rules for language: {language}"
        raise ValueError(msg)
    from mcp_zen_of_languages.analyzers.registry import REGISTRY

    detectors = REGISTRY.configs_from_rules(lang_zen)
    return cls(language=language, detectors=detectors)

def model_post_init(self, __context: object, /) -> None:
    """Default empty violation selectors to the wildcard selector."""
    if not self.violation_selectors:
        self.violation_selectors = ["*"]
    if self.verified_dogma_ids is None:
        self.verified_dogma_ids = (
            list(self.dogma_ids) if len(self.dogma_ids) == 1 else []
        )
    if self.verified_testing_ids is None:
        self.verified_testing_ids = (
            list(self.testing_ids) if len(self.testing_ids) == 1 else []
        )
    if self.verified_projection_ids is None:
        self.verified_projection_ids = (
            list(self.projection_ids) if len(self.projection_ids) == 1 else []
        )

def build_bundle(self, language: str) -> BindingPerspectiveBundle:
    """Build the rule-perspective bundle for a rule detector binding."""
    from mcp_zen_of_languages.analyzers.registry import DetectorMetadata

    dogma_rule_map: dict[str, list[str]] = {}
    dogma_verified_rule_map: dict[str, list[str]] = {}
    testing_rule_map: dict[str, list[str]] = {}
    testing_verified_rule_map: dict[str, list[str]] = {}
    projection_rule_map: dict[str, list[str]] = {}
    projection_verified_rule_map: dict[str, list[str]] = {}
    for rule_id in self.rule_ids:
        for dogma_id in self.rule_dogma_map.get(rule_id, []):
            dogma_rule_map.setdefault(dogma_id, []).append(rule_id)
        for dogma_id in self.rule_verified_dogma_map.get(rule_id, []):
            dogma_verified_rule_map.setdefault(dogma_id, []).append(rule_id)
        for testing_id in self.rule_testing_map.get(rule_id, []):
            testing_rule_map.setdefault(testing_id, []).append(rule_id)
        for testing_id in self.rule_verified_testing_map.get(rule_id, []):
            testing_verified_rule_map.setdefault(testing_id, []).append(rule_id)
        for projection_id in self.rule_projection_map.get(rule_id, []):
            projection_rule_map.setdefault(projection_id, []).append(rule_id)
        for projection_id in self.rule_verified_projection_map.get(rule_id, []):
            projection_verified_rule_map.setdefault(projection_id, []).append(
                rule_id
            )

    return BindingPerspectiveBundle(
        rule_model=DetectorMetadata(
            detector_id=self.detector_id,
            detector_class=self.detector_class,
            config_model=self.config_model,
            language=language,
            rule_ids=list(self.rule_ids),
            rule_map=dict(self.rule_map),
            rule_dogma_map=dict(self.rule_dogma_map),
            rule_verified_dogma_map=dict(self.rule_verified_dogma_map),
            default_order=self.default_order,
            enabled_by_default=self.enabled_by_default,
        ),
        dogma_model=DogmaPerspectiveModel(
            detector_id=self.detector_id,
            language=language,
            dogma_ids=list(
                dict.fromkeys(
                    [*dogma_rule_map.keys(), *dogma_verified_rule_map.keys()]
                ),
            ),
            dogma_rule_map=dogma_rule_map,
            dogma_verified_rule_map=dogma_verified_rule_map,
        ),
        testing_model=TestingPerspectiveModel(
            detector_id=self.detector_id,
            language=language,
            testing_ids=list(
                dict.fromkeys(
                    [*testing_rule_map.keys(), *testing_verified_rule_map.keys()],
                ),
            ),
            testing_rule_map=testing_rule_map,
            testing_verified_rule_map=testing_verified_rule_map,
        ),
        projection_model=ProjectionPerspectiveModel(
            detector_id=self.detector_id,
            language=language,
            projection_ids=list(
                dict.fromkeys(
                    [
                        *projection_rule_map.keys(),
                        *projection_verified_rule_map.keys(),
                    ],
                ),
            ),
            projection_rule_map=projection_rule_map,
            projection_verified_rule_map=projection_verified_rule_map,
        ),
    )

def build_bundle(self, language: str) -> BindingPerspectiveBundle:
    """Build the rule-perspective bundle for a non-rule detector binding."""
    from mcp_zen_of_languages.analyzers.registry import DetectorMetadata

    return BindingPerspectiveBundle(
        rule_model=DetectorMetadata(
            detector_id=self.detector_id,
            detector_class=self.detector_class,
            config_model=self.config_model,
            language=language,
            rule_ids=[],
            rule_map={},
            rule_dogma_map={},
            rule_verified_dogma_map={},
            default_order=self.default_order,
            enabled_by_default=self.enabled_by_default,
        ),
    )