Functions for adding conditions/observables/parameter to Problem

petab/v1/mapping.py

@@ -1,4 +1,5 @@
 """Functionality related to the PEtab entity mapping table"""
+# TODO: Move to petab.v2.mapping
 from pathlib import Path
 
 import pandas as pd

petab/v1/problem.py

@@ -3,8 +3,9 @@
 
 import os
 import tempfile
-from collections.abc import Iterable
+from collections.abc import Iterable, Sequence
 from math import nan
+from numbers import Number
 from pathlib import Path, PurePosixPath
 from typing import TYPE_CHECKING
 from warnings import warn
@@ -1005,3 +1006,181 @@ def n_priors(self) -> int:
             return 0
 
         return self.parameter_df[OBJECTIVE_PRIOR_PARAMETERS].notna().sum()
+
+    def add_condition(self, id_: str, name: str = None, **kwargs):
+        """Add a simulation condition to the problem.
+
+        Arguments:
+            id_: The condition id
+            name: The condition name
+            kwargs: Parameter, value pairs to add to the condition table.
+        """
+        record = {CONDITION_ID: [id_], **kwargs}
+        if name is not None:
+            record[CONDITION_NAME] = name
+        tmp_df = pd.DataFrame(record).set_index([CONDITION_ID])
+        self.condition_df = (
+            pd.concat([self.condition_df, tmp_df])
+            if self.condition_df is not None
+            else tmp_df
+        )
+
+    def add_observable(
+        self,
+        id_: str,
+        formula: str | float | int,
+        noise_formula: str | float | int = None,
+        noise_distribution: str = None,
+        transform: str = None,
+        name: str = None,
+        **kwargs,
+    ):
+        """Add an observable to the problem.
+
+        Arguments:
+            id_: The observable id
+            formula: The observable formula
+            noise_formula: The noise formula
+            noise_distribution: The noise distribution
+            transform: The observable transformation
+            name: The observable name
+            kwargs: additional columns/values to add to the observable table
+
+        """
+        record = {
+            OBSERVABLE_ID: [id_],
+            OBSERVABLE_FORMULA: [formula],
+        }
+        if name is not None:
+            record[OBSERVABLE_NAME] = [name]
+        if noise_formula is not None:
+            record[NOISE_FORMULA] = [noise_formula]
+        if noise_distribution is not None:
+            record[NOISE_DISTRIBUTION] = [noise_distribution]
+        if transform is not None:
+            record[OBSERVABLE_TRANSFORMATION] = [transform]
+        record.update(kwargs)
+
+        tmp_df = pd.DataFrame(record).set_index([OBSERVABLE_ID])
+        self.observable_df = (
+            pd.concat([self.observable_df, tmp_df])
+            if self.observable_df is not None
+            else tmp_df
+        )
+
+    def add_parameter(
+        self,
+        id_: str,
+        estimated: bool | str | int = True,
+        nominal_value=None,
+        scale: str = None,
+        lb: Number = None,
+        ub: Number = None,
+        init_prior_type: str = None,
+        init_prior_pars: str | Sequence = None,
+        obj_prior_type: str = None,
+        obj_prior_pars: str | Sequence = None,
+        **kwargs,
+    ):
+        """Add a parameter to the problem.
+
+        Arguments:
+            id_: The parameter id
+            estimated: Whether the parameter is estimated
+            nominal_value: The nominal value of the parameter
+            scale: The parameter scale
+            lb: The lower bound of the parameter
+            ub: The upper bound of the parameter
+            init_prior_type: The type of the initialization prior distribution
+            init_prior_pars: The parameters of the initialization prior
+                distribution
+            obj_prior_type: The type of the objective prior distribution
+            obj_prior_pars: The parameters of the objective prior distribution
+            kwargs: additional columns/values to add to the parameter table
+        """
+        record = {
+            PARAMETER_ID: [id_],
+        }
+        if estimated is not None:
+            record[ESTIMATE] = [
+                int(estimated)
+                if isinstance(estimated, bool | int)
+                else estimated
+            ]
+        if nominal_value is not None:
+            record[NOMINAL_VALUE] = [nominal_value]
+        if scale is not None:
+            record[PARAMETER_SCALE] = [scale]
+        if lb is not None:
+            record[LOWER_BOUND] = [lb]
+        if ub is not None:
+            record[UPPER_BOUND] = [ub]
+        if init_prior_type is not None:
+            record[INITIALIZATION_PRIOR_TYPE] = [init_prior_type]
+        if init_prior_pars is not None:
+            if not isinstance(init_prior_pars, str):
+                init_prior_pars = PARAMETER_SEPARATOR.join(
+                    map(str, init_prior_pars)
+                )
+            record[INITIALIZATION_PRIOR_PARAMETERS] = [init_prior_pars]
+        if obj_prior_type is not None:
+            record[OBJECTIVE_PRIOR_TYPE] = [obj_prior_type]
+        if obj_prior_pars is not None:
+            if not isinstance(obj_prior_pars, str):
+                obj_prior_pars = PARAMETER_SEPARATOR.join(
+                    map(str, obj_prior_pars)
+                )
+            record[OBJECTIVE_PRIOR_PARAMETERS] = [obj_prior_pars]
+        record.update(kwargs)
+
+        tmp_df = pd.DataFrame(record).set_index([PARAMETER_ID])
+        self.parameter_df = (
+            pd.concat([self.parameter_df, tmp_df])
+            if self.parameter_df is not None
+            else tmp_df
+        )
+
+    def add_measurement(
+        self,
+        obs_id: str,
+        sim_cond_id: str,
+        time: float,
+        measurement: float,
+        observable_parameters: Sequence[str] = None,
+        noise_parameters: Sequence[str] = None,
+        preeq_cond_id: str = None,
+    ):
+        """Add a measurement to the problem.
+
+        Arguments:
+            obs_id: The observable ID
+            sim_cond_id: The simulation condition ID
+            time: The measurement time
+            measurement: The measurement value
+            observable_parameters: The observable parameters
+            noise_parameters: The noise parameters
+            preeq_cond_id: The pre-equilibration condition ID
+        """
+        record = {
+            OBSERVABLE_ID: [obs_id],
+            SIMULATION_CONDITION_ID: [sim_cond_id],
+            TIME: [time],
+            MEASUREMENT: [measurement],
+        }
+        if observable_parameters is not None:
+            record[OBSERVABLE_PARAMETERS] = [
+                PARAMETER_SEPARATOR.join(observable_parameters)
+            ]
+        if noise_parameters is not None:
+            record[NOISE_PARAMETERS] = [
+                PARAMETER_SEPARATOR.join(noise_parameters)
+            ]
+        if preeq_cond_id is not None:
+            record[PREEQUILIBRATION_CONDITION_ID] = [preeq_cond_id]
+
+        tmp_df = pd.DataFrame(record)
+        self.measurement_df = (
+            pd.concat([self.measurement_df, tmp_df])
+            if self.measurement_df is not None
+            else tmp_df
+        )

petab/v2/petab1to2.py

@@ -2,6 +2,7 @@
 import shutil
 from itertools import chain
 from pathlib import Path
+from urllib.parse import urlparse
 
 from pandas.io.common import get_handle, is_url
 
@@ -76,7 +77,7 @@ def petab1to2(yaml_config: Path | str, output_dir: Path | str = None):
     # condition tables, observable tables, SBML files, parameter table:
     #  no changes - just copy
     file = yaml_config[C.PARAMETER_FILE]
-    _copy_file(get_src_path(file), get_dest_path(file))
+    _copy_file(get_src_path(file), Path(get_dest_path(file)))
 
     for problem_config in yaml_config[C.PROBLEMS]:
         for file in chain(
@@ -89,7 +90,7 @@ def petab1to2(yaml_config: Path | str, output_dir: Path | str = None):
             problem_config.get(C.MEASUREMENT_FILES, []),
             problem_config.get(C.VISUALIZATION_FILES, []),
         ):
-            _copy_file(get_src_path(file), get_dest_path(file))
+            _copy_file(get_src_path(file), Path(get_dest_path(file)))
 
     # TODO: Measurements: preequilibration to experiments/timecourses once
     #  finalized
@@ -131,15 +132,23 @@ def _update_yaml(yaml_config: dict) -> dict:
     return yaml_config
 
 
-def _copy_file(src: Path | str, dest: Path | str):
+def _copy_file(src: Path | str, dest: Path):
     """Copy file."""
-    src = str(src)
-    dest = str(dest)
+    # src might be a URL - convert to Path if local
+    src_url = urlparse(src)
+    if not src_url.scheme:
+        src = Path(src)
+    elif src_url.scheme == "file" and not src_url.netloc:
+        src = Path(src.removeprefix("file:/"))
 
     if is_url(src):
         with get_handle(src, mode="r") as src_handle:
             with open(dest, "w") as dest_handle:
                 dest_handle.write(src_handle.handle.read())
         return
 
-    shutil.copy(str(src), str(dest))
+    try:
+        if dest.samefile(src):
+            return
+    except FileNotFoundError:
+        shutil.copy(str(src), str(dest))