diff --git a/.devcontainer/compose.yaml b/.devcontainer/compose.yaml index 3b29512..a96b772 100644 --- a/.devcontainer/compose.yaml +++ b/.devcontainer/compose.yaml @@ -60,3 +60,10 @@ services: volumes: postgres-data: pgadmin-data: + +networks: + # Keep the shared development network independent of the Compose project + # name so sibling devcontainers can join it reliably. + default: + name: hemonc-alchemy_default + external: true diff --git a/.devcontainer/devcontainer.json b/.devcontainer/devcontainer.json index 09b3bde..611c4ec 100644 --- a/.devcontainer/devcontainer.json +++ b/.devcontainer/devcontainer.json @@ -5,6 +5,7 @@ "workspaceFolder": "/workspace/hemonc-alchemy", "remoteUser": "vscode", "updateRemoteUserUID": true, + "initializeCommand": "docker network create hemonc-alchemy_default || true", "postCreateCommand": "bash .devcontainer/post-create.sh", // also on start, so containers created before the kernel fix get it "postStartCommand": "bash .devcontainer/register-kernel.sh", diff --git a/docs/getting-started/local-development.md b/docs/getting-started/local-development.md index 269fe76..125e1d9 100644 --- a/docs/getting-started/local-development.md +++ b/docs/getting-started/local-development.md @@ -13,9 +13,13 @@ The repository's `.devcontainer` is a disposable environment for exploring the m From the repository root: ```bash +docker network inspect hemonc-alchemy_default >/dev/null 2>&1 || \ + docker network create hemonc-alchemy_default docker compose -f .devcontainer/compose.yaml up -d ``` +The network is shared with SCOOP and must exist before Compose starts. + Open the repository in VS Code's Dev Container. The project interpreter inside the Python service is `/opt/venv/bin/python`; the notebook kernel should use that environment. ## Inspect the database with pgAdmin diff --git a/docs/toolkit/scheduling.md b/docs/toolkit/scheduling.md index 529b21a..5ecf09b 100644 --- a/docs/toolkit/scheduling.md +++ b/docs/toolkit/scheduling.md @@ -27,10 +27,41 @@ clinic = frame[frame.route_group == "IV"] matrix = administration_matrix(frame) ``` -The frame has one row per drug per explicit cycle day. `decay_days=0` gives dosing days only; the default decay adds an intensity tail to the following days for occupancy-style views. Passing a list of variants produces one combined frame, which should be grouped by `variant_cui`, not the human-readable `variant` label. +The frame has one row per drug per explicit cycle day. `decay_days=0` gives dosing days only; the default decay adds an intensity tail to the following days for occupancy-style views. Decay rows have `intensity < 1` and are not extra doses. Passing a list of variants produces one combined frame, which should be grouped by `variant_cui`, not the human-readable `variant` label. + +In `administration_frame`, `indefinite` marks days continuing beyond the explicit days in `alldays`, while `cycle_indefinite` marks cycles continuing beyond the explicit cycles in `timing_sequence`. These columns preserve separate source markers; neither extends the returned rows. `route_group == "IV"` is a historical shorthand for clinic-administered routes and includes more than intravenous administration. `"PO"` represents home-administered routes. Unrecognized or unspecified routes are excluded from administration projections rather than guessed at, so the frame may cover less than the source variant. ## Preserve source-shaped values `ScheduleEvent` keeps fields such as dose and cycle-length bounds in their source form. A value like `"1.5-2"` needs an application decision before it becomes numeric. Generated enum fields remain enum members. Convert these values at the boundary where your application can state its handling of null, uncertain, or non-numeric source values. + +## Roll out a complete variant timeline + +`roll_out_variant` composes explicit cycle numbers, block-specific cycle lengths, and phase boundaries into a deterministic treatment timeline: + +```python +from datetime import date + +from hemonc_alchemy.toolkit.analytics.treatment.scheduling import roll_out_variant + +timeline = roll_out_variant(variant) +timeline[["sig_id", "component", "phase", "cycle_number", "day", "phase_elapsed_day", "elapsed_day", "timing_status"]] +``` + +`elapsed_day` is an integer relative to variant start, where day 0 is the variant start. `phase_elapsed_day` is relative to the start of each phase, with day 0 at that phase's first explicit cycle. When a missing surgery or another unresolved boundary makes the variant-relative date unknown, `elapsed_day` and `calendar_date` stay null while `phase_elapsed_day` retains the within-phase timing where it can be calculated. An unresolved block within the phase also has a null `phase_elapsed_day`. Pass `start_date` to receive known `calendar_date` values as timezone-free `datetime.date` objects: + +```python +timeline = roll_out_variant(variant, start_date=date(2026, 1, 1)) +``` + +Blocks with adjacent cycle-number ranges are anchored sequentially. Blocks with overlapping cycle numbers share the relevant anchor, while genuinely ambiguous source timing remains visible in `timing_status`. When differing cycle lengths overlap across more than one shared cycle, the later block is unresolved. When blocks ending the preceding cycle differ in cycle length, the following block is unresolved. A phase can continue cycle numbering from its predecessor when its first cycle immediately follows the predecessor's last cycle. The status is `"resolved"`, `"resolved_via_fallback: ..."`, or `"unresolved: ..."`. Phases ordered by the documented fallback rule still chain from the previous phase's computed end when that end is known. + +Unresolved phase ordering leaves every date in the variant null. A gap in explicit `phase_step` values leaves dates null from the later phase onward. Optional cycles are counted as given and marked `optional=True`; dates from the first optional cycle and every later phase carry a `resolved_via_fallback` status. + +Each rollout row includes `sig_id` (the source `Sigs.id`), `timing_sequence`, `cycle_length_lb`, `cycle_length_ub`, `cycle_length_unit`, and the `cycle_length_selection` used for this rollout. These fields identify the sig and its cycle definition when a later consumer can supply a missing phase duration. With `start_date`, a phase containing calendar-month or calendar-year cycles has null phase-relative offsets if its actual start date is unknown, since the number of days depends on that date. Without `start_date`, the existing 30-day month and 365-day year approximations apply. + +The timeline includes `modality` (`"systemic"`, `"radiation"`, or null for an unclassified sig), plus the source `component` and `component_cui`. Radiation rows can have a null `drug`. Pass `systemic_only=True` to omit radiation rows from the result; radiation phases still contribute to the timing of later phases. The timeline's `day_indefinite` and `cycle_indefinite` columns distinguish continuing days within a cycle from continuing cycles. + +`administration_frame` retains its cycle-local `day` column and adds `elapsed_day` and `timing_status`. It combines sigs for the same drug and day, so use `roll_out_variant` when phase-relative timing or sig identity is needed. Variants without resolvable cycle metadata still produce explicitly resolvable administration rows, with a null `elapsed_day` and an unresolved status rather than an invented calendar position. Numeric `(+k)` means continuation every k cycles in `timing_sequence` or every k days in `alldays`; the interval is retained without sampling future events. A continuing cycle marker prevents dates from being chained into a later phase. diff --git a/hemonc_alchemy/toolkit/analytics/treatment/classification.py b/hemonc_alchemy/toolkit/analytics/treatment/classification.py index 357f1ae..a675497 100644 --- a/hemonc_alchemy/toolkit/analytics/treatment/classification.py +++ b/hemonc_alchemy/toolkit/analytics/treatment/classification.py @@ -12,7 +12,7 @@ RAD_SIG_CLASS_VALUE = Sigs_Class_fieldEnum.RAD_SIG -def _sig_class_value(sig_or_value): +def sig_class_value(sig_or_value): value = getattr(sig_or_value, "class_field", sig_or_value) if isinstance(value, Sigs_Class_fieldEnum): return value @@ -26,7 +26,7 @@ def _sig_class_value(sig_or_value): def has_radiation_sig(variant) -> bool: """Whether any of a variant's component sigs are a radiation sig.""" - return any(_sig_class_value(sig) == RAD_SIG_CLASS_VALUE for sig in variant.component_sigs) + return any(sig_class_value(sig) == RAD_SIG_CLASS_VALUE for sig in variant.component_sigs) def has_non_radiation_sig(variant) -> bool: @@ -37,14 +37,14 @@ def has_non_radiation_sig(variant) -> bool: classification. """ return any( - (value := _sig_class_value(sig)) is not None + (value := sig_class_value(sig)) is not None and value != RAD_SIG_CLASS_VALUE for sig in variant.component_sigs ) def _has_unclassified_sig(variant) -> bool: - return any(_sig_class_value(sig) is None for sig in variant.component_sigs) + return any(sig_class_value(sig) is None for sig in variant.component_sigs) def is_concurrent_chemort(variant) -> bool: diff --git a/hemonc_alchemy/toolkit/analytics/treatment/scheduling/__init__.py b/hemonc_alchemy/toolkit/analytics/treatment/scheduling/__init__.py index 6356ecf..2a12dbb 100644 --- a/hemonc_alchemy/toolkit/analytics/treatment/scheduling/__init__.py +++ b/hemonc_alchemy/toolkit/analytics/treatment/scheduling/__init__.py @@ -11,23 +11,41 @@ home_administered_sigs_by_drug, schedule_events, ) +from .rollout import ( + AnchoredBlock, + CycleBlock, + TimedEvent, + UnresolvedTiming, + anchor_blocks, + group_into_blocks, + roll_out_phase, + roll_out_variant, +) from .routes import route_group from .tokens import Choice, Day, Indefinite, Range __all__ = [ + "AnchoredBlock", "Choice", + "CycleBlock", "Day", "Indefinite", "Range", "ResolvedSchedule", "ScheduleEvent", + "TimedEvent", + "UnresolvedTiming", "administration_frame", "administration_matrix", + "anchor_blocks", "cancer_services_drugs", "cancer_services_sigs_by_drug", + "group_into_blocks", "home_administered_drugs", "home_administered_sigs_by_drug", "resolve_all_days", + "roll_out_phase", + "roll_out_variant", "route_group", "schedule_events", ] diff --git a/hemonc_alchemy/toolkit/analytics/treatment/scheduling/handling.py b/hemonc_alchemy/toolkit/analytics/treatment/scheduling/handling.py index 6cd7dac..97ad10c 100644 --- a/hemonc_alchemy/toolkit/analytics/treatment/scheduling/handling.py +++ b/hemonc_alchemy/toolkit/analytics/treatment/scheduling/handling.py @@ -81,7 +81,10 @@ def parse_scalar_list(token: str): elif "|" in part: out.append(parse_choice(part)) else: - out.append(Day(int(part))) + try: + out.append(Day(int(part))) + except ValueError: + logger.warning("Unparseable dosing token %r dropped", part) return out @@ -102,6 +105,8 @@ def parse_optional(token: str): inner = token[1:-1] if inner.startswith("+"): + if match := re.fullmatch(r"\+([1-9][0-9]*)", inner): + return [Indefinite("+k", interval=int(match.group(1)))] match = re.fullmatch(r"\+([a-zA-Z])(\d+)?", inner) if not match: logger.warning("Unparseable indefinite-dosing token %r", token) @@ -148,9 +153,8 @@ def expand(parsed) -> ResolvedSchedule: ) else: indefinite = item - logger.warning( - "Indefinite-dosing marker %r found (continue until progression/indefinitely); " - "explicit days list is not the complete schedule", + logger.debug( + "Indefinite-dosing marker %r found; explicit days list is not the complete schedule", item, ) diff --git a/hemonc_alchemy/toolkit/analytics/treatment/scheduling/properties.py b/hemonc_alchemy/toolkit/analytics/treatment/scheduling/properties.py index c9ff604..1d3aa37 100644 --- a/hemonc_alchemy/toolkit/analytics/treatment/scheduling/properties.py +++ b/hemonc_alchemy/toolkit/analytics/treatment/scheduling/properties.py @@ -32,7 +32,7 @@ Sigs_FrequencyEnum, Sigs_PhaseEnum, ) -from .handling import Day, Indefinite, apply_sig_to_series, resolve_all_days +from .handling import Day, Indefinite, resolve_all_days from .routes import route_group DEFAULT_DECAY_DAYS = 2 @@ -48,6 +48,9 @@ "intensity", "optional", "indefinite", + "cycle_indefinite", + "elapsed_day", + "timing_status", ] @@ -143,6 +146,42 @@ def home_administered_sigs_by_drug(variant) -> dict: return _sigs_by_drug_where(variant, "PO") +def _rollout_frame_records( + variant, + *, + decay_days: int, + decay_factor: float, +) -> list[dict]: + from .rollout import roll_out_variant + + rolled = roll_out_variant( + variant, + decay_days=decay_days, + decay_factor=decay_factor, + ) + records = [] + for row in rolled.itertuples(index=False): + if row.route_group is None or row.drug_cui is None: + continue + records.append( + { + "variant_cui": row.variant_cui, + "variant": row.variant, + "route_group": row.route_group, + "drug_cui": row.drug_cui, + "drug": row.drug, + "day": row.day, + "intensity": row.intensity, + "optional": row.optional, + "indefinite": row.day_indefinite, + "cycle_indefinite": row.cycle_indefinite, + "elapsed_day": row.elapsed_day, + "timing_status": row.timing_status, + } + ) + return records + + def administration_frame( variants, *, @@ -159,9 +198,12 @@ def administration_frame( | `route_group` | `"IV"` (clinic) or `"PO"` (home) | | `drug_cui`, `drug` | the drug, by identifier and by name | | `day` | day of cycle; can be negative for lead-in dosing | + | `elapsed_day` | variant-relative day when cross-cycle timing resolves | | `intensity` | 1.0 on a dosing day, tapering over `decay_days` after | | `optional` | whether the dosing day itself was marked optional | - | `indefinite` | set when the sig continues past its stated days | + | `indefinite` | day-level marker when days continue past those stated | + | `cycle_indefinite` | cycle-level marker when cycles continue past those stated | + | `timing_status` | whether the rollout is resolved or needs review | `intensity` tapers after each dose by `decay_factor` per day for `decay_days`, so a treatment day and the days it encroaches on both @@ -169,8 +211,8 @@ def administration_frame( Rows whose route is unrecognised or not specified are excluded, as are sigs with no resolvable days -- including open-ended `EOC` ranges, so a - variant can legitimately produce no rows. Where `indefinite` is set, the - days present are only the part that was written down. + variant can legitimately produce no rows. Where `indefinite` or + `cycle_indefinite` is set, only the written days or cycles are returned. """ # Duck-typed rather than `isinstance(variants, Iterable)`: entities inherit # __iter__ from orm-loader's serialisation interface, so a single variant @@ -181,34 +223,11 @@ def administration_frame( records: list[dict] = [] for variant in variants: - for event in schedule_events(variant): - drug = event.drug_object - if event.route_group is None or drug is None or not event.days: - continue - - series: dict[int, float] = defaultdict(float) - apply_sig_to_series( - series, - list(event.days), - decay_days=decay_days, - decay_factor=decay_factor, - ) - optional_days = {day.value for day in event.days if day.optional} - - for day, intensity in series.items(): - records.append( - { - "variant_cui": variant.variant_cui, - "variant": variant.variant, - "route_group": event.route_group, - "drug_cui": drug.drug_cui, - "drug": drug.drug, - "day": day, - "intensity": intensity, - "optional": day in optional_days, - "indefinite": event.indefinite, - } - ) + records.extend(_rollout_frame_records( + variant, + decay_days=decay_days, + decay_factor=decay_factor, + )) if not records: return pd.DataFrame(columns=_FRAME_COLUMNS) @@ -219,15 +238,20 @@ def administration_frame( # decay tails can land on the same day; keep the strongest. grouped = ( frame.groupby( - ["variant_cui", "variant", "route_group", "drug_cui", "drug", "day"], + [ + "variant_cui", "variant", "route_group", "drug_cui", "drug", + "day", "elapsed_day", + ], as_index=False, dropna=False, ) .agg(intensity=("intensity", "max"), optional=("optional", "all"), - indefinite=("indefinite", "first")) + indefinite=("indefinite", "first"), + cycle_indefinite=("cycle_indefinite", "first"), + timing_status=("timing_status", "first")) ) return grouped[_FRAME_COLUMNS].sort_values( - ["variant_cui", "route_group", "drug", "day"], ignore_index=True + ["variant_cui", "route_group", "drug", "elapsed_day", "day"], ignore_index=True ) diff --git a/hemonc_alchemy/toolkit/analytics/treatment/scheduling/rollout.py b/hemonc_alchemy/toolkit/analytics/treatment/scheduling/rollout.py new file mode 100644 index 0000000..6dc382a --- /dev/null +++ b/hemonc_alchemy/toolkit/analytics/treatment/scheduling/rollout.py @@ -0,0 +1,902 @@ +"""Roll sig schedules out onto a variant-relative treatment timeline. + +The source ``day`` value is local to one cycle. This module adds the missing +context: cycle numbers, cycle lengths, block anchors, and phase boundaries. +It deliberately keeps the rollout deterministic. An indefinite source +marker is retained as metadata; it is never sampled here. + +Phases are ordered by the minimum ``phase_step`` present on their sigs. The +small fallback below is used only for missing or untested phase labels, and +its use is reported in ``timing_status``. Ties or internally inconsistent +phase steps remain unresolved rather than being guessed. +""" + +from __future__ import annotations + +import calendar +from collections import defaultdict +from collections.abc import Iterable +from dataclasses import dataclass +from datetime import date, datetime, timedelta +from decimal import Decimal, InvalidOperation +from typing import Any + +import pandas as pd # type: ignore[import-untyped] + +from .....model.enums import Sigs_Cycle_length_unitEnum, Sigs_PhaseEnum +from ..classification import RAD_SIG_CLASS_VALUE, sig_class_value +from .handling import Indefinite, apply_sig_to_series, resolve_all_days +from .properties import ScheduleEvent, schedule_events + +_FALLBACK_PHASE_RANKS = { + Sigs_PhaseEnum.PRE_TO_PHASE: -1.0, + Sigs_PhaseEnum.PERIOPERATIVE: 1.0, + Sigs_PhaseEnum.INTERIM_MAINTENANCE: 3.5, + Sigs_PhaseEnum.LATE_INTENSIFICATION: 4.5, + Sigs_PhaseEnum.CONTINUATION: 5.0, +} + + +@dataclass(frozen=True) +class UnresolvedTiming: + """A source-timing problem that must remain visible to callers.""" + + reason: str + + @property + def status(self) -> str: + return f"unresolved: {self.reason}" + + +@dataclass(frozen=True) +class CycleBlock: + """Events sharing a cycle length and a contiguous cycle-number set.""" + + events: tuple[ScheduleEvent, ...] + cycle_numbers: frozenset[int] + cycle_length_lb: str | None + cycle_length_ub: str | None + cycle_length_unit: Sigs_Cycle_length_unitEnum | None + timing_indefinite: Indefinite | None = None + optional_cycles: frozenset[int] = frozenset() + + @property + def first_cycle(self) -> int | None: + return min(self.cycle_numbers) if self.cycle_numbers else None + + @property + def last_cycle(self) -> int | None: + return max(self.cycle_numbers) if self.cycle_numbers else None + + @property + def is_contiguous(self) -> bool: + if not self.cycle_numbers: + return False + return len(self.cycle_numbers) == self.last_cycle - self.first_cycle + 1 + + +@dataclass(frozen=True) +class AnchoredBlock: + """A block plus the relationship that determines its start point.""" + + block: CycleBlock + anchor_kind: str + anchor_block: CycleBlock | None = None + anchor_cycle: int | None = None + unresolved: UnresolvedTiming | None = None + + @property + def timing_status(self) -> str: + if self.unresolved is not None: + return self.unresolved.status + return "resolved" + + +@dataclass(frozen=True) +class TimedEvent: + """One event/day result from a rolled-out block.""" + + schedule_event: ScheduleEvent + phase: Sigs_PhaseEnum | None + phase_step: int | None + cycle_number: int | None + day: int + elapsed_day: int | None + calendar_date: date | None + intensity: float + optional: bool + cycle_indefinite: Indefinite | None + timing_status: str + + +def _event_series( + event: ScheduleEvent, + *, + decay_days: int, + decay_factor: float, +) -> dict[int, float]: + series: dict[int, float] = defaultdict(float) + apply_sig_to_series( + series, + list(event.days), + decay_days=decay_days, + decay_factor=decay_factor, + ) + return series + + +def _timing_numbers( + value: str | None, +) -> tuple[frozenset[int], Indefinite | None, frozenset[int]]: + try: + resolved = resolve_all_days(value) + except (TypeError, ValueError): + return frozenset(), None, frozenset() + return ( + frozenset(day.value for day in resolved.days), + resolved.indefinite, + frozenset(day.value for day in resolved.days if day.optional), + ) + + +def group_into_blocks(events: Iterable[ScheduleEvent]) -> list[CycleBlock]: + """Group schedule events by cycle length and parsed cycle numbers.""" + + grouped: dict[tuple[Any, ...], list[ScheduleEvent]] = {} + for event in events: + cycle_numbers, indefinite, optional_cycles = _timing_numbers(event.timing_sequence) + key = ( + event.cycle_length_lb, + event.cycle_length_ub, + event.cycle_length_unit, + cycle_numbers, + indefinite, + optional_cycles, + ) + grouped.setdefault(key, []).append(event) + + blocks = [] + for key, block_events in grouped.items(): + cycle_numbers, indefinite, optional_cycles = key[3:] + blocks.append( + CycleBlock( + events=tuple(block_events), + cycle_numbers=cycle_numbers, + cycle_length_lb=key[0], + cycle_length_ub=key[1], + cycle_length_unit=key[2], + timing_indefinite=indefinite, + optional_cycles=optional_cycles, + ) + ) + return sorted( + blocks, + key=lambda block: ( + block.first_cycle is None, + block.first_cycle if block.first_cycle is not None else 0, + block.last_cycle if block.last_cycle is not None else 0, + ), + ) + + +def _overlap_conflict(block: CycleBlock, overlaps: list[AnchoredBlock]) -> UnresolvedTiming | None: + length = (block.cycle_length_lb, block.cycle_length_ub, block.cycle_length_unit) + shared = set().union(*( + prior.block.cycle_numbers & block.cycle_numbers for prior in overlaps + )) + if len(shared) > 1 and any( + (prior.block.cycle_length_lb, prior.block.cycle_length_ub, prior.block.cycle_length_unit) + != length for prior in overlaps + ): + return UnresolvedTiming( + f"cycle lengths differ across shared cycles {min(shared)}-{max(shared)}" + ) + return None + + +def anchor_blocks( + blocks: Iterable[CycleBlock], *, preceding_cycle: int | None = None +) -> list[AnchoredBlock]: + """Assign sequential, overlapping, or unresolved block anchors.""" + + ordered = sorted( + blocks, + key=lambda block: ( + block.first_cycle is None, + block.first_cycle if block.first_cycle is not None else 0, + block.last_cycle if block.last_cycle is not None else 0, + ), + ) + anchored: list[AnchoredBlock] = [] + + for index, block in enumerate(ordered): + if not block.cycle_numbers: + anchored.append( + AnchoredBlock( + block=block, + anchor_kind="unresolved", + unresolved=UnresolvedTiming("missing cycle numbers"), + ) + ) + continue + if not block.is_contiguous: + anchored.append( + AnchoredBlock( + block=block, + anchor_kind="unresolved", + unresolved=UnresolvedTiming( + f"non-contiguous cycle numbers {sorted(block.cycle_numbers)}" + ), + ) + ) + continue + if index == 0: + # Cycle numbers may continue when the preceding phase ends immediately before. + if block.first_cycle > 1 and preceding_cycle != block.first_cycle - 1: + anchored.append( + AnchoredBlock( + block=block, + anchor_kind="unresolved", + unresolved=UnresolvedTiming( + f"missing preceding cycles before cycle {block.first_cycle}" + ), + ) + ) + continue + anchored.append( + AnchoredBlock( + block=block, + anchor_kind="phase_start", + anchor_cycle=block.first_cycle, + ) + ) + continue + + overlaps = [ + prior + for prior in anchored + if prior.block.cycle_numbers & block.cycle_numbers + ] + if overlaps: + if conflict := _overlap_conflict(block, overlaps): + anchored.append( + AnchoredBlock( + block=block, + anchor_kind="unresolved", + unresolved=conflict, + ) + ) + continue + prior = overlaps[-1] + intersection = prior.block.cycle_numbers & block.cycle_numbers + anchored.append( + AnchoredBlock( + block=block, + anchor_kind="overlap", + anchor_block=prior.block, + anchor_cycle=min(intersection), + ) + ) + continue + + previous = ordered[index - 1] + if previous.last_cycle is not None and block.first_cycle == previous.last_cycle + 1: + predecessors = [ + prior.block for prior in anchored + if prior.block.last_cycle == previous.last_cycle + ] + lengths = { + (prior.cycle_length_lb, prior.cycle_length_ub, prior.cycle_length_unit) + for prior in predecessors + } + if len(lengths) > 1: + anchored.append( + AnchoredBlock( + block=block, + anchor_kind="unresolved", + unresolved=UnresolvedTiming( + f"blocks ending cycle {previous.last_cycle} differ in cycle length" + ), + ) + ) + continue + anchored.append( + AnchoredBlock( + block=block, + anchor_kind="after", + anchor_block=previous, + anchor_cycle=block.first_cycle, + ) + ) + continue + + anchored.append( + AnchoredBlock( + block=block, + anchor_kind="unresolved", + unresolved=UnresolvedTiming( + f"cycle gap or contradiction before cycles {sorted(block.cycle_numbers)}" + ), + ) + ) + + return anchored + + +def _unit_value(unit: Sigs_Cycle_length_unitEnum | str | None) -> str | None: + if unit is None: + return None + return getattr(unit, "value", unit).lower() + + +def _length_number(block: CycleBlock, selection: str) -> Decimal: + if selection not in {"lb", "ub", "mean"}: + raise ValueError("cycle_length_selection must be 'lb', 'ub', or 'mean'") + try: + lower = Decimal(str(block.cycle_length_lb)) + upper = Decimal(str(block.cycle_length_ub or block.cycle_length_lb)) + except (InvalidOperation, TypeError): + raise ValueError("cycle length is not numeric") from None + if selection == "lb": + return lower + if selection == "ub": + return upper + return (lower + upper) / 2 + + +def _calendar_add(value: date, number: int, unit: str) -> date: + if unit == "day": + return value + timedelta(days=number) + if unit == "week": + return value + timedelta(days=number * 7) + if unit == "month": + month_index = value.month - 1 + number + year, month_index = divmod(value.year * 12 + month_index, 12) + month = month_index + 1 + day = min(value.day, calendar.monthrange(year, month)[1]) + return date(year, month, day) + if unit == "year": + year = value.year + number + day = min(value.day, calendar.monthrange(year, value.month)[1]) + return date(year, value.month, day) + raise ValueError(f"unsupported calendar cycle unit {unit!r}") + + +def _advance( + start: int | date, + block: CycleBlock, + cycles: int, + *, + selection: str, +) -> int | date: + unit = _unit_value(block.cycle_length_unit) + if unit not in {"day", "week", "month", "year"}: + raise ValueError("cycle length has no resolvable unit") + number = _length_number(block, selection) + + if isinstance(start, date): + if number != number.to_integral_value(): + raise ValueError("calendar cycle length is not a whole number") + if unit in {"month", "year"}: + units = int(number) * cycles + return _calendar_add(start, units, unit) + days = int(number) * (7 if unit == "week" else 1) * cycles + return start + timedelta(days=days) + + if unit == "month": + number *= 30 + elif unit == "week": + number *= 7 + elif unit == "year": + number *= 365 + if number != number.to_integral_value(): + raise ValueError("elapsed cycle length is not a whole number of days") + return start + int(number) * cycles + + +def _block_start( + anchored: AnchoredBlock, + starts: dict[int, int | date], + ends: dict[int, int | date], + phase_start: int | date, + *, + selection: str, +) -> int | date | None: + if anchored.unresolved is not None: + return None + if anchored.anchor_kind == "phase_start": + return phase_start + if anchored.anchor_block is None: + return None + parent_start = starts.get(id(anchored.anchor_block)) + if parent_start is None: + return None + if anchored.anchor_kind == "after": + return ends.get(id(anchored.anchor_block)) + if anchored.anchor_kind == "overlap": + try: + cycle_delta = anchored.anchor_cycle - min(anchored.anchor_block.cycle_numbers) + return _advance( + parent_start, + anchored.anchor_block, + cycle_delta, + selection=selection, + ) + except ValueError: + return None + return None + + +def _phase_end( + anchored_blocks: list[AnchoredBlock], + phase_start: int | date, + *, + selection: str, +) -> int | date | None: + starts: dict[int, int | date] = {} + ends: dict[int, int | date] = {} + for anchored in anchored_blocks: + start = _block_start(anchored, starts, ends, phase_start, selection=selection) + if start is None: + continue + starts[id(anchored.block)] = start + if anchored.block.timing_indefinite is not None: + continue + try: + end = _advance( + start, + anchored.block, + anchored.block.last_cycle - anchored.anchor_cycle + 1, + selection=selection, + ) + except (TypeError, ValueError): + continue + ends[id(anchored.block)] = end + if not anchored_blocks or any( + block.unresolved is not None or block.block.timing_indefinite is not None + for block in anchored_blocks + ): + return None + if len(ends) != len(anchored_blocks): + return None + return max(ends.values()) + + +def _optional_for_day( + event: ScheduleEvent, day: int, block: CycleBlock, cycle_number: int | None +) -> bool: + return cycle_number in block.optional_cycles or any( + source_day.value == day and source_day.optional for source_day in event.days + ) + + +def _unresolved_block_events( + block: CycleBlock, + status: UnresolvedTiming, + *, + decay_days: int, + decay_factor: float, +) -> list[TimedEvent]: + output = [] + cycle_numbers = sorted(block.cycle_numbers) or [None] + for event in block.events: + series = _event_series(event, decay_days=decay_days, decay_factor=decay_factor) + for cycle_number in cycle_numbers: + for day, intensity in series.items(): + output.append( + TimedEvent( + schedule_event=event, + phase=event.phase, + phase_step=event.phase_step, + cycle_number=cycle_number, + day=day, + elapsed_day=None, + calendar_date=None, + intensity=intensity, + optional=_optional_for_day(event, day, block, cycle_number), + cycle_indefinite=block.timing_indefinite, + timing_status=status.status, + ) + ) + return output + + +def _resolved_block_events( + anchored: AnchoredBlock, + start: int | date, + *, + selection: str, + decay_days: int, + decay_factor: float, +) -> list[TimedEvent]: + output = [] + block = anchored.block + for event in block.events: + series = _event_series(event, decay_days=decay_days, decay_factor=decay_factor) + for cycle_number in sorted(block.cycle_numbers): + cycle_start = _advance( + start, + block, + cycle_number - anchored.anchor_cycle, + selection=selection, + ) + for day, intensity in series.items(): + if isinstance(cycle_start, date): + calendar_date = cycle_start + timedelta(days=day - 1) + elapsed_day = None + else: + calendar_date = None + elapsed_day = cycle_start + day - 1 + output.append( + TimedEvent( + schedule_event=event, + phase=event.phase, + phase_step=event.phase_step, + cycle_number=cycle_number, + day=day, + elapsed_day=elapsed_day, + calendar_date=calendar_date, + intensity=intensity, + optional=_optional_for_day(event, day, block, cycle_number), + cycle_indefinite=block.timing_indefinite, + timing_status=anchored.timing_status, + ) + ) + return output + + +def roll_out_phase( + blocks: Iterable[CycleBlock] | Iterable[AnchoredBlock], + phase_start: int | date, + *, + cycle_length_selection: str = "lb", + decay_days: int = 2, + decay_factor: float = 0.5, +) -> list[TimedEvent]: + """Roll one phase out from its own start point. + + ``phase_start`` is an integer elapsed-day offset or a ``date``. Unresolved + blocks still produce rows with null timeline values and an explicit status. + """ + + block_list = list(blocks) + anchored = ( + block_list + if all(isinstance(block, AnchoredBlock) for block in block_list) + else anchor_blocks(block_list) # type: ignore[arg-type] + ) + starts: dict[int, int | date] = {} + ends: dict[int, int | date] = {} + output: list[TimedEvent] = [] + + for anchored_block in anchored: + block = anchored_block.block + start = _block_start( + anchored_block, + starts, + ends, + phase_start, + selection=cycle_length_selection, + ) + if start is not None: + try: + end = _advance( + start, + block, + 1 if block.timing_indefinite is not None + else block.last_cycle - anchored_block.anchor_cycle + 1, + selection=cycle_length_selection, + ) + except (TypeError, ValueError): + start = None + else: + starts[id(block)] = start + if block.timing_indefinite is None: + ends[id(block)] = end + + if start is None: + status = anchored_block.unresolved or UnresolvedTiming( + "anchor depends on unresolved timing" + ) + output.extend( + _unresolved_block_events( + block, + status, + decay_days=decay_days, + decay_factor=decay_factor, + ) + ) + continue + + output.extend( + _resolved_block_events( + anchored_block, + start, + selection=cycle_length_selection, + decay_days=decay_days, + decay_factor=decay_factor, + ) + ) + return output + + +def _phase_groups(events: list[ScheduleEvent]) -> list[tuple[Any, list[ScheduleEvent]]]: + groups: dict[Any, list[ScheduleEvent]] = {} + for event in events: + groups.setdefault(event.phase, []).append(event) + return list(groups.items()) + + +def _phase_order( + groups: list[tuple[Any, list[ScheduleEvent]]], +) -> tuple[list[tuple[Any, list[ScheduleEvent], str]], str]: + if len(groups) <= 1: + return [(phase, events, "resolved") for phase, events in groups], "resolved" + + phase_steps: dict[Any, int | None] = {} + inconsistent: list[Any] = [] + for phase, events in groups: + steps = {event.phase_step for event in events} + if len(steps) != 1: + inconsistent.append(phase) + phase_steps[phase] = None + else: + phase_steps[phase] = next(iter(steps)) + + known = {phase: step for phase, step in phase_steps.items() if step is not None} + if inconsistent: + reason = "phase_step inconsistent within " + ", ".join(map(str, inconsistent)) + status = f"unresolved: {reason}" + return [(phase, events, status) for phase, events in groups], status + if len(set(known.values())) != len(known): + tied = [str(phase) for phase, step in phase_steps.items() if list(phase_steps.values()).count(step) > 1] + reason = "phase_step tie between " + ", ".join(tied) + status = f"unresolved: {reason}" + return [(phase, events, status) for phase, events in groups], status + + fallback_used = any(phase in _FALLBACK_PHASE_RANKS for phase, _ in groups) + ranks: dict[Any, float] = {} + for phase, step in phase_steps.items(): + if fallback_used and phase in _FALLBACK_PHASE_RANKS: + ranks[phase] = _FALLBACK_PHASE_RANKS[phase] + elif step is not None: + ranks[phase] = float(step) + else: + reason = f"no phase_step or fallback rule for {phase}" + status = f"unresolved: {reason}" + return [(p, e, status) for p, e in groups], status + + if len(set(ranks.values())) != len(ranks): + reason = "fallback phase order tie between " + ", ".join(map(str, ranks)) + status = f"unresolved: {reason}" + return [(phase, events, status) for phase, events in groups], status + + status = "resolved" + if fallback_used: + fallback_labels = ", ".join( + phase.value for phase, _ in groups if phase in _FALLBACK_PHASE_RANKS + ) + status = ( + f"resolved_via_fallback: {fallback_labels} ordered by documented " + "convention, not phase_step" + ) + ordered = sorted(groups, key=lambda item: ranks[item[0]]) + return [(phase, events, status) for phase, events in ordered], status + + +def _combine_status(*statuses: str) -> str: + if any(status.startswith("unresolved:") for status in statuses): + reasons = [ + reason + for status in statuses if status.startswith("unresolved:") + for reason in status.removeprefix("unresolved: ").split("; ") + ] + return "unresolved: " + "; ".join(dict.fromkeys(reasons)) + fallbacks = [ + reason + for status in statuses if status.startswith("resolved_via_fallback:") + for reason in status.removeprefix("resolved_via_fallback: ").split("; ") + ] + if fallbacks: + return "resolved_via_fallback: " + "; ".join(dict.fromkeys(fallbacks)) + return "resolved" + + +def _optional_cycle_status(cycles: Iterable[int]) -> str: + numbers = sorted(set(cycles)) + if not numbers: + return "resolved" + label = "cycle" if len(numbers) == 1 else "cycles" + values = ", ".join(map(str, numbers)) + return f"resolved_via_fallback: optional {label} {values} assumed given" + + +def _phase_end_failure(blocks: Iterable[CycleBlock]) -> str: + continuing = next( + (block.timing_indefinite for block in blocks if block.timing_indefinite is not None), + None, + ) + if continuing is None: + return "unresolved: previous phase end is unresolved" + if continuing.interval is None: + return "unresolved: previous phase continues indefinitely" + unit = "cycle" if continuing.interval == 1 else "cycles" + return f"unresolved: previous phase continues (every {continuing.interval} {unit})" + + +def _as_date(value: date | datetime) -> date: + return value.date() if isinstance(value, datetime) else value + + +def _phase_relative_day( + event: TimedEvent, phase_start: int | date, *, calendar_start_known: bool +) -> int | None: + if isinstance(phase_start, date): + if not calendar_start_known or event.calendar_date is None: + return None + return (event.calendar_date - phase_start).days + if event.elapsed_day is None: + return None + return event.elapsed_day - phase_start + + +def roll_out_variant( + variant, + *, + start_date: date | datetime | None = None, + cycle_length_selection: str = "lb", + decay_days: int = 2, + decay_factor: float = 0.5, + systemic_only: bool = False, +) -> pd.DataFrame: + """Compose all phases of ``variant`` into one deterministic timeline.""" + + events = schedule_events(variant) + groups = _phase_groups(events) + ordered, order_status = _phase_order(groups) + phase_start: int | date = _as_date(start_date) if start_date is not None else 0 + all_timed: list[tuple[TimedEvent, int | None]] = [] + timeline_failure = order_status if order_status.startswith("unresolved:") else None + optional_assumptions: list[str] = [] + previous_last_cycle: int | None = None + phase_steps = {phase_events[0].phase_step for _, phase_events, _ in ordered} + known_steps = sorted(step for step in phase_steps if step is not None) + missing_steps = ( + sorted(set(range(known_steps[0], known_steps[-1] + 1)) - phase_steps) + if known_steps else [] + ) + + for phase, phase_events, phase_status in ordered: + current_step = phase_events[0].phase_step + missing_before = [ + step for step in missing_steps if step < current_step + ] if current_step is not None else [] + if timeline_failure is None and missing_before: + missing = ", ".join(map(str, missing_before)) + timeline_failure = ( + f"unresolved: phase_step gap before step {current_step} " + f"(no sigs for step {missing})" + ) + blocks = group_into_blocks(phase_events) + optional_cycles = sorted({ + cycle for block in blocks for cycle in block.optional_cycles + }) + phase_anchored = anchor_blocks(blocks, preceding_cycle=previous_last_cycle) + timed = roll_out_phase( + phase_anchored, + phase_start, + cycle_length_selection=cycle_length_selection, + decay_days=decay_days, + decay_factor=decay_factor, + ) + calendar_start_known = timeline_failure is None or all( + _unit_value(block.cycle_length_unit) not in {"month", "year"} + for block in blocks + ) + phase_optional_status = _optional_cycle_status(optional_cycles) + for event in timed: + row_optional_status = ( + phase_optional_status + if optional_cycles + and event.cycle_number is not None + and event.cycle_number >= optional_cycles[0] + else "resolved" + ) + status = _combine_status( + event.timing_status, + phase_status, + timeline_failure or "resolved", + *optional_assumptions, + row_optional_status, + ) + calendar_date = event.calendar_date if timeline_failure is None else None + elapsed_day = event.elapsed_day if timeline_failure is None else None + phase_elapsed_day = _phase_relative_day( + event, phase_start, calendar_start_known=calendar_start_known + ) + if start_date is not None and calendar_date is not None: + elapsed_day = (calendar_date - _as_date(start_date)).days + all_timed.append( + ( + TimedEvent( + schedule_event=event.schedule_event, + phase=phase, + phase_step=event.phase_step, + cycle_number=event.cycle_number, + day=event.day, + elapsed_day=elapsed_day, + calendar_date=calendar_date, + intensity=event.intensity, + optional=event.optional, + cycle_indefinite=event.cycle_indefinite, + timing_status=status, + ), + phase_elapsed_day, + ) + ) + + phase_end = _phase_end(phase_anchored, phase_start, selection=cycle_length_selection) + if phase_end is None and timeline_failure is None: + timeline_failure = _phase_end_failure(blocks) + if phase_end is not None and timeline_failure is None: + phase_start = phase_end + previous_last_cycle = max(block.last_cycle for block in blocks) + else: + previous_last_cycle = None + if optional_cycles: + optional_assumptions.append(phase_optional_status) + + records = [] + for timed, phase_elapsed_day in all_timed: + event = timed.schedule_event + sig_class = sig_class_value(event.sig) + modality = ( + "radiation" if sig_class == RAD_SIG_CLASS_VALUE + else "systemic" if sig_class is not None else None + ) + if systemic_only and modality == "radiation": + continue + drug = event.drug_object + records.append( + { + "variant_cui": variant.variant_cui, + "variant": variant.variant, + "phase": timed.phase, + "phase_step": timed.phase_step, + "cycle_number": timed.cycle_number, + "sig_id": event.sig.id, + "timing_sequence": event.timing_sequence, + "cycle_length_lb": event.cycle_length_lb, + "cycle_length_ub": event.cycle_length_ub, + "cycle_length_unit": event.cycle_length_unit, + "cycle_length_selection": cycle_length_selection, + "route_group": event.route_group, + "modality": modality, + "component_cui": event.sig.component_cui, + "component": event.sig.component, + "drug_cui": drug.drug_cui if drug is not None else None, + "drug": drug.drug if drug is not None else None, + "day": timed.day, + "elapsed_day": timed.elapsed_day, + "phase_elapsed_day": phase_elapsed_day, + "calendar_date": timed.calendar_date, + "intensity": timed.intensity, + "optional": timed.optional, + "day_indefinite": event.indefinite, + "cycle_indefinite": timed.cycle_indefinite, + "timing_status": timed.timing_status, + } + ) + columns = [ + "variant_cui", "variant", "phase", "phase_step", "cycle_number", + "sig_id", "timing_sequence", "cycle_length_lb", "cycle_length_ub", + "cycle_length_unit", "cycle_length_selection", + "route_group", "modality", "component_cui", "component", + "drug_cui", "drug", "day", "elapsed_day", "phase_elapsed_day", + "calendar_date", "intensity", "optional", "day_indefinite", + "cycle_indefinite", "timing_status", + ] + frame = pd.DataFrame.from_records(records, columns=columns) + if order_status != "resolved" and frame.empty: + frame.attrs["timing_status"] = order_status + return frame diff --git a/hemonc_alchemy/toolkit/analytics/treatment/scheduling/tokens.py b/hemonc_alchemy/toolkit/analytics/treatment/scheduling/tokens.py index 13c3e30..ee708b0 100644 --- a/hemonc_alchemy/toolkit/analytics/treatment/scheduling/tokens.py +++ b/hemonc_alchemy/toolkit/analytics/treatment/scheduling/tokens.py @@ -9,7 +9,8 @@ | `[1,5,1]` | every day from 1 to 5, step 1 | a `Range` | | `(4)` | optional day | `Day(4, optional=True)` | | `1\\|2` | one of these days, unspecified which | a `Choice` | -| `(+c)`, `(+n21)` | continues indefinitely, optionally capped in days | an `Indefinite` | +| `(+c)`, `(+n21)` | continues beyond the explicit days, optionally capped | an `Indefinite` | +| `(+2)` | continues every two days in `alldays` or every two cycles in `timing_sequence` | an `Indefinite(interval=2)` | Day numbers can be negative (`-14`), meaning that many days before day 1 — conditioning or lead-in dosing. @@ -52,8 +53,9 @@ class Range: @dataclass(frozen=True) class Indefinite: - kind: str # "+n" or "+c" + kind: str # "+n", "+c", or "+k" max_days: int | None = None + interval: int | None = None # cycles in timing_sequence, days in alldays TOKEN_RE = re.compile( diff --git a/tests/test_schedule.py b/tests/test_schedule.py index aef4ba8..6c5864d 100644 --- a/tests/test_schedule.py +++ b/tests/test_schedule.py @@ -2,6 +2,8 @@ from __future__ import annotations +import logging + from hemonc_alchemy.model.enums import Sigs_RouteEnum from hemonc_alchemy.toolkit.analytics.treatment.scheduling import ( Day, @@ -70,15 +72,42 @@ def test_range_expansion(self): resolved = resolve_all_days("[1,21,7]") assert resolved.days == (Day(1), Day(8), Day(15)) - def test_indefinite_marker_is_preserved_not_dropped(self, caplog): + def test_indefinite_marker_is_preserved_and_logged_at_debug(self, caplog): # Previously: expand() silently `continue`d past Indefinite tokens, - # so a caller had no way to tell a maintenance/continuation regimen - # was truncated. Now the marker survives and a warning is logged. - resolved = resolve_all_days("1,8,15,(+n)") + # so a caller had no way to tell a maintenance regimen was truncated. + with caplog.at_level(logging.DEBUG): + resolved = resolve_all_days("1,8,15,(+n)") assert resolved.days == (Day(1), Day(8), Day(15)) assert resolved.indefinite == Indefinite(kind="+n", max_days=None) assert bool(resolved) is True - assert any("indefinite" in message.lower() for message in caplog.messages) + assert any("Indefinite-dosing marker" in record.message for record in caplog.records) + assert all(record.levelno < logging.WARNING for record in caplog.records) + + def test_known_numeric_markers_do_not_warn(self, caplog): + resolve_all_days("1,(+2)") + resolve_all_days("1,(+2)") + assert not caplog.records + + def test_unparseable_marker_still_warns(self, caplog): + resolve_all_days("1,(+bogus)") + assert any(record.levelno == logging.WARNING for record in caplog.records) + + def test_zero_continuation_interval_is_rejected(self, caplog): + resolved = resolve_all_days("1,(+0)") + assert resolved.days == (Day(1),) + assert resolved.indefinite is None + assert any("Unparseable indefinite-dosing token" in message for message in caplog.messages) + + def test_unspecified_scalar_is_dropped_with_warning(self, caplog): + resolved = resolve_all_days("NS") + assert resolved.days == () + assert resolved.indefinite is None + assert any("NS" in message for message in caplog.messages) + + def test_bare_eoc_is_dropped_without_losing_valid_days(self, caplog): + resolved = resolve_all_days("1,EOC") + assert resolved.days == (Day(1),) + assert any("EOC" in message for message in caplog.messages) def test_indefinite_only_schedule_is_still_truthy(self): resolved = resolve_all_days("(+c5)") @@ -86,6 +115,11 @@ def test_indefinite_only_schedule_is_still_truthy(self): assert resolved.indefinite == Indefinite(kind="+c", max_days=5) assert bool(resolved) is True + def test_numeric_continuation_keeps_interval_without_expanding(self): + resolved = resolve_all_days("1,(+2)") + assert resolved.days == (Day(1),) + assert resolved.indefinite == Indefinite(kind="+k", interval=2) + def test_empty_input(self): resolved = resolve_all_days(None) assert resolved.days == () diff --git a/tests/test_schedule_properties.py b/tests/test_schedule_properties.py index e6305c9..1824947 100644 --- a/tests/test_schedule_properties.py +++ b/tests/test_schedule_properties.py @@ -7,7 +7,7 @@ from __future__ import annotations -from datetime import UTC, datetime +from datetime import UTC, date, datetime import pytest import sqlalchemy as sa @@ -16,12 +16,22 @@ from hemonc_alchemy.model.base import Base from hemonc_alchemy.model.entities import Drugs, Sigs, Variants +from hemonc_alchemy.model.enums import ( + Sigs_Class_fieldEnum, + Sigs_Cycle_length_unitEnum, + Sigs_PhaseEnum, +) from hemonc_alchemy.toolkit.analytics.treatment.scheduling import ( + Indefinite, + UnresolvedTiming, administration_frame, administration_matrix, + anchor_blocks, cancer_services_drugs, cancer_services_sigs_by_drug, + group_into_blocks, home_administered_drugs, + roll_out_variant, schedule_events, ) @@ -31,6 +41,9 @@ ) _D = datetime(2020, 1, 1, tzinfo=UTC) +_IV_SIG = "IV_INTERMITTENT_CANONICAL_SIG" +_NONIV_SIG = "NON_TO_IV_CANONICAL_SIG" +_IV_CONT_SIG = "IV_CONTINUOUS_CANONICAL_SIG" @pytest.fixture @@ -66,18 +79,61 @@ def _drug(session, drug_cui: int, name: str) -> Drugs: return drug -def _sig(session, *, sig_id: int, variant_cui: int, drug_cui: int, route: str, alldays: str) -> Sigs: +def _sig( + session, + *, + sig_id: int, + variant_cui: int, + drug_cui: int, + route: str, + alldays: str, + **overrides, +) -> Sigs: + timing_sequence = overrides.pop("timing_sequence", None) + cycle_length_lb = overrides.pop("cycle_length_lb", None) + cycle_length_ub = overrides.pop("cycle_length_ub", None) + cycle_length_unit = overrides.pop("cycle_length_unit", None) + phase = overrides.pop("phase", None) + phase_step = overrides.pop("phase_step", 1) + class_field = overrides.pop("class_field", "iv intermittent canonical sig") + component = overrides.pop("component", f"c{drug_cui}") + assert not overrides sig = Sigs( - id=sig_id, variant_cui=variant_cui, component_cui=drug_cui, component=f"c{drug_cui}", - class_field="iv intermittent canonical sig", component_role="primary systemic", + id=sig_id, variant_cui=variant_cui, component_cui=drug_cui, component=component, + class_field=class_field, component_role="primary systemic", portion="1", regimen="R", regimen_cui=1, step_number="1", divided=False, - phase_step=1, variant=f"v{variant_cui}", route=route, alldays=alldays, date_added=_D, + phase=phase, phase_step=phase_step, variant=f"v{variant_cui}", route=route, + alldays=alldays, timing_sequence=timing_sequence, + cycle_length_lb=cycle_length_lb, cycle_length_ub=cycle_length_ub, + cycle_length_unit=cycle_length_unit, date_added=_D, ) session.add(sig) session.flush() return sig +def _source_variant(session, variant_cui, rows): + """Mirror selected development database sigs in sqlite.""" + variant = _variant(session, variant_cui) + for component_cui, component in dict.fromkeys((row[1], row[0]) for row in rows): + _drug(session, component_cui, component) + for sig_id, ( + component, component_cui, alldays, timing_sequence, length, + unit, route, sig_class, phase, phase_step, + ) in enumerate(rows, start=1): + _sig( + session, sig_id=sig_id, variant_cui=variant_cui, drug_cui=component_cui, + route=route, alldays=alldays, timing_sequence=timing_sequence, + cycle_length_lb=length, cycle_length_ub=length, + cycle_length_unit=Sigs_Cycle_length_unitEnum[unit], + class_field=Sigs_Class_fieldEnum[sig_class], component=component, + phase=Sigs_PhaseEnum[phase] if phase is not None else None, + phase_step=phase_step, + ) + session.expire_all() + return variant + + class TestScheduleEvents: def test_one_event_per_sig_with_its_days_resolved(self, session): variant = _variant(session, 1) @@ -102,6 +158,934 @@ def test_an_indefinite_schedule_is_flagged_not_dropped(self, session): assert [day.value for day in event.days] == [1] +class TestRollout: + def _two_block_variant(self, session, variant_cui=90): + variant = _variant(session, variant_cui) + _drug(session, 1, "docetaxel") + _drug(session, 2, "trastuzumab") + _sig( + session, sig_id=1, variant_cui=variant_cui, drug_cui=1, + route="INTRAVENOUS", alldays="1", + timing_sequence="1,2,3,4,5,6,7,8", + cycle_length_lb="2", cycle_length_ub="2", + cycle_length_unit=Sigs_Cycle_length_unitEnum.WEEK, + ) + _sig( + session, sig_id=2, variant_cui=variant_cui, drug_cui=2, + route="INTRAVENOUS", alldays="1", + timing_sequence="9,10,11,12,13,14,15,16,17,18,19,20", + cycle_length_lb="3", cycle_length_ub="3", + cycle_length_unit=Sigs_Cycle_length_unitEnum.WEEK, + ) + session.expire_all() + return variant + + def test_groups_by_cycle_length_and_cycle_numbers(self, session): + variant = self._two_block_variant(session) + + blocks = group_into_blocks(schedule_events(variant)) + + assert len(blocks) == 2 + assert [block.cycle_numbers for block in blocks] == [ + frozenset(range(1, 9)), frozenset(range(9, 21)), + ] + + def test_sequential_and_unresolved_anchors(self, session): + variant = self._two_block_variant(session, variant_cui=91) + blocks = group_into_blocks(schedule_events(variant)) + + anchored = anchor_blocks(blocks) + assert [item.anchor_kind for item in anchored] == ["phase_start", "after"] + + variant_gap = _variant(session, 92) + _drug(session, 3, "cyclophosphamide") + _sig( + session, sig_id=3, variant_cui=92, drug_cui=3, + route="INTRAVENOUS", alldays="1", timing_sequence="1,2", + cycle_length_lb="14", cycle_length_ub="14", + cycle_length_unit=Sigs_Cycle_length_unitEnum.DAY, + ) + _sig( + session, sig_id=4, variant_cui=92, drug_cui=3, + route="INTRAVENOUS", alldays="1", timing_sequence="4,5", + cycle_length_lb="14", cycle_length_ub="14", + cycle_length_unit=Sigs_Cycle_length_unitEnum.DAY, + ) + session.expire_all() + gap = anchor_blocks(group_into_blocks(schedule_events(variant_gap))) + assert isinstance(gap[-1].unresolved, UnresolvedTiming) + + def test_first_block_after_cycle_one_has_no_variant_start_date(self, session): + variant = _variant(session, 109) + _drug(session, 1, "cycle-three-drug") + _drug(session, 2, "cycle-four-drug") + for sig_id, drug_cui, cycles in ((1, 1, "3"), (2, 2, "4")): + _sig( + session, sig_id=sig_id, variant_cui=109, drug_cui=drug_cui, + route="INTRAVENOUS", alldays="1", timing_sequence=cycles, + cycle_length_lb="21", cycle_length_ub="21", + cycle_length_unit=Sigs_Cycle_length_unitEnum.DAY, + ) + session.expire_all() + + anchored = anchor_blocks(group_into_blocks(schedule_events(variant))) + assert anchored[0].unresolved is not None + frame = roll_out_variant(variant, start_date=date(2020, 1, 1), decay_days=0) + assert set(frame["cycle_number"]) == {3, 4} + assert frame["elapsed_day"].isna().all() + assert frame["phase_elapsed_day"].isna().all() + assert frame["calendar_date"].isna().all() + assert frame["timing_status"].str.startswith("unresolved:").all() + + def test_phase_cycle_numbers_can_continue_from_previous_phase(self, session): + # LVP Sandwich Variant #01 carries cycle numbers across its phases. + variant = _variant(session, 111) + for cui in (1, 2, 3): + _drug(session, cui, f"drug-{cui}") + for sig_id, drug_cui, cycles, phase, step in ( + (1, 1, "1,2", Sigs_PhaseEnum.INDUCTION, 1), + (2, 2, "3", Sigs_PhaseEnum.DEFINITIVE, 2), + (3, 3, "4,5", Sigs_PhaseEnum.CONSOLIDATION, 3), + ): + _sig( + session, sig_id=sig_id, variant_cui=111, drug_cui=drug_cui, + route="INTRAVENOUS", alldays="1", timing_sequence=cycles, + cycle_length_lb="21", cycle_length_ub="21", + cycle_length_unit=Sigs_Cycle_length_unitEnum.DAY, + phase=phase, phase_step=step, + ) + session.expire_all() + + frame = roll_out_variant(variant, decay_days=0) + assert frame.loc[frame["phase"] == Sigs_PhaseEnum.DEFINITIVE, "elapsed_day"].iloc[0] == 42 + assert frame.loc[frame["phase"] == Sigs_PhaseEnum.CONSOLIDATION, "elapsed_day"].min() == 63 + assert set(frame["timing_status"]) == {"resolved"} + + def test_phase_cycle_number_jump_remains_unresolved(self, session): + variant = _variant(session, 112) + _drug(session, 1, "induction-drug") + _drug(session, 2, "consolidation-drug") + for sig_id, drug_cui, cycles, phase, step in ( + (1, 1, "1,2", Sigs_PhaseEnum.INDUCTION, 1), + (2, 2, "4", Sigs_PhaseEnum.CONSOLIDATION, 2), + ): + _sig( + session, sig_id=sig_id, variant_cui=112, drug_cui=drug_cui, + route="INTRAVENOUS", alldays="1", timing_sequence=cycles, + cycle_length_lb="21", cycle_length_ub="21", + cycle_length_unit=Sigs_Cycle_length_unitEnum.DAY, + phase=phase, phase_step=step, + ) + session.expire_all() + + frame = roll_out_variant(variant, decay_days=0) + later = frame[frame["phase"] == Sigs_PhaseEnum.CONSOLIDATION] + assert later["elapsed_day"].isna().all() + assert later["timing_status"].str.startswith("unresolved:").all() + + def test_overlapping_blocks_share_an_anchor(self, session): + variant = _variant(session, 93) + _drug(session, 1, "carfilzomib") + _drug(session, 2, "lenalidomide") + _sig( + session, sig_id=1, variant_cui=93, drug_cui=1, + route="INTRAVENOUS", alldays="1", timing_sequence="1,2,3", + cycle_length_lb="28", cycle_length_ub="28", + cycle_length_unit=Sigs_Cycle_length_unitEnum.DAY, + ) + _sig( + session, sig_id=2, variant_cui=93, drug_cui=2, + route="ORAL", alldays="1", timing_sequence="1", + cycle_length_lb="14", cycle_length_ub="14", + cycle_length_unit=Sigs_Cycle_length_unitEnum.DAY, + ) + session.expire_all() + + anchored = anchor_blocks(group_into_blocks(schedule_events(variant))) + assert [item.anchor_kind for item in anchored] == ["phase_start", "overlap"] + + def test_three_overlaps_reject_conflicting_cycle_starts(self, session): + variant = _variant(session, 119) + for cui in (1, 2, 3): + _drug(session, cui, f"drug-{cui}") + for sig_id, start, stop, length in ((1, 1, 5, 7), (2, 3, 7, 14), (3, 5, 9, 21)): + _sig( + session, sig_id=sig_id, variant_cui=119, drug_cui=sig_id, + route="INTRAVENOUS", alldays="1", + timing_sequence=",".join(map(str, range(start, stop + 1))), + cycle_length_lb=str(length), cycle_length_ub=str(length), + cycle_length_unit=Sigs_Cycle_length_unitEnum.DAY, + ) + session.expire_all() + + frame = roll_out_variant(variant, decay_days=0) + first = frame[frame["drug"] == "drug-1"] + second = frame[frame["drug"] == "drug-2"] + third = frame[frame["drug"] == "drug-3"] + assert first.loc[first["cycle_number"] == 5, "elapsed_day"].iloc[0] == 28 + assert set(first["timing_status"]) == {"resolved"} + assert second["elapsed_day"].isna().all() + assert set(second["timing_status"]) == { + "unresolved: cycle lengths differ across shared cycles 3-5" + } + assert third["elapsed_day"].isna().all() + assert set(third["timing_status"]) == { + "unresolved: cycle lengths differ across shared cycles 5-7" + } + + @pytest.mark.parametrize( + ("later_cycles", "expected_status"), + [ + ("2,3,4", "unresolved: cycle lengths differ across shared cycles 2-3"), + ("3,4,5", "resolved"), + ], + ) + def test_different_length_overlap_needs_multiple_shared_cycles( + self, session, later_cycles, expected_status + ): + variant = _variant(session, 120) + _drug(session, 1, "earlier-drug") + _drug(session, 2, "later-drug") + for sig_id, drug_cui, cycles, length in ( + (1, 1, "1,2,3", 7), + (2, 2, later_cycles, 14), + ): + _sig( + session, sig_id=sig_id, variant_cui=120, drug_cui=drug_cui, + route="INTRAVENOUS", alldays="1", timing_sequence=cycles, + cycle_length_lb=str(length), cycle_length_ub=str(length), + cycle_length_unit=Sigs_Cycle_length_unitEnum.DAY, + ) + session.expire_all() + + frame = roll_out_variant(variant, decay_days=0) + later = frame[frame["drug"] == "later-drug"] + assert set(later["timing_status"]) == {expected_status} + if expected_status == "resolved": + assert later.loc[later["cycle_number"] == 3, "elapsed_day"].iloc[0] == 14 + else: + assert later["elapsed_day"].isna().all() + + def test_single_shared_cycle_with_two_prior_blocks_stays_resolved(self, session): + variant = _variant(session, 125) + for cui, name in ((1, "carboplatin"), (2, "paclitaxel"), (3, "ipilimumab")): + _drug(session, cui, name) + for sig_id, cycles, length in ( + (1, "1,2,3,4", 3), + (2, "1,2,3,4,5,6", 3), + (3, "3", 12), + ): + _sig( + session, sig_id=sig_id, variant_cui=125, drug_cui=sig_id, + route="INTRAVENOUS", alldays="1", timing_sequence=cycles, + cycle_length_lb=str(length), cycle_length_ub=str(length), + cycle_length_unit=Sigs_Cycle_length_unitEnum.WEEK, + ) + session.expire_all() + + frame = roll_out_variant(variant, decay_days=0) + ipilimumab = frame[frame["drug"] == "ipilimumab"] + assert ipilimumab["elapsed_day"].iloc[0] == 42 + assert set(ipilimumab["timing_status"]) == {"resolved"} + + @pytest.mark.parametrize( + ("predecessor_lengths", "expected_day", "expected_status"), + [ + ((4, 3, 1), None, "unresolved: blocks ending cycle 1 differ in cycle length"), + ((4, 4, 4), 28, "resolved"), + ], + ) + def test_following_cycle_checks_all_predecessor_lengths( + self, session, predecessor_lengths, expected_day, expected_status + ): + # Variant 131178 has three cycle-1 blocks; sort order puts the one-week + # block last, but cycle 2 cannot be anchored to that block alone. + variant = _variant(session, 127) + for cui, name in ( + (1, "cobimetinib"), + (2, "vemurafenib-early"), + (3, "vemurafenib-late"), + (4, "atezolizumab"), + ): + _drug(session, cui, name) + for sig_id, cycles, length in ( + (1, "1,(+1)", predecessor_lengths[0]), + (2, "1", predecessor_lengths[1]), + (3, "1", predecessor_lengths[2]), + (4, "2,(+1)", 4), + ): + _sig( + session, sig_id=sig_id, variant_cui=127, drug_cui=sig_id, + route="ORAL", alldays="1", timing_sequence=cycles, + cycle_length_lb=str(length), cycle_length_ub=str(length), + cycle_length_unit=Sigs_Cycle_length_unitEnum.WEEK, + ) + session.expire_all() + + frame = roll_out_variant(variant, decay_days=0) + cycle_two = frame[frame["drug"] == "atezolizumab"] + if expected_day is None: + assert cycle_two["elapsed_day"].isna().all() + else: + assert cycle_two["elapsed_day"].iloc[0] == expected_day + assert set(cycle_two["timing_status"]) == {expected_status} + + def test_three_blocks_sharing_only_cycle_five_stay_resolved(self, session): + variant = _variant(session, 126) + for cui in (1, 2, 3): + _drug(session, cui, f"drug-{cui}") + for sig_id, cycles, length in ( + (1, "1,2,3,4,5", 7), + (2, "5", 14), + (3, "5,6,7,8,9", 21), + ): + _sig( + session, sig_id=sig_id, variant_cui=126, drug_cui=sig_id, + route="INTRAVENOUS", alldays="1", timing_sequence=cycles, + cycle_length_lb=str(length), cycle_length_ub=str(length), + cycle_length_unit=Sigs_Cycle_length_unitEnum.DAY, + ) + session.expire_all() + + frame = roll_out_variant(variant, decay_days=0) + cycle_five = frame[frame["cycle_number"] == 5] + assert set(cycle_five["elapsed_day"]) == {28} + assert set(cycle_five["timing_status"]) == {"resolved"} + + def test_same_cycle_length_pattern_change_forms_two_blocks(self, session): + variant = _variant(session, 99) + _drug(session, 1, "carfilzomib") + _drug(session, 2, "dexamethasone") + _sig( + session, sig_id=1, variant_cui=99, drug_cui=1, + route="INTRAVENOUS", alldays="1,2,8,9,15,16", + timing_sequence="2,3,4,5,6,7,8,9,10,11,12", + cycle_length_lb="4", cycle_length_ub="4", + cycle_length_unit=Sigs_Cycle_length_unitEnum.WEEK, + ) + _sig( + session, sig_id=2, variant_cui=99, drug_cui=2, + route="ORAL", alldays="1,15", + timing_sequence="13,14,15,16,17,18", + cycle_length_lb="4", cycle_length_ub="4", + cycle_length_unit=Sigs_Cycle_length_unitEnum.WEEK, + ) + session.expire_all() + + blocks = group_into_blocks(schedule_events(variant)) + + assert len(blocks) == 2 + assert {block.cycle_length_lb for block in blocks} == {"4"} + + def test_rollout_uses_previous_block_duration(self, session): + variant = self._two_block_variant(session, variant_cui=94) + + frame = roll_out_variant(variant) + + first = frame[(frame["drug"] == "docetaxel") & (frame["cycle_number"] == 8)] + second = frame[(frame["drug"] == "trastuzumab") & (frame["cycle_number"] == 9)] + assert first["elapsed_day"].iloc[0] == 98 + assert second["elapsed_day"].iloc[0] == 112 + assert set(frame["timing_status"]) == {"resolved"} + + def test_rollout_can_return_calendar_dates(self, session): + variant = self._two_block_variant(session, variant_cui=95) + + frame = roll_out_variant(variant, start_date=date(2020, 1, 1)) + + second = frame[(frame["drug"] == "trastuzumab") & (frame["cycle_number"] == 9)] + assert second["calendar_date"].iloc[0] == date(2020, 4, 22) + assert second["elapsed_day"].iloc[0] == 112 + + def test_phase_rollout_chains_at_full_cycle_end(self, session): + variant = _variant(session, 96) + _drug(session, 1, "induction-drug") + _drug(session, 2, "maintenance-drug") + _sig( + session, sig_id=1, variant_cui=96, drug_cui=1, + route="INTRAVENOUS", alldays="1", timing_sequence="1,2", + cycle_length_lb="14", cycle_length_ub="14", + cycle_length_unit=Sigs_Cycle_length_unitEnum.DAY, + phase=Sigs_PhaseEnum.INDUCTION, phase_step=1, + ) + _sig( + session, sig_id=2, variant_cui=96, drug_cui=2, + route="INTRAVENOUS", alldays="1", timing_sequence="1,2", + cycle_length_lb="21", cycle_length_ub="21", + cycle_length_unit=Sigs_Cycle_length_unitEnum.DAY, + phase=Sigs_PhaseEnum.MAINTENANCE, phase_step=2, + ) + session.expire_all() + + frame = roll_out_variant(variant) + maintenance = frame[frame["phase"] == Sigs_PhaseEnum.MAINTENANCE] + assert maintenance["elapsed_day"].min() == 28 + assert set(frame["timing_status"]) == {"resolved"} + + @pytest.mark.parametrize("start_date", [None, date(2020, 1, 1)]) + def test_phase_step_gap_nulls_later_phases(self, session, start_date): + variant = _variant(session, 113) + for cui in (1, 2, 3): + _drug(session, cui, f"drug-{cui}") + for sig_id, drug_cui, phase, step in ( + (1, 1, Sigs_PhaseEnum.NEOADJUVANT, 1), + (2, 2, Sigs_PhaseEnum.ADJUVANT, 3), + (3, 3, Sigs_PhaseEnum.MAINTENANCE, 5), + ): + _sig( + session, sig_id=sig_id, variant_cui=113, drug_cui=drug_cui, + route="INTRAVENOUS", alldays="1", timing_sequence="1,2", + cycle_length_lb="21", cycle_length_ub="21", + cycle_length_unit=Sigs_Cycle_length_unitEnum.DAY, + phase=phase, phase_step=step, + ) + session.expire_all() + + frame = roll_out_variant(variant, start_date=start_date, decay_days=0) + assert frame.loc[frame["phase"] == Sigs_PhaseEnum.NEOADJUVANT, "elapsed_day"].iloc[0] == 0 + later = frame[frame["phase"] != Sigs_PhaseEnum.NEOADJUVANT] + assert later["elapsed_day"].isna().all() + assert later["calendar_date"].isna().all() + assert set(later["timing_status"]) == { + "unresolved: phase_step gap before step 3 (no sigs for step 2)" + } + for phase in (Sigs_PhaseEnum.NEOADJUVANT, Sigs_PhaseEnum.ADJUVANT, + Sigs_PhaseEnum.MAINTENANCE): + phase_rows = frame[frame["phase"] == phase] + assert set(phase_rows["phase_elapsed_day"]) == {0, 21} + assert set(frame["sig_id"]) == {1, 2, 3} + assert set(frame["cycle_length_lb"]) == {"21"} + assert set(frame["cycle_length_ub"]) == {"21"} + assert set(frame["cycle_length_unit"]) == {Sigs_Cycle_length_unitEnum.DAY} + assert set(frame["cycle_length_selection"]) == {"lb"} + + def test_phase_step_gap_with_fallback_order_nulls_later_phase(self, session): + variant = _variant(session, 122) + _drug(session, 1, "perioperative-drug") + _drug(session, 2, "adjuvant-drug") + for sig_id, drug_cui, phase, step in ( + (1, 1, Sigs_PhaseEnum.PERIOPERATIVE, 1), + (2, 2, Sigs_PhaseEnum.ADJUVANT, 3), + ): + _sig( + session, sig_id=sig_id, variant_cui=122, drug_cui=drug_cui, + route="INTRAVENOUS", alldays="1", timing_sequence="1,2", + cycle_length_lb="21", cycle_length_ub="21", + cycle_length_unit=Sigs_Cycle_length_unitEnum.DAY, + phase=phase, phase_step=step, + ) + session.expire_all() + + frame = roll_out_variant(variant, decay_days=0) + perioperative = frame[frame["phase"] == Sigs_PhaseEnum.PERIOPERATIVE] + adjuvant = frame[frame["phase"] == Sigs_PhaseEnum.ADJUVANT] + assert set(perioperative["elapsed_day"]) == {0, 21} + assert adjuvant["elapsed_day"].isna().all() + assert set(adjuvant["timing_status"]) == { + "unresolved: phase_step gap before step 3 (no sigs for step 2)" + } + + @pytest.mark.parametrize( + ("start_date", "expected_phase_days"), + [(None, {0, 30}), (date(2020, 1, 31), None)], + ) + def test_unknown_calendar_month_start_has_no_exact_phase_offset( + self, session, start_date, expected_phase_days + ): + variant = _source_variant(session, 129, [ + ("neoadjuvant-drug", 1, "1", "1,2", "1", "DAY", "INTRAVENOUS", _IV_SIG, "NEOADJUVANT", 1), + ("adjuvant-drug", 2, "1", "1,2", "1", "MONTH", "INTRAVENOUS", _IV_SIG, "ADJUVANT", 3), + ]) + + frame = roll_out_variant(variant, start_date=start_date, decay_days=0) + adjuvant = frame[frame["phase"] == Sigs_PhaseEnum.ADJUVANT] + assert adjuvant["elapsed_day"].isna().all() + if expected_phase_days is None: + assert adjuvant["phase_elapsed_day"].isna().all() + else: + assert set(adjuvant["phase_elapsed_day"]) == expected_phase_days + + def test_known_calendar_month_start_uses_actual_month_length(self, session): + variant = _source_variant(session, 130, [ + ("monthly-drug", 1, "1", "1,2", "1", "MONTH", "INTRAVENOUS", _IV_SIG, "MAINTENANCE", 1), + ]) + + frame = roll_out_variant(variant, start_date=date(2020, 1, 31), decay_days=0) + assert set(frame["phase_elapsed_day"]) == {0, 29} + assert set(frame["elapsed_day"]) == {0, 29} + + def test_phase_offset_records_selected_cycle_length_bound(self, session): + variant = _variant(session, 131) + _drug(session, 1, "ranged-cycle-drug") + _sig( + session, sig_id=1, variant_cui=131, drug_cui=1, + route="INTRAVENOUS", alldays="1", timing_sequence="1,2", + cycle_length_lb="14", cycle_length_ub="21", + cycle_length_unit=Sigs_Cycle_length_unitEnum.DAY, + ) + session.expire_all() + + frame = roll_out_variant(variant, cycle_length_selection="ub", decay_days=0) + assert set(frame["phase_elapsed_day"]) == {0, 21} + assert set(frame["cycle_length_selection"]) == {"ub"} + + def test_fallback_gap_checks_step_span_not_loop_neighbours(self, session): + variant = _variant(session, 123) + _drug(session, 1, "perioperative-drug") + _drug(session, 2, "adjuvant-drug") + for sig_id, drug_cui, phase, step in ( + (1, 1, Sigs_PhaseEnum.PERIOPERATIVE, 4), + (2, 2, Sigs_PhaseEnum.ADJUVANT, 2), + ): + _sig( + session, sig_id=sig_id, variant_cui=123, drug_cui=drug_cui, + route="INTRAVENOUS", alldays="1", timing_sequence="1", + cycle_length_lb="21", cycle_length_ub="21", + cycle_length_unit=Sigs_Cycle_length_unitEnum.DAY, + phase=phase, phase_step=step, + ) + session.expire_all() + + frame = roll_out_variant(variant, decay_days=0) + assert frame["elapsed_day"].isna().all() + assert set(frame["timing_status"]) == { + "unresolved: phase_step gap before step 4 (no sigs for step 3)" + } + + def test_optional_cycle_marks_rows_and_following_phase(self, session): + variant = _variant(session, 114) + _drug(session, 1, "induction-drug") + _drug(session, 2, "maintenance-drug") + for sig_id, drug_cui, cycles, phase, step in ( + (1, 1, "1,2,(3),(4)", Sigs_PhaseEnum.INDUCTION, 1), + (2, 2, "1", Sigs_PhaseEnum.MAINTENANCE, 2), + ): + _sig( + session, sig_id=sig_id, variant_cui=114, drug_cui=drug_cui, + route="INTRAVENOUS", alldays="1", timing_sequence=cycles, + cycle_length_lb="14", cycle_length_ub="14", + cycle_length_unit=Sigs_Cycle_length_unitEnum.DAY, + phase=phase, phase_step=step, + ) + session.expire_all() + + frame = roll_out_variant(variant, decay_days=0) + induction = frame[frame["phase"] == Sigs_PhaseEnum.INDUCTION] + assert induction.loc[induction["cycle_number"] == 3, "optional"].all() + assert set(induction.loc[induction["cycle_number"] < 3, "timing_status"]) == {"resolved"} + assert set(induction.loc[induction["cycle_number"] == 3, "timing_status"]) == { + "resolved_via_fallback: optional cycles 3, 4 assumed given" + } + assert set(induction.loc[induction["cycle_number"] == 4, "timing_status"]) == { + "resolved_via_fallback: optional cycles 3, 4 assumed given" + } + maintenance = frame[frame["phase"] == Sigs_PhaseEnum.MAINTENANCE] + assert maintenance["elapsed_day"].iloc[0] == 56 + assert set(maintenance["timing_status"]) == { + "resolved_via_fallback: optional cycles 3, 4 assumed given" + } + + def test_later_block_in_same_phase_inherits_optional_cycle_assumption(self, session): + variant = _variant(session, 115) + _drug(session, 1, "optional-drug") + _drug(session, 2, "later-drug") + for sig_id, drug_cui, cycles in ((1, 1, "1,(2)"), (2, 2, "3")): + _sig( + session, sig_id=sig_id, variant_cui=115, drug_cui=drug_cui, + route="INTRAVENOUS", alldays="1", timing_sequence=cycles, + cycle_length_lb="14", cycle_length_ub="14", + cycle_length_unit=Sigs_Cycle_length_unitEnum.DAY, + ) + session.expire_all() + + frame = roll_out_variant(variant, decay_days=0) + assert frame.loc[frame["cycle_number"] == 1, "timing_status"].iloc[0] == "resolved" + assert set(frame.loc[frame["cycle_number"] >= 2, "timing_status"]) == { + "resolved_via_fallback: optional cycle 2 assumed given" + } + + def test_optional_cycle_and_phase_order_keep_both_fallback_reasons(self, session): + variant = _variant(session, 116) + _drug(session, 1, "perioperative-drug") + _drug(session, 2, "later-drug") + for sig_id, drug_cui, cycles, phase, step in ( + (1, 1, "1,(2)", Sigs_PhaseEnum.PERIOPERATIVE, 1), + (2, 2, "1", Sigs_PhaseEnum.MAINTENANCE, 2), + ): + _sig( + session, sig_id=sig_id, variant_cui=116, drug_cui=drug_cui, + route="INTRAVENOUS", alldays="1", timing_sequence=cycles, + cycle_length_lb="14", cycle_length_ub="14", + cycle_length_unit=Sigs_Cycle_length_unitEnum.DAY, + phase=phase, phase_step=step, + ) + session.expire_all() + + frame = roll_out_variant(variant, decay_days=0) + assert set(frame.loc[frame["phase"] == Sigs_PhaseEnum.MAINTENANCE, "timing_status"]) == { + "resolved_via_fallback: perioperative ordered by documented convention, not phase_step; optional cycle 2 assumed given" + } + + def test_numeric_continuation_prevents_later_phase_chaining(self, session): + variant = _variant(session, 117) + _drug(session, 1, "continuing-drug") + _drug(session, 2, "later-drug") + for sig_id, drug_cui, cycles, phase, step in ( + (1, 1, "1,(+2)", Sigs_PhaseEnum.INDUCTION, 1), + (2, 2, "1", Sigs_PhaseEnum.MAINTENANCE, 2), + ): + _sig( + session, sig_id=sig_id, variant_cui=117, drug_cui=drug_cui, + route="INTRAVENOUS", alldays="1", timing_sequence=cycles, + cycle_length_lb="21", cycle_length_ub="21", + cycle_length_unit=Sigs_Cycle_length_unitEnum.DAY, + phase=phase, phase_step=step, + ) + session.expire_all() + + frame = roll_out_variant(variant, decay_days=0) + induction = frame[frame["phase"] == Sigs_PhaseEnum.INDUCTION] + maintenance = frame[frame["phase"] == Sigs_PhaseEnum.MAINTENANCE] + assert induction["elapsed_day"].iloc[0] == 0 + assert induction["cycle_indefinite"].iloc[0] == Indefinite(kind="+k", interval=2) + assert maintenance["elapsed_day"].isna().all() + assert maintenance["phase_elapsed_day"].iloc[0] == 0 + assert set(maintenance["timing_status"]) == { + "unresolved: previous phase continues (every 2 cycles)" + } + + @pytest.mark.parametrize( + ("length", "unit"), + [("21", None), ("NS", Sigs_Cycle_length_unitEnum.DAY)], + ) + def test_indefinite_block_with_bad_cycle_length_is_unresolved(self, session, length, unit): + variant = _variant(session, 124) + _drug(session, 1, "continuing-drug") + _sig( + session, sig_id=1, variant_cui=124, drug_cui=1, + route="INTRAVENOUS", alldays="1", timing_sequence="1,(+1)", + cycle_length_lb=length, cycle_length_ub=length, + cycle_length_unit=unit, + ) + session.expire_all() + + for frame in ( + roll_out_variant(variant, decay_days=0), + administration_frame(variant, decay_days=0), + ): + assert len(frame) == 1 + assert frame["elapsed_day"].isna().all() + assert frame["timing_status"].str.startswith("unresolved:").all() + + def test_numeric_day_continuation_is_preserved(self, session): + variant = _variant(session, 118) + _drug(session, 1, "continuing-days-drug") + _sig( + session, sig_id=1, variant_cui=118, drug_cui=1, + route="INTRAVENOUS", alldays="1,(+30)", timing_sequence="1", + cycle_length_lb="30", cycle_length_ub="30", + cycle_length_unit=Sigs_Cycle_length_unitEnum.DAY, + ) + session.expire_all() + + frame = roll_out_variant(variant, decay_days=0) + assert frame["day_indefinite"].iloc[0] == Indefinite(kind="+k", interval=30) + assert set(frame["day"]) == {1} + + def test_phase_end_uses_longest_overlapping_block(self, session): + variant = _variant(session, 101) + _drug(session, 1, "long-course") + _drug(session, 2, "cycle-three") + _drug(session, 3, "next-phase") + for sig_id, drug_cui, cycles, phase, step in ( + (1, 1, "1,2,3,4,5,6", Sigs_PhaseEnum.INDUCTION, 1), + (2, 2, "3", Sigs_PhaseEnum.INDUCTION, 1), + (3, 3, "1", Sigs_PhaseEnum.MAINTENANCE, 2), + ): + _sig( + session, sig_id=sig_id, variant_cui=101, drug_cui=drug_cui, + route="INTRAVENOUS", alldays="1", timing_sequence=cycles, + cycle_length_lb="21", cycle_length_ub="21", + cycle_length_unit=Sigs_Cycle_length_unitEnum.DAY, + phase=phase, phase_step=step, + ) + session.expire_all() + + frame = roll_out_variant(variant, decay_days=0) + assert frame.loc[frame["drug"] == "next-phase", "elapsed_day"].iloc[0] == 126 + + def test_unresolved_block_end_prevents_phase_chaining(self, session): + variant = _variant(session, 107) + for cui in (1, 2, 3): + _drug(session, cui, f"drug-{cui}") + for sig_id, drug_cui, cycles, length, phase, step in ( + (1, 1, "1,2,3", "21", Sigs_PhaseEnum.INDUCTION, 1), + (2, 2, "2", "not-a-number", Sigs_PhaseEnum.INDUCTION, 1), + (3, 3, "1", "21", Sigs_PhaseEnum.MAINTENANCE, 2), + ): + _sig( + session, sig_id=sig_id, variant_cui=107, drug_cui=drug_cui, + route="INTRAVENOUS", alldays="1", timing_sequence=cycles, + cycle_length_lb=length, cycle_length_ub=length, + cycle_length_unit=Sigs_Cycle_length_unitEnum.DAY, + phase=phase, phase_step=step, + ) + session.expire_all() + + frame = roll_out_variant(variant, decay_days=0) + next_phase = frame[frame["phase"] == Sigs_PhaseEnum.MAINTENANCE] + assert set(next_phase["timing_status"]) == { + "unresolved: previous phase end is unresolved" + } + assert next_phase["elapsed_day"].isna().all() + assert next_phase["phase_elapsed_day"].iloc[0] == 0 + + def test_fallback_ordered_phases_chain(self, session): + variant = _variant(session, 102) + _drug(session, 1, "perioperative-drug") + _drug(session, 2, "adjuvant-drug") + for sig_id, drug_cui, phase, step in ( + (1, 1, Sigs_PhaseEnum.PERIOPERATIVE, 1), + (2, 2, Sigs_PhaseEnum.ADJUVANT, 2), + ): + _sig( + session, sig_id=sig_id, variant_cui=102, drug_cui=drug_cui, + route="INTRAVENOUS", alldays="1", timing_sequence="1,2", + cycle_length_lb="14", cycle_length_ub="14", + cycle_length_unit=Sigs_Cycle_length_unitEnum.DAY, + phase=phase, phase_step=step, + ) + session.expire_all() + + frame = roll_out_variant(variant, decay_days=0) + adjuvant = frame[frame["phase"] == Sigs_PhaseEnum.ADJUVANT] + assert adjuvant["elapsed_day"].min() == 28 + assert adjuvant["timing_status"].str.startswith("resolved_via_fallback:").all() + + def test_empty_rollout_keeps_prefixed_order_status(self, session): + variant = _variant(session, 103) + _drug(session, 1, "perioperative-drug") + _drug(session, 2, "adjuvant-drug") + for sig_id, drug_cui, phase, step in ( + (1, 1, Sigs_PhaseEnum.PERIOPERATIVE, 1), + (2, 2, Sigs_PhaseEnum.ADJUVANT, 2), + ): + _sig( + session, sig_id=sig_id, variant_cui=103, drug_cui=drug_cui, + route="INTRAVENOUS", alldays="[1,EOC,7]", timing_sequence="1", + cycle_length_lb="14", cycle_length_ub="14", + cycle_length_unit=Sigs_Cycle_length_unitEnum.DAY, + phase=phase, phase_step=step, + ) + session.expire_all() + + frame = roll_out_variant(variant) + assert frame.empty + assert frame.attrs["timing_status"].startswith("resolved_via_fallback:") + + @pytest.mark.parametrize( + ("variant_cui", "alldays", "timing_sequence", "day_indefinite", "cycle_indefinite"), + [ + (104, "1,(+c)", "1", True, False), + (105, "1", "1,(+n)", False, True), + ], + ) + def test_indefinite_markers_identify_their_scope( + self, session, variant_cui, alldays, timing_sequence, + day_indefinite, cycle_indefinite, + ): + variant = _variant(session, variant_cui) + _drug(session, 1, "continuing-drug") + _sig( + session, sig_id=1, variant_cui=variant_cui, drug_cui=1, + route="ORAL", alldays=alldays, timing_sequence=timing_sequence, + cycle_length_lb="21", cycle_length_ub="21", + cycle_length_unit=Sigs_Cycle_length_unitEnum.DAY, + ) + session.expire_all() + + rolled = roll_out_variant(variant, decay_days=0) + administered = administration_frame(variant, decay_days=0) + assert (rolled["day_indefinite"].notna().all()) == day_indefinite + assert (rolled["cycle_indefinite"].notna().all()) == cycle_indefinite + assert (administered["indefinite"].notna().all()) == day_indefinite + assert (administered["cycle_indefinite"].notna().all()) == cycle_indefinite + + def test_radiation_rows_are_labeled_and_anchor_later_phases(self, session): + variant = _variant(session, 106) + _drug(session, 1, "adjuvant-drug") + _sig( + session, sig_id=1, variant_cui=106, drug_cui=999, + route="NS", alldays="1", timing_sequence="1,2", + cycle_length_lb="14", cycle_length_ub="14", + cycle_length_unit=Sigs_Cycle_length_unitEnum.DAY, + phase=Sigs_PhaseEnum.DEFINITIVE, phase_step=1, + class_field=Sigs_Class_fieldEnum.RAD_SIG, component="radiotherapy", + ) + _sig( + session, sig_id=2, variant_cui=106, drug_cui=1, + route="INTRAVENOUS", alldays="1", timing_sequence="1", + cycle_length_lb="21", cycle_length_ub="21", + cycle_length_unit=Sigs_Cycle_length_unitEnum.DAY, + phase=Sigs_PhaseEnum.ADJUVANT, phase_step=2, + component="adjuvant-drug", + ) + session.expire_all() + + frame = roll_out_variant(variant, decay_days=0) + radiation = frame[frame["modality"] == "radiation"] + assert len(radiation) == 2 + assert radiation["drug"].isna().all() + assert set(radiation["component"]) == {"radiotherapy"} + assert set(radiation["component_cui"]) == {999} + assert frame.loc[frame["modality"] == "systemic", "elapsed_day"].iloc[0] == 28 + + systemic = roll_out_variant(variant, decay_days=0, systemic_only=True) + assert set(systemic["modality"]) == {"systemic"} + assert systemic["elapsed_day"].iloc[0] == 28 + + def test_unclassified_sig_keeps_unknown_modality(self, session): + variant = _variant(session, 108) + _drug(session, 1, "unknown-class-drug") + _sig( + session, sig_id=1, variant_cui=108, drug_cui=1, + route="INTRAVENOUS", alldays="1", timing_sequence="1", + cycle_length_lb="21", cycle_length_ub="21", + cycle_length_unit=Sigs_Cycle_length_unitEnum.DAY, + class_field=None, + ) + session.expire_all() + + assert roll_out_variant(variant, systemic_only=True)["modality"].isna().all() + + def test_phase_step_tie_is_unresolved(self, session): + variant = _variant(session, 97) + _drug(session, 1, "induction-drug") + _drug(session, 2, "consolidation-drug") + for sig_id, drug_cui, phase in ( + (1, 1, Sigs_PhaseEnum.INDUCTION), + (2, 2, Sigs_PhaseEnum.CONSOLIDATION), + ): + _sig( + session, sig_id=sig_id, variant_cui=97, drug_cui=drug_cui, + route="INTRAVENOUS", alldays="1", timing_sequence="1", + cycle_length_lb="14", cycle_length_ub="14", + cycle_length_unit=Sigs_Cycle_length_unitEnum.DAY, + phase=phase, phase_step=1, + ) + session.expire_all() + + frame = roll_out_variant(variant, start_date=date(2020, 1, 1)) + assert frame["timing_status"].str.startswith("unresolved:").all() + assert frame["elapsed_day"].isna().all() + assert frame["calendar_date"].isna().all() + assert set(frame["phase_elapsed_day"]) == {0, 1, 2} + + +class TestRealVariantSnapshots: + """Source rows from the development HemOnc database on 2026-09-24.""" + + def test_ddfec_ddth_cycle_length_change(self, session): + variant = _source_variant(session, 131576, [ + ("Cyclophosphamide", 122, "1", "1,2,3,4", "2", "WEEK", "INTRAVENOUS", _IV_SIG, None, 1), + ("Docetaxel", 164, "1", "5,6,7,8", "2", "WEEK", "INTRAVENOUS", _IV_SIG, None, 1), + ("Epirubicin", 191, "1", "1,2,3,4", "2", "WEEK", "INTRAVENOUS", _IV_SIG, None, 1), + ("Filgrastim", 220, "[3,10,1]", "1,2,3,4,5,6,7,8", "2", "WEEK", "SUBCUTANEOUS", _NONIV_SIG, None, 1), + ("Fluorouracil", 225, "1", "1,2,3,4", "2", "WEEK", "INTRAVENOUS", _IV_SIG, None, 1), + ("Trastuzumab", 512, "1", "5", "2", "WEEK", "INTRAVENOUS", _IV_SIG, None, 1), + ("Trastuzumab", 512, "1", "6,7,8", "2", "WEEK", "INTRAVENOUS", _IV_SIG, None, 1), + ("Trastuzumab", 512, "1", "9,10,11,12,13,14,15,16,17,18,19,20", "3", "WEEK", "INTRAVENOUS", _IV_SIG, None, 1), + ]) + frame = roll_out_variant(variant, start_date=date(2020, 1, 1), decay_days=0) + cycle_nine = frame[(frame["component"] == "Trastuzumab") & (frame["cycle_number"] == 9)] + # Eight two-week cycles end 112 days after 2020-01-01. + assert cycle_nine["elapsed_day"].iloc[0] == 112 + assert cycle_nine["calendar_date"].iloc[0] == date(2020, 4, 22) + assert set(frame["timing_status"]) == {"resolved"} + + def test_dara_krd_pattern_changes_within_four_week_cycles(self, session): + variant = _source_variant(session, 131485, [ + ("Carfilzomib", 89, "[8,15,7]", "1", "4", "WEEK", "INTRAVENOUS", _IV_SIG, None, 1), + ("Carfilzomib", 89, "1", "1", "4", "WEEK", "INTRAVENOUS", _IV_SIG, None, 1), + ("Carfilzomib", 89, "[1,15,7]", "2,3,4,5,6,7,8", "4", "WEEK", "INTRAVENOUS", _IV_SIG, None, 1), + ("Daratumumab", 139, "[1,22,7]", "1,2", "4", "WEEK", "INTRAVENOUS", _IV_SIG, None, 1), + ("Daratumumab", 139, "[1,15,14]", "3,4,5,6", "4", "WEEK", "INTRAVENOUS", _IV_SIG, None, 1), + ("Daratumumab", 139, "1", "7,8", "4", "WEEK", "INTRAVENOUS", _IV_SIG, None, 1), + ("Dexamethasone", 156, "[1,22,7]", "1,2,3,4", "4", "WEEK", "INTRAVENOUS", _IV_SIG, None, 1), + ("Dexamethasone", 156, "[1,22,7]", "5,6,7,8", "4", "WEEK", "ORAL", _NONIV_SIG, None, 1), + ("Lenalidomide", 299, "[1,21,1]", "1,2,3,4,5,6,7,8", "4", "WEEK", "ORAL", _NONIV_SIG, None, 1), + ]) + frame = roll_out_variant(variant, decay_days=0) + daratumumab = frame[frame["component"] == "Daratumumab"] + # Cycle 3 starts at 2 × 28 = 56 days; day 15 is day 70. + assert set(daratumumab[daratumumab["cycle_number"] == 3]["elapsed_day"]) == {56, 70} + # Cycle 7 starts at 6 × 28 = 168 days and has only day 1. + assert set(daratumumab[daratumumab["cycle_number"] == 7]["elapsed_day"]) == {168} + assert set(frame["timing_status"]) == {"resolved"} + + def test_five_plus_two_day_control(self, session): + variant = _source_variant(session, 129498, [ + ("Cytarabine", 126, "1", "1", "5", "DAY", "INTRAVENOUS", _IV_CONT_SIG, None, 1), + ("Daunorubicin", 143, "[1,2,1]", "1", "5", "DAY", "INTRAVENOUS", _IV_SIG, None, 1), + ]) + frame = roll_out_variant(variant, decay_days=0) + # A single five-day block begins at day 0; its explicit days are 1 and 2. + assert set(frame["elapsed_day"]) == {0, 1} + assert set(frame["timing_status"]) == {"resolved"} + + def test_tislelizumab_neoadjuvant_to_adjuvant(self, session): + variant = _source_variant(session, 150930, [ + ("Carboplatin", 88, "1", "1,2,3,(4)", "3", "WEEK", "INTRAVENOUS", _IV_SIG, "NEOADJUVANT", 1), + ("Paclitaxel", 379, "1", "1,2,3,(4)", "3", "WEEK", "INTRAVENOUS", _IV_SIG, "NEOADJUVANT", 1), + ("Tislelizumab", 64463, "1", "1,2,3,(4)", "3", "WEEK", "INTRAVENOUS", _IV_SIG, "NEOADJUVANT", 1), + ("Tislelizumab", 64463, "1", "1,2,3,4,5,6,7,8", "6", "WEEK", "INTRAVENOUS", _IV_SIG, "ADJUVANT", 3), + ]) + frame = roll_out_variant(variant, decay_days=0) + neoadjuvant = frame[frame["phase"] == Sigs_PhaseEnum.NEOADJUVANT] + adjuvant = frame[frame["phase"] == Sigs_PhaseEnum.ADJUVANT] + # Optional cycle 4 begins after three 21-day cycles. + assert neoadjuvant.loc[neoadjuvant["cycle_number"] == 4, "elapsed_day"].iloc[0] == 63 + assert neoadjuvant.loc[neoadjuvant["cycle_number"] == 4, "optional"].all() + assert set(neoadjuvant.loc[neoadjuvant["cycle_number"] < 4, "timing_status"]) == {"resolved"} + assert set(neoadjuvant.loc[neoadjuvant["cycle_number"] == 4, "timing_status"]) == { + "resolved_via_fallback: optional cycle 4 assumed given" + } + # Step 2 has no sig, so adjuvant dates are unknown. + assert adjuvant["elapsed_day"].isna().all() + assert set(adjuvant["timing_status"]) == { + "unresolved: phase_step gap before step 3 (no sigs for step 2)" + } + + def test_osimertinib_compact_daily_days(self, session): + variant = _source_variant(session, 136672, [ + ("Carboplatin", 88, "1", "1,2,3,4", "3", "WEEK", "INTRAVENOUS", _IV_SIG, None, 1), + ("Osimertinib", 375, "[1,21,1]", "1,(+1)", "3", "WEEK", "ORAL", _NONIV_SIG, None, 1), + ("Pemetrexed", 395, "1", "1,(+1)", "3", "WEEK", "INTRAVENOUS", _IV_SIG, None, 1), + ]) + frame = roll_out_variant(variant, decay_days=0) + oral = frame[frame["component"] == "Osimertinib"] + # The source's explicit range covers days 1 through 21 in cycle 1. + assert set(oral["elapsed_day"]) == set(range(21)) + # Carboplatin cycle 4 begins after 3 × 21 = 63 days. + assert frame[(frame["component"] == "Carboplatin") & (frame["cycle_number"] == 4)]["elapsed_day"].iloc[0] == 63 + assert set(frame["timing_status"]) == {"resolved"} + assert set(oral["cycle_indefinite"]) == {Indefinite(kind="+k", interval=1)} + + def test_ipilimumab_cycle_three_overlaps_six_cycle_cp(self, session): + variant = _source_variant(session, 130411, [ + ("Carboplatin", 88, "1", "1,2,3,4,5,6", "3", "WEEK", "INTRAVENOUS", _IV_SIG, None, 1), + ("Ipilimumab", 279, "1", "3,(+1)", "12", "WEEK", "INTRAVENOUS", _IV_SIG, None, 1), + ("Paclitaxel", 379, "1", "1,2,3,4,5,6", "3", "WEEK", "INTRAVENOUS", _IV_SIG, None, 1), + ]) + frame = roll_out_variant(variant, decay_days=0) + # Ipilimumab cycle 3 starts after two three-week CP cycles: 42 days. + assert frame.loc[frame["component"] == "Ipilimumab", "elapsed_day"].iloc[0] == 42 + # CP cycle 6 begins after five three-week cycles: 105 days. + assert frame[(frame["component"] == "Paclitaxel") & (frame["cycle_number"] == 6)]["elapsed_day"].iloc[0] == 105 + assert set(frame["timing_status"]) == {"resolved"} + assert frame.loc[frame["component"] == "Ipilimumab", "cycle_indefinite"].iloc[0] == Indefinite(kind="+k", interval=1) + + class TestAdministrationFrame: def _nsclc_ish(self, session): variant = _variant(session, 10) @@ -116,7 +1100,8 @@ def test_one_row_per_drug_per_day(self, session): frame = administration_frame(self._nsclc_ish(session), decay_days=0) assert list(frame.columns) == [ "variant_cui", "variant", "route_group", "drug_cui", "drug", - "day", "intensity", "optional", "indefinite", + "day", "intensity", "optional", "indefinite", "cycle_indefinite", + "elapsed_day", "timing_status", ] assert len(frame) == 4 # carboplatin d1, etoposide d1-3 assert set(frame["route_group"]) == {"IV", "PO"} @@ -176,6 +1161,25 @@ def test_overlapping_sigs_for_one_drug_keep_the_strongest_day(self, session): day_2 = frame[frame["day"] == 2] assert len(day_2) == 1 # not one row per sig assert day_2["intensity"].iloc[0] == 1.0 # dosing day beats the other's tail + assert set(roll_out_variant(variant)["sig_id"]) == {1, 2} + + @pytest.mark.parametrize("indefinite_first", [False, True]) + def test_collapsed_sigs_preserve_cycle_indefinite(self, session, indefinite_first): + variant = _variant(session, 110) + _drug(session, 1, "continuing-drug") + sequences = ("1,(+n)", "1") if indefinite_first else ("1", "1,(+n)") + for sig_id, sequence in enumerate(sequences, start=1): + _sig( + session, sig_id=sig_id, variant_cui=110, drug_cui=1, + route="INTRAVENOUS", alldays="1", timing_sequence=sequence, + cycle_length_lb="21", cycle_length_ub="21", + cycle_length_unit=Sigs_Cycle_length_unitEnum.DAY, + ) + session.expire_all() + + frame = administration_frame(variant, decay_days=0) + assert len(frame) == 1 + assert frame["cycle_indefinite"].iloc[0] is not None def test_many_variants_come_back_in_one_frame(self, session): _drug(session, 1, "cisplatin") @@ -216,6 +1220,18 @@ def record(conn, cursor, statement, parameters, context, executemany): finally: sa.event.remove(engine, "before_cursor_execute", record) + def test_elapsed_day_exposes_cross_block_timing(self, session): + variant = TestRollout()._two_block_variant(session, variant_cui=98) + + frame = administration_frame(variant, decay_days=0) + first_block = frame[frame["drug"] == "docetaxel"] + second_block = frame[frame["drug"] == "trastuzumab"] + + assert first_block["elapsed_day"].min() == 0 + assert second_block["elapsed_day"].min() == 112 + assert set(frame["day"]) == {1} + assert set(frame["timing_status"]) == {"resolved"} + class TestAdministrationMatrix: def test_pivots_to_the_drug_by_day_grid(self, session):