Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions domain-specific-terms.txt
Original file line number Diff line number Diff line change
@@ -1,5 +1,8 @@
# contains 1 lower case word per line which are ignored in the spell_check
# German MaKo domain terms / identifiers that codespell mistakes for English typos
aktion
als
frist
oder
crossreference
vor
7 changes: 6 additions & 1 deletion src/makoralle/models/process.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,11 +10,16 @@
class DeadlineRule(BaseModel):
"""Structured deadline derived from free-text Frist."""

type: str # "unverzüglich", "parallel", "none", "complex"
type: str # "unverzüglich", "parallel", "none", "complex",
# "terminiert" (fixed relative to an external anchor), "reference" (real but
# irreducible — kept as a note, without a [REVIEW] flag).
latest_time: str | None = None # e.g. "07:00"
business_days: int | None = None # e.g. 1 for "1. WT"
reference_step: int | None = None # step number this deadline is relative to
reference_event: str | None = None # "ÜT" (Übertragungstag) or "ÜZ" (Übertragungszeitpunkt)
direction: Literal["vor", "nach"] | None = None # WT count before/after the anchor
anchor: str | None = None # external anchor when it is not a step ("Änderungstermin", "Zahlungsziel", …)
recurring: bool = False # "täglich …" recurring obligation
raw: str = "" # original free-text deadline


Expand Down
39 changes: 25 additions & 14 deletions src/makoralle/serialization/markdown.py
Original file line number Diff line number Diff line change
Expand Up @@ -150,22 +150,33 @@ def _pid_table(sd: dict[str, Any]) -> list[str]:

def _deadline_legend(sd: dict[str, Any]) -> list[str]:
"""A short vocabulary legend for the deadline tags/notes rendered on the SD
image. Emitted only when at least one step carries a deadline that shows up
on the diagram (an inline tag for unverzüglich/parallel, or a [REVIEW] note
for complex)."""
image. Only the marker kinds actually present on this diagram are listed:
inline tags (unverzüglich/parallel/terminiert), an (i) reference note, and/or
a [REVIEW] note for a still-unstructured complex Frist."""
steps = sd.get("steps", [])
shown = any((s.get("deadline_rule") or {}).get("type") in ("unverzüglich", "parallel", "complex") for s in steps)
if not shown:
types = {(s.get("deadline_rule") or {}).get("type") for s in steps}
has_tags = bool(types & {"unverzüglich", "parallel", "terminiert"})
has_reference = "reference" in types
has_complex = "complex" in types
if not (has_tags or has_reference or has_complex):
return []
return [
"**Fristen (Legende der Diagramm-Markierungen):**",
"",
"- `{u}` — unverzüglich",
"- `{∥#N}` — parallel zu Schritt N",
"- `{≤HH:MM nWT ÜZ#N}` — spätestens HH:MM, n Werktage nach dem ÜZ/ÜT von Schritt N",
"- `(!) … [REVIEW]` (Notiz) — komplexe Frist, noch nicht strukturiert geparst",
"",
]
lines = ["**Fristen (Legende der Diagramm-Markierungen):**", ""]
if has_tags:
lines += [
"- `{u}` — unverzüglich",
"- `{∥#N}` — parallel zu Schritt N",
"- `{≤HH:MM nWT ÜZ#N}` — spätestens HH:MM, n Werktage nach dem ÜZ/ÜT von Schritt N",
"- `{≤nWT vor|nach Anker}` — terminierte Frist, n Werktage vor/nach einem Termin "
"(z. B. Zahlungsziel, Änderungstermin)",
]
if has_reference:
lines.append(
"- `(i) …` (Notiz) — Frist als Verweis auf eine Tabelle / ein SD / den Rahmenvertrag oder mit Bedingung"
)
if has_complex:
lines.append("- `(!) … [REVIEW]` (Notiz) — komplexe Frist, noch nicht strukturiert geparst")
lines.append("")
return lines


def _render_sequence_diagram(sd: dict[str, Any]) -> list[str]: # pylint: disable=too-many-locals
Expand Down
32 changes: 30 additions & 2 deletions src/makoralle/serialization/wsd.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,9 +41,31 @@ def _deadline_tag(rule: DeadlineRule | None) -> str:
evt = rule.reference_event or ""
parts.append(f"{evt}#{rule.reference_step}")
return "{" + " ".join(parts) + "}" if parts else "{u}"
if rule.type == "terminiert":
return "{" + _terminiert_core(rule) + "}"
return ""


def _terminiert_core(rule: DeadlineRule) -> str:
"""Inner text of a ``terminiert`` tag — a 'spätester' deadline fixed relative to
an external anchor. Always leads with ``≤`` (or ``täglich``). Examples:
``≤20WT vor Änderungstermin``, ``≤11WT nach #2``, ``≤Zahlungsziel``,
``täglich ≤14:00``."""
if rule.recurring:
return f"täglich ≤{rule.latest_time}" if rule.latest_time else "täglich"
anchor = f"#{rule.reference_step}" if rule.reference_step else rule.anchor
if rule.business_days is not None:
core = f"≤{rule.business_days}WT"
if rule.direction:
core += f" {rule.direction}"
return f"{core} {anchor}" if anchor else core
if rule.latest_time:
return f"≤{rule.latest_time} {anchor}" if anchor else f"≤{rule.latest_time}"
if anchor:
return f"≤{anchor}"
return "terminiert"


def _clean_note_text(text: str) -> str:
"""Collapse whitespace/newlines so raw Frist text is a single safe note line
(the WSD parser treats newlines as statement breaks). See p17 escaping TODO."""
Expand All @@ -57,15 +79,21 @@ def _deadline_note(step: SDStep, participants: list[str]) -> str | None:
subprocess step that is the sender lifeline (via ``_ref_lifeline``), since
Vision often mis-guesses the receiver of a ref."""
dl = step.deadline_rule
if not dl or dl.type != "complex" or not dl.raw:
if not dl or dl.type not in ("complex", "reference") or not dl.raw:
return None
if (step.message or "").strip().lower().startswith("ref "):
who = _ref_lifeline(step, participants)
else:
who = step.receiver if step.receiver and step.receiver != "?" else step.sender
if not who or who == "?":
return None
return f"note right of {who}: (!) Frist: {_clean_note_text(dl.raw)} [REVIEW]"
text = _clean_note_text(dl.raw)
# A "reference" deadline is real but irreducible (points to another table/SD/
# contract, or is conditional): keep it visible as an (i) note, but WITHOUT the
# [REVIEW] flag so extract_review_notes never surfaces it as "Prüfung nötig".
if dl.type == "reference":
return f"note right of {who}: (i) Frist: {text}"
return f"note right of {who}: (!) Frist: {text} [REVIEW]"


def _build_step_paths(fragments: list[SDFragment]) -> dict[int, list[tuple[SDFragment, int]]]:
Expand Down
11 changes: 11 additions & 0 deletions unittests/test_build_webapp_data.py
Original file line number Diff line number Diff line change
Expand Up @@ -128,6 +128,17 @@ def test_extract_review_notes_parses_wsd_lines() -> None:
assert extract_review_notes(wsd) == ["(!) Frist: Deaktivierung"]


def test_extract_review_notes_ignores_info_frist_notes() -> None:
"""A 'reference' deadline renders as an (i) note without [REVIEW]; it must NOT be
surfaced as a review item ("Prüfung nötig")."""
wsd = (
"title X\n"
"note right of NB: (i) Frist: Gemäß Rahmenvertrag.\n"
"note right of LF: (!) Frist: Echtes Review [REVIEW]\n"
)
assert extract_review_notes(wsd) == ["(!) Frist: Echtes Review"]


def test_sd_source_hash_ignores_whitespace_churn_but_not_content() -> None:
base = "title X\nNB->ÜNB: Anmeldung\n"
# leading/trailing whitespace + CRLF line endings normalize away
Expand Down
29 changes: 29 additions & 0 deletions unittests/test_domain_models.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
from makoralle.models.ebd import DecisionStep, DecisionTree
from makoralle.models.pid import PIDMapping
from makoralle.models.process import (
DeadlineRule,
NamedSD,
Process,
SDBranch,
Expand All @@ -13,6 +14,34 @@
)


def test_deadline_rule_defaults_for_new_fields() -> None:
"""The anchor/direction/recurring fields default to None/None/False so existing
(pre-extension) rules deserialize unchanged."""
rule = DeadlineRule(type="unverzüglich", raw="Unverzüglich")
assert rule.direction is None
assert rule.anchor is None
assert rule.recurring is False


def test_deadline_rule_terminiert_roundtrips_external_anchor() -> None:
"""A 'terminiert' rule carrying a WT count relative to an external, non-step
anchor round-trips through model_dump/model_validate."""
rule = DeadlineRule(
type="terminiert",
direction="vor",
business_days=20,
reference_event="ÜT",
anchor="Änderungstermin",
raw="Spätester ÜT ist der 20. WT vor dem gewünschten Änderungstermin.",
)
back = DeadlineRule.model_validate(rule.model_dump())
assert back.type == "terminiert"
assert back.direction == "vor"
assert back.business_days == 20
assert back.anchor == "Änderungstermin"
assert back.recurring is False


def test_use_case() -> None:
uc = UseCase(
goal="Zuordnung eines LF zu einer Marktlokation",
Expand Down
14 changes: 14 additions & 0 deletions unittests/test_p15_emit_markdown.py
Original file line number Diff line number Diff line change
Expand Up @@ -116,6 +116,20 @@ def test_deadline_legend_explains_vocabulary() -> None:
assert "ÜZ" in text or "ÜT" in text


def test_deadline_legend_emitted_for_terminiert() -> None:
sd = {"steps": [{"nr": 1, "deadline_rule": {"type": "terminiert", "anchor": "Zahlungsziel"}}]}
text = "\n".join(_deadline_legend(sd))
assert text # legend shown
assert "vor" in text and "nach" in text # explains the terminiert direction vocabulary


def test_deadline_legend_emitted_for_reference_note() -> None:
sd = {"steps": [{"nr": 1, "deadline_rule": {"type": "reference", "raw": "Gemäß Rahmenvertrag."}}]}
text = "\n".join(_deadline_legend(sd))
assert "(i)" in text # explains the info note
assert "REVIEW" not in text # a pure-reference SD needs no [REVIEW] legend line


def test_sd_table_keeps_full_complex_deadline() -> None:
long_raw = "spätestens 5 Werktage vor dem geplanten Zuordnungsbeginn der Marktlokation X"
sd = {"steps": [{"nr": 1, "sender": "NB", "receiver": "LF", "message": "Prüfung", "deadline": long_raw}]}
Expand Down
47 changes: 46 additions & 1 deletion unittests/test_p17_emit_wsd.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,28 @@
from makoralle.models.process import DeadlineRule, SDBranch, SDFragment, SDNote, SDStep, SequenceDiagram
from makoralle.serialization.wsd import _deadline_tag, emit_wsd
from makoralle.serialization.wsd import _deadline_note, _deadline_tag, emit_wsd


def _step_with_deadline(rule: DeadlineRule) -> SDStep:
return SDStep(nr=1, sender="LF", receiver="NB", message="Foo", deadline_rule=rule)


def test_deadline_note_complex_keeps_review_flag() -> None:
note = _deadline_note(_step_with_deadline(DeadlineRule(type="complex", raw="Gemäß irgendwas.")), ["LF", "NB"])
assert note == "note right of NB: (!) Frist: Gemäß irgendwas. [REVIEW]"


def test_deadline_note_reference_is_info_without_review_flag() -> None:
"""A 'reference' deadline (real but irreducible) stays a note, but with an (i)
marker and NO [REVIEW] — so extract_review_notes never pulls it into review."""
note = _deadline_note(_step_with_deadline(DeadlineRule(type="reference", raw="Gemäß Rahmenvertrag.")), ["LF", "NB"])
assert note == "note right of NB: (i) Frist: Gemäß Rahmenvertrag."
assert "[REVIEW]" not in note


def test_deadline_note_terminiert_and_structured_get_no_note() -> None:
"""Structured deadlines (tag-rendered) never produce a note."""
for t in ("terminiert", "unverzüglich", "parallel", "none"):
assert _deadline_note(_step_with_deadline(DeadlineRule(type=t, raw="x")), ["LF", "NB"]) is None


def test_emit_flat() -> None:
Expand Down Expand Up @@ -399,6 +422,28 @@ def test_deadline_tag_unverzueglich_with_clock_omits_null_pieces() -> None:
assert _deadline_tag(DeadlineRule(type="unverzüglich", business_days=3, raw="...")) == "{3WT}"


def test_deadline_tag_terminiert_wt_before_external_anchor() -> None:
rule = DeadlineRule(
type="terminiert", direction="vor", business_days=20, reference_event="ÜT", anchor="Änderungstermin", raw="..."
)
assert _deadline_tag(rule) == "{≤20WT vor Änderungstermin}"


def test_deadline_tag_terminiert_wt_after_reference_step() -> None:
rule = DeadlineRule(type="terminiert", direction="nach", business_days=11, reference_step=2, raw="...")
assert _deadline_tag(rule) == "{≤11WT nach #2}"


def test_deadline_tag_terminiert_anchor_only() -> None:
rule = DeadlineRule(type="terminiert", anchor="Zahlungsziel", raw="Spätester ÜT ist zum angegebenen Zahlungsziel.")
assert _deadline_tag(rule) == "{≤Zahlungsziel}"


def test_deadline_tag_terminiert_recurring_with_time() -> None:
rule = DeadlineRule(type="terminiert", recurring=True, latest_time="14:00", raw="Täglich … bis spätestens 14 Uhr.")
assert _deadline_tag(rule) == "{täglich ≤14:00}"


def test_emit_appends_deadline_tag_after_pid_suffix() -> None:
sd = SequenceDiagram(
participants=["NB", "LF"],
Expand Down