TrajectoryCollection

Author: joseph-hellerstein

docs/model_based_design.md

@@ -63,21 +63,21 @@ This class does linear prediction and evaluations of these predictions. It is co
 This class represents a collection of ``Trajectory`` with the same
 ``Model``. Methods include:
 
-* ``plotTimecourse`` pieces together the timecourse for each ``DynamicModel``, and plots it with a vertical dashed line separating each ``DynamicModel``.
-* ``__eq__`` which checks if it's the same as another ``DynamicModelCollection`` by comparing each ``DynamicModel``.
-* ``__lt`` if the last timepoint of the first Trajector equals the first timepoint of the second trajectory.
-* ``split`` takes as input timepoints to create multiple Trajectory objects. A timepoint specifies the last time for the preceeding Trajectory and the first time for the next Trajectory.
+* Constructor takes a list of ``Tracjectory`` all of which have the same ``Model``. Error checking is done to ensure that timepoints do not overlap. The list is sorted using ``Trajectory.__lt__``. The constructor does not verify that adjacent ``Trajectory`` overlap their end_time and start_time.
+* ``plotTimecourse`` pieces together the timecourse for each ``DynamicModel``, and plots it with a vertical dashed line separating each ``Trajectory``. The arguments to this method are the kwards used by PlotOptions. Internally, the method uses PlotOptions. It should return PlotOptions.
+* ``__eq__`` which checks if it's the same as another ``TrajectoryCollection`` by comparing each ``Trajectory``.
+* ``split`` is a class method that takes as input timepoints to create multiple Trajectory objects. Its signature is ``split(cls, trajectory: Trajectory, timepoints: List[float])``.
+    * A timepoint specifies the last time for the preceeding Trajectory and the first time for the next Trajectory.
+    * Returns ``TrajectoryCollection``.
+    * If a split time falls between existing timepoints (i.e., not exactly in timepoint_arr), it snaps to the nearest timepoint.
 
 ### ``MultipleLinearPredictor``
 
 This class performs piece-wise linear prediction. The times at which
-there is a partition of the linear model is a "split point". Using slicing, we can easily construct the DynamicModels for a collection of split points. (Of course, all will have the same StaticModel.) If there are n split points, then there are n + 1 linear models.
-Note that the old Trajectory.sequentialPartition / nonsequentialPartition aren't mentioned beause their implementation is deferred.
+there is a partition of the linear model is a "split point".
 
-* When splitting a ``DynamicModel``, slicing is used, not simulation.
-* ``split`` can be called with specific split times or without any split time specified. A split time t1 specifies the time at which the previous DynamicModel ends and the second timepoint of the new Dynamics model.
-**Does this belong here?**
-* ``predict``
+* Constructor has the arguments ``TrajectoryCollection`` and ``num_step``, the number of steps ahead for which prediction is done.
+* ``predict`` uses LinearPredictor.predict for each Trajectory.
 * ``score``
 * ``plotPrediction`` The plot shows predicted and actual (simulated) values with vertical dashed lines to indicate regions for submodels.
 

src/trajectory_collection.py

@@ -0,0 +1,120 @@
+"""Collection of Trajectory objects sharing the same Model."""
+
+import src.constants as cn  # type: ignore
+from src.plot_options import PlotOptions  # type: ignore
+from src.trajectory import Trajectory  # type: ignore
+
+import numpy as np  # type: ignore
+from typing import List
+
+
+class TrajectoryCollection(object):
+    """An ordered, non-overlapping collection of Trajectory sharing one Model.
+
+    Use split() to partition a single Trajectory into a TrajectoryCollection.
+    """
+
+    def __init__(self, trajectories: List[Trajectory]) -> None:
+        """
+        Parameters
+        ----------
+        trajectories : List[Trajectory]
+            All must share the same Model. Sorted by Trajectory.__lt__;
+            overlapping time ranges raise ValueError.
+        """
+        # Error checking
+        if not trajectories:
+            raise ValueError("trajectories must not be empty.")
+        self.trajectories = sorted(trajectories)
+        self.model = self.trajectories[0].model
+        for traj in trajectories[1:]:
+            if traj.model != self.model:
+                raise ValueError(
+                        "All trajectories must share the same Model.")
+        self.start_time = self.trajectories[0].start_time
+        self.end_time = self.trajectories[-1].end_time
+
+    def isConsecutive(self) -> bool:
+        """True iff trajectories are consecutive (no gaps or overlaps)."""
+        for t1, t2 in zip(self.trajectories, self.trajectories[1:]):
+            if not np.isclose(t1.end_time, t2.start_time):
+                return False
+        return True 
+
+    def __eq__(self, other: object) -> bool:
+        if not isinstance(other, TrajectoryCollection):
+            raise ValueError("Can only compare TrajectoryCollection to another ")
+        if len(self.trajectories) != len(other.trajectories):
+            return False
+        return all(t1 == t2 for t1, t2 in zip(self.trajectories, other.trajectories))
+
+    def plotTimecourse(self, **kwargs) -> PlotOptions:
+        """Plot the pieced-together timecourse with dashed vertical separators.
+
+        Each trajectory's timecourse is drawn continuously; a vertical dashed
+        line marks the boundary between adjacent trajectories.  One legend
+        entry per species.
+
+        Parameters
+        ----------
+        **kwargs
+            Passed to PlotOptions. Supported keys: ax, fig, title, xlabel,
+            ylabel, legend, xlim, ylim, model_name.
+
+        Returns
+        -------
+        PlotOptions
+        """
+        plot_options = PlotOptions(**kwargs)
+        ax = plot_options.ax
+        for i, name in enumerate(self.model.species_names):
+            color = f"C{i}"
+            for j, traj in enumerate(self.trajectories):
+                label = name if j == 0 else None
+                ax.plot(
+                        traj.timecourse_df.index,
+                        traj.timecourse_df[name],
+                        color=color,
+                        label=label,
+                )
+        for traj in self.trajectories[:-1]:
+            ax.axvline(x=traj.end_time, color="black", linestyle="--",
+                    linewidth=0.8)
+        plot_options.apply()
+        return plot_options
+
+    @classmethod
+    def split(cls,
+            trajectory: Trajectory,
+            timepoints: List[float]) -> "TrajectoryCollection":
+        """Partition a Trajectory at the given split times.
+
+        Each split time becomes the shared end/start boundary between adjacent
+        sub-trajectories.  Times not exactly in timepoint_arr snap to the
+        nearest available timepoint.  Split times that snap to start_time or
+        end_time are ignored.
+
+        Parameters
+        ----------
+        trajectory : Trajectory
+        timepoints : List[float]
+
+        Returns
+        -------
+        TrajectoryCollection
+        """
+        tp_arr = trajectory.timepoint_arr
+        snapped = []
+        for t in timepoints:
+            idx = int(np.argmin(np.abs(tp_arr - t)))
+            snapped.append(float(tp_arr[idx]))
+        snapped = sorted({
+                t for t in snapped
+                if trajectory.start_time < t < trajectory.end_time
+        })
+        boundaries = [trajectory.start_time] + snapped + [trajectory.end_time]
+        sub_trajectories = [
+                trajectory.makeSubmodel(boundaries[i], boundaries[i + 1])
+                for i in range(len(boundaries) - 1)
+        ]
+        return cls(sub_trajectories)