CLARISSA Tutorial 06: Deck Generator¶

Learning Objectives:

Generate ECLIPSE decks from templates
Use structured LLM output for deck sections
Build decks incrementally with validation
Implement feedback loops for error correction

Prerequisites: Notebooks 01-05

Estimated Time: 60 minutes

Overview¶

The Deck Generator combines everything we've learned:

Intent from LLM Conversation (04)
Validation from Constraint Engine (05)
Knowledge from Knowledge Layer (03)
Syntax from ECLIPSE Fundamentals (01)

This tutorial shows how these components work together.

In [ ]:

Copied!





from dataclasses import dataclass, field
from typing import List, Dict, Optional, Any, Callable
from enum import Enum, auto
from string import Template
import json
import re

print("Deck Generator initialized")
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Any, Callable
from enum import Enum, auto
from string import Template
import json
import re

print("Deck Generator initialized")

Section 1: Template System¶

We use templates for predictable, validated deck sections.

In [ ]:

Copied!





# ECLIPSE section templates
TEMPLATES = {
    'RUNSPEC': Template('''RUNSPEC

TITLE
${title}

-- Phases
${phases}

-- Units
${units}

-- Grid dimensions
DIMENS
  ${nx} ${ny} ${nz} /

-- Table dimensions
TABDIMS
  1 1 20 20 1 20 /

-- Well dimensions
WELLDIMS
  ${max_wells} ${max_connections} ${max_groups} ${max_wells} /

START
  ${start_day} ${start_month} ${start_year} /
'''),

    'GRID': Template('''GRID

-- Cell dimensions
DX
  ${total_cells}*${dx} /
DY
  ${total_cells}*${dy} /
DZ
  ${total_cells}*${dz} /

-- Top depth
TOPS
  ${top_cells}*${tops} /

-- Porosity
PORO
  ${total_cells}*${poro} /

-- Permeability
PERMX
  ${total_cells}*${permx} /
PERMY
  ${total_cells}*${permy} /
PERMZ
  ${total_cells}*${permz} /
'''),

    'PROPS_BLACKOIL': Template('''PROPS

-- Water-Oil relative permeability
SWOF
${swof_table}
/

-- PVT: Water
PVTW
  ${pref} ${bw} ${cw} ${visc_w} 0.0 /

-- PVT: Dead Oil
PVDO
${pvdo_table}
/

-- Rock compressibility
ROCK
  ${pref} ${cr} /

-- Fluid densities
DENSITY
  ${rho_oil} ${rho_water} ${rho_gas} /
'''),

    'SOLUTION': Template('''SOLUTION

EQUIL
-- Datum  Pres    WOC   Pcow  GOC   Pcog  Init
   ${datum_depth}  ${datum_pressure}  ${woc}  0  ${goc}  0  1 /
'''),

    'SCHEDULE': Template('''SCHEDULE

${well_specs}

${completions}

${controls}

TSTEP
  ${timesteps} /

END
''')
}

print(f"Loaded {len(TEMPLATES)} section templates")
# ECLIPSE section templates
TEMPLATES = {
    'RUNSPEC': Template('''RUNSPEC

TITLE
${title}

-- Phases
${phases}

-- Units
${units}

-- Grid dimensions
DIMENS
  ${nx} ${ny} ${nz} /

-- Table dimensions
TABDIMS
  1 1 20 20 1 20 /

-- Well dimensions
WELLDIMS
  ${max_wells} ${max_connections} ${max_groups} ${max_wells} /

START
  ${start_day} ${start_month} ${start_year} /
'''),

    'GRID': Template('''GRID

-- Cell dimensions
DX
  ${total_cells}*${dx} /
DY
  ${total_cells}*${dy} /
DZ
  ${total_cells}*${dz} /

-- Top depth
TOPS
  ${top_cells}*${tops} /

-- Porosity
PORO
  ${total_cells}*${poro} /

-- Permeability
PERMX
  ${total_cells}*${permx} /
PERMY
  ${total_cells}*${permy} /
PERMZ
  ${total_cells}*${permz} /
'''),

    'PROPS_BLACKOIL': Template('''PROPS

-- Water-Oil relative permeability
SWOF
${swof_table}
/

-- PVT: Water
PVTW
  ${pref} ${bw} ${cw} ${visc_w} 0.0 /

-- PVT: Dead Oil
PVDO
${pvdo_table}
/

-- Rock compressibility
ROCK
  ${pref} ${cr} /

-- Fluid densities
DENSITY
  ${rho_oil} ${rho_water} ${rho_gas} /
'''),

    'SOLUTION': Template('''SOLUTION

EQUIL
-- Datum  Pres    WOC   Pcow  GOC   Pcog  Init
   ${datum_depth}  ${datum_pressure}  ${woc}  0  ${goc}  0  1 /
'''),

    'SCHEDULE': Template('''SCHEDULE

${well_specs}

${completions}

${controls}

TSTEP
  ${timesteps} /

END
''')
}

print(f"Loaded {len(TEMPLATES)} section templates")

In [ ]:

Copied!





# Default values for common scenarios
DEFAULTS = {
    'waterflood': {
        'phases': 'OIL\nWATER',
        'units': 'FIELD',
        'start_day': 1,
        'start_month': 'JAN',
        'start_year': 2024,
        'max_wells': 20,
        'max_connections': 50,
        'max_groups': 5,
        'poro': 0.20,
        'permx': 100,
        'permy': 100,
        'permz': 10,
        'bw': 1.01,
        'cw': 3.0e-6,
        'visc_w': 0.5,
        'cr': 3.0e-6,
        'rho_oil': 45.0,
        'rho_water': 64.0,
        'rho_gas': 0.06,
    },
    'gas_reservoir': {
        'phases': 'GAS\nWATER',
        'units': 'FIELD',
        'poro': 0.15,
        'permx': 50,
    }
}

# Standard relative permeability tables
RELPERM_TABLES = {
    'standard_water_wet': '''-- Sw      Krw      Krow     Pcow
   0.20    0.0000   1.0000   0.0
   0.30    0.0200   0.6000   0.0
   0.40    0.0500   0.3500   0.0
   0.50    0.1000   0.2000   0.0
   0.60    0.2000   0.0900   0.0
   0.70    0.3500   0.0200   0.0
   0.80    0.5000   0.0000   0.0''',

    'mixed_wet': '''-- Sw      Krw      Krow     Pcow
   0.15    0.0000   1.0000   0.0
   0.25    0.0100   0.7000   0.0
   0.35    0.0400   0.4500   0.0
   0.50    0.1200   0.2000   0.0
   0.65    0.2500   0.0500   0.0
   0.75    0.4000   0.0000   0.0'''
}

# Standard PVT tables
PVT_TABLES = {
    'light_oil': '''-- P       Bo        Vo
   1000    1.250     1.5
   2000    1.200     1.2
   3000    1.150     1.0
   4000    1.100     0.8
   5000    1.060     0.7''',

    'heavy_oil': '''-- P       Bo        Vo
   1000    1.050     50.0
   2000    1.045     30.0
   3000    1.040     20.0
   4000    1.035     15.0
   5000    1.030     12.0'''
}

print("Default configurations loaded")
# Default values for common scenarios
DEFAULTS = {
    'waterflood': {
        'phases': 'OIL\nWATER',
        'units': 'FIELD',
        'start_day': 1,
        'start_month': 'JAN',
        'start_year': 2024,
        'max_wells': 20,
        'max_connections': 50,
        'max_groups': 5,
        'poro': 0.20,
        'permx': 100,
        'permy': 100,
        'permz': 10,
        'bw': 1.01,
        'cw': 3.0e-6,
        'visc_w': 0.5,
        'cr': 3.0e-6,
        'rho_oil': 45.0,
        'rho_water': 64.0,
        'rho_gas': 0.06,
    },
    'gas_reservoir': {
        'phases': 'GAS\nWATER',
        'units': 'FIELD',
        'poro': 0.15,
        'permx': 50,
    }
}

# Standard relative permeability tables
RELPERM_TABLES = {
    'standard_water_wet': '''-- Sw      Krw      Krow     Pcow
   0.20    0.0000   1.0000   0.0
   0.30    0.0200   0.6000   0.0
   0.40    0.0500   0.3500   0.0
   0.50    0.1000   0.2000   0.0
   0.60    0.2000   0.0900   0.0
   0.70    0.3500   0.0200   0.0
   0.80    0.5000   0.0000   0.0''',

    'mixed_wet': '''-- Sw      Krw      Krow     Pcow
   0.15    0.0000   1.0000   0.0
   0.25    0.0100   0.7000   0.0
   0.35    0.0400   0.4500   0.0
   0.50    0.1200   0.2000   0.0
   0.65    0.2500   0.0500   0.0
   0.75    0.4000   0.0000   0.0'''
}

# Standard PVT tables
PVT_TABLES = {
    'light_oil': '''-- P       Bo        Vo
   1000    1.250     1.5
   2000    1.200     1.2
   3000    1.150     1.0
   4000    1.100     0.8
   5000    1.060     0.7''',

    'heavy_oil': '''-- P       Bo        Vo
   1000    1.050     50.0
   2000    1.045     30.0
   3000    1.040     20.0
   4000    1.035     15.0
   5000    1.030     12.0'''
}

print("Default configurations loaded")

Section 2: Structured Model Specification¶

The LLM outputs a structured specification that drives deck generation.

In [ ]:

Copied!





@dataclass
class WellSpec:
    """Specification for a single well."""
    name: str
    i: int
    j: int
    k1: int = 1
    k2: int = 5
    well_type: str = 'producer'  # producer, injector
    phase: str = 'OIL'  # OIL, WATER, GAS
    control: str = 'ORAT'  # ORAT, WRAT, GRAT, BHP
    target: float = 1000.0  # Rate or BHP
    bhp_limit: float = 1000.0

@dataclass
class ModelSpec:
    """Complete model specification for deck generation."""
    # Identification
    title: str = "CLARISSA Generated Model"
    scenario: str = 'waterflood'
    
    # Grid
    nx: int = 10
    ny: int = 10
    nz: int = 5
    dx: float = 100.0
    dy: float = 100.0
    dz: float = 20.0
    tops: float = 8000.0
    
    # Rock properties
    porosity: float = 0.20
    permx: float = 100.0
    permy: float = 100.0
    permz: float = 10.0
    
    # Fluid system
    oil_type: str = 'light_oil'
    wettability: str = 'standard_water_wet'
    
    # Initial conditions
    datum_depth: float = 8000.0
    datum_pressure: float = 3500.0
    woc: float = 9000.0
    goc: float = 7000.0
    
    # Wells
    wells: List[WellSpec] = field(default_factory=list)
    
    # Schedule
    simulation_days: int = 365
    report_interval: int = 30

# Example: Parse LLM output into ModelSpec
def parse_llm_output(llm_json: str) -> ModelSpec:
    """Parse structured LLM output into ModelSpec."""
    data = json.loads(llm_json)
    
    # Parse wells
    wells = []
    for w in data.get('wells', []):
        wells.append(WellSpec(**w))
    
    # Create spec (with defaults for missing fields)
    spec_data = {k: v for k, v in data.items() if k != 'wells'}
    spec_data['wells'] = wells
    
    return ModelSpec(**spec_data)

# Demo: LLM-style JSON output
llm_output = '''{
    "title": "Permian Waterflood Demo",
    "scenario": "waterflood",
    "nx": 20, "ny": 20, "nz": 5,
    "dx": 100, "dy": 100, "dz": 20,
    "tops": 8500,
    "porosity": 0.22,
    "permx": 150, "permy": 150, "permz": 15,
    "datum_depth": 8500,
    "datum_pressure": 3800,
    "woc": 9500,
    "wells": [
        {"name": "PROD1", "i": 10, "j": 10, "well_type": "producer", "target": 500},
        {"name": "INJ1", "i": 1, "j": 1, "well_type": "injector", "phase": "WATER", "target": 600},
        {"name": "INJ2", "i": 1, "j": 20, "well_type": "injector", "phase": "WATER", "target": 600},
        {"name": "INJ3", "i": 20, "j": 1, "well_type": "injector", "phase": "WATER", "target": 600},
        {"name": "INJ4", "i": 20, "j": 20, "well_type": "injector", "phase": "WATER", "target": 600}
    ],
    "simulation_days": 730
}'''

spec = parse_llm_output(llm_output)
print(f"Parsed model: {spec.title}")
print(f"Grid: {spec.nx}x{spec.ny}x{spec.nz}")
print(f"Wells: {len(spec.wells)}")
@dataclass
class WellSpec:
    """Specification for a single well."""
    name: str
    i: int
    j: int
    k1: int = 1
    k2: int = 5
    well_type: str = 'producer'  # producer, injector
    phase: str = 'OIL'  # OIL, WATER, GAS
    control: str = 'ORAT'  # ORAT, WRAT, GRAT, BHP
    target: float = 1000.0  # Rate or BHP
    bhp_limit: float = 1000.0

@dataclass
class ModelSpec:
    """Complete model specification for deck generation."""
    # Identification
    title: str = "CLARISSA Generated Model"
    scenario: str = 'waterflood'
    
    # Grid
    nx: int = 10
    ny: int = 10
    nz: int = 5
    dx: float = 100.0
    dy: float = 100.0
    dz: float = 20.0
    tops: float = 8000.0
    
    # Rock properties
    porosity: float = 0.20
    permx: float = 100.0
    permy: float = 100.0
    permz: float = 10.0
    
    # Fluid system
    oil_type: str = 'light_oil'
    wettability: str = 'standard_water_wet'
    
    # Initial conditions
    datum_depth: float = 8000.0
    datum_pressure: float = 3500.0
    woc: float = 9000.0
    goc: float = 7000.0
    
    # Wells
    wells: List[WellSpec] = field(default_factory=list)
    
    # Schedule
    simulation_days: int = 365
    report_interval: int = 30

# Example: Parse LLM output into ModelSpec
def parse_llm_output(llm_json: str) -> ModelSpec:
    """Parse structured LLM output into ModelSpec."""
    data = json.loads(llm_json)
    
    # Parse wells
    wells = []
    for w in data.get('wells', []):
        wells.append(WellSpec(**w))
    
    # Create spec (with defaults for missing fields)
    spec_data = {k: v for k, v in data.items() if k != 'wells'}
    spec_data['wells'] = wells
    
    return ModelSpec(**spec_data)

# Demo: LLM-style JSON output
llm_output = '''{
    "title": "Permian Waterflood Demo",
    "scenario": "waterflood",
    "nx": 20, "ny": 20, "nz": 5,
    "dx": 100, "dy": 100, "dz": 20,
    "tops": 8500,
    "porosity": 0.22,
    "permx": 150, "permy": 150, "permz": 15,
    "datum_depth": 8500,
    "datum_pressure": 3800,
    "woc": 9500,
    "wells": [
        {"name": "PROD1", "i": 10, "j": 10, "well_type": "producer", "target": 500},
        {"name": "INJ1", "i": 1, "j": 1, "well_type": "injector", "phase": "WATER", "target": 600},
        {"name": "INJ2", "i": 1, "j": 20, "well_type": "injector", "phase": "WATER", "target": 600},
        {"name": "INJ3", "i": 20, "j": 1, "well_type": "injector", "phase": "WATER", "target": 600},
        {"name": "INJ4", "i": 20, "j": 20, "well_type": "injector", "phase": "WATER", "target": 600}
    ],
    "simulation_days": 730
}'''

spec = parse_llm_output(llm_output)
print(f"Parsed model: {spec.title}")
print(f"Grid: {spec.nx}x{spec.ny}x{spec.nz}")
print(f"Wells: {len(spec.wells)}")

Section 3: Deck Generator Class¶

The main generator that combines templates, defaults, and validation.

In [ ]:

Copied!





class DeckGenerator:
    """Generates ECLIPSE decks from specifications."""
    
    def __init__(self):
        self.templates = TEMPLATES
        self.defaults = DEFAULTS
        self.sections: Dict[str, str] = {}
        self.assumptions: List[str] = []
    
    def generate(self, spec: ModelSpec) -> str:
        """Generate complete deck from specification."""
        self.assumptions = []
        self.sections = {}
        
        # Generate each section
        self.sections['RUNSPEC'] = self._generate_runspec(spec)
        self.sections['GRID'] = self._generate_grid(spec)
        self.sections['PROPS'] = self._generate_props(spec)
        self.sections['SOLUTION'] = self._generate_solution(spec)
        self.sections['SCHEDULE'] = self._generate_schedule(spec)
        
        # Combine sections
        deck = '\n'.join(self.sections.values())
        
        return deck
    
    def _get_default(self, spec: ModelSpec, key: str, fallback: Any = None) -> Any:
        """Get default value for scenario."""
        scenario_defaults = self.defaults.get(spec.scenario, {})
        value = scenario_defaults.get(key, fallback)
        if value != fallback:
            self.assumptions.append(f"Using default {key}={value} for {spec.scenario}")
        return value
    
    def _generate_runspec(self, spec: ModelSpec) -> str:
        """Generate RUNSPEC section."""
        params = {
            'title': spec.title,
            'phases': self._get_default(spec, 'phases', 'OIL\nWATER'),
            'units': self._get_default(spec, 'units', 'FIELD'),
            'nx': spec.nx,
            'ny': spec.ny,
            'nz': spec.nz,
            'max_wells': max(len(spec.wells) * 2, 10),
            'max_connections': spec.nz * 10,
            'max_groups': 5,
            'start_day': self._get_default(spec, 'start_day', 1),
            'start_month': self._get_default(spec, 'start_month', 'JAN'),
            'start_year': self._get_default(spec, 'start_year', 2024),
        }
        return self.templates['RUNSPEC'].substitute(params)
    
    def _generate_grid(self, spec: ModelSpec) -> str:
        """Generate GRID section."""
        total_cells = spec.nx * spec.ny * spec.nz
        top_cells = spec.nx * spec.ny
        
        params = {
            'total_cells': total_cells,
            'top_cells': top_cells,
            'dx': spec.dx,
            'dy': spec.dy,
            'dz': spec.dz,
            'tops': spec.tops,
            'poro': spec.porosity,
            'permx': spec.permx,
            'permy': spec.permy,
            'permz': spec.permz,
        }
        return self.templates['GRID'].substitute(params)
    
    def _generate_props(self, spec: ModelSpec) -> str:
        """Generate PROPS section."""
        swof = RELPERM_TABLES.get(spec.wettability, RELPERM_TABLES['standard_water_wet'])
        pvdo = PVT_TABLES.get(spec.oil_type, PVT_TABLES['light_oil'])
        
        if spec.wettability not in RELPERM_TABLES:
            self.assumptions.append(f"Using standard_water_wet relperm (requested: {spec.wettability})")
        
        params = {
            'swof_table': swof,
            'pvdo_table': pvdo,
            'pref': spec.datum_pressure,
            'bw': self._get_default(spec, 'bw', 1.01),
            'cw': self._get_default(spec, 'cw', 3.0e-6),
            'visc_w': self._get_default(spec, 'visc_w', 0.5),
            'cr': self._get_default(spec, 'cr', 3.0e-6),
            'rho_oil': self._get_default(spec, 'rho_oil', 45.0),
            'rho_water': self._get_default(spec, 'rho_water', 64.0),
            'rho_gas': self._get_default(spec, 'rho_gas', 0.06),
        }
        return self.templates['PROPS_BLACKOIL'].substitute(params)
    
    def _generate_solution(self, spec: ModelSpec) -> str:
        """Generate SOLUTION section."""
        params = {
            'datum_depth': spec.datum_depth,
            'datum_pressure': spec.datum_pressure,
            'woc': spec.woc,
            'goc': spec.goc,
        }
        return self.templates['SOLUTION'].substitute(params)
    
    def _generate_schedule(self, spec: ModelSpec) -> str:
        """Generate SCHEDULE section."""
        # WELSPECS
        welspecs_lines = ['WELSPECS']
        for w in spec.wells:
            welspecs_lines.append(f"  {w.name:8} G1 {w.i:3} {w.j:3} 1* {w.phase} /")
        welspecs_lines.append('/')
        
        # COMPDAT
        compdat_lines = ['COMPDAT']
        for w in spec.wells:
            compdat_lines.append(f"  {w.name:8} {w.i:3} {w.j:3} {w.k1:3} {w.k2:3} OPEN 1* 0.5 /")
        compdat_lines.append('/')
        
        # Controls
        control_lines = []
        
        # Producers
        producers = [w for w in spec.wells if w.well_type == 'producer']
        if producers:
            control_lines.append('WCONPROD')
            for w in producers:
                control_lines.append(f"  {w.name:8} OPEN {w.control} {w.target:.0f} 4* {w.bhp_limit:.0f} /")
            control_lines.append('/')
        
        # Injectors
        injectors = [w for w in spec.wells if w.well_type == 'injector']
        if injectors:
            control_lines.append('WCONINJE')
            for w in injectors:
                control_lines.append(f"  {w.name:8} {w.phase} OPEN RATE {w.target:.0f} 1* 5000 /")
            control_lines.append('/')
        
        # Time steps
        num_steps = spec.simulation_days // spec.report_interval
        timesteps = f"{num_steps}*{spec.report_interval}"
        
        params = {
            'well_specs': '\n'.join(welspecs_lines),
            'completions': '\n'.join(compdat_lines),
            'controls': '\n'.join(control_lines),
            'timesteps': timesteps,
        }
        return self.templates['SCHEDULE'].substitute(params)
    
    def get_assumptions(self) -> List[str]:
        """Return list of assumptions made during generation."""
        return self.assumptions

# Generate deck from spec
generator = DeckGenerator()
deck = generator.generate(spec)

print(f"Generated deck: {len(deck)} characters")
print(f"Assumptions made: {len(generator.get_assumptions())}")
for assumption in generator.get_assumptions()[:5]:
    print(f"  - {assumption}")
class DeckGenerator:
    """Generates ECLIPSE decks from specifications."""
    
    def __init__(self):
        self.templates = TEMPLATES
        self.defaults = DEFAULTS
        self.sections: Dict[str, str] = {}
        self.assumptions: List[str] = []
    
    def generate(self, spec: ModelSpec) -> str:
        """Generate complete deck from specification."""
        self.assumptions = []
        self.sections = {}
        
        # Generate each section
        self.sections['RUNSPEC'] = self._generate_runspec(spec)
        self.sections['GRID'] = self._generate_grid(spec)
        self.sections['PROPS'] = self._generate_props(spec)
        self.sections['SOLUTION'] = self._generate_solution(spec)
        self.sections['SCHEDULE'] = self._generate_schedule(spec)
        
        # Combine sections
        deck = '\n'.join(self.sections.values())
        
        return deck
    
    def _get_default(self, spec: ModelSpec, key: str, fallback: Any = None) -> Any:
        """Get default value for scenario."""
        scenario_defaults = self.defaults.get(spec.scenario, {})
        value = scenario_defaults.get(key, fallback)
        if value != fallback:
            self.assumptions.append(f"Using default {key}={value} for {spec.scenario}")
        return value
    
    def _generate_runspec(self, spec: ModelSpec) -> str:
        """Generate RUNSPEC section."""
        params = {
            'title': spec.title,
            'phases': self._get_default(spec, 'phases', 'OIL\nWATER'),
            'units': self._get_default(spec, 'units', 'FIELD'),
            'nx': spec.nx,
            'ny': spec.ny,
            'nz': spec.nz,
            'max_wells': max(len(spec.wells) * 2, 10),
            'max_connections': spec.nz * 10,
            'max_groups': 5,
            'start_day': self._get_default(spec, 'start_day', 1),
            'start_month': self._get_default(spec, 'start_month', 'JAN'),
            'start_year': self._get_default(spec, 'start_year', 2024),
        }
        return self.templates['RUNSPEC'].substitute(params)
    
    def _generate_grid(self, spec: ModelSpec) -> str:
        """Generate GRID section."""
        total_cells = spec.nx * spec.ny * spec.nz
        top_cells = spec.nx * spec.ny
        
        params = {
            'total_cells': total_cells,
            'top_cells': top_cells,
            'dx': spec.dx,
            'dy': spec.dy,
            'dz': spec.dz,
            'tops': spec.tops,
            'poro': spec.porosity,
            'permx': spec.permx,
            'permy': spec.permy,
            'permz': spec.permz,
        }
        return self.templates['GRID'].substitute(params)
    
    def _generate_props(self, spec: ModelSpec) -> str:
        """Generate PROPS section."""
        swof = RELPERM_TABLES.get(spec.wettability, RELPERM_TABLES['standard_water_wet'])
        pvdo = PVT_TABLES.get(spec.oil_type, PVT_TABLES['light_oil'])
        
        if spec.wettability not in RELPERM_TABLES:
            self.assumptions.append(f"Using standard_water_wet relperm (requested: {spec.wettability})")
        
        params = {
            'swof_table': swof,
            'pvdo_table': pvdo,
            'pref': spec.datum_pressure,
            'bw': self._get_default(spec, 'bw', 1.01),
            'cw': self._get_default(spec, 'cw', 3.0e-6),
            'visc_w': self._get_default(spec, 'visc_w', 0.5),
            'cr': self._get_default(spec, 'cr', 3.0e-6),
            'rho_oil': self._get_default(spec, 'rho_oil', 45.0),
            'rho_water': self._get_default(spec, 'rho_water', 64.0),
            'rho_gas': self._get_default(spec, 'rho_gas', 0.06),
        }
        return self.templates['PROPS_BLACKOIL'].substitute(params)
    
    def _generate_solution(self, spec: ModelSpec) -> str:
        """Generate SOLUTION section."""
        params = {
            'datum_depth': spec.datum_depth,
            'datum_pressure': spec.datum_pressure,
            'woc': spec.woc,
            'goc': spec.goc,
        }
        return self.templates['SOLUTION'].substitute(params)
    
    def _generate_schedule(self, spec: ModelSpec) -> str:
        """Generate SCHEDULE section."""
        # WELSPECS
        welspecs_lines = ['WELSPECS']
        for w in spec.wells:
            welspecs_lines.append(f"  {w.name:8} G1 {w.i:3} {w.j:3} 1* {w.phase} /")
        welspecs_lines.append('/')
        
        # COMPDAT
        compdat_lines = ['COMPDAT']
        for w in spec.wells:
            compdat_lines.append(f"  {w.name:8} {w.i:3} {w.j:3} {w.k1:3} {w.k2:3} OPEN 1* 0.5 /")
        compdat_lines.append('/')
        
        # Controls
        control_lines = []
        
        # Producers
        producers = [w for w in spec.wells if w.well_type == 'producer']
        if producers:
            control_lines.append('WCONPROD')
            for w in producers:
                control_lines.append(f"  {w.name:8} OPEN {w.control} {w.target:.0f} 4* {w.bhp_limit:.0f} /")
            control_lines.append('/')
        
        # Injectors
        injectors = [w for w in spec.wells if w.well_type == 'injector']
        if injectors:
            control_lines.append('WCONINJE')
            for w in injectors:
                control_lines.append(f"  {w.name:8} {w.phase} OPEN RATE {w.target:.0f} 1* 5000 /")
            control_lines.append('/')
        
        # Time steps
        num_steps = spec.simulation_days // spec.report_interval
        timesteps = f"{num_steps}*{spec.report_interval}"
        
        params = {
            'well_specs': '\n'.join(welspecs_lines),
            'completions': '\n'.join(compdat_lines),
            'controls': '\n'.join(control_lines),
            'timesteps': timesteps,
        }
        return self.templates['SCHEDULE'].substitute(params)
    
    def get_assumptions(self) -> List[str]:
        """Return list of assumptions made during generation."""
        return self.assumptions

# Generate deck from spec
generator = DeckGenerator()
deck = generator.generate(spec)

print(f"Generated deck: {len(deck)} characters")
print(f"Assumptions made: {len(generator.get_assumptions())}")
for assumption in generator.get_assumptions()[:5]:
    print(f"  - {assumption}")

In [ ]:

Copied!





# Show generated deck (first 2500 chars)
print("Generated Deck Preview:")
print("=" * 60)
print(deck[:2500])
print("...")
# Show generated deck (first 2500 chars)
print("Generated Deck Preview:")
print("=" * 60)
print(deck[:2500])
print("...")

Section 4: Incremental Generation with Feedback¶

Build the deck step by step, validating each section.

In [ ]:

Copied!





class IncrementalDeckBuilder:
    """Builds deck incrementally with validation at each step."""
    
    def __init__(self):
        self.sections: Dict[str, str] = {}
        self.validation_errors: List[str] = []
        self.generator = DeckGenerator()
    
    def set_grid(self, nx: int, ny: int, nz: int, 
                 dx: float, dy: float, dz: float,
                 tops: float) -> Tuple[bool, str]:
        """Set grid parameters with validation."""
        # Validate
        total_cells = nx * ny * nz
        if total_cells > 1_000_000:
            return False, f"Grid too large: {total_cells:,} cells. Consider coarsening."
        if any(d <= 0 for d in [dx, dy, dz]):
            return False, "Cell dimensions must be positive"
        
        self.sections['GRID'] = {
            'nx': nx, 'ny': ny, 'nz': nz,
            'dx': dx, 'dy': dy, 'dz': dz,
            'tops': tops
        }
        return True, f"Grid set: {nx}x{ny}x{nz} = {total_cells:,} cells"
    
    def set_rock_properties(self, poro: float, permx: float, 
                            permy: float, permz: float) -> Tuple[bool, str]:
        """Set rock properties with validation."""
        if not (0 < poro < 1):
            return False, f"Porosity {poro} out of range (0, 1)"
        if any(k <= 0 for k in [permx, permy, permz]):
            return False, "Permeability must be positive"
        if permz > permx:
            return False, f"kv ({permz}) > kh ({permx}) is unusual. Confirm if intentional."
        
        self.sections['ROCK'] = {
            'poro': poro, 'permx': permx, 'permy': permy, 'permz': permz
        }
        return True, f"Rock properties set: phi={poro:.0%}, k={permx:.0f} md"
    
    def set_initial_conditions(self, depth: float, pressure: float,
                               woc: float) -> Tuple[bool, str]:
        """Set initial conditions with validation."""
        if pressure <= 0:
            return False, "Pressure must be positive"
        
        # Check pressure gradient
        gradient = pressure / depth
        if gradient < 0.3 or gradient > 0.6:
            return False, f"Pressure gradient {gradient:.3f} psi/ft unusual. Expected 0.3-0.6"
        
        if woc < depth:
            return False, f"WOC ({woc} ft) above datum ({depth} ft). Check contacts."
        
        self.sections['INIT'] = {
            'depth': depth, 'pressure': pressure, 'woc': woc
        }
        return True, f"Initial conditions set: {pressure:.0f} psi at {depth:.0f} ft"
    
    def add_well(self, name: str, i: int, j: int, 
                 well_type: str, rate: float) -> Tuple[bool, str]:
        """Add a well with validation."""
        if 'GRID' not in self.sections:
            return False, "Define grid before adding wells"
        
        grid = self.sections['GRID']
        if i > grid['nx'] or j > grid['ny']:
            return False, f"Well {name} at ({i},{j}) outside grid ({grid['nx']},{grid['ny']})"
        
        if 'WELLS' not in self.sections:
            self.sections['WELLS'] = []
        
        self.sections['WELLS'].append({
            'name': name, 'i': i, 'j': j, 'type': well_type, 'rate': rate
        })
        return True, f"Added {well_type} well {name} at ({i},{j}) with rate {rate}"
    
    def get_status(self) -> str:
        """Get current build status."""
        status = ["Build Status:"]
        status.append(f"  Grid: {'defined' if 'GRID' in self.sections else 'MISSING'}")
        status.append(f"  Rock: {'defined' if 'ROCK' in self.sections else 'MISSING'}")
        status.append(f"  Init: {'defined' if 'INIT' in self.sections else 'MISSING'}")
        wells = self.sections.get('WELLS', [])
        status.append(f"  Wells: {len(wells)}")
        return '\n'.join(status)

# Demo incremental building
builder = IncrementalDeckBuilder()

print("Incremental Deck Building:")
print("=" * 50)

# Step 1: Grid
success, msg = builder.set_grid(20, 20, 5, 100, 100, 20, 8500)
print(f"Step 1 - Grid: {msg}")

# Step 2: Rock properties
success, msg = builder.set_rock_properties(0.22, 150, 150, 15)
print(f"Step 2 - Rock: {msg}")

# Step 3: Initial conditions (try invalid first)
success, msg = builder.set_initial_conditions(8500, 100, 9500)  # Too low pressure
print(f"Step 3a - Init (invalid): {msg}")

success, msg = builder.set_initial_conditions(8500, 3800, 9500)  # Valid
print(f"Step 3b - Init (valid): {msg}")

# Step 4: Wells
success, msg = builder.add_well("PROD1", 10, 10, "producer", 500)
print(f"Step 4 - Well: {msg}")

success, msg = builder.add_well("INJ1", 1, 1, "injector", 600)
print(f"Step 5 - Well: {msg}")

print("\n" + builder.get_status())
class IncrementalDeckBuilder:
    """Builds deck incrementally with validation at each step."""
    
    def __init__(self):
        self.sections: Dict[str, str] = {}
        self.validation_errors: List[str] = []
        self.generator = DeckGenerator()
    
    def set_grid(self, nx: int, ny: int, nz: int, 
                 dx: float, dy: float, dz: float,
                 tops: float) -> Tuple[bool, str]:
        """Set grid parameters with validation."""
        # Validate
        total_cells = nx * ny * nz
        if total_cells > 1_000_000:
            return False, f"Grid too large: {total_cells:,} cells. Consider coarsening."
        if any(d <= 0 for d in [dx, dy, dz]):
            return False, "Cell dimensions must be positive"
        
        self.sections['GRID'] = {
            'nx': nx, 'ny': ny, 'nz': nz,
            'dx': dx, 'dy': dy, 'dz': dz,
            'tops': tops
        }
        return True, f"Grid set: {nx}x{ny}x{nz} = {total_cells:,} cells"
    
    def set_rock_properties(self, poro: float, permx: float, 
                            permy: float, permz: float) -> Tuple[bool, str]:
        """Set rock properties with validation."""
        if not (0 < poro < 1):
            return False, f"Porosity {poro} out of range (0, 1)"
        if any(k <= 0 for k in [permx, permy, permz]):
            return False, "Permeability must be positive"
        if permz > permx:
            return False, f"kv ({permz}) > kh ({permx}) is unusual. Confirm if intentional."
        
        self.sections['ROCK'] = {
            'poro': poro, 'permx': permx, 'permy': permy, 'permz': permz
        }
        return True, f"Rock properties set: phi={poro:.0%}, k={permx:.0f} md"
    
    def set_initial_conditions(self, depth: float, pressure: float,
                               woc: float) -> Tuple[bool, str]:
        """Set initial conditions with validation."""
        if pressure <= 0:
            return False, "Pressure must be positive"
        
        # Check pressure gradient
        gradient = pressure / depth
        if gradient < 0.3 or gradient > 0.6:
            return False, f"Pressure gradient {gradient:.3f} psi/ft unusual. Expected 0.3-0.6"
        
        if woc < depth:
            return False, f"WOC ({woc} ft) above datum ({depth} ft). Check contacts."
        
        self.sections['INIT'] = {
            'depth': depth, 'pressure': pressure, 'woc': woc
        }
        return True, f"Initial conditions set: {pressure:.0f} psi at {depth:.0f} ft"
    
    def add_well(self, name: str, i: int, j: int, 
                 well_type: str, rate: float) -> Tuple[bool, str]:
        """Add a well with validation."""
        if 'GRID' not in self.sections:
            return False, "Define grid before adding wells"
        
        grid = self.sections['GRID']
        if i > grid['nx'] or j > grid['ny']:
            return False, f"Well {name} at ({i},{j}) outside grid ({grid['nx']},{grid['ny']})"
        
        if 'WELLS' not in self.sections:
            self.sections['WELLS'] = []
        
        self.sections['WELLS'].append({
            'name': name, 'i': i, 'j': j, 'type': well_type, 'rate': rate
        })
        return True, f"Added {well_type} well {name} at ({i},{j}) with rate {rate}"
    
    def get_status(self) -> str:
        """Get current build status."""
        status = ["Build Status:"]
        status.append(f"  Grid: {'defined' if 'GRID' in self.sections else 'MISSING'}")
        status.append(f"  Rock: {'defined' if 'ROCK' in self.sections else 'MISSING'}")
        status.append(f"  Init: {'defined' if 'INIT' in self.sections else 'MISSING'}")
        wells = self.sections.get('WELLS', [])
        status.append(f"  Wells: {len(wells)}")
        return '\n'.join(status)

# Demo incremental building
builder = IncrementalDeckBuilder()

print("Incremental Deck Building:")
print("=" * 50)

# Step 1: Grid
success, msg = builder.set_grid(20, 20, 5, 100, 100, 20, 8500)
print(f"Step 1 - Grid: {msg}")

# Step 2: Rock properties
success, msg = builder.set_rock_properties(0.22, 150, 150, 15)
print(f"Step 2 - Rock: {msg}")

# Step 3: Initial conditions (try invalid first)
success, msg = builder.set_initial_conditions(8500, 100, 9500)  # Too low pressure
print(f"Step 3a - Init (invalid): {msg}")

success, msg = builder.set_initial_conditions(8500, 3800, 9500)  # Valid
print(f"Step 3b - Init (valid): {msg}")

# Step 4: Wells
success, msg = builder.add_well("PROD1", 10, 10, "producer", 500)
print(f"Step 4 - Well: {msg}")

success, msg = builder.add_well("INJ1", 1, 1, "injector", 600)
print(f"Step 5 - Well: {msg}")

print("\n" + builder.get_status())

Section 5: Error Recovery and Suggestions¶

When generation fails, provide helpful suggestions.

In [ ]:

Copied!





@dataclass
class GenerationError:
    """Error during deck generation with suggestions."""
    section: str
    message: str
    suggestions: List[str]
    auto_fix: Optional[Callable] = None

class SmartDeckGenerator(DeckGenerator):
    """Deck generator with error recovery."""
    
    def __init__(self):
        super().__init__()
        self.errors: List[GenerationError] = []
    
    def generate_with_recovery(self, spec: ModelSpec) -> Tuple[str, List[GenerationError]]:
        """Generate deck with automatic error recovery."""
        self.errors = []
        
        # Pre-validate and fix common issues
        spec = self._auto_fix(spec)
        
        # Generate
        try:
            deck = self.generate(spec)
            return deck, self.errors
        except Exception as e:
            self.errors.append(GenerationError(
                section='GENERAL',
                message=str(e),
                suggestions=['Check all required parameters are provided']
            ))
            return '', self.errors
    
    def _auto_fix(self, spec: ModelSpec) -> ModelSpec:
        """Automatically fix common issues."""
        
        # Fix permeability anisotropy
        if spec.permz > spec.permx:
            self.errors.append(GenerationError(
                section='GRID',
                message=f'kv ({spec.permz}) > kh ({spec.permx})',
                suggestions=[f'Auto-fixed: set kv = kh * 0.1 = {spec.permx * 0.1}']
            ))
            spec.permz = spec.permx * 0.1
        
        # Fix well locations outside grid
        for well in spec.wells:
            if well.i > spec.nx:
                self.errors.append(GenerationError(
                    section='SCHEDULE',
                    message=f'Well {well.name} i={well.i} > nx={spec.nx}',
                    suggestions=[f'Auto-fixed: set i = {spec.nx}']
                ))
                well.i = spec.nx
            if well.j > spec.ny:
                self.errors.append(GenerationError(
                    section='SCHEDULE',
                    message=f'Well {well.name} j={well.j} > ny={spec.ny}',
                    suggestions=[f'Auto-fixed: set j = {spec.ny}']
                ))
                well.j = spec.ny
        
        return spec

# Demo with problematic spec
bad_spec = ModelSpec(
    title="Problematic Model",
    nx=10, ny=10, nz=5,
    permx=100, permy=100, permz=200,  # kv > kh (bad)
    wells=[
        WellSpec(name="P1", i=15, j=15),  # Outside grid (bad)
    ]
)

smart_gen = SmartDeckGenerator()
deck, errors = smart_gen.generate_with_recovery(bad_spec)

print("Smart Generation with Auto-Fix:")
print("=" * 50)
if errors:
    print(f"Fixed {len(errors)} issues:")
    for err in errors:
        print(f"  [{err.section}] {err.message}")
        for sug in err.suggestions:
            print(f"    -> {sug}")
print(f"\nGenerated deck: {len(deck)} characters")
@dataclass
class GenerationError:
    """Error during deck generation with suggestions."""
    section: str
    message: str
    suggestions: List[str]
    auto_fix: Optional[Callable] = None

class SmartDeckGenerator(DeckGenerator):
    """Deck generator with error recovery."""
    
    def __init__(self):
        super().__init__()
        self.errors: List[GenerationError] = []
    
    def generate_with_recovery(self, spec: ModelSpec) -> Tuple[str, List[GenerationError]]:
        """Generate deck with automatic error recovery."""
        self.errors = []
        
        # Pre-validate and fix common issues
        spec = self._auto_fix(spec)
        
        # Generate
        try:
            deck = self.generate(spec)
            return deck, self.errors
        except Exception as e:
            self.errors.append(GenerationError(
                section='GENERAL',
                message=str(e),
                suggestions=['Check all required parameters are provided']
            ))
            return '', self.errors
    
    def _auto_fix(self, spec: ModelSpec) -> ModelSpec:
        """Automatically fix common issues."""
        
        # Fix permeability anisotropy
        if spec.permz > spec.permx:
            self.errors.append(GenerationError(
                section='GRID',
                message=f'kv ({spec.permz}) > kh ({spec.permx})',
                suggestions=[f'Auto-fixed: set kv = kh * 0.1 = {spec.permx * 0.1}']
            ))
            spec.permz = spec.permx * 0.1
        
        # Fix well locations outside grid
        for well in spec.wells:
            if well.i > spec.nx:
                self.errors.append(GenerationError(
                    section='SCHEDULE',
                    message=f'Well {well.name} i={well.i} > nx={spec.nx}',
                    suggestions=[f'Auto-fixed: set i = {spec.nx}']
                ))
                well.i = spec.nx
            if well.j > spec.ny:
                self.errors.append(GenerationError(
                    section='SCHEDULE',
                    message=f'Well {well.name} j={well.j} > ny={spec.ny}',
                    suggestions=[f'Auto-fixed: set j = {spec.ny}']
                ))
                well.j = spec.ny
        
        return spec

# Demo with problematic spec
bad_spec = ModelSpec(
    title="Problematic Model",
    nx=10, ny=10, nz=5,
    permx=100, permy=100, permz=200,  # kv > kh (bad)
    wells=[
        WellSpec(name="P1", i=15, j=15),  # Outside grid (bad)
    ]
)

smart_gen = SmartDeckGenerator()
deck, errors = smart_gen.generate_with_recovery(bad_spec)

print("Smart Generation with Auto-Fix:")
print("=" * 50)
if errors:
    print(f"Fixed {len(errors)} issues:")
    for err in errors:
        print(f"  [{err.section}] {err.message}")
        for sug in err.suggestions:
            print(f"    -> {sug}")
print(f"\nGenerated deck: {len(deck)} characters")

Summary¶

In this tutorial, we learned:

Template System: Structured templates for each ECLIPSE section
Structured Output: Parse LLM JSON into ModelSpec dataclass
Deck Generator: Combine templates, defaults, and specifications
Incremental Building: Step-by-step with validation
Error Recovery: Auto-fix common issues and provide suggestions

Key Insight: Template-based generation with validation catches errors that pure LLM generation would miss.

Next Tutorial: 07_RL_Agent.ipynb - Reinforcement learning for action optimization