Merge pull request #73 from multiscale/flatten_timeline_support

Flatten timeline support

Author: LourensVeen

ymmsl/v0_2/model.py

@@ -4,6 +4,7 @@
 from typing import Any, List, Sequence, TypeAlias, cast
 
 import yatiml
+from typing_extensions import Self
 
 from ymmsl.v0_2.component import Component
 from ymmsl.v0_2.identity import Identifier, Reference
@@ -331,6 +332,35 @@ def __init__(
     def __contains__(self, timeline: Timeline) -> bool:
         return timeline in self.matches
 
+    def __copy__(self) -> "MatchingTimelines":
+        """Shallow-copy the object.
+
+        Since this class effectively models a container, a shallow copy should create a
+        new container with the same items in it, to which items can be added or removed
+        without affecting the original. By default, copy.copy() won't do that however,
+        as it will return a new MatchingTimelines containing the same self.matches. So
+        we override it to give more natural semantics.
+        """
+        return MatchingTimelines(self.head, list(self.matches))
+
+    def __ior__(self, other: "MatchingTimelines") -> Self:
+        """Merge another MatchingTimelines into this one.
+
+        This takes all of the matching timelines from other and adds them to this
+        object. If the head timeline of other is nested less deeply than the current
+        head, then it will replace the current head, otherwise the head remains
+        unchanged.
+        """
+
+        self.matches |= other.matches
+
+        this_depth = max([len(ref) for ref in self.head])
+        other_depth = max([len(ref) for ref in other.head])
+        if other_depth < this_depth:
+            self.head = other.head
+
+        return self
+
     def _yatiml_attributes(self) -> OrderedDict:
         matches: str | list[str] = list(map(str, sorted(self.matches - {self.head})))
         if len(matches) < 6 and sum(len(m) for m in matches) < 60:

ymmsl/v0_2/ports.py

@@ -180,6 +180,15 @@ def __init__(
         self.operator = operator
         if timeline is None:
             timeline = Timeline("")
+
+        lengthy = [ref for ref in timeline if len(ref) > 1]
+        if lengthy:
+            raise ValueError(
+                "To avoid confusion, timeline annotations must have a single"
+                " identifier between colons, so {timeline} is not valid. Please"
+                " use names without periods instead of e.g. {lengthy[0]}."
+            )
+
         self.timeline = timeline
 
     def __eq__(self, other: Any) -> bool:

ymmsl/v0_2/tests/test_model.py

@@ -1,3 +1,4 @@
+from copy import copy
 from typing import Callable
 
 import pytest
@@ -194,7 +195,7 @@ def test_dump_multicast_conduits() -> None:
     assert text == ("sender: init.out\nreceiver:\n- c1.in\n- repeat pad c2.in\n")
 
 
-def test_create_matching_timeline() -> None:
+def test_create_matching_timelines() -> None:
     mt = MatchingTimelines(Timeline("tl1"), "tl2")
     assert isinstance(mt.head, Timeline)
     assert mt.head == Timeline("tl1")
@@ -225,6 +226,49 @@ def test_create_matching_timeline() -> None:
     }
 
 
+def test_copy_matching_timelines() -> None:
+    tl1 = Timeline("tl1")
+    tl2 = Timeline("tl2")
+    mt1 = MatchingTimelines(tl1, [tl2])
+
+    mt2 = copy(mt1)
+
+    assert mt2.head is mt1.head
+    assert mt2.matches is not mt1.matches
+    for m2 in mt2.matches:
+        assert len([m1 for m1 in mt1.matches if m1 is m2]) > 0
+
+
+def test_merge_matching_timelines() -> None:
+    mt1 = MatchingTimelines(Timeline("tl1"), "tl2")
+    mt2 = MatchingTimelines(Timeline("tl2"), "tl3")
+
+    mt1 |= mt2
+    assert mt1.head == "tl1"
+    assert mt1.matches == {Timeline("tl1"), Timeline("tl2"), Timeline("tl3")}
+
+    mt3 = MatchingTimelines(Timeline("tl4"), "tl5")
+    mt1 |= mt3
+    assert mt1.head == "tl1"
+    assert mt1.matches == {
+        Timeline("tl1"),
+        Timeline("tl2"),
+        Timeline("tl3"),
+        Timeline("tl4"),
+        Timeline("tl5"),
+    }
+
+    mt4 = MatchingTimelines(Timeline("c1.c2.tl1"), "c1.c3.tl1")
+    mt4 |= mt3
+    assert mt4.head == "tl4"
+    assert mt4.matches == {
+        Timeline("tl4"),
+        Timeline("tl5"),
+        Timeline("c1.c2.tl1"),
+        Timeline("c1.c3.tl1"),
+    }
+
+
 def test_load_matching_timelines() -> None:
     load = yatiml.load_function(MatchingTimelines, Timeline)
 

ymmsl/v0_2/timeline_resolver.py

@@ -273,8 +273,8 @@ def timeline_for_port(
         This returns the name of the timeline the given port communicates on. These are
         as follows:
 
-        - for a model port without timeline annotation, empty
-        - for a model port with a timeline annotation, that annotation
+        - for a model port without timeline annotation, <model>
+        - for a model port with a timeline annotation tl1:tl2, <model>.tl1:<model>.tl2
         - for a component port without timeline annotation, <parent_tl>:<component>
         - for a component O_I or S port with timeline annotation "subtl1",
           <parent_tl>:<component>.subtl1
@@ -296,9 +296,9 @@ def timeline_for_port(
             assert isinstance(model_port, Identifier)
             port = self._model.ports[model_port]
             if port.timeline:
-                result = port.timeline
+                result = Timeline([self._model.name + tl for tl in port.timeline])
             else:
-                result = ROOT_TIMELINE
+                result = Timeline([self._model.name])
 
         else:
             # Component port
@@ -377,6 +377,8 @@ def format_timelines(self) -> str:
         """Create a formatted list of determined timelines per component."""
         return "\n".join(
             f"- Component '{comp}' has timeline '{tl}:{comp}'"
+            if tl
+            else f"- Component '{comp}' has timeline '{comp}'"
             for comp, tl in self._parent_timeline.items()
             if len(comp) > 0  # Ony print actual components
         )