Spaces:
Paused
Paused
File size: 5,558 Bytes
5fb57bc a1a4d73 5fb57bc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 | """Deterministic schema-6 protocol pairs for fixed validation banks."""
from __future__ import annotations
import hashlib
from dataclasses import dataclass, replace
from typing import Callable
from pff_pk.data.studies import Study, StudyFamily
from pff_pk.tasks.batch import FlowTask
from pff_pk.tasks.config import TaskConfig
from pff_pk.tasks.generation import sample_task
class ProtocolPairUnavailable(ValueError):
"""Raised when one family cannot instantiate a requested validation case."""
ArmPredicate = Callable[[Study], bool]
def _kind(name: str) -> ArmPredicate:
return lambda arm: arm.metadata.get("protocol_arm_kind") == name
def _global_direction(direction: str) -> ArmPredicate:
def predicate(arm: Study) -> bool:
if arm.metadata.get("protocol_arm_kind") != "global_scale":
return False
scale = arm.metadata.get("relative_global_scale")
if scale is None:
return False
return float(scale) < 1.0 if direction == "lower" else float(scale) > 1.0
return predicate
def _add_on_position(position: str) -> ArmPredicate:
def predicate(arm: Study) -> bool:
if arm.metadata.get("protocol_arm_kind") != "add_on":
return False
event = arm.metadata.get("added_dose")
if not isinstance(event, dict) or event.get("time") is None:
return False
return float(event["time"]) < 0.5 if position == "early" else float(event["time"]) >= 0.5
return predicate
@dataclass(frozen=True)
class ProtocolPairCase:
name: str
context: ArmPredicate
target: ArmPredicate
same_arm: bool = False
distinct_arms: bool = False
PROTOCOL_PAIR_CASES = (
ProtocolPairCase("reference_identity", _kind("reference"), _kind("reference"), same_arm=True),
ProtocolPairCase("global_identity", _kind("global_scale"), _kind("global_scale"), same_arm=True),
ProtocolPairCase("add_on_identity", _kind("add_on"), _kind("add_on"), same_arm=True),
ProtocolPairCase("reference_to_lower_scale", _kind("reference"), _global_direction("lower")),
ProtocolPairCase("reference_to_higher_scale", _kind("reference"), _global_direction("higher")),
ProtocolPairCase("lower_scale_to_reference", _global_direction("lower"), _kind("reference")),
ProtocolPairCase("higher_scale_to_reference", _global_direction("higher"), _kind("reference")),
ProtocolPairCase(
"global_scale_cross",
_kind("global_scale"),
_kind("global_scale"),
distinct_arms=True,
),
ProtocolPairCase("reference_to_early_add_on", _kind("reference"), _add_on_position("early")),
ProtocolPairCase("reference_to_late_add_on", _kind("reference"), _add_on_position("late")),
ProtocolPairCase("add_on_to_reference", _kind("add_on"), _kind("reference")),
ProtocolPairCase("add_on_cross", _kind("add_on"), _kind("add_on"), distinct_arms=True),
ProtocolPairCase("global_to_add_on", _kind("global_scale"), _kind("add_on")),
ProtocolPairCase("add_on_to_global", _kind("add_on"), _kind("global_scale")),
)
def _choice(values: list[int], token: str) -> int:
digest = hashlib.blake2b(token.encode(), digest_size=8).digest()
return values[int.from_bytes(digest, "little") % len(values)]
def select_protocol_pair(
family: StudyFamily,
case: ProtocolPairCase,
*,
draw: int = 0,
) -> tuple[int, int]:
"""Select a reproducible ordered arm pair satisfying ``case``."""
contexts = [index for index, arm in enumerate(family.arms) if case.context(arm)]
targets = [index for index, arm in enumerate(family.arms) if case.target(arm)]
if case.same_arm:
common = sorted(set(contexts) & set(targets))
if not common:
raise ProtocolPairUnavailable(f"{family.identifier} lacks {case.name}")
selected = _choice(common, f"{family.identifier}:{case.name}:{draw}:same")
return selected, selected
pairs = [
(context, target)
for context in contexts
for target in targets
if not case.distinct_arms or context != target
]
if not pairs:
raise ProtocolPairUnavailable(f"{family.identifier} lacks {case.name}")
return pairs[
_choice(list(range(len(pairs))), f"{family.identifier}:{case.name}:{draw}:pair")
]
@dataclass(frozen=True)
class FixedProtocolValidationSampler:
"""Build one reproducible protocol-pair/acquisition validation stratum."""
case: ProtocolPairCase
acquisition: str
def __call__(
self,
study: Study | StudyFamily,
config: TaskConfig,
*,
draw: int = 0,
) -> FlowTask:
if not isinstance(study, StudyFamily):
raise ProtocolPairUnavailable("v6 protocol validation requires schema-6 families")
pair = select_protocol_pair(study, self.case, draw=draw)
fixed = replace(config, acquisition_weights={self.acquisition: 1.0})
return sample_task(study, fixed, draw=draw, protocol_arm_indices=pair)
def protocol_validation_definitions() -> tuple[tuple[str, float, FixedProtocolValidationSampler], ...]:
"""Return the balanced 14-pair by 3-acquisition validation design."""
acquisitions = ("unscheduled", "exact", "pseudo_scheduled")
count = len(PROTOCOL_PAIR_CASES) * len(acquisitions)
return tuple(
(
f"{case.name}__{acquisition}",
1.0 / count,
FixedProtocolValidationSampler(case, acquisition),
)
for case in PROTOCOL_PAIR_CASES
for acquisition in acquisitions
)
|