CLARISSA Portal: OPM Flow Playground¶

Interactive playground for reservoir simulation with OPM Flow

This notebook provides hands-on access to ECLIPSE deck manipulation and OPM Flow simulation:

• 📝 Deck Files - Create, read, modify ECLIPSE input decks
• 🔬 Simulation - Run OPM Flow, monitor progress
• 📊 Results - Parse and visualize simulation output
• 🧪 SPE Benchmarks - Load and run standard test cases

---

🛠️ Environment Setup¶

OPM Flow requires either:

• Docker (recommended for Colab)
• Native installation (Linux)

In Colab, we'll use a lightweight simulation mode or Docker if available.

In [ ]:

# Configuration
import os
import json
import subprocess
import tempfile
from pathlib import Path
from datetime import datetime
from typing import Dict, List, Optional, Tuple

# Check environment
IN_COLAB = 'google.colab' in str(get_ipython()) if 'get_ipython' in dir() else False
WORK_DIR = Path(tempfile.mkdtemp()) if IN_COLAB else Path('./opm_workspace')
WORK_DIR.mkdir(exist_ok=True)

print(f"📁 Working directory: {WORK_DIR}")
print(f"🌐 Environment: {'Google Colab' if IN_COLAB else 'Local'}")
print(f"📅 Session: {datetime.now().isoformat()}")

# Configuration import os import json import subprocess import tempfile from pathlib import Path from datetime import datetime from typing import Dict, List, Optional, Tuple # Check environment IN_COLAB = 'google.colab' in str(get_ipython()) if 'get_ipython' in dir() else False WORK_DIR = Path(tempfile.mkdtemp()) if IN_COLAB else Path('./opm_workspace') WORK_DIR.mkdir(exist_ok=True) print(f"📁 Working directory: {WORK_DIR}") print(f"🌐 Environment: {'Google Colab' if IN_COLAB else 'Local'}") print(f"📅 Session: {datetime.now().isoformat()}")

📦 Install OPM Flow (Colab)¶

OPM Flow can be installed directly in Google Colab. Choose one method:

• Method 1: Native installation via apt (recommended, ~2 min)
• Method 2: Docker container (if Docker available)
• Method 3: Python bindings only (limited functionality)

In [ ]:

# Method 1: Native Installation (Ubuntu/Colab)
# This installs OPM Flow directly - works in Google Colab!

def install_opm_native():
    """Install OPM Flow natively via apt."""
    import subprocess
    import sys
    
    print("📦 Installing OPM Flow (this takes ~2-3 minutes)...")
    print()
    
    commands = [
        # Add OPM repository
        "sudo apt-get update -qq",
        "sudo apt-get install -y -qq software-properties-common",
        "sudo apt-add-repository -y ppa:opm/ppa",
        "sudo apt-get update -qq",
        # Install OPM Flow (CORRECT PACKAGE NAME!)
        "sudo apt-get install -y libopm-simulators-bin",
    ]
    
    for i, cmd in enumerate(commands, 1):
        print(f"Step {i}/{len(commands)}: {cmd.split()[-1]}...")
        result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
        if result.returncode != 0:
            print(f"❌ Failed: {result.stderr[:200]}")
            return False
    
    # Verify installation
    result = subprocess.run(["flow", "--version"], capture_output=True, text=True)
    if result.returncode == 0:
        print(f"
✅ OPM Flow installed successfully!")
        print(f"   Version: {result.stdout.strip()}")
        return True
    else:
        print("❌ Installation verification failed")
        return False

# Uncomment to install:
# install_opm_native()

# Method 1: Native Installation (Ubuntu/Colab) # This installs OPM Flow directly - works in Google Colab! def install_opm_native(): """Install OPM Flow natively via apt.""" import subprocess import sys print("📦 Installing OPM Flow (this takes ~2-3 minutes)...") print() commands = [ # Add OPM repository "sudo apt-get update -qq", "sudo apt-get install -y -qq software-properties-common", "sudo apt-add-repository -y ppa:opm/ppa", "sudo apt-get update -qq", # Install OPM Flow (CORRECT PACKAGE NAME!) "sudo apt-get install -y libopm-simulators-bin", ] for i, cmd in enumerate(commands, 1): print(f"Step {i}/{len(commands)}: {cmd.split()[-1]}...") result = subprocess.run(cmd, shell=True, capture_output=True, text=True) if result.returncode != 0: print(f"❌ Failed: {result.stderr[:200]}") return False # Verify installation result = subprocess.run(["flow", "--version"], capture_output=True, text=True) if result.returncode == 0: print(f" ✅ OPM Flow installed successfully!") print(f" Version: {result.stdout.strip()}") return True else: print("❌ Installation verification failed") return False # Uncomment to install: # install_opm_native()

In [ ]:

# Method 2: Docker Installation (if Docker available)

def install_opm_docker():
    """Pull OPM Flow Docker image."""
    import subprocess
    
    print("🐳 Pulling OPM Flow Docker image...")
    result = subprocess.run(
        ["docker", "pull", "opmproject/flow:latest"],
        capture_output=True, text=True
    )
    
    if result.returncode == 0:
        print("✅ Docker image ready!")
        print("   Image: opmproject/flow:latest")
        return True
    else:
        print(f"❌ Docker pull failed: {result.stderr}")
        return False

# Uncomment to install:
# install_opm_docker()

# Method 2: Docker Installation (if Docker available) def install_opm_docker(): """Pull OPM Flow Docker image.""" import subprocess print("🐳 Pulling OPM Flow Docker image...") result = subprocess.run( ["docker", "pull", "opmproject/flow:latest"], capture_output=True, text=True ) if result.returncode == 0: print("✅ Docker image ready!") print(" Image: opmproject/flow:latest") return True else: print(f"❌ Docker pull failed: {result.stderr}") return False # Uncomment to install: # install_opm_docker()

In [ ]:

# Method 3: Python Bindings (opm package)
# Note: This provides Python API but NOT the flow executable

def install_opm_python():
    """Install OPM Python bindings."""
    import subprocess
    import sys
    
    print("🐍 Installing OPM Python bindings...")
    
    # Install dependencies first
    subprocess.run([sys.executable, "-m", "pip", "install", "-q", "numpy"])
    
    # Try to install opm
    result = subprocess.run(
        [sys.executable, "-m", "pip", "install", "-q", "opm"],
        capture_output=True, text=True
    )
    
    if result.returncode == 0:
        print("✅ OPM Python bindings installed!")
        print("   Usage: from opm.io.ecl import EclFile")
        print("   Note: This does NOT include the flow simulator.")
        return True
    else:
        print(f"❌ Installation failed: {result.stderr}")
        print("   Try Method 1 (native) instead.")
        return False

# Uncomment to install:
# install_opm_python()

# Method 3: Python Bindings (opm package) # Note: This provides Python API but NOT the flow executable def install_opm_python(): """Install OPM Python bindings.""" import subprocess import sys print("🐍 Installing OPM Python bindings...") # Install dependencies first subprocess.run([sys.executable, "-m", "pip", "install", "-q", "numpy"]) # Try to install opm result = subprocess.run( [sys.executable, "-m", "pip", "install", "-q", "opm"], capture_output=True, text=True ) if result.returncode == 0: print("✅ OPM Python bindings installed!") print(" Usage: from opm.io.ecl import EclFile") print(" Note: This does NOT include the flow simulator.") return True else: print(f"❌ Installation failed: {result.stderr}") print(" Try Method 1 (native) instead.") return False # Uncomment to install: # install_opm_python()

In [ ]:

# 🚀 Quick Install - One Click!
# This automatically chooses the best method for your environment.

def auto_install_opm():
    """Automatically install OPM Flow using the best available method."""
    import subprocess
    import os
    
    # Check if already installed
    try:
        result = subprocess.run(["flow", "--version"], capture_output=True, text=True, timeout=5)
        if result.returncode == 0:
            print(f"✅ OPM Flow already installed: {result.stdout.strip()}")
            return True
    except:
        pass
    
    # Detect environment
    in_colab = "google.colab" in str(globals().get("get_ipython", lambda: "")())
    has_docker = subprocess.run(["which", "docker"], capture_output=True).returncode == 0
    is_linux = os.uname().sysname == "Linux"
    
    print(f"🔍 Environment: Colab={in_colab}, Docker={has_docker}, Linux={is_linux}")
    print()
    
    if in_colab or is_linux:
        print("📦 Using native installation (recommended for Colab/Linux)...")
        return install_opm_native()
    elif has_docker:
        print("🐳 Using Docker installation...")
        return install_opm_docker()
    else:
        print("⚠️ No suitable installation method found.")
        print("   Please install OPM Flow manually or use Docker.")
        return False

# 🚀 UNCOMMENT THIS LINE TO INSTALL OPM FLOW:
# auto_install_opm()

# 🚀 Quick Install - One Click! # This automatically chooses the best method for your environment. def auto_install_opm(): """Automatically install OPM Flow using the best available method.""" import subprocess import os # Check if already installed try: result = subprocess.run(["flow", "--version"], capture_output=True, text=True, timeout=5) if result.returncode == 0: print(f"✅ OPM Flow already installed: {result.stdout.strip()}") return True except: pass # Detect environment in_colab = "google.colab" in str(globals().get("get_ipython", lambda: "")()) has_docker = subprocess.run(["which", "docker"], capture_output=True).returncode == 0 is_linux = os.uname().sysname == "Linux" print(f"🔍 Environment: Colab={in_colab}, Docker={has_docker}, Linux={is_linux}") print() if in_colab or is_linux: print("📦 Using native installation (recommended for Colab/Linux)...") return install_opm_native() elif has_docker: print("🐳 Using Docker installation...") return install_opm_docker() else: print("⚠️ No suitable installation method found.") print(" Please install OPM Flow manually or use Docker.") return False # 🚀 UNCOMMENT THIS LINE TO INSTALL OPM FLOW: # auto_install_opm()

⏱️ Installation Times¶

Method	Time	Notes
Native (apt)	~2 min	✅ Recommended for Colab
Docker pull	~1 min	Requires Docker runtime
Python bindings	~30 sec	No simulator, just file I/O

---

In [ ]:

# Check for OPM Flow
def check_opm_flow() -> Tuple[bool, str]:
    """Check if OPM Flow is available."""
    # Try native
    try:
        result = subprocess.run(['flow', '--version'], capture_output=True, text=True, timeout=5)
        if result.returncode == 0:
            return True, f"Native: {result.stdout.strip()}"
    except:
        pass
    
    # Try Docker
    try:
        result = subprocess.run(
            ['docker', 'run', '--rm', 'opmproject/flow:latest', 'flow', '--version'],
            capture_output=True, text=True, timeout=30
        )
        if result.returncode == 0:
            return True, f"Docker: {result.stdout.strip()}"
    except:
        pass
    
    return False, "Not available (will use mock mode)"

OPM_AVAILABLE, OPM_VERSION = check_opm_flow()
print(f"🔬 OPM Flow: {OPM_VERSION}")

if not OPM_AVAILABLE:
    print("
⚠️  Running in MOCK MODE - deck operations work, simulation returns dummy results.")
    print("    To use real simulation, install OPM Flow or use Docker.")

# Check for OPM Flow def check_opm_flow() -> Tuple[bool, str]: """Check if OPM Flow is available.""" # Try native try: result = subprocess.run(['flow', '--version'], capture_output=True, text=True, timeout=5) if result.returncode == 0: return True, f"Native: {result.stdout.strip()}" except: pass # Try Docker try: result = subprocess.run( ['docker', 'run', '--rm', 'opmproject/flow:latest', 'flow', '--version'], capture_output=True, text=True, timeout=30 ) if result.returncode == 0: return True, f"Docker: {result.stdout.strip()}" except: pass return False, "Not available (will use mock mode)" OPM_AVAILABLE, OPM_VERSION = check_opm_flow() print(f"🔬 OPM Flow: {OPM_VERSION}") if not OPM_AVAILABLE: print(" ⚠️ Running in MOCK MODE - deck operations work, simulation returns dummy results.") print(" To use real simulation, install OPM Flow or use Docker.")

In [ ]:

# OPM Flow Manager
class OPMFlowManager:
    """Manager for ECLIPSE decks and OPM Flow simulations."""
    
    def __init__(self, work_dir: Path, use_docker: bool = True):
        self.work_dir = Path(work_dir)
        self.work_dir.mkdir(exist_ok=True)
        self.use_docker = use_docker
        self.current_deck = None
        self.current_deck_path = None
    
    # ============ CREATE ============
    def create_deck(self, name: str, content: str) -> Dict:
        """Create a new ECLIPSE deck file."""
        deck_path = self.work_dir / f"{name}.DATA"
        deck_path.write_text(content)
        self.current_deck = content
        self.current_deck_path = deck_path
        return {
            'status': 'created',
            'path': str(deck_path),
            'size': len(content),
            'lines': content.count('
') + 1
        }
    
    def create_minimal_deck(self, title: str = "Minimal Model") -> Dict:
        """Create a minimal valid ECLIPSE deck."""
        deck = f"""-- Minimal ECLIPSE Deck
-- Created by CLARISSA OPM Playground
-- {datetime.now().isoformat()}

RUNSPEC

TITLE
{title} /

DIMENS
10 10 3 /

OIL
WATER

METRIC

TABDIMS
1 1 20 20 /

WELLDIMS
2 10 1 2 /

START
1 'JAN' 2026 /

GRID

DX
300*100 /

DY
300*100 /

DZ
300*10 /

TOPS
100*2000 /

PORO
300*0.2 /

PERMX
300*100 /

PERMY
300*100 /

PERMZ
300*10 /

PROPS

SWOF
0.2  0.0   1.0  0.0
0.3  0.05  0.6  0.0
0.5  0.2   0.3  0.0
0.7  0.5   0.1  0.0
0.8  1.0   0.0  0.0
/

PVTW
300 1.0 4.0E-5 0.5 0.0 /

PVDO
100 1.05 2.0
300 1.00 2.5
/

ROCK
300 4.0E-5 /

DENSITY
800 1025 1.0 /

SOLUTION

EQUIL
2025 300 2100 0 2000 0 /

SUMMARY

FOPR
FWPR
FWCT
FOPT
FWPT
FPR

SCHEDULE

WELSPECS
'PROD' 'G1' 5 5 2025 'OIL' /
'INJ'  'G1' 1 1 2025 'WATER' /
/

COMPDAT
'PROD' 5 5 1 3 'OPEN' 2* 0.2 /
'INJ'  1 1 1 3 'OPEN' 2* 0.2 /
/

WCONPROD
'PROD' 'OPEN' 'ORAT' 500 4* 100 /
/

WCONINJE
'INJ' 'WATER' 'OPEN' 'RATE' 600 1* 400 /
/

TSTEP
30 30 30 30 30 30 30 30 30 30 30 30 /

END
"""
        return self.create_deck('minimal', deck)
    
    # ============ READ ============
    def read_deck(self, path: str = None) -> Dict:
        """Read an ECLIPSE deck file."""
        deck_path = Path(path) if path else self.current_deck_path
        if not deck_path or not deck_path.exists():
            return {'error': 'No deck file found'}
        
        content = deck_path.read_text()
        self.current_deck = content
        self.current_deck_path = deck_path
        
        return {
            'path': str(deck_path),
            'content': content,
            'lines': content.count('
') + 1,
            'sections': self._parse_sections(content)
        }
    
    def _parse_sections(self, content: str) -> List[str]:
        """Extract section names from deck."""
        sections = []
        for line in content.split('
'):
            line = line.strip()
            if line in ['RUNSPEC', 'GRID', 'EDIT', 'PROPS', 'REGIONS', 'SOLUTION', 'SUMMARY', 'SCHEDULE']:
                sections.append(line)
        return sections
    
    def get_keywords(self) -> Dict:
        """Extract keywords from current deck."""
        if not self.current_deck:
            return {'error': 'No deck loaded'}
        
        keywords = []
        for line in self.current_deck.split('
'):
            line = line.strip()
            if line and not line.startswith('--') and line.isupper() and ' ' not in line:
                if not any(c.isdigit() for c in line) and '/' not in line:
                    keywords.append(line)
        
        return {'keywords': list(set(keywords)), 'count': len(set(keywords))}
    
    def list_decks(self) -> List[str]:
        """List all deck files in work directory."""
        return [str(f) for f in self.work_dir.glob('*.DATA')]
    
    # ============ UPDATE ============
    def update_keyword(self, keyword: str, new_value: str) -> Dict:
        """Update a keyword value in the deck."""
        if not self.current_deck:
            return {'error': 'No deck loaded'}
        
        lines = self.current_deck.split('
')
        updated = False
        new_lines = []
        skip_until_slash = False
        
        for i, line in enumerate(lines):
            if skip_until_slash:
                if '/' in line:
                    skip_until_slash = False
                continue
            
            if line.strip() == keyword:
                new_lines.append(line)
                new_lines.append(new_value)
                skip_until_slash = True
                updated = True
            else:
                new_lines.append(line)
        
        if updated:
            self.current_deck = '
'.join(new_lines)
            if self.current_deck_path:
                self.current_deck_path.write_text(self.current_deck)
            return {'status': 'updated', 'keyword': keyword}
        else:
            return {'status': 'not_found', 'keyword': keyword}
    
    def append_schedule(self, schedule_content: str) -> Dict:
        """Append content to SCHEDULE section."""
        if not self.current_deck:
            return {'error': 'No deck loaded'}
        
        # Find END and insert before it
        if 'END' in self.current_deck:
            self.current_deck = self.current_deck.replace('END', f'{schedule_content}

END')
        else:
            self.current_deck += f'
{schedule_content}
'
        
        if self.current_deck_path:
            self.current_deck_path.write_text(self.current_deck)
        
        return {'status': 'appended', 'lines_added': schedule_content.count('
') + 1}
    
    # ============ DELETE ============
    def delete_deck(self, name: str) -> Dict:
        """Delete a deck file."""
        deck_path = self.work_dir / f"{name}.DATA"
        if deck_path.exists():
            deck_path.unlink()
            if self.current_deck_path == deck_path:
                self.current_deck = None
                self.current_deck_path = None
            return {'status': 'deleted', 'path': str(deck_path)}
        return {'status': 'not_found', 'path': str(deck_path)}
    
    def clear_workspace(self) -> Dict:
        """Remove all files from workspace."""
        count = 0
        for f in self.work_dir.iterdir():
            f.unlink()
            count += 1
        self.current_deck = None
        self.current_deck_path = None
        return {'status': 'cleared', 'files_removed': count}
    
    # ============ SIMULATE ============
    def run_simulation(self, deck_path: str = None) -> Dict:
        """Run OPM Flow simulation."""
        path = Path(deck_path) if deck_path else self.current_deck_path
        if not path or not path.exists():
            return {'error': 'No deck file specified'}
        
        if not OPM_AVAILABLE:
            # Mock simulation
            return self._mock_simulation(path)
        
        # Real simulation
        try:
            if self.use_docker:
                cmd = [
                    'docker', 'run', '--rm',
                    '-v', f'{path.parent}:/data',
                    'opmproject/flow:latest',
                    'flow', f'/data/{path.name}'
                ]
            else:
                cmd = ['flow', str(path)]
            
            result = subprocess.run(cmd, capture_output=True, text=True, timeout=300)
            
            return {
                'status': 'success' if result.returncode == 0 else 'failed',
                'returncode': result.returncode,
                'stdout': result.stdout[-2000:] if result.stdout else '',
                'stderr': result.stderr[-1000:] if result.stderr else ''
            }
        except subprocess.TimeoutExpired:
            return {'status': 'timeout', 'error': 'Simulation exceeded 5 minutes'}
        except Exception as e:
            return {'status': 'error', 'error': str(e)}
    
    def _mock_simulation(self, path: Path) -> Dict:
        """Return mock simulation results."""
        return {
            'status': 'mock',
            'message': 'Mock simulation (OPM Flow not available)',
            'deck': str(path),
            'mock_results': {
                'FOPR': [500, 480, 460, 440, 420, 400, 380, 360, 340, 320, 300, 280],
                'FWPR': [0, 20, 50, 80, 120, 160, 200, 250, 300, 350, 400, 450],
                'FWCT': [0.0, 0.04, 0.10, 0.15, 0.22, 0.29, 0.34, 0.41, 0.47, 0.52, 0.57, 0.62],
                'FPR': [300, 295, 288, 280, 271, 261, 250, 238, 225, 211, 196, 180]
            },
            'timesteps': list(range(0, 360, 30))
        }
    
    def get_results(self) -> Dict:
        """Parse simulation results."""
        if not self.current_deck_path:
            return {'error': 'No simulation run'}
        
        # Look for summary file
        base = self.current_deck_path.stem
        smspec = self.work_dir / f"{base}.SMSPEC"
        unsmry = self.work_dir / f"{base}.UNSMRY"
        
        if smspec.exists() and unsmry.exists():
            return {'status': 'results_available', 'files': [str(smspec), str(unsmry)]}
        else:
            return {'status': 'no_results', 'hint': 'Run simulation first or use mock mode'}

# Initialize manager
opm = OPMFlowManager(WORK_DIR)
print("✅ OPMFlowManager initialized")

# OPM Flow Manager class OPMFlowManager: """Manager for ECLIPSE decks and OPM Flow simulations.""" def __init__(self, work_dir: Path, use_docker: bool = True): self.work_dir = Path(work_dir) self.work_dir.mkdir(exist_ok=True) self.use_docker = use_docker self.current_deck = None self.current_deck_path = None # ============ CREATE ============ def create_deck(self, name: str, content: str) -> Dict: """Create a new ECLIPSE deck file.""" deck_path = self.work_dir / f"{name}.DATA" deck_path.write_text(content) self.current_deck = content self.current_deck_path = deck_path return { 'status': 'created', 'path': str(deck_path), 'size': len(content), 'lines': content.count(' ') + 1 } def create_minimal_deck(self, title: str = "Minimal Model") -> Dict: """Create a minimal valid ECLIPSE deck.""" deck = f"""-- Minimal ECLIPSE Deck -- Created by CLARISSA OPM Playground -- {datetime.now().isoformat()} RUNSPEC TITLE {title} / DIMENS 10 10 3 / OIL WATER METRIC TABDIMS 1 1 20 20 / WELLDIMS 2 10 1 2 / START 1 'JAN' 2026 / GRID DX 300*100 / DY 300*100 / DZ 300*10 / TOPS 100*2000 / PORO 300*0.2 / PERMX 300*100 / PERMY 300*100 / PERMZ 300*10 / PROPS SWOF 0.2 0.0 1.0 0.0 0.3 0.05 0.6 0.0 0.5 0.2 0.3 0.0 0.7 0.5 0.1 0.0 0.8 1.0 0.0 0.0 / PVTW 300 1.0 4.0E-5 0.5 0.0 / PVDO 100 1.05 2.0 300 1.00 2.5 / ROCK 300 4.0E-5 / DENSITY 800 1025 1.0 / SOLUTION EQUIL 2025 300 2100 0 2000 0 / SUMMARY FOPR FWPR FWCT FOPT FWPT FPR SCHEDULE WELSPECS 'PROD' 'G1' 5 5 2025 'OIL' / 'INJ' 'G1' 1 1 2025 'WATER' / / COMPDAT 'PROD' 5 5 1 3 'OPEN' 2* 0.2 / 'INJ' 1 1 1 3 'OPEN' 2* 0.2 / / WCONPROD 'PROD' 'OPEN' 'ORAT' 500 4* 100 / / WCONINJE 'INJ' 'WATER' 'OPEN' 'RATE' 600 1* 400 / / TSTEP 30 30 30 30 30 30 30 30 30 30 30 30 / END """ return self.create_deck('minimal', deck) # ============ READ ============ def read_deck(self, path: str = None) -> Dict: """Read an ECLIPSE deck file.""" deck_path = Path(path) if path else self.current_deck_path if not deck_path or not deck_path.exists(): return {'error': 'No deck file found'} content = deck_path.read_text() self.current_deck = content self.current_deck_path = deck_path return { 'path': str(deck_path), 'content': content, 'lines': content.count(' ') + 1, 'sections': self._parse_sections(content) } def _parse_sections(self, content: str) -> List[str]: """Extract section names from deck.""" sections = [] for line in content.split(' '): line = line.strip() if line in ['RUNSPEC', 'GRID', 'EDIT', 'PROPS', 'REGIONS', 'SOLUTION', 'SUMMARY', 'SCHEDULE']: sections.append(line) return sections def get_keywords(self) -> Dict: """Extract keywords from current deck.""" if not self.current_deck: return {'error': 'No deck loaded'} keywords = [] for line in self.current_deck.split(' '): line = line.strip() if line and not line.startswith('--') and line.isupper() and ' ' not in line: if not any(c.isdigit() for c in line) and '/' not in line: keywords.append(line) return {'keywords': list(set(keywords)), 'count': len(set(keywords))} def list_decks(self) -> List[str]: """List all deck files in work directory.""" return [str(f) for f in self.work_dir.glob('*.DATA')] # ============ UPDATE ============ def update_keyword(self, keyword: str, new_value: str) -> Dict: """Update a keyword value in the deck.""" if not self.current_deck: return {'error': 'No deck loaded'} lines = self.current_deck.split(' ') updated = False new_lines = [] skip_until_slash = False for i, line in enumerate(lines): if skip_until_slash: if '/' in line: skip_until_slash = False continue if line.strip() == keyword: new_lines.append(line) new_lines.append(new_value) skip_until_slash = True updated = True else: new_lines.append(line) if updated: self.current_deck = ' '.join(new_lines) if self.current_deck_path: self.current_deck_path.write_text(self.current_deck) return {'status': 'updated', 'keyword': keyword} else: return {'status': 'not_found', 'keyword': keyword} def append_schedule(self, schedule_content: str) -> Dict: """Append content to SCHEDULE section.""" if not self.current_deck: return {'error': 'No deck loaded'} # Find END and insert before it if 'END' in self.current_deck: self.current_deck = self.current_deck.replace('END', f'{schedule_content} END') else: self.current_deck += f' {schedule_content} ' if self.current_deck_path: self.current_deck_path.write_text(self.current_deck) return {'status': 'appended', 'lines_added': schedule_content.count(' ') + 1} # ============ DELETE ============ def delete_deck(self, name: str) -> Dict: """Delete a deck file.""" deck_path = self.work_dir / f"{name}.DATA" if deck_path.exists(): deck_path.unlink() if self.current_deck_path == deck_path: self.current_deck = None self.current_deck_path = None return {'status': 'deleted', 'path': str(deck_path)} return {'status': 'not_found', 'path': str(deck_path)} def clear_workspace(self) -> Dict: """Remove all files from workspace.""" count = 0 for f in self.work_dir.iterdir(): f.unlink() count += 1 self.current_deck = None self.current_deck_path = None return {'status': 'cleared', 'files_removed': count} # ============ SIMULATE ============ def run_simulation(self, deck_path: str = None) -> Dict: """Run OPM Flow simulation.""" path = Path(deck_path) if deck_path else self.current_deck_path if not path or not path.exists(): return {'error': 'No deck file specified'} if not OPM_AVAILABLE: # Mock simulation return self._mock_simulation(path) # Real simulation try: if self.use_docker: cmd = [ 'docker', 'run', '--rm', '-v', f'{path.parent}:/data', 'opmproject/flow:latest', 'flow', f'/data/{path.name}' ] else: cmd = ['flow', str(path)] result = subprocess.run(cmd, capture_output=True, text=True, timeout=300) return { 'status': 'success' if result.returncode == 0 else 'failed', 'returncode': result.returncode, 'stdout': result.stdout[-2000:] if result.stdout else '', 'stderr': result.stderr[-1000:] if result.stderr else '' } except subprocess.TimeoutExpired: return {'status': 'timeout', 'error': 'Simulation exceeded 5 minutes'} except Exception as e: return {'status': 'error', 'error': str(e)} def _mock_simulation(self, path: Path) -> Dict: """Return mock simulation results.""" return { 'status': 'mock', 'message': 'Mock simulation (OPM Flow not available)', 'deck': str(path), 'mock_results': { 'FOPR': [500, 480, 460, 440, 420, 400, 380, 360, 340, 320, 300, 280], 'FWPR': [0, 20, 50, 80, 120, 160, 200, 250, 300, 350, 400, 450], 'FWCT': [0.0, 0.04, 0.10, 0.15, 0.22, 0.29, 0.34, 0.41, 0.47, 0.52, 0.57, 0.62], 'FPR': [300, 295, 288, 280, 271, 261, 250, 238, 225, 211, 196, 180] }, 'timesteps': list(range(0, 360, 30)) } def get_results(self) -> Dict: """Parse simulation results.""" if not self.current_deck_path: return {'error': 'No simulation run'} # Look for summary file base = self.current_deck_path.stem smspec = self.work_dir / f"{base}.SMSPEC" unsmry = self.work_dir / f"{base}.UNSMRY" if smspec.exists() and unsmry.exists(): return {'status': 'results_available', 'files': [str(smspec), str(unsmry)]} else: return {'status': 'no_results', 'hint': 'Run simulation first or use mock mode'} # Initialize manager opm = OPMFlowManager(WORK_DIR) print("✅ OPMFlowManager initialized")

---

📝 CREATE: Build a Deck¶

In [ ]:

# CREATE: Minimal deck
result = opm.create_minimal_deck("Playground Test Model")
print("📝 Created deck:")
print(json.dumps(result, indent=2))

# CREATE: Minimal deck result = opm.create_minimal_deck("Playground Test Model") print("📝 Created deck:") print(json.dumps(result, indent=2))

In [ ]:

# CREATE: Custom deck
custom_deck = """
-- Custom ECLIPSE Deck
RUNSPEC

TITLE
My Custom Model /

DIMENS
5 5 2 /

-- Add more sections as needed...
"""

# Uncomment to create:
# result = opm.create_deck('custom', custom_deck)
# print(json.dumps(result, indent=2))

print("⚠️ Custom deck creation is commented out. Uncomment to test.")

# CREATE: Custom deck custom_deck = """ -- Custom ECLIPSE Deck RUNSPEC TITLE My Custom Model / DIMENS 5 5 2 / -- Add more sections as needed... """ # Uncomment to create: # result = opm.create_deck('custom', custom_deck) # print(json.dumps(result, indent=2)) print("⚠️ Custom deck creation is commented out. Uncomment to test.")

---

📖 READ: Inspect the Deck¶

In [ ]:

# READ: Deck info
deck_info = opm.read_deck()
print(f"📄 Deck: {deck_info.get('path')}")
print(f"   Lines: {deck_info.get('lines')}")
print(f"   Sections: {deck_info.get('sections')}")

# READ: Deck info deck_info = opm.read_deck() print(f"📄 Deck: {deck_info.get('path')}") print(f" Lines: {deck_info.get('lines')}") print(f" Sections: {deck_info.get('sections')}")

In [ ]:

# READ: Keywords
keywords = opm.get_keywords()
print(f"🔑 Keywords ({keywords.get('count')}):")
for kw in sorted(keywords.get('keywords', [])):
    print(f"   {kw}")

# READ: Keywords keywords = opm.get_keywords() print(f"🔑 Keywords ({keywords.get('count')}):") for kw in sorted(keywords.get('keywords', [])): print(f" {kw}")

In [ ]:

# READ: Show deck content (first 30 lines)
deck_info = opm.read_deck()
if 'content' in deck_info:
    lines = deck_info['content'].split('
')[:30]
    print("📜 Deck content (first 30 lines):")
    for i, line in enumerate(lines, 1):
        print(f"{i:3}: {line}")

# READ: Show deck content (first 30 lines) deck_info = opm.read_deck() if 'content' in deck_info: lines = deck_info['content'].split(' ')[:30] print("📜 Deck content (first 30 lines):") for i, line in enumerate(lines, 1): print(f"{i:3}: {line}")

---

✏️ UPDATE: Modify the Deck¶

In [ ]:

# UPDATE: Change permeability
result = opm.update_keyword('PERMX', '300*200 /')
print(f"✏️ Update PERMX: {result}")

# UPDATE: Change permeability result = opm.update_keyword('PERMX', '300*200 /') print(f"✏️ Update PERMX: {result}")

In [ ]:

# UPDATE: Add more timesteps
additional_schedule = """
TSTEP
30 30 30 30 30 30 /
"""
result = opm.append_schedule(additional_schedule)
print(f"✏️ Append schedule: {result}")

# UPDATE: Add more timesteps additional_schedule = """ TSTEP 30 30 30 30 30 30 / """ result = opm.append_schedule(additional_schedule) print(f"✏️ Append schedule: {result}")

---

🔬 SIMULATE: Run OPM Flow¶

In [ ]:

# RUN: Execute simulation
print("🚀 Running simulation...")
sim_result = opm.run_simulation()
print(json.dumps(sim_result, indent=2))

# RUN: Execute simulation print("🚀 Running simulation...") sim_result = opm.run_simulation() print(json.dumps(sim_result, indent=2))

In [ ]:

# VISUALIZE: Plot results (mock or real)
try:
    import matplotlib.pyplot as plt
    
    if 'mock_results' in sim_result:
        data = sim_result['mock_results']
        time = sim_result['timesteps']
        
        fig, axes = plt.subplots(2, 2, figsize=(12, 8))
        
        axes[0, 0].plot(time, data['FOPR'], 'g-', linewidth=2)
        axes[0, 0].set_title('Oil Production Rate (FOPR)')
        axes[0, 0].set_xlabel('Days')
        axes[0, 0].set_ylabel('SM3/DAY')
        axes[0, 0].grid(True)
        
        axes[0, 1].plot(time, data['FWPR'], 'b-', linewidth=2)
        axes[0, 1].set_title('Water Production Rate (FWPR)')
        axes[0, 1].set_xlabel('Days')
        axes[0, 1].set_ylabel('SM3/DAY')
        axes[0, 1].grid(True)
        
        axes[1, 0].plot(time, data['FWCT'], 'r-', linewidth=2)
        axes[1, 0].set_title('Water Cut (FWCT)')
        axes[1, 0].set_xlabel('Days')
        axes[1, 0].set_ylabel('Fraction')
        axes[1, 0].grid(True)
        
        axes[1, 1].plot(time, data['FPR'], 'm-', linewidth=2)
        axes[1, 1].set_title('Field Pressure (FPR)')
        axes[1, 1].set_xlabel('Days')
        axes[1, 1].set_ylabel('BARSA')
        axes[1, 1].grid(True)
        
        plt.tight_layout()
        plt.show()
    else:
        print("📊 Run simulation first to see results.")
        
except ImportError:
    print("⚠️ matplotlib not available. Install with: pip install matplotlib")

# VISUALIZE: Plot results (mock or real) try: import matplotlib.pyplot as plt if 'mock_results' in sim_result: data = sim_result['mock_results'] time = sim_result['timesteps'] fig, axes = plt.subplots(2, 2, figsize=(12, 8)) axes[0, 0].plot(time, data['FOPR'], 'g-', linewidth=2) axes[0, 0].set_title('Oil Production Rate (FOPR)') axes[0, 0].set_xlabel('Days') axes[0, 0].set_ylabel('SM3/DAY') axes[0, 0].grid(True) axes[0, 1].plot(time, data['FWPR'], 'b-', linewidth=2) axes[0, 1].set_title('Water Production Rate (FWPR)') axes[0, 1].set_xlabel('Days') axes[0, 1].set_ylabel('SM3/DAY') axes[0, 1].grid(True) axes[1, 0].plot(time, data['FWCT'], 'r-', linewidth=2) axes[1, 0].set_title('Water Cut (FWCT)') axes[1, 0].set_xlabel('Days') axes[1, 0].set_ylabel('Fraction') axes[1, 0].grid(True) axes[1, 1].plot(time, data['FPR'], 'm-', linewidth=2) axes[1, 1].set_title('Field Pressure (FPR)') axes[1, 1].set_xlabel('Days') axes[1, 1].set_ylabel('BARSA') axes[1, 1].grid(True) plt.tight_layout() plt.show() else: print("📊 Run simulation first to see results.") except ImportError: print("⚠️ matplotlib not available. Install with: pip install matplotlib")

---

🗑️ DELETE: Cleanup¶

In [ ]:

# DELETE: List files before cleanup
print("📁 Files in workspace:")
for f in opm.list_decks():
    print(f"   {f}")

# DELETE: List files before cleanup print("📁 Files in workspace:") for f in opm.list_decks(): print(f" {f}")

In [ ]:

# DELETE: Clear workspace
# Uncomment to delete all files:

# result = opm.clear_workspace()
# print(f"🗑️ Cleared: {result}")

print("⚠️ Workspace cleanup is commented out. Uncomment to delete all files.")

# DELETE: Clear workspace # Uncomment to delete all files: # result = opm.clear_workspace() # print(f"🗑️ Cleared: {result}") print("⚠️ Workspace cleanup is commented out. Uncomment to delete all files.")

---

🎮 Interactive Playground¶

In [ ]:

# 🎮 YOUR PLAYGROUND
# Try your own OPM Flow operations here!

# Examples:
# opm.create_minimal_deck("My Test")
# opm.update_keyword('PORO', '300*0.25 /')
# opm.run_simulation()

print("🎮 Edit this cell to try your own operations!")

# 🎮 YOUR PLAYGROUND # Try your own OPM Flow operations here! # Examples: # opm.create_minimal_deck("My Test") # opm.update_keyword('PORO', '300*0.25 /') # opm.run_simulation() print("🎮 Edit this cell to try your own operations!")

In [ ]:

# 📋 Quick Reference
print("""
╔══════════════════════════════════════════════════════════════════╗
║                  OPM Flow Quick Reference                        ║
╠══════════════════════════════════════════════════════════════════╣
║ CREATE                                                           ║
║   opm.create_deck(name, content)       → Create deck file        ║
║   opm.create_minimal_deck(title)       → Create minimal model    ║
╠══════════════════════════════════════════════════════════════════╣
║ READ                                                             ║
║   opm.read_deck(path)                  → Read deck content       ║
║   opm.get_keywords()                   → List keywords           ║
║   opm.list_decks()                     → List all decks          ║
╠══════════════════════════════════════════════════════════════════╣
║ UPDATE                                                           ║
║   opm.update_keyword(kw, value)        → Modify keyword          ║
║   opm.append_schedule(content)         → Add schedule steps      ║
╠══════════════════════════════════════════════════════════════════╣
║ DELETE                                                           ║
║   opm.delete_deck(name)                → Delete deck file        ║
║   opm.clear_workspace()                → Remove all files        ║
╠══════════════════════════════════════════════════════════════════╣
║ SIMULATE                                                         ║
║   opm.run_simulation(path)             → Run OPM Flow            ║
║   opm.get_results()                    → Parse output            ║
╚══════════════════════════════════════════════════════════════════╝
""")

# 📋 Quick Reference print(""" ╔══════════════════════════════════════════════════════════════════╗ ║ OPM Flow Quick Reference ║ ╠══════════════════════════════════════════════════════════════════╣ ║ CREATE ║ ║ opm.create_deck(name, content) → Create deck file ║ ║ opm.create_minimal_deck(title) → Create minimal model ║ ╠══════════════════════════════════════════════════════════════════╣ ║ READ ║ ║ opm.read_deck(path) → Read deck content ║ ║ opm.get_keywords() → List keywords ║ ║ opm.list_decks() → List all decks ║ ╠══════════════════════════════════════════════════════════════════╣ ║ UPDATE ║ ║ opm.update_keyword(kw, value) → Modify keyword ║ ║ opm.append_schedule(content) → Add schedule steps ║ ╠══════════════════════════════════════════════════════════════════╣ ║ DELETE ║ ║ opm.delete_deck(name) → Delete deck file ║ ║ opm.clear_workspace() → Remove all files ║ ╠══════════════════════════════════════════════════════════════════╣ ║ SIMULATE ║ ║ opm.run_simulation(path) → Run OPM Flow ║ ║ opm.get_results() → Parse output ║ ╚══════════════════════════════════════════════════════════════════╝ """)

---

📚 Resources¶

• 📖 OPM Flow Documentation
• 🐳 OPM Docker Images
• 📝 ECLIPSE Reference Manual
• 🧪 SPE Comparative Solution Project

---

CLARISSA Portal - OPM Flow Playground