orchestrata_ultimate.factory.detectors.clause

Reference › orchestrata_ultimate › factory › detectors › clause

clause is the clause segmenter behind the C3 detector set (SPEC-033 row B22, ADR-0029 rule 1). The comparative and negation detectors must skip conditional clauses, which needs the start and end of each conditional clause and not only the presence of a word like "if". sentences splits text into (line number, sentence) pairs; a newline always ends a sentence, and periods end one except after listed abbreviations and inside decimal numbers (_is_abbreviation_boundary, _is_decimal_point, _is_sentence_boundary). segment splits one sentence into Segment records of kind conditional, rationale or main: first on sentence-internal punctuation (_piece_offsets), then at conditional and rationale markers from lexicons found inside a piece (_marker_boundaries), classifying each part by its first token (_classify). The docstring states the limits: no real grammar, no attachment or coordination resolution, with a grammar-based upgrade named as the future path. rules is the only importer; tests are in tests/factory/test_detector_clause.py.

At a glance

Class diagram

Dependency diagram

API

Classes

Segment (dataclass)

Decorators: dataclass(frozen=True) · Line 72

One clause segment of a sentence, with offsets into that sentence.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/factory/detectors/clause.py

"""Clause segmenter for the SPEC-033 C3 detector set (B22, ADR-0029 rule 1).

A parser, not a word list: exempting conditional clauses from the
comparative/negation detectors ("~48% of comparative false positives were
comparisons against absolute numbers inside a condition") requires knowing
where a conditional clause *starts and ends*, not just whether the word "if"
appears somewhere in the sentence.

Two functions:

- :func:`sentences` -- line/offset-aware sentence splitting. Deliberately
  simple (periods + newlines, with a small abbreviation exception list) --
  not a full sentence boundary detector. A newline always ends a sentence,
  matching this codebase's one-requirement-per-line convention.
- :func:`segment` -- splits one sentence into ``conditional``/``rationale``/
  ``main`` clauses on sentence-internal punctuation (``,``, ``;``, em-dash)
  and on :data:`~.lexicons.CONDITIONAL_MARKERS` /
  :data:`~.lexicons.RATIONALE_MARKERS` occurring *inside* a punctuation-
  delimited piece (a marker there starts a fresh segment running to the next
  boundary, so "X, and if Y then Z" splits into three segments, not two).

Known, documented limits: no real grammar (no attachment resolution, no
coordination scope). A DCG-based upgrade able to resolve nested/coordinated
clauses precisely is the stated future path (ADR-0029 accepts this cost
explicitly: "a real implementation cost, bought with a measured ~48%
reduction in comparative false positives").
"""

from __future__ import annotations

import re
from dataclasses import dataclass
from typing import Literal

from orchestrata_ultimate.factory.detectors.lexicons import (
    CONDITIONAL_MARKERS,
    RATIONALE_MARKERS,
    compile_lexicon,
)

SegmentKind = Literal["conditional", "rationale", "main"]

# Abbreviations whose trailing "." must never be treated as a sentence
# terminator. Matched case-insensitively against the token immediately
# preceding the candidate terminator.
_ABBREVIATIONS = frozenset(
    {
        "e.g.",
        "i.e.",
        "etc.",
        "vs.",
        "cf.",
        "et al.",
        "no.",
        "fig.",
        "approx.",
        "dr.",
        "mr.",
        "mrs.",
        "ms.",
    }
)

_LAST_TOKEN_RE = re.compile(r"(\S+)$")
_PUNCT_SPLIT_RE = re.compile(r",|;|--|—")

_CONDITIONAL_RE = compile_lexicon(CONDITIONAL_MARKERS)
_RATIONALE_RE = compile_lexicon(RATIONALE_MARKERS)


@dataclass(frozen=True)
class Segment:
    """One clause segment of a sentence, with offsets into that sentence."""

    text: str
    start: int
    end: int
    kind: SegmentKind


def _is_abbreviation_boundary(line: str, terminator_end: int) -> bool:
    """True if the ``.``/``!``/``?`` ending at ``terminator_end`` is part of
    a known abbreviation (e.g. the second "." of "e.g.") and must not split.
    """
    before = line[:terminator_end]
    match = _LAST_TOKEN_RE.search(before)
    token = match.group(1).lower() if match else ""
    return token in _ABBREVIATIONS


def _is_sentence_boundary(line: str, terminator_end: int) -> bool:
    """True if the char after (skipping spaces) starts a new sentence."""
    rest = line[terminator_end:].lstrip(" ")
    return rest == "" or rest[0].isupper() or rest[0].isdigit()


def _is_decimal_point(line: str, terminator_end: int) -> bool:
    """True if this "." sits between two digits (e.g. "3.14") -- never a boundary."""
    before_ok = terminator_end >= 2 and line[terminator_end - 2].isdigit()
    after_ok = terminator_end < len(line) and line[terminator_end].isdigit()
    return before_ok and after_ok


def _split_line_into_sentences(line: str) -> list[str]:
    """Split one physical line into sentence fragments."""
    split_points: list[int] = []
    for match in re.finditer(r"[.!?]", line):
        pos = match.end()
        if match.group(0) == "." and _is_decimal_point(line, pos):
            continue
        if _is_abbreviation_boundary(line, pos):
            continue
        if _is_sentence_boundary(line, pos):
            split_points.append(pos)
    parts: list[str] = []
    start = 0
    for pos in split_points:
        parts.append(line[start:pos])
        start = pos
    parts.append(line[start:])
    return [p.strip() for p in parts if p.strip()]


def sentences(text: str) -> list[tuple[int, str]]:
    """Split ``text`` into ``(1-indexed line number, sentence text)`` pairs.

    A newline always ends a sentence (this codebase's AC/requirement lines
    are one-sentence-per-line by convention); within a line, ``.``/``!``/``?``
    end a sentence only when followed by whitespace-then-uppercase/digit (or
    end of line) AND the preceding token is not a known abbreviation.
    """
    return [
        (line_no, sentence)
        for line_no, line in enumerate(text.splitlines(), start=1)
        for sentence in _split_line_into_sentences(line)
    ]


def _piece_offsets(sentence: str) -> list[tuple[int, int]]:
    """Offsets of the sentence-internal-punctuation-delimited pieces."""
    pieces: list[tuple[int, int]] = []
    start = 0
    for match in _PUNCT_SPLIT_RE.finditer(sentence):
        pieces.append((start, match.start()))
        start = match.end()
    pieces.append((start, len(sentence)))
    return pieces


def _marker_boundaries(piece_text: str) -> list[int]:
    """Offsets (relative to ``piece_text``) where a clause marker starts a
    fresh sub-segment -- always includes 0.
    """
    offsets = {0}
    for m in _CONDITIONAL_RE.finditer(piece_text):
        offsets.add(m.start())
    for m in _RATIONALE_RE.finditer(piece_text):
        offsets.add(m.start())
    return sorted(offsets)


def _classify(piece_text: str) -> SegmentKind:
    """Classify a sub-segment by what its first token matches."""
    if _CONDITIONAL_RE.match(piece_text):
        return "conditional"
    if _RATIONALE_RE.match(piece_text):
        return "rationale"
    return "main"


def segment(sentence: str) -> list[Segment]:
    """Split ``sentence`` into clause :class:`Segment`\\ s.

    Sentence-internal punctuation (``,``, ``;``, em-dash) delimits the first
    cut; within each resulting piece, any occurrence of a
    :data:`~.lexicons.CONDITIONAL_MARKERS`/:data:`~.lexicons.RATIONALE_MARKERS`
    entry that is not already at the piece's start begins a further
    sub-segment running to the next boundary. Each final segment is
    classified by whether it begins with a conditional marker, a rationale
    marker, or neither (``main``). Empty (whitespace-only) segments are
    dropped; offsets are trimmed to the stripped text they describe.
    """
    results: list[Segment] = []
    for piece_start, piece_end in _piece_offsets(sentence):
        piece_text = sentence[piece_start:piece_end]
        boundaries = _marker_boundaries(piece_text)
        for i, sub_start in enumerate(boundaries):
            sub_end = boundaries[i + 1] if i + 1 < len(boundaries) else len(piece_text)
            raw = piece_text[sub_start:sub_end]
            stripped = raw.strip()
            if not stripped:
                continue
            lead_ws = len(raw) - len(raw.lstrip())
            trail_ws = len(raw) - len(raw.rstrip())
            abs_start = piece_start + sub_start + lead_ws
            abs_end = piece_start + sub_end - trail_ws
            results.append(
                Segment(
                    text=stripped,
                    start=abs_start,
                    end=abs_end,
                    kind=_classify(stripped),
                )
            )
    return results