Orion Events: Add group member events

GrpcInterface/Python/rips/orion_events.py

@@ -79,6 +79,8 @@ Notes on the grammar:
   group name is injected as the ``GROUP`` item when each event is applied.
   A bare ``SCHEDULE`` line opens a block of schedule-level keyword events not
   tied to any well (RPTRST, GRUPTREE, TUNING, ...). Empty blocks are legal.
+  ``MEMBER MEMBERS="A,B"`` inside a GROUP block is shorthand for one GRUPTREE
+  record per unique member, with the enclosing group as parent.
 * ``REPORT <date_expr>`` (one date per line, anywhere after the header) names
   a date that should appear as a bare ``DATES`` keyword in the generated
   schedule even when no events fall on it — in Eclipse/Flow a ``DATES`` entry
@@ -1235,6 +1237,54 @@ def _suspected_typo(event_type: str) -> Optional[str]:
     return close[0] if close else None
 
 
+def _apply_member_event(
+    event: OrionEvent,
+    timeline: Any,
+    report: ApplyReport,
+    group_name: Optional[str],
+) -> None:
+    """Expand a GROUP MEMBER event into one GRUPTREE event per member."""
+    if group_name is None:
+        report.errors.append(f"Line {event.loc.line}: MEMBER needs a GROUP block")
+        report.events_skipped += 1
+        return
+
+    unknown = set(event.attributes) - {"MEMBERS", "COMMENT"}
+    if unknown:
+        report.errors.append(
+            f"Line {event.loc.line}: unknown MEMBER attribute(s): "
+            f"{', '.join(sorted(unknown))}"
+        )
+        report.events_skipped += 1
+        return
+    if "MEMBERS" not in event.attributes:
+        report.errors.append(
+            f"Line {event.loc.line}: MEMBER missing required attribute: MEMBERS"
+        )
+        report.events_skipped += 1
+        return
+
+    raw_members = str(event.attributes["MEMBERS"].value)
+    members = [member.strip() for member in raw_members.split(",")]
+    if not members or any(not member for member in members):
+        report.errors.append(
+            f"Line {event.loc.line}: MEMBERS must be a comma-delimited list of "
+            "non-empty names"
+        )
+        report.events_skipped += 1
+        return
+
+    unique_members = list(dict.fromkeys(members))
+    for member in unique_members:
+        timeline_event = timeline.add_keyword_event(
+            event_date=_iso_event_date(event.event_date),
+            keyword_name="GRUPTREE",
+            keyword_data={"CHILD_GROUP": member, "PARENT_GROUP": group_name},
+        )
+        _apply_event_comment(event, timeline_event)
+        report.events_applied += 1
+
+
 def _apply_schedule_event(
     event: OrionEvent,
     timeline: Any,
@@ -1243,6 +1293,9 @@ def _apply_schedule_event(
 ) -> None:
     """Apply one GROUP- or SCHEDULE-block event as an Eclipse keyword."""
     event_type = event.event_type.upper()
+    if event_type == "MEMBER":
+        _apply_member_event(event, timeline, report, group_name)
+        return
     if event_type in _COMPLETION_EVENT_TYPES:
         report.errors.append(
             f"Line {event.loc.line}: {event_type} is a completion event and "

GrpcInterface/Python/rips/tests/test_orion_events.py

@@ -1082,6 +1082,50 @@ class TestApplying:
         assert timeline.schedule_keyword_calls[1]["keyword_data"]["GROUP"] == "OP"
         assert timeline.schedule_keyword_calls[2]["keyword_data"]["GROUP"] == "WI"
 
+    def test_member_event_expands_to_unique_grouptree_events(self):
+        text = (
+            'ORIONEVENTS 2.0\nGROUP "PRODUCERS"\n'
+            '  @2024-01-01 MEMBER MEMBERS="WELL_A, WELL_B,WELL_A" '
+            'COMMENT="Group membership"\n'
+        )
+        timeline, report = self._apply(text)
+
+        assert report.events_applied == 2
+        assert report.errors == []
+        assert [call["keyword_data"] for call in timeline.schedule_keyword_calls] == [
+            {"CHILD_GROUP": "WELL_A", "PARENT_GROUP": "PRODUCERS"},
+            {"CHILD_GROUP": "WELL_B", "PARENT_GROUP": "PRODUCERS"},
+        ]
+        assert all(
+            event.comment == "Group membership" for event in timeline.created_events
+        )
+
+    @pytest.mark.parametrize(
+        "block,event,expected_error",
+        [
+            ("SCHEDULE", 'MEMBER MEMBERS="A"', "needs a GROUP block"),
+            ('GROUP "G"', "MEMBER", "missing required attribute"),
+            (
+                'GROUP "G"',
+                'MEMBER MEMBERS="A,,B"',
+                "non-empty names",
+            ),
+            (
+                'GROUP "G"',
+                'MEMBER MEMBERS="A" EXTRA=1',
+                "unknown MEMBER attribute",
+            ),
+        ],
+    )
+    def test_invalid_member_event_is_skipped(self, block, event, expected_error):
+        text = f"ORIONEVENTS 2.0\n{block}\n  @2024-01-01 {event}\n"
+        timeline, report = self._apply(text)
+
+        assert report.events_applied == 0
+        assert report.events_skipped == 1
+        assert timeline.schedule_keyword_calls == []
+        assert expected_error in report.errors[0]
+
     def test_completion_event_in_schedule_block_is_error(self):
         text = (
             "ORIONEVENTS 2.0\nSCHEDULE\n  @2024-01-01 PERFORATION MDSTART=1 MDEND=2\n"
@@ -1368,6 +1412,27 @@ class TestOrionEventsIntegration:
         assert "'OP'" in schedule
         assert "'WI'" in schedule
 
+    def test_member_event_generates_grouptree(self, project_with_case_and_wells):
+        project, case, timeline = project_with_case_and_wells
+        well = project.well_paths()[0]
+        document = parse_orion_events(
+            "ORIONEVENTS 2.0\n"
+            f'WELL "{well.name}"\n'
+            "  @2024-01-01 WCONHIST STATUS=OPEN\n"
+            'GROUP "PRODUCERS"\n'
+            '  @2024-01-01 MEMBER MEMBERS="WELL_A,WELL_B"\n'
+        )
+
+        report = apply_orion_document(document, timeline, project)
+        assert report.errors == []
+        assert report.events_applied == 3
+
+        schedule = timeline.generate_schedule_text(eclipse_case=case)
+        grouptree_block = schedule.split("GRUPTREE", 1)[1].split("\n/\n", 1)[0]
+        normalized_block = " ".join(grouptree_block.split())
+        assert "'WELL_A' 'PRODUCERS'" in normalized_block
+        assert "'WELL_B' 'PRODUCERS'" in normalized_block
+
     def test_apply_creates_perforations_and_schedule(self, project_with_case_and_wells):
         """End-to-end: parse -> apply -> set_timestamp -> generate schedule."""
         project, case, timeline = project_with_case_and_wells