""" SIMEVENTS well-event-timeline parser and applier. This module reads a SIMEVENTS text file describing well completion and control events over time, and applies them to a ResInsight project through the existing :class:`WellEventTimeline` API (``add_perf_event``, ``add_well_keyword_event``, ...). It is split into two independent layers: * **Layer A - parser** (:func:`parse_simulator_events` / :func:`parse_simulator_events_file`): a pure-Python tokenizer that turns the text into an intermediate representation (:class:`SimulatorEventsDocument`). It has no dependency on a running ResInsight instance and can be unit-tested standalone. It doubles as a standalone validator: ``python3 -m rips.simulator_events ...``. Several files can be combined with :func:`parse_simulator_events_files` / :func:`merge_simulator_events_documents`. * **Layer B - applier** (:func:`apply_simulator_events_document` / :func:`apply_simulator_events_file` / :func:`apply_simulator_events_files`): takes the intermediate representation plus a live ``rips`` project/timeline and calls the ``WellEventTimeline`` API, performing all semantic mapping and validation. File format grammar, version 1.2 (EBNF-ish):: document = header , { statement } ; header = "SIMEVENTS" , "1.2" ; (* first meaningful line *) statement = unit_directive | declaration | well_block_open | group_block_open | schedule_block_open | event_line | raw_text_event ; unit_directive = "UNIT" , ( "METRIC" | "FIELD" | "LAB" ) ; declaration = date_decl | duration_decl | well_decl | filter_decl ; date_decl = "DATE" , ident , "=" , date_expr ; (* DATE X = 2018-03-01 + 9d *) duration_decl = "DURATION" , ident , "=" , duration_expr ; (* DURATION RAMP = 5d12h *) well_decl = "WELL" , ident , "=" , quoted_string ; (* WELL A1 = "55_33-A-1" *) filter_decl = "FILTER" , ident , "=" , '"' , filter_expr , '"' ; (* FILTER POROPERM = "PORO > 0.4 AND PERMX > 100.0" *) filter_expr = filter_term , { ( "AND" | "OR" ) , filter_term } ; (* one combine mode; mixing AND and OR is an error *) filter_term = [ result_type , "." ] , ident , comp_op , number ; comp_op = ">" | ">=" | "<" | "<=" ; well_block_open = "WELL" , ( quoted_string | ident ) ; (* no "=" present *) group_block_open = "GROUP" , quoted_string ; (* group keyword events *) schedule_block_open = "SCHEDULE" ; (* well-less keyword events *) event_line = date_expr , event_type , { attribute } ; insert_date_event = date_expr , "INSERT_DATE" , (* SCHEDULE block only *) [ "EVERY=" , duration_expr ] , [ "UNTIL=" , date_expr ] , [ "COMMENT=" , quoted_string ] ; (* UNTIL requires EVERY; quote a value containing spaces *) raw_text_event = date_expr , "RAW_TEXT" , raw_text_attributes , newline, { raw_line , newline } , "END_RAW_TEXT" ; raw_text_attributes = "PLACEMENT=" , ( "AFTER_DATE" | "BEFORE_KEYWORD" | "AFTER_KEYWORD" | "END_OF_DATE" ) , [ "ANCHOR=" , ident ] , [ "PRIORITY=" , integer ] ; date_expr = ( iso_datetime | date_ident ) , { sign , term } ; duration_expr = term , { sign , term } ; term = duration_lit | dur_ident ; (* e.g. 2d, -12h30m, RAMP *) sign = "+" | "-" ; duration_lit = [ sign ] , component , { component } ; (* units strictly descending, each at most once *) component = number , ( "mon" | "d" | "h" | "m" | "s" ) ; number = digits | digits , "." , digits ; (* a fraction is allowed only on the last component and never on "mon" or "s" *) iso_datetime = 4digit , "-" , 2digit , "-" , 2digit , [ "T" , 2digit , ":" , 2digit , ":" , 2digit , [ "." , digits ] ] ; (* 2024-05-15 or 2024-05-15T14:45:30 *) ident = letter_or_underscore , { word_char } ; attribute = ident , "=" , ( quoted_string | bareword ) ; comment = "#" , rest-of-line ; (* line or trailing *) Notes on the grammar: * The format is line-oriented; every non-blank line is dispatched on its first token: ``SIMEVENTS`` (once), ``UNIT``, ``DATE``, ``DURATION``, ``WELL``, ``GROUP``, ``SCHEDULE`` or an event date. Anything else is an error. Keywords are uppercase and case-sensitive; duration units are lowercase. * Comments start with ``#`` (outside of double quotes) and run to end of line. * Variables are **typed**: ``DATE``, ``DURATION``, ``WELL`` (well-name alias) and ``FILTER`` (cell filter expression) declarations share one namespace and must precede use. Using a variable of the wrong type is an error that cites both the use and the declaration site. Redeclaring a name with the same type warns and the last value wins; redeclaring with a different type is an error. * Every ``DATE`` value is a :class:`datetime.datetime` with second resolution. A date-only literal means midnight; a time-of-day is written with a ``T`` separator (``2024-05-15T14:45:30``). Fractional seconds are accepted and rounded to the nearest second. The schedule generator emits a non-midnight time as the optional TIME field of the DATES keyword. * A ``DURATION`` is a Go-style literal: one or more ```` components written without spaces, units in strictly descending order and each used at most once — ``mon`` (calendar month), ``d``, ``h``, ``m`` (minute), ``s``. Examples: ``5d``, ``12h30m``, ``1mon2d``, ``-3d``, ``1.5h``. A fraction is allowed only on the last component, and never on ``mon`` or ``s``; the result is rounded to whole seconds. Unitless numbers and the legacy ``DAYS`` suffix are errors. Because ``mon`` is calendar-relative, a :class:`Duration` keeps months separate from its fixed ``timedelta`` part. Adding months clamps to the last day of a shorter month (``2024-01-31 + 1mon`` is ``2024-02-29``). * Date arithmetic is a chain of signed terms, each a duration literal or a ``DURATION`` variable: ``START + RAMP - 2d``. An operand may carry its own sign, so ``START + -2d`` is accepted and means the same as ``START - 2d``. Whitespace around ``+``/``-`` is optional but conventional. * ``WELL `` opens an event block for a declared ``WELL`` alias; ``WELL ""`` opens a block for the literal well name and never consults variables. A ``WELL`` line containing ``=`` is always a declaration. A bare ``GROUP ""`` opens a block of group-level Eclipse keyword events; the 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. A schedule may contain one attribute-free ``RESTART`` event; it truncates generated schedule output before its timestamp and is not itself emitted as a keyword. * `` INSERT_DATE`` inside a ``SCHEDULE`` block 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 ensures a summary report at that date. ``EVERY=`` makes it recurring (``EVERY=1mon``, ``EVERY=30d``, ``EVERY=12mon``, ``EVERY=RAMP``), and an inclusive ``UNTIL=`` sets the end date (``UNTIL=START+90d`` or ``UNTIL="START + 90d"``). When ``UNTIL`` is omitted, the latest event date is used. Occurrence ``n`` is the start date plus ``n`` times the month part, then plus ``n`` times the fixed part, so monthly and yearly recurrences stay anchored to the initial calendar day, clamping to the end of shorter months. ``EVERY`` must be positive. ``COMMENT`` is written as a ``--`` comment directly below each generated date. A statement is expanded into one insert-date event per occurrence, each keeping the ``COMMENT`` of the statement; they are collected on :attr:`SimulatorEventsDocument.insert_date_events`. The applier turns each into a timeline event with ``WellEventTimeline.add_insert_date_event()``, so ``generate_schedule_text()`` emits the dates without further arguments. An ``INSERT_DATE`` event is not tied to any well and generates no keyword. * ``RAW_TEXT`` is valid only inside a ``SCHEDULE`` block. Its body is copied without parsing or formatting through the mandatory standalone ``END_RAW_TEXT`` line. ``PLACEMENT`` is ``AFTER_DATE``, ``BEFORE_KEYWORD``, ``AFTER_KEYWORD`` or ``END_OF_DATE``. Before/after-keyword placement requires ``ANCHOR=``; the other placements forbid it. ``PRIORITY`` is an optional integer (default 0); lower values are emitted first and source order breaks ties. * Double quotes are used everywhere: well names, filter expressions and attribute values, e.g. ``FILTER="SOIL > 0.8 AND PERMX > 200"``. * Every attribute is ``KEY=VALUE``; bare positional tokens are rejected. * Event types inside a WELL block are either the built-in completion events ``PERFORATION``, ``SEGMENT``, ``VALVE``, ``STATE`` and ``WELSPECS``, or any Eclipse well keyword (``WCONHIST``, ``WELTARG``, ``WRFTPLT``, ``WCONPROD``, ...), which is passed through generically with the well name injected as WELL. Event ``WELSPECS`` accepts partial updates to ``GROUP``, ``CROSSFLOW``, ``REFDEPTH``, ``PHASE`` (OIL/GAS/WATER/LIQUID) and ``PRESSURE_COMPONENTS`` (H--/HF-/HFA, the WELSEGS pressure drop components). Omitted values inherit the previous WELSPECS state, initially using the well's completion export settings. There may be multiple WELSPECS events for a well, but not at the same timestamp. Each emits a WELSPECS record with the cumulative state. Event values are materialized back onto completion settings by ``timeline.set_timestamp()``. Event types inside a GROUP block are Eclipse group keywords with the group name injected as GROUP. Event types inside a SCHEDULE block are Eclipse schedule keywords passed through as-is. An event type that closely resembles a built-in is treated as a typo per the ``on_unknown_event`` policy instead of being passed through. * On a PERFORATION event, ``FILTER=`` references a declared ``FILTER`` variable and ``FILTER=""`` is an inline anonymous filter expression. The applier materializes each used filter as a case-level combined data filter and attaches it to the perforation event. A filter term is ``[TYPE.]NAME NUMBER`` with ``>``, ``>=``, ``<`` or ``<=``; bounds are inclusive, so ``>`` behaves as ``>=``. An unqualified result name is searched in STATIC_NATIVE, then DYNAMIC_NATIVE, then GENERATED results; a ``TYPE.`` qualifier (``STATIC``/``DYNAMIC``/``GENERATED`` or the full ``*_NATIVE`` form, case-insensitive) restricts the search to that type. * Any other attribute key parses; ``FILTER`` on events other than PERFORATION is ignored with a warning when applied. * The parser recovers per line and reports **all** errors in one pass: the raised :class:`SimulatorEventsParseError` carries one :class:`ParseIssue` per problem. Unknown names come with "did you mean" suggestions where possible. * Legacy ``SET`` variables and single-quoted well names are not supported. * Several files may be imported together. Each file is parsed on its own (variables are file-local), then the documents are concatenated in the given order. Keyword events with the same owner (well, group or schedule), type and timestamp are merged, whether they come from one file or several files. A repeated attribute takes the later value, with a warning that names both source locations when the values differ. All files must use the same ``UNIT``. """ from __future__ import annotations import calendar import copy import datetime import difflib import os import re from dataclasses import dataclass, field, replace from typing import Any, Callable, Dict, List, Optional, Set, Tuple, Union from .exception import RipsError AttrScalar = Union[str, int, float, bool] # --------------------------------------------------------------------------- # Exceptions and intermediate representation (Layer A output) # --------------------------------------------------------------------------- @dataclass(frozen=True) class SourceLoc: """Location of a construct in the source file (1-based line number). ``source`` names the originating file when several files are combined (see :func:`merge_simulator_events_documents`); None for a single text. """ line: int text: str source: Optional[str] = None def _loc_label(loc: SourceLoc) -> str: """Return a message prefix such as ``Line 12`` or ``a.events, line 12``.""" if loc.source is None: return f"Line {loc.line}" return f"{loc.source}, line {loc.line}" def _loc_reference(loc: SourceLoc) -> str: """Return an inline reference such as ``line 12`` or ``line 12 of a.events``.""" if loc.source is None: return f"line {loc.line}" return f"line {loc.line} of {loc.source}" @dataclass(frozen=True) class ParseIssue: """A single fatal problem found while parsing.""" message: str loc: Optional[SourceLoc] class SimulatorEventsParseError(Exception): """Raised for structural or type errors while parsing. ``errors`` holds one :class:`ParseIssue` per problem; a single parse pass collects every bad line before raising. """ def __init__( self, message: Optional[str] = None, loc: Optional[SourceLoc] = None, errors: Optional[List[ParseIssue]] = None, ) -> None: if errors is not None: self.errors = list(errors) else: self.errors = [ParseIssue(message=message or "", loc=loc)] self.loc = loc lines = [ f"{_loc_label(issue.loc)}: {issue.message}" if issue.loc else issue.message for issue in self.errors ] super().__init__("\n".join(lines)) @dataclass(frozen=True) class ParseWarning: """A non-fatal issue encountered while parsing.""" message: str loc: SourceLoc @dataclass(frozen=True) class FilterTerm: """One comparison in a filter expression: ``[TYPE.]NAME value``. ``result_type`` is ``STATIC_NATIVE``, ``DYNAMIC_NATIVE`` or ``GENERATED`` when the term is qualified, or None to search all three (in that order). """ result_name: str result_type: Optional[str] op: str value: float @dataclass(frozen=True, order=False) class Duration: """A signed calendar-aware duration: whole months plus a fixed time span. ``months`` is calendar-relative (``mon``); ``delta`` holds the ``d``, ``h``, ``m`` and ``s`` components with second resolution. ``str()`` renders the Go-style form used in the file format, e.g. ``1mon5d12h30m`` or ``-3d``. """ months: int = 0 delta: datetime.timedelta = datetime.timedelta(0) def __add__(self, other: "Duration") -> "Duration": if not isinstance(other, Duration): return NotImplemented return Duration(self.months + other.months, self.delta + other.delta) def __sub__(self, other: "Duration") -> "Duration": if not isinstance(other, Duration): return NotImplemented return Duration(self.months - other.months, self.delta - other.delta) def __neg__(self) -> "Duration": return Duration(-self.months, -self.delta) def __bool__(self) -> bool: return bool(self.months) or bool(self.delta) def __radd__(self, other: datetime.datetime) -> datetime.datetime: if not isinstance(other, datetime.datetime): return NotImplemented return _add_months(other, self.months) + self.delta def __rsub__(self, other: datetime.datetime) -> datetime.datetime: if not isinstance(other, datetime.datetime): return NotImplemented return other + (-self) def __str__(self) -> str: negative = self.months < 0 or (self.months == 0 and self.delta < _ZERO_DELTA) months = abs(self.months) total_seconds = abs(int(self.delta.total_seconds())) # A mixed-sign duration (e.g. 1mon - 1d) cannot be written as a single # literal; render it as an explicit expression instead. if ( self.months and self.delta and (self.months < 0) != (self.delta < _ZERO_DELTA) ): return f"{Duration(self.months)} {'-' if self.delta < _ZERO_DELTA else '+'} {Duration(0, abs(self.delta))}" days, rem = divmod(total_seconds, 86400) hours, rem = divmod(rem, 3600) minutes, seconds = divmod(rem, 60) parts = [ f"{value}{unit}" for value, unit in ( (months, "mon"), (days, "d"), (hours, "h"), (minutes, "m"), (seconds, "s"), ) if value ] if not parts: return "0s" return ("-" if negative else "") + "".join(parts) _ZERO_DELTA = datetime.timedelta(0) def _add_months(value: datetime.datetime, months: int) -> datetime.datetime: """Shift a datetime by whole months, clamping to the end of shorter months.""" if not months: return value month_index = value.year * 12 + value.month - 1 + months year, zero_based_month = divmod(month_index, 12) month = zero_based_month + 1 day = min(value.day, calendar.monthrange(year, month)[1]) return value.replace(year=year, month=month, day=day) @dataclass(frozen=True) class FilterExpr: """A parsed filter expression: comparison terms with one combine mode.""" terms: Tuple[FilterTerm, ...] combine_mode: str raw: str def __str__(self) -> str: return self.raw @dataclass(frozen=True) class EventFilter: """A cell filter attached to a PERFORATION event. ``name`` is the FILTER declaration name, or None for an inline expression. """ name: Optional[str] expr: FilterExpr @dataclass(frozen=True) class SimulatorEventValue: """A typed variable: kind is ``DATE``, ``DURATION``, ``WELL`` or ``FILTER``. A ``DATE`` value is a :class:`datetime.datetime` (midnight when declared without a time-of-day), a ``DURATION`` value is a :class:`Duration`, a ``WELL`` value is a string and a ``FILTER`` value is a :class:`FilterExpr`. """ kind: str value: Union[datetime.datetime, Duration, str, FilterExpr] loc: SourceLoc @dataclass(frozen=True) class AttrValue: """A single ``KEY=VALUE`` attribute with its type-inferred value.""" raw: str value: AttrScalar quoted: bool @dataclass class WellSpecState: """Fully resolved cumulative state for one WELSPECS event.""" group: str crossflow: bool refdepth: Optional[float] phase: str pressure_components: str @dataclass class SimulatorEvent: """One event line in an enclosing WELL, GROUP or SCHEDULE block.""" event_type: str event_date: datetime.datetime attributes: Dict[str, AttrValue] loc: SourceLoc filter: Optional[EventFilter] = None well_spec: Optional[WellSpecState] = None raw_text: Optional[str] = None raw_placement: Optional[str] = None raw_anchor: Optional[str] = None raw_priority: int = 0 scope: str = "SCHEDULE" scope_name: Optional[str] = None @dataclass class WellBlock: """A ``WELL`` block header followed by its events.""" well_name: str events: List[SimulatorEvent] = field(default_factory=list) loc: SourceLoc = SourceLoc(0, "") @dataclass class GroupBlock: """A ``GROUP`` block header followed by its keyword events.""" group_name: str events: List[SimulatorEvent] = field(default_factory=list) loc: SourceLoc = SourceLoc(0, "") @dataclass(frozen=True) class _ReportSpec: """One INSERT_DATE event, expanded after all event dates are known.""" event: SimulatorEvent every: Optional[Duration] end: Optional[datetime.datetime] @property def start(self) -> datetime.datetime: return self.event.event_date @property def loc(self) -> SourceLoc: return self.event.loc @dataclass class SimulatorEventsDocument: """Parsed, lossless representation of a SIMEVENTS file.""" version: str unit_system: str = "METRIC" variables: Dict[str, SimulatorEventValue] = field(default_factory=dict) wells: List[WellBlock] = field(default_factory=list) groups: List[GroupBlock] = field(default_factory=list) schedule_events: List[SimulatorEvent] = field(default_factory=list) insert_date_events: List[SimulatorEvent] = field(default_factory=list) warnings: List[ParseWarning] = field(default_factory=list) source: Optional[str] = None @property def insert_dates(self) -> List[datetime.datetime]: """The dates of the expanded ``INSERT_DATE`` events, in document order.""" return [event.event_date for event in self.insert_date_events] def _iso_event_date(event_date: datetime.datetime) -> str: """Format an event date for the timeline API. Midnight timestamps are written as a plain ``YYYY-MM-DD``; any other time of day is written as ``YYYY-MM-DDTHH:MM:SS`` (second resolution). """ if event_date.time() == datetime.time.min: return event_date.date().isoformat() return event_date.isoformat(timespec="seconds") def _event_context(event: SimulatorEvent) -> str: """Return the source scope and timestamp identifying an event.""" scope = event.scope if event.scope_name is not None: scope += f' "{event.scope_name}"' return f"[{scope}, date {_iso_event_date(event.event_date)}]" def _event_message(event: SimulatorEvent, message: str) -> str: """Add source line, scope and timestamp to an event-level message.""" return f"{_loc_label(event.loc)} {_event_context(event)}: {message}" def _set_event_scopes( wells: List[WellBlock], groups: List[GroupBlock], schedule_events: List[SimulatorEvent], ) -> None: """Attach enclosing block information to parsed events.""" for well in wells: for event in well.events: event.scope = "WELL" event.scope_name = well.well_name for group in groups: for event in group.events: event.scope = "GROUP" event.scope_name = group.group_name for event in schedule_events: event.scope = "SCHEDULE" event.scope_name = None # --------------------------------------------------------------------------- # Layer A: pure parser # --------------------------------------------------------------------------- _KEYWORDS = ( "SIMEVENTS", "UNIT", "DATE", "DURATION", "WELL", "FILTER", "GROUP", "SCHEDULE", ) _IDENT = r"[A-Za-z_]\w*" _ISO_DATE = r"\d{4}-\d{2}-\d{2}(?:T\d{2}:\d{2}:\d{2}(?:\.\d+)?)?" _DATE_BASE = rf"(?P{_ISO_DATE}|{_IDENT})" # A duration literal is Go-style: e.g. 5d, 12h30m, -1mon2d, 1.5h. The regex is # deliberately loose (any number/unit-letters run); _parse_duration_literal # validates unit names, ordering and fractions and produces the diagnostics. # Require unit letters between numbers to avoid ambiguous digit-run splits. _DURATION_LIT = r"[-+]?\d+(?:\.\d+)?(?:[A-Za-z]+\d+(?:\.\d+)?)*[A-Za-z]*" # A term operand may carry its own sign, so "START + -2d" subtracts two days. _TERMS_BODY = rf"(?:\s*[-+]\s*(?:{_DURATION_LIT}|{_IDENT}))*" _TERMS = rf"(?P{_TERMS_BODY})" _HEADER_RE = re.compile(r"^SIMEVENTS\s+(?P\d+\.\d+)$") _UNIT_RE = re.compile(r"^UNIT\s+(?PMETRIC|FIELD|LAB)$") _DATE_DECL_RE = re.compile(rf"^DATE\s+(?P{_IDENT})\s*=\s*{_DATE_BASE}{_TERMS}$") _DURATION_DECL_RE = re.compile( rf"^DURATION\s+(?P{_IDENT})\s*=\s*(?P{_DURATION_LIT}|{_IDENT}){_TERMS}$" ) _WELL_DECL_RE = re.compile(rf'^WELL\s+(?P{_IDENT})\s*=\s*"(?P[^"]*)"$') # Pre-1.2 INSERT_DATE line syntax, only recognized to suggest the new form. _LEGACY_INSERT_DATE_RE = re.compile( rf"^INSERT_DATE\s+{_DATE_BASE}{_TERMS}" rf"(?:\s+EVERY\s+(?:(?P\d+)\s+)?" rf"(?PDAY|DAYS|MONTH|MONTHS|YEAR|YEARS)" rf"(?:\s+UNTIL\s+(?P{_ISO_DATE}|{_IDENT})" rf"(?P{_TERMS_BODY})" rf")?)?$" ) _FILTER_DECL_RE = re.compile(rf'^FILTER\s+(?P{_IDENT})\s*=\s*"(?P[^"]*)"$') _FILTER_SPLIT_RE = re.compile(r"\s+(AND|OR)\s+") _NUMBER = r"[-+]?(?:\d+\.?\d*|\.\d+)(?:[eE][-+]?\d+)?" _FILTER_TERM_RE = re.compile( rf"^(?:(?P{_IDENT})\.)?(?P{_IDENT})\s*" rf"(?P>=|<=|>|<)\s*(?P{_NUMBER})$" ) # Result-type qualifiers accepted in filter terms (case-insensitive). _RESULT_TYPE_ALIASES = { "STATIC": "STATIC_NATIVE", "STATIC_NATIVE": "STATIC_NATIVE", "DYNAMIC": "DYNAMIC_NATIVE", "DYNAMIC_NATIVE": "DYNAMIC_NATIVE", "GENERATED": "GENERATED", } _WELL_BLOCK_RE = re.compile(rf'^WELL\s+(?:"(?P[^"]*)"|(?P{_IDENT}))$') _GROUP_BLOCK_RE = re.compile(r'^GROUP\s+"(?P[^"]*)"$') _EVENT_RE = re.compile(rf"^{_DATE_BASE}{_TERMS}\s+(?P.+)$") # Attribute values of INSERT_DATE: EVERY=, UNTIL=. _DURATION_EXPR_RE = re.compile(rf"^\s*(?P{_DURATION_LIT}|{_IDENT}){_TERMS}\s*$") _DATE_EXPR_RE = re.compile(rf"^\s*{_DATE_BASE}{_TERMS}\s*$") _TERM_RE = re.compile(rf"([-+])\s*({_DURATION_LIT}|{_IDENT})") _DURATION_COMPONENT_RE = re.compile(r"(?P\d+(?:\.\d+)?)(?P[A-Za-z]*)") _TIME_OF_DAY_RE = re.compile(r"\d{2}:\d{2}:\d{2}(?:\.\d+)?") # Duration units in the order they must appear, with their length in seconds # (None for the calendar-relative month). _DURATION_UNITS: Tuple[Tuple[str, Optional[int]], ...] = ( ("mon", None), ("d", 86400), ("h", 3600), ("m", 60), ("s", 1), ) _DURATION_UNIT_ORDER = {unit: index for index, (unit, _) in enumerate(_DURATION_UNITS)} _DURATION_UNIT_LIST = ", ".join(unit for unit, _ in _DURATION_UNITS) _DURATION_NO_FRACTION_UNITS = {"mon", "s"} _ATTR_RE = re.compile(r'(?P[A-Za-z_]\w*)\s*=\s*(?:"(?P[^"]*)"|(?P\S+))') def parse_simulator_events_file( path: Union[str, "os.PathLike[str]"], *, source: Optional[str] = None, ) -> SimulatorEventsDocument: """Parse a SIMEVENTS file from disk into an :class:`SimulatorEventsDocument`. ``source`` is an optional label (typically the file name) attached to every source location, so diagnostics can identify the file. """ with open(path, "r", encoding="utf-8") as handle: return parse_simulator_events(handle.read(), source=source) def _source_labels(paths: List[Union[str, "os.PathLike[str]"]]) -> List[str]: """Return file names as labels, falling back to full paths when ambiguous.""" names = [os.path.basename(os.fspath(path)) for path in paths] if len(set(names)) == len(names): return names return [os.fspath(path) for path in paths] def parse_simulator_events_files( paths: List[Union[str, "os.PathLike[str]"]], ) -> SimulatorEventsDocument: """Parse several SIMEVENTS files and merge them into one document. Every file is parsed before failing, so the raised :class:`SimulatorEventsParseError` reports the errors of all files. Source locations carry the file name. See :func:`merge_simulator_events_documents`. """ documents: List[SimulatorEventsDocument] = [] errors: List[ParseIssue] = [] for path, label in zip(paths, _source_labels(paths)): try: documents.append(parse_simulator_events_file(path, source=label)) except SimulatorEventsParseError as exc: errors.extend( issue if issue.loc is not None else ParseIssue(f"{label}: {issue.message}", None) for issue in exc.errors ) if errors: raise SimulatorEventsParseError(errors=errors) return merge_simulator_events_documents(documents) def _version_key(version: str) -> Tuple[int, ...]: return tuple(int(part) for part in version.split(".")) def merge_simulator_events_documents( documents: List[SimulatorEventsDocument], ) -> SimulatorEventsDocument: """Combine several parsed documents into one, in the given order. Blocks, events, inserted dates and warnings are concatenated; the inputs are not modified. Matching events from different files are merged afterwards by :func:`coalesce_simulator_events_document` exactly as if they were written in one file in the given order, so a later file overrides repeated attributes (with a warning when the values differ). Each file is parsed independently: variables are file-local, and a recurring ``INSERT_DATE`` without ``UNTIL`` ends at the last event of its own file. Raises: SimulatorEventsParseError: if the documents use different unit systems, or the combined events violate document rules (more than one ``RESTART``, or several ``WELSPECS`` for a well at one timestamp). """ if not documents: raise ValueError("At least one document is required") sources = copy.deepcopy(documents) first = sources[0] for document in sources[1:]: if document.unit_system != first.unit_system: raise SimulatorEventsParseError( f"Unit system mismatch: {document.source or 'document'} uses " f"{document.unit_system}, but {first.source or 'the first document'} " f"uses {first.unit_system}" ) result = SimulatorEventsDocument( version=max((document.version for document in sources), key=_version_key), unit_system=first.unit_system, ) for document in sources: result.warnings.extend(document.warnings) for name, value in document.variables.items(): previous = result.variables.get(name) if ( previous is not None and previous.kind == value.kind == "FILTER" and previous.value != value.value ): result.warnings.append( ParseWarning( f"FILTER '{name}' is also declared on " f"{_loc_reference(previous.loc)} with a different " "expression; each file keeps its own definition", value.loc, ) ) result.variables[name] = value result.wells.extend(document.wells) result.groups.extend(document.groups) result.schedule_events.extend(document.schedule_events) result.insert_date_events.extend(document.insert_date_events) errors = _restart_validation_issues( result.wells, result.groups, result.schedule_events ) errors.extend(_wellspec_validation_issues(result.wells)) errors.extend(_segment_validation_issues(result.wells)) if errors: raise SimulatorEventsParseError(errors=errors) return result def parse_simulator_events( text: str, *, source: Optional[str] = None ) -> SimulatorEventsDocument: """Parse SIMEVENTS 1.2 text into an :class:`SimulatorEventsDocument`. ``source`` is an optional label (typically the file name) stored on every :class:`SourceLoc` and on the document. Raises: SimulatorEventsParseError: carrying every error found in the file. A missing or unsupported ``SIMEVENTS`` header aborts immediately. """ version: Optional[str] = None unit_holder = ["METRIC"] # mutable so _parse_line can update it variables: Dict[str, SimulatorEventValue] = {} wells: List[WellBlock] = [] groups: List[GroupBlock] = [] schedule_events: List[SimulatorEvent] = [] report_specs: List[_ReportSpec] = [] warnings: List[ParseWarning] = [] errors: List[ParseIssue] = [] # Event lines append to the current sink: a WellBlock's event list or the # document-level schedule_events list. current_events: Optional[List[SimulatorEvent]] = None source_lines = text.splitlines() line_index = 0 while line_index < len(source_lines): raw_line = source_lines[line_index] loc = SourceLoc(line=line_index + 1, text=raw_line, source=source) line = _strip_comment(raw_line).strip() line_index += 1 if not line: continue # The first meaningful line must be the header; header problems are # fatal since nothing else can be interpreted without a version. if version is None: match = _HEADER_RE.match(line) if match is None: raise SimulatorEventsParseError( "File must start with 'SIMEVENTS '", loc ) version = match.group("version") _check_version(version, loc) continue if _is_raw_text_header(line): end_index = line_index while ( end_index < len(source_lines) and source_lines[end_index].strip() != "END_RAW_TEXT" ): end_index += 1 if end_index == len(source_lines): errors.append(ParseIssue("Unterminated RAW_TEXT block", loc)) break body_lines = source_lines[line_index:end_index] line_index = end_index + 1 try: if current_events is not schedule_events: raise SimulatorEventsParseError( "RAW_TEXT is only valid in a SCHEDULE block", loc ) current_events.append( _parse_raw_text_event(line, body_lines, variables, loc) ) except SimulatorEventsParseError as exc: errors.extend(exc.errors) continue try: current_events = _parse_line( line, loc, variables, wells, groups, schedule_events, report_specs, warnings, current_events, unit_holder, ) except SimulatorEventsParseError as exc: errors.extend(exc.errors) if line.split(None, 1)[0] in ("WELL", "GROUP", "SCHEDULE") or ( _EVENT_RE.match(line) is not None and current_events is None ): # Suppress cascading errors from lines belonging to a broken # (or missing) block: swallow them into a discarded list. current_events = [] if version is None: raise SimulatorEventsParseError("Empty file: missing 'SIMEVENTS' header") _set_event_scopes(wells, groups, schedule_events) errors.extend(_restart_validation_issues(wells, groups, schedule_events)) errors.extend(_wellspec_validation_issues(wells)) errors.extend(_segment_validation_issues(wells)) insert_date_events, report_errors = _expand_report_specs( report_specs, wells, groups, schedule_events ) errors.extend(report_errors) if errors: raise SimulatorEventsParseError(errors=errors) return SimulatorEventsDocument( version=version, unit_system=unit_holder[0], variables=variables, wells=wells, groups=groups, schedule_events=schedule_events, insert_date_events=insert_date_events, warnings=warnings, source=source, ) def _report_occurrence( start: datetime.datetime, every: Duration, occurrence: int ) -> datetime.datetime: """Return occurrence ``n`` of a series: ``start + n*months + n*delta``. The month part is applied relative to ``start`` so a monthly series stays anchored to its initial calendar day, clamping to shorter months. """ return _add_months(start, every.months * occurrence) + every.delta * occurrence def _expand_report_specs( report_specs: List[_ReportSpec], wells: List[WellBlock], groups: List[GroupBlock], schedule_events: List[SimulatorEvent], ) -> Tuple[List[SimulatorEvent], List[ParseIssue]]: """Expand every ``INSERT_DATE`` spec into one event per occurrence. Each occurrence is a copy of the source event, so an attached ``COMMENT`` is kept for every date of a recurring series. """ event_dates = [event.event_date for well in wells for event in well.events] event_dates.extend(event.event_date for group in groups for event in group.events) event_dates.extend(event.event_date for event in schedule_events) last_event_date = max(event_dates) if event_dates else None def occurrence_event( spec: _ReportSpec, event_date: datetime.datetime ) -> SimulatorEvent: attributes = { name: value for name, value in spec.event.attributes.items() if name == "COMMENT" } return replace(spec.event, event_date=event_date, attributes=attributes) events: List[SimulatorEvent] = [] issues: List[ParseIssue] = [] for spec in report_specs: if spec.every is None: events.append(occurrence_event(spec, spec.start)) continue end = spec.end if spec.end is not None else last_event_date if end is None: issues.append( ParseIssue( "Recurring INSERT_DATE without UNTIL requires at least one event", spec.loc, ) ) continue if end < spec.start: issues.append( ParseIssue( "INSERT_DATE end date must not precede its start date", spec.loc ) ) continue occurrence = 0 while True: try: value = _report_occurrence(spec.start, spec.every, occurrence) except (OverflowError, ValueError): break if value > end: break events.append(occurrence_event(spec, value)) occurrence += 1 return events, issues def _restart_validation_issues( wells: List[WellBlock], groups: List[GroupBlock], schedule_events: List[SimulatorEvent], ) -> List[ParseIssue]: """Validate placement, cardinality and shape of RESTART events.""" issues: List[ParseIssue] = [] for well_block in wells: for event in well_block.events: if event.event_type.upper() == "RESTART": issues.append( ParseIssue( f"{_event_context(event)}: RESTART is only valid in a " "SCHEDULE block", event.loc, ) ) for group_block in groups: for event in group_block.events: if event.event_type.upper() == "RESTART": issues.append( ParseIssue( f"{_event_context(event)}: RESTART is only valid in a " "SCHEDULE block", event.loc, ) ) restart_events = [ event for event in schedule_events if event.event_type.upper() == "RESTART" ] for event in restart_events: if event.attributes: issues.append( ParseIssue( f"{_event_context(event)}: RESTART takes no attributes", event.loc ) ) for event in restart_events[1:]: issues.append( ParseIssue( f"{_event_context(event)}: Only one RESTART event is allowed per " "schedule", event.loc, ) ) return issues def _wellspec_validation_issues(wells: List[WellBlock]) -> List[ParseIssue]: """Reject multiple WELSPECS events for one well at the same timestamp.""" seen: Dict[Tuple[str, datetime.datetime], SimulatorEvent] = {} issues: List[ParseIssue] = [] for well in wells: for event in well.events: if event.event_type.upper() != "WELSPECS": continue key = (well.well_name, event.event_date) previous = seen.get(key) if previous is not None: issues.append( ParseIssue( f"{_event_context(event)}: WELSPECS already defined " f"(first definition on {_loc_reference(previous.loc)})", event.loc, ) ) else: seen[key] = event return issues def _segment_validation_issues(wells: List[WellBlock]) -> List[ParseIssue]: """Validate the segment-length attributes of SEGMENT events. Lengths must be positive numbers, a fixed length cannot be combined with min/max lengths, and the min length cannot exceed the max length. """ issues: List[ParseIssue] = [] for well in wells: for event in well.events: if event.event_type.upper() != "SEGMENT": continue lengths: Dict[str, float] = {} for name in _SEGMENT_LENGTH_KWARGS: attr = event.attributes.get(name) if attr is None: continue value = attr.value if isinstance(value, bool) or not isinstance(value, (int, float)): issues.append( ParseIssue( f"{_event_context(event)}: {name} must be a number, " f"got {attr.raw!r}", event.loc, ) ) elif value <= 0: issues.append( ParseIssue( f"{_event_context(event)}: {name} must be greater " f"than zero, got {attr.raw}", event.loc, ) ) else: lengths[name] = float(value) present = [ name for name in _SEGMENT_LENGTH_KWARGS if name in event.attributes ] if "FIXED_SEGMENT_LENGTH" in present and len(present) > 1: issues.append( ParseIssue( f"{_event_context(event)}: FIXED_SEGMENT_LENGTH cannot be " "combined with MIN_SEGMENT_LENGTH or MAX_SEGMENT_LENGTH", event.loc, ) ) min_length = lengths.get("MIN_SEGMENT_LENGTH") max_length = lengths.get("MAX_SEGMENT_LENGTH") if ( min_length is not None and max_length is not None and min_length > max_length ): issues.append( ParseIssue( f"{_event_context(event)}: MIN_SEGMENT_LENGTH ({min_length:g}) " f"exceeds MAX_SEGMENT_LENGTH ({max_length:g})", event.loc, ) ) return issues _SUPPORTED_VERSION = "1.2" def _check_version(version: str, loc: SourceLoc) -> None: if version == _SUPPORTED_VERSION: return message = ( f"Unsupported SIMEVENTS version '{version}'; expected {_SUPPORTED_VERSION}" ) if version == "1.1": message += ( " (1.2 writes INSERT_DATE as a SCHEDULE event, e.g. " "'INSERT_DATE 2024-01-01 EVERY 3 MONTHS' -> " "'2024-01-01 INSERT_DATE EVERY=3mon')" ) if version == "1.0": message += ( " (durations require a unit, e.g. '5 DAYS' -> '5d', dates use 'T' " "between date and time, and INSERT_DATE is written as a SCHEDULE " "event, e.g. '2024-01-01 INSERT_DATE EVERY=3mon')" ) raise SimulatorEventsParseError(message, loc) def _parse_line( line: str, loc: SourceLoc, variables: Dict[str, SimulatorEventValue], wells: List[WellBlock], groups: List[GroupBlock], schedule_events: List[SimulatorEvent], report_specs: List[_ReportSpec], warnings: List[ParseWarning], current_events: Optional[List[SimulatorEvent]], unit_holder: List[str], ) -> Optional[List[SimulatorEvent]]: """Dispatch one non-header line; returns the current event sink.""" first = line.split(None, 1)[0] if first == "UNIT": match = _UNIT_RE.match(line) if match is None: raise SimulatorEventsParseError( f"Malformed UNIT line: {line!r} (expected UNIT METRIC|FIELD|LAB)", loc ) unit_holder[0] = match.group("unit") return current_events if first == "GROUP": match = _GROUP_BLOCK_RE.match(line) if match is None: raise SimulatorEventsParseError( f'Malformed GROUP line: {line!r} (expected GROUP "")', loc, ) new_group = GroupBlock(group_name=match.group("name"), loc=loc) groups.append(new_group) return new_group.events if first == "SCHEDULE": if line != "SCHEDULE": raise SimulatorEventsParseError( f"Malformed SCHEDULE line: {line!r} (SCHEDULE takes no arguments)", loc ) return schedule_events if first == "INSERT_DATE": raise SimulatorEventsParseError(_legacy_insert_date_message(line), loc) if first == "DATE": match = _DATE_DECL_RE.match(line) if match is None: hint = "" if re.search(rf"\d{{4}}-\d{{2}}-\d{{2}}\s+{_TIME_OF_DAY_RE.pattern}", line): hint = "; a time-of-day must be joined to the date with 'T'" raise SimulatorEventsParseError( f"Malformed DATE declaration: {line!r} " "(expected DATE NAME = [+|- ...])" f"{hint}", loc, ) value = _eval_date_expr( match.group("base"), match.group("terms"), variables, loc ) _declare( match.group("name"), SimulatorEventValue("DATE", value, loc), variables, warnings, loc, ) return current_events if first == "DURATION": match = _DURATION_DECL_RE.match(line) if match is None: hint = "" if re.search(r"\b(?:DAYS|days)\b", line): hint = "; the DAYS suffix is not supported, write e.g. 5d" raise SimulatorEventsParseError( f"Malformed DURATION declaration: {line!r} " "(expected DURATION NAME = [+|- ...], " f"e.g. DURATION RAMP = 5d12h){hint}", loc, ) duration = _eval_duration_expr( match.group("base"), match.group("terms"), variables, loc ) _declare( match.group("name"), SimulatorEventValue("DURATION", duration, loc), variables, warnings, loc, ) return current_events if first == "FILTER": match = _FILTER_DECL_RE.match(line) if match is None: raise SimulatorEventsParseError( f"Malformed FILTER declaration: {line!r} " '(expected FILTER NAME = " [AND|OR ...]")', loc, ) expr = _parse_filter_expr(match.group("expr"), loc) _declare( match.group("name"), SimulatorEventValue("FILTER", expr, loc), variables, warnings, loc, ) return current_events if first == "WELL": decl_match = _WELL_DECL_RE.match(line) if decl_match is not None: _declare( decl_match.group("name"), SimulatorEventValue("WELL", decl_match.group("well"), loc), variables, warnings, loc, ) return current_events block_match = _WELL_BLOCK_RE.match(line) if block_match is None: raise SimulatorEventsParseError( f"Malformed WELL line: {line!r} (well names containing special " 'characters must be double-quoted, e.g. WELL "55_33-A-1")', loc, ) name = block_match.group("qname") if name is None: name = str( _lookup_var(block_match.group("ref"), "WELL", variables, loc).value ) new_well = WellBlock(well_name=name, loc=loc) wells.append(new_well) return new_well.events if first == "REPORT": raise SimulatorEventsParseError( "REPORT has been renamed to INSERT_DATE and is only valid in a " "SCHEDULE block", loc, ) if current_events is not None and _EVENT_RE.match(line): event = _parse_event_line(line, variables, loc) if event.event_type.upper() == "INSERT_DATE": if current_events is not schedule_events: raise SimulatorEventsParseError( "INSERT_DATE is only valid in a SCHEDULE block", loc ) report_specs.append(_insert_date_spec(event, variables)) return current_events current_events.append(event) return current_events if first[0].isdigit() and _EVENT_RE.match(line): raise SimulatorEventsParseError( "Event line found before any WELL or SCHEDULE block", loc ) raise SimulatorEventsParseError(_unrecognized_line_message(line, first), loc) _INSERT_DATE_ATTRS = {"EVERY", "UNTIL", "COMMENT"} _LEGACY_PERIOD_UNITS = {"DAY": ("d", 1), "MONTH": ("mon", 1), "YEAR": ("mon", 12)} def _legacy_insert_date_message(line: str) -> str: """Explain the SIMEVENTS 1.2 INSERT_DATE form, rewriting the line if possible.""" message = ( "INSERT_DATE is written as a SCHEDULE event since SIMEVENTS 1.2: " " INSERT_DATE [EVERY=] [UNTIL=]" ) match = _LEGACY_INSERT_DATE_RE.match(line) if match is None: return message start = (match.group("base") + match.group("terms")).strip() parts = [start, "INSERT_DATE"] period = match.group("period") if period: unit, factor = _LEGACY_PERIOD_UNITS[period.rstrip("S")] parts.append(f"EVERY={int(match.group('count') or 1) * factor}{unit}") if match.group("end_base"): until = (match.group("end_base") + match.group("end_terms")).strip() parts.append(f'UNTIL="{until}"' if " " in until else f"UNTIL={until}") return f"{message}; write {' '.join(parts)!r}" def _insert_date_spec( event: SimulatorEvent, variables: Dict[str, SimulatorEventValue] ) -> _ReportSpec: """Build the report specification for an ``INSERT_DATE`` schedule event. ``EVERY`` is a duration expression and ``UNTIL`` an inclusive date expression; quote either value when it contains spaces. ``COMMENT`` is accepted but not used. """ loc = event.loc unknown = set(event.attributes) - _INSERT_DATE_ATTRS if unknown: raise SimulatorEventsParseError( f"Unknown INSERT_DATE attribute(s): {', '.join(sorted(unknown))} " "(expected EVERY, UNTIL)", loc, ) every: Optional[Duration] = None if "EVERY" in event.attributes: raw = event.attributes["EVERY"].raw match = _DURATION_EXPR_RE.match(raw) if match is None: raise SimulatorEventsParseError( f"INSERT_DATE EVERY must be a duration, e.g. EVERY=1mon or " f"EVERY=30d, got {raw!r}", loc, ) every = _eval_duration_expr( match.group("base"), match.group("terms"), variables, loc ) if every.months < 0 or every.delta < _ZERO_DELTA or not every: raise SimulatorEventsParseError( f"INSERT_DATE EVERY must be a positive duration, got {raw!r}", loc ) end: Optional[datetime.datetime] = None if "UNTIL" in event.attributes: if every is None: raise SimulatorEventsParseError("INSERT_DATE UNTIL requires EVERY", loc) raw = event.attributes["UNTIL"].raw match = _DATE_EXPR_RE.match(raw) if match is None: raise SimulatorEventsParseError( f"INSERT_DATE UNTIL must be a date expression, got {raw!r}", loc ) end = _eval_date_expr(match.group("base"), match.group("terms"), variables, loc) return _ReportSpec(event=event, every=every, end=end) def _unrecognized_line_message(line: str, first: str) -> str: if first == "SIMEVENTS": return "Duplicate SIMEVENTS header" if first == "SET": return ( "SET is not supported; declare a typed variable instead, " "e.g. DATE NAME = 2018-01-01" ) if line.startswith("'"): return 'Single-quoted well names are not supported; open a well block with WELL "name"' message = f"Unrecognized line: {line!r}" close = difflib.get_close_matches(first, _KEYWORDS, n=1, cutoff=0.6) if close: message += f"; did you mean '{close[0]}'?" return message def _declare( name: str, value: SimulatorEventValue, variables: Dict[str, SimulatorEventValue], warnings: List[ParseWarning], loc: SourceLoc, ) -> None: existing = variables.get(name) if existing is not None: if existing.kind != value.kind: raise SimulatorEventsParseError( f"'{name}' is already declared as {existing.kind} " f"(line {existing.loc.line}); cannot redeclare as {value.kind}", loc, ) warnings.append(ParseWarning(f"Duplicate {value.kind} '{name}'", loc)) variables[name] = value def _lookup_var( name: str, expected_kind: str, variables: Dict[str, SimulatorEventValue], loc: SourceLoc, ) -> SimulatorEventValue: """Resolve a typed variable reference; the single typed-error site.""" value = variables.get(name) if value is None: same_kind = [n for n, v in variables.items() if v.kind == expected_kind] close = difflib.get_close_matches( name, same_kind or list(variables), n=1, cutoff=0.6 ) hint = f"; did you mean '{close[0]}'?" if close else "" raise SimulatorEventsParseError(f"Unknown variable '{name}'{hint}", loc) if value.kind != expected_kind: raise SimulatorEventsParseError( f"Variable '{name}' is a {value.kind} (declared line {value.loc.line}) " f"but a {expected_kind} is required here", loc, ) return value def _parse_duration_literal(text: str, loc: SourceLoc) -> Duration: """Parse a Go-style duration literal such as ``5d``, ``-12h30m`` or ``1.5h``. Units must be ``mon``, ``d``, ``h``, ``m`` or ``s``, appear in strictly descending order and at most once each. A fraction is allowed only on the last component and never on ``mon`` or ``s``. The fixed part is rounded to whole seconds. """ body = text negative = False if body[:1] in "+-": negative = body[0] == "-" body = body[1:] def fail(reason: str) -> SimulatorEventsParseError: return SimulatorEventsParseError(f"Invalid duration {text!r}: {reason}", loc) if not body or not body[0].isdigit(): raise fail( f"expected components with units {_DURATION_UNIT_LIST}" ) components: List[Tuple[str, str]] = [] pos = 0 while pos < len(body): match = _DURATION_COMPONENT_RE.match(body, pos) if match is None: raise fail(f"unexpected text {body[pos:]!r}") components.append((match.group("number"), match.group("unit"))) pos = match.end() months = 0 seconds = 0.0 previous_order = -1 for index, (number, unit) in enumerate(components): if not unit: if len(components) == 1: raise fail( f"a unit is required ({_DURATION_UNIT_LIST}), e.g. {number}d" ) raise fail(f"missing unit after {number!r}") if unit.upper() in ("DAY", "DAYS"): raise fail(f"the DAYS suffix is not supported; write {number}d") order = _DURATION_UNIT_ORDER.get(unit) if order is None: close = difflib.get_close_matches( unit.lower(), list(_DURATION_UNIT_ORDER), n=1, cutoff=0.6 ) hint = f"; did you mean '{close[0]}'?" if close else "" raise fail(f"unknown unit {unit!r} (expected {_DURATION_UNIT_LIST}){hint}") if order == previous_order: raise fail(f"unit '{unit}' given more than once") if order < previous_order: raise fail( f"units must be in descending order ({_DURATION_UNIT_LIST}); " f"'{unit}' cannot follow '{components[index - 1][1]}'" ) previous_order = order is_fraction = "." in number if is_fraction: if index != len(components) - 1: raise fail("a fraction is only allowed on the last component") if unit in _DURATION_NO_FRACTION_UNITS: raise fail(f"a fraction is not allowed on '{unit}'") unit_seconds = dict(_DURATION_UNITS)[unit] if unit_seconds is None: months += int(number) else: seconds += float(number) * unit_seconds duration = Duration(months, datetime.timedelta(seconds=round(seconds))) return -duration if negative else duration def _eval_term( term: str, variables: Dict[str, SimulatorEventValue], loc: SourceLoc ) -> Duration: """Evaluate one term: a duration literal or a DURATION variable.""" # Identifiers never start with a sign or a digit, so this stays unambiguous. if term[0].isdigit() or term[0] in "+-": return _parse_duration_literal(term, loc) value = _lookup_var(term, "DURATION", variables, loc).value assert isinstance(value, Duration) return value def _eval_terms( terms: str, variables: Dict[str, SimulatorEventValue], loc: SourceLoc ) -> Duration: """Evaluate a signed chain of duration terms to a net :class:`Duration`.""" total = Duration() for match in _TERM_RE.finditer(terms): sign, term = match.groups() value = _eval_term(term, variables, loc) total = total + value if sign == "+" else total - value return total def _parse_iso_datetime(base: str, loc: SourceLoc) -> datetime.datetime: """Parse ``YYYY-MM-DD[THH:MM:SS[.fff]]`` to a second-resolution datetime.""" try: if "T" in base: result = datetime.datetime.fromisoformat(base) else: result = datetime.datetime.combine( datetime.date.fromisoformat(base), datetime.time.min ) except ValueError as exc: raise SimulatorEventsParseError(f"Invalid date '{base}': {exc}", loc) if result.microsecond: result = result.replace(microsecond=0) + datetime.timedelta( seconds=round(result.microsecond / 1_000_000) ) return result def _eval_date_expr( base: str, terms: str, variables: Dict[str, SimulatorEventValue], loc: SourceLoc ) -> datetime.datetime: """Evaluate an ISO datetime or DATE variable plus optional signed duration terms.""" if base[0].isdigit(): result = _parse_iso_datetime(base, loc) else: value = _lookup_var(base, "DATE", variables, loc).value assert isinstance(value, datetime.datetime) result = value return result + _eval_terms(terms, variables, loc) def _eval_duration_expr( base: str, terms: str, variables: Dict[str, SimulatorEventValue], loc: SourceLoc ) -> Duration: """Evaluate a duration literal or DURATION variable plus optional signed terms.""" return _eval_term(base, variables, loc) + _eval_terms(terms, variables, loc) def _parse_filter_expr(text: str, loc: SourceLoc) -> FilterExpr: """Parse a filter expression into typed comparison terms.""" stripped = text.strip() if not stripped: raise SimulatorEventsParseError("Empty filter expression", loc) parts = _FILTER_SPLIT_RE.split(stripped) chunks = parts[0::2] connectors = parts[1::2] if len(set(connectors)) > 1: raise SimulatorEventsParseError( "Filter expression mixes AND and OR; a combined filter has a " "single combine mode", loc, ) combine_mode = connectors[0] if connectors else "AND" terms = tuple(_parse_filter_term(chunk, loc) for chunk in chunks) return FilterExpr(terms=terms, combine_mode=combine_mode, raw=stripped) def _parse_filter_term(chunk: str, loc: SourceLoc) -> FilterTerm: chunk = chunk.strip() match = _FILTER_TERM_RE.match(chunk) if match is None: raise SimulatorEventsParseError(_malformed_filter_term_message(chunk), loc) qual = match.group("qual") result_type: Optional[str] = None if qual is not None: result_type = _RESULT_TYPE_ALIASES.get(qual.upper()) if result_type is None: close = difflib.get_close_matches( qual.upper(), sorted(set(_RESULT_TYPE_ALIASES)), n=1, cutoff=0.6 ) hint = f"; did you mean '{close[0]}'?" if close else "" raise SimulatorEventsParseError( f"Unknown result type '{qual}' in filter term {chunk!r}{hint}", loc ) return FilterTerm( result_name=match.group("name").upper(), result_type=result_type, op=match.group("op"), value=float(match.group("value")), ) def _malformed_filter_term_message(chunk: str) -> str: if re.search(r"\b(and|or)\b", chunk): return ( f"Malformed filter term {chunk!r}: combine keywords must be " "uppercase AND / OR" ) if re.search(r"(?])=", chunk): return ( f"Malformed filter term {chunk!r}: only >, >=, < and <= are " "supported in filter expressions (bounds are inclusive, so > " "behaves as >=)" ) return ( f"Malformed filter term {chunk!r} " "(expected NAME NUMBER with >, >=, < or <=, optionally TYPE.NAME)" ) def _strip_comment(line: str) -> str: """Remove a ``#`` comment, honoring double-quoted spans.""" in_quote = False result: List[str] = [] for char in line: if char == '"': in_quote = not in_quote if char == "#" and not in_quote: break result.append(char) return "".join(result) def _is_raw_text_header(line: str) -> bool: """Return whether an event line starts a multiline RAW_TEXT block.""" match = _EVENT_RE.match(line) if match is None: return False return match.group("rest").split(None, 1)[0].upper() == "RAW_TEXT" def _parse_raw_text_event( line: str, body_lines: List[str], variables: Dict[str, SimulatorEventValue], loc: SourceLoc, ) -> SimulatorEvent: """Parse and validate a RAW_TEXT header and attach its unparsed body.""" event = _parse_event_line(line, variables, loc) allowed = {"PLACEMENT", "ANCHOR", "PRIORITY"} unknown = set(event.attributes) - allowed if unknown: raise SimulatorEventsParseError( f"Unknown RAW_TEXT attribute(s): {', '.join(sorted(unknown))}", loc ) if "PLACEMENT" not in event.attributes: raise SimulatorEventsParseError("RAW_TEXT requires PLACEMENT", loc) if not body_lines: raise SimulatorEventsParseError("RAW_TEXT body must not be empty", loc) placement_value = event.attributes["PLACEMENT"].value placement = placement_value.upper() if isinstance(placement_value, str) else "" valid_placements = { "AFTER_DATE", "BEFORE_KEYWORD", "AFTER_KEYWORD", "END_OF_DATE", } if placement not in valid_placements: raise SimulatorEventsParseError( "RAW_TEXT PLACEMENT must be AFTER_DATE, BEFORE_KEYWORD, " "AFTER_KEYWORD, or END_OF_DATE", loc, ) anchor: Optional[str] = None if "ANCHOR" in event.attributes: anchor_value = event.attributes["ANCHOR"].value if not isinstance(anchor_value, str) or not anchor_value.strip(): raise SimulatorEventsParseError( "RAW_TEXT ANCHOR must be a keyword name", loc ) anchor = anchor_value.strip().upper() anchored = placement in {"BEFORE_KEYWORD", "AFTER_KEYWORD"} if anchored and anchor is None: raise SimulatorEventsParseError( "RAW_TEXT ANCHOR is required for BEFORE_KEYWORD and AFTER_KEYWORD", loc, ) if not anchored and anchor is not None: raise SimulatorEventsParseError( "RAW_TEXT ANCHOR is only valid for BEFORE_KEYWORD and AFTER_KEYWORD", loc, ) priority = 0 if "PRIORITY" in event.attributes: priority_value = event.attributes["PRIORITY"].value if isinstance(priority_value, bool) or not isinstance(priority_value, int): raise SimulatorEventsParseError("RAW_TEXT PRIORITY must be an integer", loc) priority = priority_value event.raw_text = "\n".join(body_lines) + "\n" event.raw_placement = placement event.raw_anchor = anchor event.raw_priority = priority return event def _parse_event_line( line: str, variables: Dict[str, SimulatorEventValue], loc: SourceLoc ) -> SimulatorEvent: match = _EVENT_RE.match(line) if match is None: raise SimulatorEventsParseError(f"Malformed event line: {line!r}", loc) event_date = _eval_date_expr( match.group("base"), match.group("terms"), variables, loc ) rest = match.group("rest").strip() parts = rest.split(None, 1) event_type = parts[0] if _TIME_OF_DAY_RE.fullmatch(event_type): raise SimulatorEventsParseError( f"Malformed event line: {line!r} (a time-of-day must be joined to the " f"date with 'T', e.g. {match.group('base')}T{event_type})", loc, ) attr_str = parts[1] if len(parts) > 1 else "" attributes = _parse_attributes(attr_str, loc) event_filter: Optional[EventFilter] = None if event_type.upper() == "PERFORATION" and "FILTER" in attributes: event_filter = _resolve_event_filter(attributes["FILTER"], variables, loc) return SimulatorEvent( event_type=event_type, event_date=event_date, attributes=attributes, loc=loc, filter=event_filter, ) def _resolve_event_filter( attr: AttrValue, variables: Dict[str, SimulatorEventValue], loc: SourceLoc ) -> EventFilter: """Resolve a PERFORATION FILTER attribute to a declared or inline filter.""" if attr.quoted: return EventFilter(name=None, expr=_parse_filter_expr(attr.raw, loc)) if re.fullmatch(_IDENT, attr.raw) is None: raise SimulatorEventsParseError( "FILTER must name a declared FILTER variable or be a quoted " f"expression, got {attr.raw!r}", loc, ) value = _lookup_var(attr.raw, "FILTER", variables, loc) assert isinstance(value.value, FilterExpr) return EventFilter(name=attr.raw, expr=value.value) def _parse_attributes(attr_str: str, loc: SourceLoc) -> Dict[str, AttrValue]: attributes: Dict[str, AttrValue] = {} pos = 0 for match in _ATTR_RE.finditer(attr_str): gap = attr_str[pos : match.start()] if gap.strip(): raise SimulatorEventsParseError( f"Malformed attribute near {gap.strip()!r}", loc ) key = match.group("key").upper() qval = match.group("qval") if qval is not None: attributes[key] = AttrValue(raw=qval, value=qval, quoted=True) else: raw = match.group("val") attributes[key] = AttrValue(raw=raw, value=_infer_value(raw), quoted=False) pos = match.end() trailing = attr_str[pos:] if trailing.strip(): raise SimulatorEventsParseError( f"Malformed attribute near {trailing.strip()!r}", loc ) return attributes def _infer_value(raw: str) -> AttrScalar: """Infer bool, int, then float; otherwise keep the raw string.""" upper = raw.upper() if upper == "TRUE": return True if upper == "FALSE": return False try: return int(raw) except ValueError: pass try: return float(raw) except ValueError: pass return raw # --------------------------------------------------------------------------- # Layer B: applier # --------------------------------------------------------------------------- @dataclass class ApplyReport: """Summary of applying an :class:`SimulatorEventsDocument` to a timeline.""" events_applied: int = 0 events_skipped: int = 0 warnings: List[str] = field(default_factory=list) errors: List[str] = field(default_factory=list) def _record_event_exception( event: SimulatorEvent, error: Exception, report: ApplyReport ) -> None: """Record an event failure with its Simulator Events source context.""" report.errors.append(_event_message(event, str(error))) report.events_skipped += 1 def _record_event_parse_error( event: SimulatorEvent, error: SimulatorEventsParseError, report: ApplyReport ) -> None: """Record parser-style event failures without duplicating their line number.""" for issue in error.errors: report.errors.append(_event_message(event, issue.message)) report.events_skipped += 1 # Policies for handling unresolved references: skip silently, warn, or raise. _POLICIES = ("warn", "error", "skip") # Attributes accepted on a keyword event but intentionally not emitted. _IGNORED_KEYWORD_ATTRS = {"FILTER"} # Completion event attribute handling: (required, known-optional) per type. # FILTER is applied on PERFORATION events; it is accepted on the other # completion events but ignored with a warning. _PERF_REQUIRED = ("MDSTART", "MDEND") _PERF_KNOWN = { "MDSTART", "MDEND", "DIAMETER", "SKIN", "COMPLETION_NUMBER", "FILTER", "COMMENT", } _SEGMENT_REQUIRED = ("MDSTART", "MDEND") _SEGMENT_KNOWN = { "MDSTART", "MDEND", "INNER_DIAMETER", "ROUGHNESS", "FIXED_SEGMENT_LENGTH", "MIN_SEGMENT_LENGTH", "MAX_SEGMENT_LENGTH", "COMMENT", } # SEGMENT length attribute -> add_segment_event keyword-argument name. _SEGMENT_LENGTH_KWARGS = { "FIXED_SEGMENT_LENGTH": "fixed_segment_length", "MIN_SEGMENT_LENGTH": "min_segment_length", "MAX_SEGMENT_LENGTH": "max_segment_length", } _PRESSURE_COMPONENTS = {"H--", "HF-", "HFA"} _VALVE_REQUIRED = ("MD", "TYPE") _VALVE_KNOWN = {"MD", "TYPE", "STATE", "CV", "AREA", "COMMENT"} | { "AICD_STRENGTH", "AICD_DENSITY_CALIB_FLUID", "AICD_VISCOSITY_CALIB_FLUID", "AICD_VOL_FLOW_EXP", "AICD_VISC_FUNC_EXP", } _STATE_REQUIRED = ("STATE",) _STATE_KNOWN = {"STATE", "COMMENT"} _WELLSPEC_KNOWN = { "GROUP", "CROSSFLOW", "REFDEPTH", "PHASE", "PRESSURE_COMPONENTS", "COMMENT", } _WELLSPEC_PHASES = {"OIL", "GAS", "WATER", "LIQUID"} _COMPLETION_IGNORED = {"FILTER"} _PERF_IGNORED = _COMPLETION_IGNORED # backwards-compatible alias # SIMEVENTS -> Eclipse item-name translations per keyword. _WCONHIST_FIELD_MAP = {"VFP": "VFP_TABLE"} _WELTARG_FIELD_MAP = {"VALUE": "NEW_VALUE"} def _apply_event_comment(event: SimulatorEvent, timeline_event: Any) -> None: """Copy an optional COMMENT attribute to the created timeline event.""" comment = event.attributes.get("COMMENT") if comment is None: return timeline_event.comment = str(comment.value) timeline_event.update() def apply_simulator_events_file( path: Union[str, "os.PathLike[str]"], timeline: Any, project: Any, *, case: Any = None, **options: str, ) -> ApplyReport: """Parse a SIMEVENTS file and apply it to ``timeline``.""" document = parse_simulator_events_file(path) return apply_simulator_events_document( document, timeline, project, case=case, **options ) def apply_simulator_events_files( paths: List[Union[str, "os.PathLike[str]"]], timeline: Any, project: Any, *, case: Any = None, **options: str, ) -> ApplyReport: """Parse several SIMEVENTS files, merge them and apply them to ``timeline``. Matching events from different files are merged into one timeline event; see :func:`merge_simulator_events_documents`. """ document = parse_simulator_events_files(paths) return apply_simulator_events_document( document, timeline, project, case=case, **options ) _NON_COALESCING_EVENT_TYPES = { "MEMBER", "PERFORATION", "RAW_TEXT", "RESTART", "SEGMENT", "STATE", "VALVE", "WELSPECS", } _NON_HISTORICAL_KEYWORD_ATTRS = {"COMMENT", "FILTER"} def coalesce_simulator_events_document( document: SimulatorEventsDocument, ) -> SimulatorEventsDocument: """Return a copy with matching keyword events merged. Keyword events with the same owner, type and timestamp are merged. The first event retains its position, and attributes from later matching events are applied in source order. Well keyword attributes are then carried forward chronologically to later events of the same type. Events that create or expand domain objects are kept separate so, for example, same-date perforation intervals are not lost. Inserted dates repeating both date and comment are reduced to one event. """ result = copy.deepcopy(document) def merge_events( events: List[SimulatorEvent], *, inherit_history: bool = False ) -> List[SimulatorEvent]: merged: List[SimulatorEvent] = [] by_key: Dict[Tuple[datetime.datetime, str], SimulatorEvent] = {} attribute_locs: Dict[Tuple[datetime.datetime, str], Dict[str, SourceLoc]] = {} for event in events: event_type = event.event_type.upper() if event_type in _NON_COALESCING_EVENT_TYPES: merged.append(event) continue key = (event.event_date, event_type) existing = by_key.get(key) if existing is None: by_key[key] = event attribute_locs[key] = {name: event.loc for name in event.attributes} merged.append(event) continue repeated = ( set(existing.attributes) & set(event.attributes) - _NON_HISTORICAL_KEYWORD_ATTRS ) for name in sorted(repeated): previous = existing.attributes[name] replacement = event.attributes[name] if previous.value != replacement.value: result.warnings.append( ParseWarning( f"{_event_context(event)}: conflicting {event_type} " f"attribute '{name}' (previous value on " f"{_loc_reference(attribute_locs[key][name])}); using " f"{replacement.raw!r}", event.loc, ) ) existing.attributes.update(event.attributes) attribute_locs[key].update({name: event.loc for name in event.attributes}) if "FILTER" in event.attributes: existing.filter = event.filter existing.loc = event.loc if inherit_history: history: Dict[str, Dict[str, AttrValue]] = {} for event in sorted(merged, key=lambda item: item.event_date): event_type = event.event_type.upper() if event_type in _NON_COALESCING_EVENT_TYPES: continue attributes = history.get(event_type, {}).copy() attributes.update(event.attributes) event.attributes = attributes history[event_type] = { name: value for name, value in attributes.items() if name not in _NON_HISTORICAL_KEYWORD_ATTRS } return merged def merge_well_blocks(blocks: List[WellBlock]) -> List[WellBlock]: merged_blocks: List[WellBlock] = [] by_name: Dict[str, WellBlock] = {} for block in blocks: block_events = block.events existing = by_name.get(block.well_name) if existing is None: block.events = [] by_name[block.well_name] = block merged_blocks.append(block) by_name[block.well_name].events.extend(block_events) for block in merged_blocks: block.events = merge_events(block.events, inherit_history=True) return merged_blocks def merge_group_blocks(blocks: List[GroupBlock]) -> List[GroupBlock]: merged_blocks: List[GroupBlock] = [] by_name: Dict[str, GroupBlock] = {} for block in blocks: block_events = block.events existing = by_name.get(block.group_name) if existing is None: block.events = [] by_name[block.group_name] = block merged_blocks.append(block) by_name[block.group_name].events.extend(block_events) for block in merged_blocks: block.events = merge_events(block.events) return merged_blocks def merge_insert_date_events(events: List[SimulatorEvent]) -> List[SimulatorEvent]: """Drop repeated occurrences of the same date and comment.""" merged: List[SimulatorEvent] = [] seen: Set[Tuple[datetime.datetime, str]] = set() for event in events: comment = event.attributes.get("COMMENT") key = (event.event_date, "" if comment is None else str(comment.value)) if key in seen: continue seen.add(key) merged.append(event) return merged result.wells = merge_well_blocks(result.wells) result.groups = merge_group_blocks(result.groups) result.schedule_events = merge_events(result.schedule_events) result.insert_date_events = merge_insert_date_events(result.insert_date_events) return result def _enum_text(value: Any) -> str: """Return the serialized text of a generated enum or plain string.""" return str(getattr(value, "value", value)).upper() def _prepare_wellspec_events( events: List[SimulatorEvent], completion_settings: Any, report: ApplyReport ) -> None: """Validate WELSPECS attributes and resolve partial updates chronologically.""" state = WellSpecState( group=str(completion_settings.group_name_for_export), crossflow=bool(completion_settings.allow_well_cross_flow), refdepth=completion_settings.reference_depth_for_export, phase=_enum_text(completion_settings.well_type_for_export), pressure_components=_enum_text(completion_settings.pressure_drop), ) wellspecs = sorted( (event for event in events if event.event_type.upper() == "WELSPECS"), key=lambda event: event.event_date, ) for event in wellspecs: attrs = event.attributes unknown = set(attrs) - _WELLSPEC_KNOWN if unknown: report.errors.append( _event_message( event, f"unknown WELSPECS attribute(s): {', '.join(sorted(unknown))}", ) ) report.events_skipped += 1 continue if not (set(attrs) - {"COMMENT"}): report.errors.append( _event_message(event, "WELSPECS needs at least one setting attribute") ) report.events_skipped += 1 continue next_state = copy.copy(state) errors: List[str] = [] if "GROUP" in attrs: value = attrs["GROUP"].value if not isinstance(value, str) or not value: errors.append("GROUP must be a non-empty string") else: next_state.group = value if "CROSSFLOW" in attrs: value = attrs["CROSSFLOW"].value if not isinstance(value, bool): errors.append("CROSSFLOW must be True or False") else: next_state.crossflow = value if "REFDEPTH" in attrs: value = attrs["REFDEPTH"].value if isinstance(value, bool) or not isinstance(value, (int, float)): errors.append("REFDEPTH must be numeric") else: next_state.refdepth = float(value) if "PHASE" in attrs: value = attrs["PHASE"].value phase = value.upper() if isinstance(value, str) else "" if phase not in _WELLSPEC_PHASES: errors.append("PHASE must be OIL, GAS, WATER, or LIQUID") else: next_state.phase = phase if "PRESSURE_COMPONENTS" in attrs: value = attrs["PRESSURE_COMPONENTS"].value components = value.upper() if isinstance(value, str) else "" if components not in _PRESSURE_COMPONENTS: errors.append("PRESSURE_COMPONENTS must be H--, HF-, or HFA") else: next_state.pressure_components = components if errors: report.errors.extend(_event_message(event, error) for error in errors) report.events_skipped += 1 continue state = next_state event.well_spec = copy.copy(state) def apply_simulator_events_document( document: SimulatorEventsDocument, timeline: Any, project: Any, *, case: Any = None, on_unknown_well: str = "warn", on_unknown_event: str = "warn", ) -> ApplyReport: """Apply a parsed :class:`SimulatorEventsDocument` to a ``WellEventTimeline``. Arguments: document: The parsed document (see :func:`parse_simulator_events`). timeline: A ``rips.WellEventTimeline`` object. project: A ``rips.Project`` used to resolve well names to well paths. case: The ``rips.Case`` used to resolve filter result names and to own the combined data filters created from FILTER expressions. Only needed when the document uses filters; defaults to the project's first case. Raises :class:`RipsError` if a filter is used and no case is available, or if a filter references a result that does not exist in the case (checked up front, before applying events). on_unknown_well: ``"warn"`` (default), ``"error"`` or ``"skip"`` - behavior when a well name has no matching well path. on_unknown_event: same policy values, for event types that closely resemble a misspelled built-in event type. Other unrecognized event types are passed through as generic Eclipse keywords. Returns: ApplyReport: counts plus collected warnings/errors. Each ``INSERT_DATE`` occurrence becomes an insert-date event on the timeline, so its date is emitted as a DATES keyword (with its comment, if any) by ``timeline.generate_schedule_text()``. """ _validate_policy(on_unknown_well, "on_unknown_well") _validate_policy(on_unknown_event, "on_unknown_event") document = coalesce_simulator_events_document(document) report = ApplyReport() report.warnings.extend( f"{_loc_label(warning.loc)}: {warning.message}" for warning in document.warnings ) ctx = _prepare_filter_context(document, project, case) for well in document.wells: well_path = project.well_path_by_name(well.well_name) if well_path is None: message = f"Unknown well '{well.well_name}' ({_loc_reference(well.loc)})" if on_unknown_well == "error": raise RipsError(message) if on_unknown_well == "warn": report.warnings.append(message) report.events_skipped += len(well.events) continue _prepare_wellspec_events(well.events, well_path.completion_settings(), report) for event in well.events: event_type = event.event_type.upper() dispatch = _EVENT_DISPATCH.get(event_type) if dispatch is None: typo_of = _suspected_typo(event_type) if typo_of is not None: message = _event_message( event, f"Unknown event type '{event.event_type}'; " f"did you mean '{typo_of}'?", ) if on_unknown_event == "error": raise RipsError(message) if on_unknown_event == "warn": report.warnings.append(message) report.events_skipped += 1 continue dispatch = _apply_generic_well_keyword dispatch(event, well_path, timeline, report, ctx) for group in document.groups: for event in group.events: _apply_schedule_event(event, timeline, report, group.group_name) for event in document.schedule_events: _apply_schedule_event(event, timeline, report) for event in document.insert_date_events: _apply_insert_date_event(event, timeline, report) return report # Search order for unqualified filter result names. _UNQUALIFIED_RESULT_SEARCH_ORDER = ("STATIC_NATIVE", "DYNAMIC_NATIVE", "GENERATED") @dataclass class _FilterContext: """Per-apply-run state for materializing filter expressions on a case.""" case: Any resolved_types: Dict[FilterTerm, str] = field(default_factory=dict) combined_by_key: Dict[Tuple[Optional[str], str], Any] = field(default_factory=dict) properties_by_type: Dict[str, List[str]] = field(default_factory=dict) def available(self, result_type: str) -> List[str]: if result_type not in self.properties_by_type: try: names = list(self.case.available_properties(result_type)) except RipsError: # The gRPC service reports NOT_FOUND when the case has no # results of this type at all; treat that as an empty list. names = [] self.properties_by_type[result_type] = names return self.properties_by_type[result_type] def _prepare_filter_context( document: SimulatorEventsDocument, project: Any, case: Any ) -> Optional[_FilterContext]: """Build a filter context when the document uses filters; else None. Resolves every filter term's result type up front so a missing result raises before any event has been applied. """ filtered_events: List[Tuple[SimulatorEvent, EventFilter]] = [] for well in document.wells: for event in well.events: if event.filter is not None: filtered_events.append((event, event.filter)) if not filtered_events: return None if case is None: cases = project.cases() case = cases[0] if cases else None if case is None: raise RipsError( "The document uses FILTER but no case is available; pass case= " "or load a case in the project" ) ctx = _FilterContext(case=case) missing: List[str] = [] for event, event_filter in filtered_events: label = event_filter.name or event_filter.expr.raw for term in event_filter.expr.terms: if term in ctx.resolved_types: continue resolved = _resolve_result_type(ctx, term) if resolved is not None: ctx.resolved_types[term] = resolved continue if term.result_type is not None: missing.append( _event_message( event, f"filter '{label}': result '{term.result_name}' not found " f"among {term.result_type} results", ) ) else: missing.append( _event_message( event, f"filter '{label}': result '{term.result_name}' not found " f"(searched {', '.join(_UNQUALIFIED_RESULT_SEARCH_ORDER)})", ) ) if missing: raise RipsError("\n".join(missing)) return ctx def _resolve_result_type(ctx: _FilterContext, term: FilterTerm) -> Optional[str]: """Return the result type holding this term's result, or None if missing.""" search_order = ( (term.result_type,) if term.result_type is not None else _UNQUALIFIED_RESULT_SEARCH_ORDER ) for result_type in search_order: if term.result_name in ctx.available(result_type): return result_type return None def _materialize_filter(ctx: _FilterContext, event_filter: EventFilter) -> Any: """Create (or reuse) the case-level combined filter for an event filter. Declared filters are created once per declaration name and shared between the perforations that reference them; inline filters are shared when their expression text is identical. """ # Keyed on name and expression: files merged from different sources may # declare the same filter name with different expressions. key = (event_filter.name, event_filter.expr.raw) existing = ctx.combined_by_key.get(key) if existing is not None: return existing expr = event_filter.expr combined = ctx.case.data_filter_collection().add_combined_filter( name=event_filter.name or "", combine_mode=expr.combine_mode ) for term in expr.terms: property_filter = combined.add_property_filter( result_variable=term.result_name, result_type=ctx.resolved_types[term], ) # Only one bound is set; the other keeps the default result min/max, # so '>' and '>=' are equivalent (bounds are inclusive). if term.op in (">", ">="): property_filter.lower_bound = term.value else: property_filter.upper_bound = term.value property_filter.update() ctx.combined_by_key[key] = combined return combined def _suspected_typo(event_type: str) -> Optional[str]: """Return the built-in event type this one looks like a misspelling of.""" if event_type == "TUBING": return "SEGMENT" close = difflib.get_close_matches(event_type, _EVENT_DISPATCH, n=1, cutoff=0.8) return close[0] if close else None def _apply_member_event( event: SimulatorEvent, 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(_event_message(event, "MEMBER needs a GROUP block")) report.events_skipped += 1 return unknown = set(event.attributes) - {"MEMBERS", "COMMENT"} if unknown: report.errors.append( _event_message( event, f"unknown MEMBER attribute(s): {', '.join(sorted(unknown))}", ) ) report.events_skipped += 1 return if "MEMBERS" not in event.attributes: report.errors.append( _event_message(event, "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( _event_message( event, "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: SimulatorEvent, timeline: Any, report: ApplyReport, group_name: Optional[str] = None, ) -> None: """Apply one GROUP- or SCHEDULE-block event as an Eclipse keyword.""" event_type = event.event_type.upper() if event_type == "RAW_TEXT": timeline.add_raw_text_event( event_date=_iso_event_date(event.event_date), text=event.raw_text, placement=event.raw_placement, anchor_keyword=event.raw_anchor or "", priority=event.raw_priority, ) report.events_applied += 1 return if event_type == "RESTART": timeline.add_keyword_event( event_date=_iso_event_date(event.event_date), keyword_name="RESTART", keyword_data={}, ) report.events_applied += 1 return if event_type == "MEMBER": _apply_member_event(event, timeline, report, group_name) return if event_type in _COMPLETION_EVENT_TYPES: report.errors.append( _event_message( event, f"{event_type} is a completion event and needs a WELL block, " "not GROUP or SCHEDULE", ) ) report.events_skipped += 1 return keyword_data: Dict[str, Any] = {} for key, attr in event.attributes.items(): if key == "COMMENT": continue if key in _IGNORED_KEYWORD_ATTRS: report.warnings.append( _event_message( event, f"attribute '{key}' on {event_type} is ignored (not yet supported)", ) ) continue keyword_data[key] = attr.value if group_name is not None: keyword_data["GROUP"] = group_name try: timeline_event = timeline.add_keyword_event( event_date=_iso_event_date(event.event_date), keyword_name=event_type, keyword_data=keyword_data, ) except RipsError as exc: _record_event_exception(event, exc, report) return _apply_event_comment(event, timeline_event) report.events_applied += 1 def _apply_insert_date_event( event: SimulatorEvent, timeline: Any, report: ApplyReport ) -> None: """Apply one expanded ``INSERT_DATE`` occurrence, comment included.""" comment = event.attributes.get("COMMENT") timeline.add_insert_date_event( event_date=_iso_event_date(event.event_date), comment="" if comment is None else str(comment.value), ) report.events_applied += 1 def _validate_policy(value: str, name: str) -> None: if value not in _POLICIES: raise ValueError(f"{name} must be one of {_POLICIES}, got {value!r}") def _check_completion_attrs( event: SimulatorEvent, type_name: str, known: Set[str], required: Tuple[str, ...], report: ApplyReport, ignored: Set[str] = _COMPLETION_IGNORED, ) -> bool: """Validate a completion event's attributes; False means skip the event.""" attrs = event.attributes unknown = set(attrs) - known - ignored if unknown: report.errors.append( _event_message( event, f"unknown {type_name} attribute(s): {', '.join(sorted(unknown))}", ) ) report.events_skipped += 1 return False for key in sorted(ignored & set(attrs)): report.warnings.append( _event_message( event, f"attribute '{key}' on {type_name} is ignored (not yet supported)", ) ) missing = [name for name in required if name not in attrs] if missing: report.errors.append( _event_message( event, f"{type_name} missing required attribute(s): {', '.join(missing)}", ) ) report.events_skipped += 1 return False return True def _apply_perforation( event: SimulatorEvent, well_path: Any, timeline: Any, report: ApplyReport, ctx: Optional[_FilterContext] = None, ) -> None: if not _check_completion_attrs( event, "PERFORATION", _PERF_KNOWN, _PERF_REQUIRED, report, ignored=set() ): return attrs = event.attributes try: kwargs: Dict[str, Any] = { "event_date": _iso_event_date(event.event_date), "well_path": well_path, "start_md": float(_as_number(attrs["MDSTART"], event.loc)), "end_md": float(_as_number(attrs["MDEND"], event.loc)), "state": "OPEN", } if "DIAMETER" in attrs: kwargs["diameter"] = float(_as_number(attrs["DIAMETER"], event.loc)) if "SKIN" in attrs: kwargs["skin_factor"] = float(_as_number(attrs["SKIN"], event.loc)) if "COMPLETION_NUMBER" in attrs: kwargs["completion_number"] = int( _as_number(attrs["COMPLETION_NUMBER"], event.loc) ) except SimulatorEventsParseError as exc: _record_event_parse_error(event, exc, report) return perf_event = timeline.add_perf_event(**kwargs) if event.filter is not None and ctx is not None: perf_event.add_filter(filter=_materialize_filter(ctx, event.filter)) _apply_event_comment(event, perf_event) report.events_applied += 1 def _apply_segment( event: SimulatorEvent, well_path: Any, timeline: Any, report: ApplyReport, ctx: Optional[_FilterContext] = None, ) -> None: if not _check_completion_attrs( event, "SEGMENT", _SEGMENT_KNOWN, _SEGMENT_REQUIRED, report ): return attrs = event.attributes try: start_md = float(_as_number(attrs["MDSTART"], event.loc)) end_md = float(_as_number(attrs["MDEND"], event.loc)) kwargs: Dict[str, Any] = { "event_date": _iso_event_date(event.event_date), "well_path": well_path, "start_md": start_md, "end_md": end_md, } if "INNER_DIAMETER" in attrs: kwargs["inner_diameter"] = float( _as_number(attrs["INNER_DIAMETER"], event.loc) ) if "ROUGHNESS" in attrs: kwargs["roughness"] = float(_as_number(attrs["ROUGHNESS"], event.loc)) lengths: Dict[str, float] = {} for name, kwarg in _SEGMENT_LENGTH_KWARGS.items(): if name in attrs: lengths[kwarg] = float(_as_number(attrs[name], event.loc)) except SimulatorEventsParseError as exc: _record_event_parse_error(event, exc, report) return timeline_event = timeline.add_segment_event(**kwargs, **lengths) interval = well_path.segment_collection().add_segment_interval( start_md=start_md, end_md=end_md, diameter=kwargs.get("inner_diameter", 0.152), roughness_factor=kwargs.get("roughness", 1.0e-5), ) if "min_segment_length" in lengths: interval.set_min_segment_length(length=lengths["min_segment_length"]) if "max_segment_length" in lengths: interval.set_max_segment_length(length=lengths["max_segment_length"]) if "fixed_segment_length" in lengths: interval.set_fixed_segment_length(length=lengths["fixed_segment_length"]) _apply_event_comment(event, timeline_event) report.events_applied += 1 # VALVE attribute -> add_valve_event keyword-argument name (numeric values). _VALVE_NUMERIC_KWARGS = { "CV": "flow_coefficient", "AREA": "area", "AICD_STRENGTH": "aicd_strength", "AICD_DENSITY_CALIB_FLUID": "aicd_density_calib_fluid", "AICD_VISCOSITY_CALIB_FLUID": "aicd_viscosity_calib_fluid", "AICD_VOL_FLOW_EXP": "aicd_vol_flow_exp", "AICD_VISC_FUNC_EXP": "aicd_visc_func_exp", } def _apply_valve( event: SimulatorEvent, well_path: Any, timeline: Any, report: ApplyReport, ctx: Optional[_FilterContext] = None, ) -> None: if not _check_completion_attrs( event, "VALVE", _VALVE_KNOWN, _VALVE_REQUIRED, report ): return attrs = event.attributes try: kwargs: Dict[str, Any] = { "event_date": _iso_event_date(event.event_date), "well_path": well_path, "measured_depth": float(_as_number(attrs["MD"], event.loc)), "valve_type": str(attrs["TYPE"].value), } if "STATE" in attrs: kwargs["state"] = str(attrs["STATE"].value) for attr_name, kwarg_name in _VALVE_NUMERIC_KWARGS.items(): if attr_name in attrs: kwargs[kwarg_name] = float(_as_number(attrs[attr_name], event.loc)) except SimulatorEventsParseError as exc: _record_event_parse_error(event, exc, report) return timeline_event = timeline.add_valve_event(**kwargs) _apply_event_comment(event, timeline_event) report.events_applied += 1 def _apply_state( event: SimulatorEvent, well_path: Any, timeline: Any, report: ApplyReport, ctx: Optional[_FilterContext] = None, ) -> None: if not _check_completion_attrs( event, "STATE", _STATE_KNOWN, _STATE_REQUIRED, report ): return timeline_event = timeline.add_state_event( event_date=_iso_event_date(event.event_date), well_path=well_path, well_state=str(event.attributes["STATE"].value), ) _apply_event_comment(event, timeline_event) report.events_applied += 1 def _apply_wellspec( event: SimulatorEvent, well_path: Any, timeline: Any, report: ApplyReport, ctx: Optional[_FilterContext] = None, ) -> None: if event.well_spec is None: return state = event.well_spec timeline_event = timeline.add_wellspec_event( event_date=_iso_event_date(event.event_date), well_path=well_path, group_name=state.group, allow_cross_flow=state.crossflow, reference_depth=state.refdepth, well_type=state.phase, pressure_drop=state.pressure_components, ) _apply_event_comment(event, timeline_event) report.events_applied += 1 def _apply_keyword( event: SimulatorEvent, well_path: Any, timeline: Any, report: ApplyReport, keyword_name: str, field_map: Dict[str, str], ) -> None: keyword_data: Dict[str, Any] = {"WELL": well_path.name} for key, attr in event.attributes.items(): if key == "COMMENT": continue if key in _IGNORED_KEYWORD_ATTRS: report.warnings.append( _event_message( event, f"attribute '{key}' on {keyword_name} is ignored " "(not yet supported)", ) ) continue keyword_data[field_map.get(key, key)] = attr.value try: timeline_event = timeline.add_well_keyword_event( event_date=_iso_event_date(event.event_date), well_path=well_path, keyword_name=keyword_name, keyword_data=keyword_data, ) except RipsError as exc: _record_event_exception(event, exc, report) return _apply_event_comment(event, timeline_event) report.events_applied += 1 def _apply_wconhist( event: SimulatorEvent, well_path: Any, timeline: Any, report: ApplyReport, ctx: Optional[_FilterContext] = None, ) -> None: _apply_keyword(event, well_path, timeline, report, "WCONHIST", _WCONHIST_FIELD_MAP) def _apply_weltarg( event: SimulatorEvent, well_path: Any, timeline: Any, report: ApplyReport, ctx: Optional[_FilterContext] = None, ) -> None: _apply_keyword(event, well_path, timeline, report, "WELTARG", _WELTARG_FIELD_MAP) def _as_number(attr: AttrValue, loc: SourceLoc) -> Union[int, float]: if isinstance(attr.value, bool) or not isinstance(attr.value, (int, float)): raise SimulatorEventsParseError( f"Expected a numeric value, got {attr.raw!r}", loc ) return attr.value def _apply_generic_well_keyword( event: SimulatorEvent, well_path: Any, timeline: Any, report: ApplyReport, ctx: Optional[_FilterContext] = None, ) -> None: _apply_keyword(event, well_path, timeline, report, event.event_type.upper(), {}) _EventDispatch = Callable[ [SimulatorEvent, Any, Any, ApplyReport, Optional[_FilterContext]], None ] # Built-in event types. Any other event type inside a WELL block is passed # through as a generic Eclipse well keyword (unless it looks like a typo of a # built-in, which is governed by the on_unknown_event policy). _EVENT_DISPATCH: Dict[str, _EventDispatch] = { "PERFORATION": _apply_perforation, "SEGMENT": _apply_segment, "VALVE": _apply_valve, "STATE": _apply_state, "WELSPECS": _apply_wellspec, "WCONHIST": _apply_wconhist, "WELTARG": _apply_weltarg, } # Completion event types that require a well and cannot appear in a SCHEDULE # block or be emitted as Eclipse keywords. _COMPLETION_EVENT_TYPES = ( "PERFORATION", "SEGMENT", "VALVE", "STATE", "WELSPECS", ) # --------------------------------------------------------------------------- # Standalone validator CLI: python3 -m rips.simulator_events # --------------------------------------------------------------------------- def _apply_to_running_instance(document: SimulatorEventsDocument) -> int: """Apply a parsed document to the well event timeline of a running ResInsight instance found through the RESINSIGHT_GRPC_PORT environment variable. Returns a process exit code.""" import rips resinsight = rips.Instance.find() if resinsight is None: print("Error: no running ResInsight instance found") return 1 project = resinsight.project well_path_collections = project.descendants(rips.WellPathCollection) if not well_path_collections: print("Error: the project has no well path collection") return 1 timeline = well_path_collections[0].event_timeline() # A case is only needed to materialize FILTER declarations; default to the # first case when one is loaded. cases = project.cases() case = cases[0] if cases else None try: report = apply_simulator_events_document( document, timeline, project, case=case, on_unknown_well="warn" ) except RipsError as exc: print(f"Error: {exc}") return 1 print(f" Events applied: {report.events_applied}") print(f" Events skipped: {report.events_skipped}") for warning in report.warnings: print(f" Warning: {warning}") for error in report.errors: print(f" Error: {error}") return 1 if report.errors else 0 def _cli(argv: Optional[List[str]] = None) -> int: import argparse arg_parser = argparse.ArgumentParser( prog="python3 -m rips.simulator_events", description="Validate one or more SIMEVENTS files (parse only; no " "ResInsight needed), and optionally apply them to a running ResInsight " "instance. Several files are merged into one document, in the given " "order, before matching events are coalesced.", ) arg_parser.add_argument( "files", nargs="+", metavar="file", help="path to a SIMEVENTS file" ) arg_parser.add_argument( "--apply", action="store_true", help="apply the events to a running ResInsight instance after validating", ) args = arg_parser.parse_args(argv) files: List[str] = args.files label = ", ".join(files) try: if len(files) == 1: document = parse_simulator_events_file(files[0]) else: document = parse_simulator_events_files(list(files)) except OSError as exc: print(f"Error: {exc}") return 1 except SimulatorEventsParseError as exc: for issue in exc.errors: if issue.loc is not None: print(f"{_loc_label(issue.loc)}: {issue.message}") else: print(issue.message) print(f"{label}: {len(exc.errors)} error(s) found") return 1 event_count = sum(len(well.events) for well in document.wells) group_event_count = sum(len(group.events) for group in document.groups) insert_date_count = len(document.insert_date_events) schedule_event_count = len(document.schedule_events) print(f"{label}: OK (SIMEVENTS {document.version}, units {document.unit_system})") print( f" {len(document.variables)} variable(s), {len(document.wells)} " f"well block(s), {event_count} well event(s), " f"{len(document.groups)} group block(s), {group_event_count} group event(s), " f"{schedule_event_count} schedule event(s), " f"{insert_date_count} inserted date(s)" ) normalized = coalesce_simulator_events_document(document) for warning in normalized.warnings: print(f" Warning {_loc_reference(warning.loc)}: {warning.message}") if args.apply: return _apply_to_running_instance(document) return 0 if __name__ == "__main__": import sys sys.exit(_cli())