Spaces:
Sleeping
Sleeping
File size: 4,624 Bytes
77e1e28 | 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 | """Task generator — produces (ValidationTaskSpec, FullLatentState) pairs
for drug-target-validation episodes.
Supports two modes:
1. Select from the curated ``SCENARIO_LIBRARY``.
2. Add procedurally-generated scenarios on top.
"""
from __future__ import annotations
from typing import List, Optional, Tuple
import numpy as np
from models import ActionType, ValidationTaskSpec
from server.simulator.latent_state import (
CreditState,
DataQualityState,
FullLatentState,
TargetProfile,
ValidationProgress,
)
from .scenarios import SCENARIO_LIBRARY, Scenario
from .procedural_generator import generate_procedural_scenarios
class TaskGenerator:
"""Generates task + latent-state pairs for environment episodes."""
def __init__(
self,
scenarios: Optional[List[Scenario]] = None,
domain_randomise: bool = True,
):
if scenarios is not None:
self.scenarios = scenarios
else:
self.scenarios = list(SCENARIO_LIBRARY) + generate_procedural_scenarios(
n=20, seed=42,
)
self.domain_randomise = domain_randomise
def generate(
self,
*,
seed: Optional[int] = None,
scenario_name: Optional[str] = None,
) -> Tuple[ValidationTaskSpec, FullLatentState]:
rng = np.random.default_rng(seed)
if scenario_name:
scenario = self._find_scenario(scenario_name)
else:
idx = int(rng.integers(0, len(self.scenarios)))
scenario = self.scenarios[idx]
task = scenario.task.model_copy(deep=True)
target = scenario.target.model_copy(deep=True)
data_quality = scenario.data_quality.model_copy(deep=True)
if self.domain_randomise:
self._randomise(rng, task, target, data_quality)
if not task.available_actions:
task.available_actions = [a.value for a in ActionType]
latent = FullLatentState(
target=target,
data_quality=data_quality,
progress=ValidationProgress(),
credits=CreditState(credits_total=task.credits_limit),
rng_seed=seed or 0,
)
return task, latent
def list_scenarios(self) -> List[str]:
return [s.name for s in self.scenarios]
# ── internals ───────────────────────────────────────────────────────
def _find_scenario(self, name: str) -> Scenario:
for s in self.scenarios:
if s.name == name:
return s
available = ", ".join(self.list_scenarios())
raise ValueError(f"Unknown scenario '{name}'. Available: {available}")
@staticmethod
def _randomise(
rng: np.random.Generator,
task: ValidationTaskSpec,
target: TargetProfile,
data_quality: DataQualityState,
) -> None:
"""Light domain randomisation that nudges noise / numerics without
flipping ``correct_decision`` or ``key_evidence_dimensions``."""
# Credit budget jitter
task.credits_limit = int(
max(15, round(task.credits_limit * float(rng.uniform(0.9, 1.1))))
)
# Data-quality jitter
data_quality.noise_level = float(np.clip(
data_quality.noise_level + rng.normal(0, 0.02), 0.02, 0.4
))
data_quality.false_positive_rate = float(np.clip(
data_quality.false_positive_rate + rng.normal(0, 0.01), 0.0, 0.3
))
data_quality.false_negative_rate = float(np.clip(
data_quality.false_negative_rate + rng.normal(0, 0.01), 0.0, 0.3
))
data_quality.database_coverage = float(np.clip(
data_quality.database_coverage + rng.normal(0, 0.03), 0.5, 1.0
))
# Target profile numerics — keep categorical fields fixed.
target.tissue_specificity = float(np.clip(
target.tissue_specificity * float(rng.uniform(0.9, 1.1)), 0.0, 1.0
))
target.disease_overexpression = float(max(
0.1, target.disease_overexpression * float(rng.uniform(0.85, 1.15))
))
target.druggability_score = float(np.clip(
target.druggability_score * float(rng.uniform(0.9, 1.1)), 0.0, 1.0
))
target.selectivity_ratio = float(max(
0.0, target.selectivity_ratio * float(rng.uniform(0.85, 1.15))
))
target.in_vitro_ic50_nM = float(max(
0.5, target.in_vitro_ic50_nM * float(rng.uniform(0.7, 1.3))
))
|