conditions

Author: dweindl

petab/v2/core.py

@@ -20,17 +20,29 @@
 
 
 class ObservableTransformation(str, Enum):
+    """Observable transformation types.
+
+    Observable transformations as used in the PEtab observables table.
+    """
+
     LIN = C.LIN
     LOG = C.LOG
     LOG10 = C.LOG10
 
 
 class NoiseDistribution(str, Enum):
+    """Noise distribution types.
+
+    Noise distributions as used in the PEtab observables table.
+    """
+
     NORMAL = C.NORMAL
     LAPLACE = C.LAPLACE
 
 
 class Observable(BaseModel):
+    """Observable definition."""
+
     id: str = Field(alias=C.OBSERVABLE_ID)
     name: str | None = Field(alias=C.OBSERVABLE_NAME, default=None)
     formula: sp.Basic | None = Field(alias=C.OBSERVABLE_FORMULA, default=None)
@@ -82,6 +94,8 @@ class Config:
 
 
 class ObservablesTable(BaseModel):
+    """PEtab observables table."""
+
     observables: list[Observable]
 
     @classmethod
@@ -107,3 +121,169 @@ 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):
+    """Operation types for model changes in the PEtab conditions table."""
+
+    # TODO update names
+    SET_CURRENT_VALUE = "setCurrentValue"
+    SET_RATE = "setRate"
+    SET_ASSIGNMENT = "setAssignment"
+    CONSTANT = "constant"
+    INITIAL = "initial"
+    ...
+
+
+class Change(BaseModel):
+    """A change to the model or model state.
+
+    A change to the model or model state, corresponding to an individual
+    row of the PEtab conditions table.
+    """
+
+    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 ChangeSet(BaseModel):
+    """A set of changes to the model or model state.
+
+    A set of simultaneously occuring changes to the model or model state,
+    corresponding to a perturbation of the underlying system. This corresponds
+    to all rows of the PEtab conditions table with the same condition ID.
+    """
+
+    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):
+    """PEtab conditions table."""
+
+    conditions: list[ChangeSet]
+
+    @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(ChangeSet(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):
+    """A period of a timecourse defined by a start time and a set changes.
+
+    This corresponds to a row of the PEtab experiments table.
+    """
+
+    start: float = Field(alias=C.TIME)
+    conditions: list[ChangeSet]
+
+    class Config:
+        populate_by_name = True
+
+
+class Experiment(BaseModel):
+    """An experiment or a timecourse defined by an ID and a set of different
+    periods.
+
+    Corresponds to a group of rows of the PEtab experiments table with the same
+    experiment ID.
+    """
+
+    id: str = Field(alias=C.EXPERIMENT_ID)
+    periods: list[ExperimentPeriod]
+
+    class Config:
+        populate_by_name = True
+        arbitrary_types_allowed = True
+
+
+class ExperimentsTable(BaseModel):
+    """PEtab experiments table."""
+
+    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)

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

petab/v2/problem.py

@@ -93,11 +93,22 @@ def __init__(
         ] = default_validation_tasks.copy()
         self.config = config
 
-        from .core import Observable, ObservablesTable
+        from .core import (
+            ChangeSet,
+            ConditionsTable,
+            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[ChangeSet] = self.conditions_table.conditions
 
     def __str__(self):
         model = f"with model ({self.model})" if self.model else "without model"

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