From f64ea80f0cdcc89f06860d6498ee1607af13f518 Mon Sep 17 00:00:00 2001 From: Daniel Weindl Date: Mon, 16 Dec 2024 21:26:48 +0100 Subject: [PATCH] conditions --- petab/v2/core.py | 137 ++++++++++++++++++++++++++++++++++++++++++ petab/v2/petab1to2.py | 2 +- petab/v2/problem.py | 19 +++++- tests/v2/test_core.py | 22 +++++-- 4 files changed, 171 insertions(+), 9 deletions(-) diff --git a/petab/v2/core.py b/petab/v2/core.py index cdf22f21..20da0c15 100644 --- a/petab/v2/core.py +++ b/petab/v2/core.py @@ -107,3 +107,140 @@ def from_tsv(cls, file_path: str | Path) -> ObservablesTable: def to_tsv(self, file_path: str | Path) -> None: df = self.to_dataframe() df.to_csv(file_path, sep="\t", index=False) + + +class OperationType(str, Enum): + # TODO update names + SET_CURRENT_VALUE = "setCurrentValue" + SET_RATE = "setRate" + SET_ASSIGNMENT = "setAssignment" + CONSTANT = "constant" + INITIAL = "initial" + ... + + +class Change(BaseModel): + target_id: str = Field(alias=C.TARGET_ID) + operation_type: OperationType = Field(alias=C.VALUE_TYPE) + target_value: sp.Basic = Field(alias=C.TARGET_VALUE) + + class Config: + populate_by_name = True + arbitrary_types_allowed = True + use_enum_values = True + + @field_validator("target_id") + @classmethod + def validate_id(cls, v): + if not v: + raise ValueError("ID must not be empty.") + if not is_valid_identifier(v): + raise ValueError(f"Invalid ID: {v}") + return v + + @field_validator("target_value", mode="before") + @classmethod + def sympify(cls, v): + if v is None or isinstance(v, sp.Basic): + return v + if isinstance(v, float) and np.isnan(v): + return None + + return sympify_petab(v) + + +class ExperimentalCondition(BaseModel): + id: str = Field(alias=C.CONDITION_ID) + changes: list[Change] + + class Config: + populate_by_name = True + + @field_validator("id") + @classmethod + def validate_id(cls, v): + if not v: + raise ValueError("ID must not be empty.") + if not is_valid_identifier(v): + raise ValueError(f"Invalid ID: {v}") + return v + + +class ConditionsTable(BaseModel): + conditions: list[ExperimentalCondition] + + @classmethod + def from_dataframe(cls, df: pd.DataFrame) -> ConditionsTable: + if df is None: + return cls(conditions=[]) + + conditions = [] + for condition_id, sub_df in df.groupby(C.CONDITION_ID): + changes = [Change(**row.to_dict()) for _, row in sub_df.iterrows()] + conditions.append( + ExperimentalCondition(id=condition_id, changes=changes) + ) + + return cls(conditions=conditions) + + def to_dataframe(self) -> pd.DataFrame: + records = [ + {C.CONDITION_ID: condition.id, **change.model_dump()} + for condition in self.conditions + for change in condition.changes + ] + return pd.DataFrame(records) + + @classmethod + def from_tsv(cls, file_path: str | Path) -> ConditionsTable: + df = pd.read_csv(file_path, sep="\t") + return cls.from_dataframe(df) + + def to_tsv(self, file_path: str | Path) -> None: + df = self.to_dataframe() + df.to_csv(file_path, sep="\t", index=False) + + +class ExperimentPeriod(BaseModel): + start: float = Field(alias=C.TIME) + conditions: list[ExperimentalCondition] + + class Config: + populate_by_name = True + + +class Experiment(BaseModel): + id: str = Field(alias=C.EXPERIMENT_ID) + periods: list[ExperimentPeriod] + + class Config: + populate_by_name = True + arbitrary_types_allowed = True + + +class ExperimentsTable(BaseModel): + experiments: list[Experiment] + + @classmethod + def from_dataframe(cls, df: pd.DataFrame) -> ExperimentsTable: + if df is None: + return cls(experiments=[]) + + experiments = [ + Experiment(**row.to_dict()) + for _, row in df.reset_index().iterrows() + ] + + return cls(experiments=experiments) + + def to_dataframe(self) -> pd.DataFrame: + return pd.DataFrame(self.model_dump()["experiments"]) + + @classmethod + def from_tsv(cls, file_path: str | Path) -> ExperimentsTable: + df = pd.read_csv(file_path, sep="\t") + return cls.from_dataframe(df) + + def to_tsv(self, file_path: str | Path) -> None: + df = self.to_dataframe() + df.to_csv(file_path, sep="\t", index=False) diff --git a/petab/v2/petab1to2.py b/petab/v2/petab1to2.py index 78304328..b6051510 100644 --- a/petab/v2/petab1to2.py +++ b/petab/v2/petab1to2.py @@ -280,7 +280,7 @@ def v1v2_condition_df( id_vars=[v1.C.CONDITION_ID], var_name=v2.C.TARGET_ID, value_name=v2.C.TARGET_VALUE, - ) + ).dropna(subset=[v2.C.TARGET_VALUE]) if condition_df.empty: # This happens if there weren't any condition-specific changes diff --git a/petab/v2/problem.py b/petab/v2/problem.py index 95ec5118..eb0a0808 100644 --- a/petab/v2/problem.py +++ b/petab/v2/problem.py @@ -92,11 +92,24 @@ def __init__( ] = default_validation_tasks.copy() self.config = config - from .core import Observable, ObservablesTable + from .core import ( + ConditionsTable, + ExperimentalCondition, + Observable, + ObservablesTable, + ) + + self.observables_table: ObservablesTable = ( + ObservablesTable.from_dataframe(self.observable_df) + ) + self.observables: list[Observable] = self.observables_table.observables - self.observables: list[Observable] = ObservablesTable.from_dataframe( - self.observable_df + self.conditions_table: ConditionsTable = ( + ConditionsTable.from_dataframe(self.condition_df) ) + self.conditions: list[ + ExperimentalCondition + ] = self.conditions_table.conditions def __str__(self): model = f"with model ({self.model})" if self.model else "without model" diff --git a/tests/v2/test_core.py b/tests/v2/test_core.py index f5987f16..76933a1c 100644 --- a/tests/v2/test_core.py +++ b/tests/v2/test_core.py @@ -1,14 +1,14 @@ import tempfile from pathlib import Path -from petab.v2.core import ObservablesTable +from petab.v2.core import ConditionsTable, ObservablesTable +from petab.v2.petab1to2 import petab1to2 + +example_dir_fujita = Path(__file__).parents[2] / "doc/example/example_Fujita" def test_observables_table(): - file = ( - Path(__file__).parents[2] - / "doc/example/example_Fujita/Fujita_observables.tsv" - ) + file = example_dir_fujita / "Fujita_observables.tsv" # read-write-read round trip observables = ObservablesTable.from_tsv(file) @@ -18,3 +18,15 @@ def test_observables_table(): observables.to_tsv(tmp_file) observables2 = ObservablesTable.from_tsv(tmp_file) assert observables == observables2 + + +def test_conditions_table(): + with tempfile.TemporaryDirectory() as tmp_dir: + petab1to2(example_dir_fujita / "Fujita.yaml", tmp_dir) + file = Path(tmp_dir, "Fujita_experimentalCondition.tsv") + # read-write-read round trip + conditions = ConditionsTable.from_tsv(file) + tmp_file = Path(tmp_dir) / "conditions.tsv" + conditions.to_tsv(tmp_file) + conditions2 = ConditionsTable.from_tsv(tmp_file) + assert conditions == conditions2