tessl/pypi-resdata

Python package for reading and writing reservoir simulator result files including RESTART, INIT, RFT, Summary and GRID files in various formats.

—

Pending

Overview

Eval results

Files

Well Data Processing

Name: tessl/pypi-resdata
Author: tessl

Comprehensive well data analysis including well states, connections, segments, production and injection rates, and time-based well performance tracking from reservoir simulation results.

Capabilities

Well Information Management

Central interface for accessing well data and states across simulation time steps.

class WellInfo:
    """Well information container and state management."""
    
    def get_ts(self, well_name: str) -> list:
        """Get time series data for well."""
    
    def get_state_from_report(self, well_name: str, report_step: int) -> WellState:
        """
        Get well state at specific report step.
        
        Args:
            well_name (str): Name of the well
            report_step (int): Report step number
            
        Returns:
            WellState: Well state at the report step
        """
    
    def get_state_from_time(self, well_name: str, time: datetime) -> WellState:
        """
        Get well state at specific time.
        
        Args:
            well_name (str): Name of the well
            time (datetime): Simulation time
            
        Returns:
            WellState: Well state at the specified time
        """
    
    def load_rstfile(self, rst_file: ResdataFile):
        """Load well data from restart file."""
    
    def well_names(self) -> list:
        """
        Get list of all well names.
        
        Returns:
            list: List of well name strings
        """
    
    def get_well_time_line(self, well_name: str) -> WellTimeLine:
        """Get time line for specific well."""

Well State Analysis

Well state at specific simulation time including production/injection data and connection information.

class WellState:
    """Well state at specific simulation time."""
    
    def name(self) -> str:
        """Get well name."""
    
    def report_nr(self) -> int:
        """Get report number."""
    
    def sim_time(self) -> float:
        """Get simulation time in days."""
    
    def has_global_connections(self) -> bool:
        """Check if well has global connections."""
    
    def get_global_connections(self) -> list:
        """Get list of global connections."""
    
    def get_connections(self) -> list:
        """Get list of well connections."""
    
    def is_producer(self) -> bool:
        """Check if well is a producer."""
    
    def is_injector(self) -> bool:
        """Check if well is an injector."""
    
    def get_production_rates(self) -> dict:
        """
        Get production rates.
        
        Returns:
            dict: Production rates with keys 'oil', 'gas', 'water', 'liquid'
        """
    
    def get_injection_rates(self) -> dict:
        """
        Get injection rates.
        
        Returns:
            dict: Injection rates with keys 'water', 'gas'
        """

Well Connection Data

Individual well connection information including perforation data and flow rates.

class WellConnection:
    """Individual well connection data."""
    
    def i(self) -> int:
        """Get I-index of connection."""
    
    def j(self) -> int:
        """Get J-index of connection."""
    
    def k(self) -> int:
        """Get K-index of connection."""
    
    def is_open(self) -> bool:
        """Check if connection is open."""
    
    def is_shut(self) -> bool:
        """Check if connection is shut."""
    
    def transmissibility_factor(self) -> float:
        """Get transmissibility factor."""
    
    def diameter(self) -> float:
        """Get connection diameter."""
    
    def connection_factor(self) -> float:
        """Get connection factor."""
    
    def depth(self) -> float:
        """Get connection depth."""
    
    def volume_rate(self) -> float:
        """Get total volume rate."""
    
    def oil_rate(self) -> float:
        """Get oil production rate."""
    
    def gas_rate(self) -> float:
        """Get gas production rate."""
    
    def water_rate(self) -> float:
        """Get water production rate."""
    
    def resv_rate(self) -> float:
        """Get reservoir volume rate."""

Well Segment Operations

Well segment information for multi-segment wells including pressure drop and flow characteristics.

class WellSegment:
    """Well segment information for multi-segment wells."""
    
    def get_id(self) -> int:
        """Get segment ID."""
    
    def get_outlet(self) -> int:
        """Get outlet segment ID."""
    
    def is_active(self) -> bool:
        """Check if segment is active."""
    
    def is_closest(self) -> bool:
        """Check if segment is closest to surface."""
    
    def is_main_stem(self) -> bool:
        """Check if segment is in main stem."""
    
    def get_length(self) -> float:
        """Get segment length."""
    
    def get_depth(self) -> float:
        """Get segment depth."""
    
    def get_diameter(self) -> float:
        """Get segment diameter."""

Well Timeline Management

Time-based well operations and state changes throughout simulation.

class WellTimeLine:
    """Time line for well operations and state changes."""
    
    def get_first_date(self) -> datetime:
        """Get first operation date."""
    
    def get_last_date(self) -> datetime:
        """Get last operation date."""
    
    def time_list(self) -> list:
        """Get list of operation times."""
    
    def get_state(self, time: datetime) -> WellState:
        """Get well state at specific time."""

Usage Examples

Basic Well Analysis

from resdata.well import WellInfo
from resdata.resfile import ResdataFile

# Load restart file and extract well information
restart = ResdataFile("SIMULATION.UNRST")
well_info = WellInfo()
well_info.load_rstfile(restart)

# Get all well names
wells = well_info.well_names()
print(f"Wells in simulation: {wells}")

# Analyze each well's final state
for well_name in wells:
    try:
        # Get final state (last report step)
        final_state = well_info.get_state_from_report(well_name, -1)
        
        print(f"\nWell: {well_name}")
        print(f"  Type: {'Producer' if final_state.is_producer() else 'Injector'}")
        print(f"  Report: {final_state.report_nr()}")
        print(f"  Sim time: {final_state.sim_time():.1f} days")
        
        # Get production/injection rates
        if final_state.is_producer():
            rates = final_state.get_production_rates()
            print(f"  Oil rate: {rates.get('oil', 0):.1f} m³/day")
            print(f"  Gas rate: {rates.get('gas', 0):.1f} m³/day")
            print(f"  Water rate: {rates.get('water', 0):.1f} m³/day")
        else:
            rates = final_state.get_injection_rates()
            print(f"  Water inj: {rates.get('water', 0):.1f} m³/day")
            print(f"  Gas inj: {rates.get('gas', 0):.1f} m³/day")
            
    except Exception as e:
        print(f"Error processing well {well_name}: {e}")

Well Connection Analysis

from resdata.well import WellInfo
from resdata.resfile import ResdataFile

# Load data
restart = ResdataFile("SIMULATION.UNRST")
well_info = WellInfo()
well_info.load_rstfile(restart)

# Analyze connections for a specific well
well_name = "PROD01"
final_state = well_info.get_state_from_report(well_name, -1)

print(f"Connection analysis for {well_name}:")
print(f"Has global connections: {final_state.has_global_connections()}")

connections = final_state.get_connections()
print(f"Number of connections: {len(connections)}")

total_oil_rate = 0.0
active_connections = 0

for i, conn in enumerate(connections):
    if conn.is_open():
        print(f"\nConnection {i+1} (Active):")
        print(f"  Grid cell: ({conn.i()}, {conn.j()}, {conn.k()})")
        print(f"  Depth: {conn.depth():.1f} m")
        print(f"  Diameter: {conn.diameter():.3f} m")
        print(f"  Trans factor: {conn.transmissibility_factor():.2e}")
        print(f"  Oil rate: {conn.oil_rate():.2f} m³/day")
        print(f"  Water rate: {conn.water_rate():.2f} m³/day")
        print(f"  Gas rate: {conn.gas_rate():.2f} m³/day")
        
        total_oil_rate += conn.oil_rate()
        active_connections += 1
    else:
        print(f"\nConnection {i+1} (Shut)")

print(f"\nSummary:")
print(f"  Active connections: {active_connections}/{len(connections)}")
print(f"  Total oil rate: {total_oil_rate:.1f} m³/day")

Well Performance Over Time

from resdata.well import WellInfo
from resdata.resfile import ResdataFile
from datetime import datetime, timedelta
import matplotlib.pyplot as plt

# Load data
restart = ResdataFile("SIMULATION.UNRST")
well_info = WellInfo() 
well_info.load_rstfile(restart)

# Get well timeline
well_name = "PROD01"
timeline = well_info.get_well_time_line(well_name)

# Get operation period
start_date = timeline.get_first_date()
end_date = timeline.get_last_date()
print(f"Well {well_name} active period: {start_date} to {end_date}")

# Sample well performance monthly
performance_data = {
    'dates': [],
    'oil_rates': [],
    'water_rates': [],
    'water_cuts': []
}

current_date = start_date
while current_date <= end_date:
    try:
        state = well_info.get_state_from_time(well_name, current_date)
        if state.is_producer():
            rates = state.get_production_rates()
            oil_rate = rates.get('oil', 0)
            water_rate = rates.get('water', 0)
            
            # Calculate water cut
            total_liquid = oil_rate + water_rate
            water_cut = water_rate / total_liquid if total_liquid > 0 else 0
            
            performance_data['dates'].append(current_date)
            performance_data['oil_rates'].append(oil_rate)
            performance_data['water_rates'].append(water_rate)
            performance_data['water_cuts'].append(water_cut * 100)  # Percentage
            
    except Exception as e:
        print(f"No data for {current_date}: {e}")
    
    # Move to next month
    if current_date.month == 12:
        current_date = current_date.replace(year=current_date.year + 1, month=1)
    else:
        current_date = current_date.replace(month=current_date.month + 1)

# Plot performance
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 10))

# Production rates
ax1.plot(performance_data['dates'], performance_data['oil_rates'], 
         'g-', linewidth=2, label='Oil Rate')
ax1.plot(performance_data['dates'], performance_data['water_rates'], 
         'b-', linewidth=2, label='Water Rate')
ax1.set_ylabel('Rate (m³/day)')
ax1.set_title(f'{well_name} Production Performance')
ax1.legend()
ax1.grid(True)

# Water cut
ax2.plot(performance_data['dates'], performance_data['water_cuts'], 
         'r-', linewidth=2)
ax2.set_ylabel('Water Cut (%)')
ax2.set_xlabel('Date')
ax2.grid(True)

plt.tight_layout()
plt.show()

# Performance statistics
if performance_data['oil_rates']:
    print(f"\nPerformance Statistics:")
    print(f"  Peak oil rate: {max(performance_data['oil_rates']):.1f} m³/day")
    print(f"  Average oil rate: {sum(performance_data['oil_rates'])/len(performance_data['oil_rates']):.1f} m³/day")
    print(f"  Final water cut: {performance_data['water_cuts'][-1]:.1f}%")

Multi-Segment Well Analysis

from resdata.well import WellInfo
from resdata.resfile import ResdataFile

# Load data (assuming multi-segment well simulation)
restart = ResdataFile("SIMULATION.UNRST")
well_info = WellInfo()
well_info.load_rstfile(restart)

# Analyze multi-segment well
well_name = "MSW01"  # Multi-segment well
final_state = well_info.get_state_from_report(well_name, -1)

print(f"Multi-segment well analysis: {well_name}")

# Note: Segment analysis would require additional well segment data
# This is a simplified example showing the interface
connections = final_state.get_connections()

print(f"Well connections: {len(connections)}")
for i, conn in enumerate(connections):
    if conn.is_open():
        print(f"  Connection {i+1}: Cell ({conn.i()}, {conn.j()}, {conn.k()})")
        print(f"    Depth: {conn.depth():.1f} m")
        print(f"    Oil rate: {conn.oil_rate():.2f} m³/day")
        print(f"    Connection factor: {conn.connection_factor():.2e}")

Well State Comparison

from resdata.well import WellInfo
from resdata.resfile import ResdataFile

# Load data
restart = ResdataFile("SIMULATION.UNRST")
well_info = WellInfo()
well_info.load_rstfile(restart)

# Compare well states at different times
well_name = "PROD01"

# Early production (report step 10)
early_state = well_info.get_state_from_report(well_name, 10)

# Late production (report step -10, i.e., 10th from end)
late_state = well_info.get_state_from_report(well_name, -10)

print(f"Well state comparison: {well_name}")
print("\nEarly Production:")
print(f"  Time: {early_state.sim_time():.1f} days")
if early_state.is_producer():
    early_rates = early_state.get_production_rates()
    print(f"  Oil rate: {early_rates.get('oil', 0):.1f} m³/day")
    print(f"  Water rate: {early_rates.get('water', 0):.1f} m³/day")

print("\nLate Production:")
print(f"  Time: {late_state.sim_time():.1f} days")
if late_state.is_producer():
    late_rates = late_state.get_production_rates()
    print(f"  Oil rate: {late_rates.get('oil', 0):.1f} m³/day")
    print(f"  Water rate: {late_rates.get('water', 0):.1f} m³/day")

# Calculate decline
if early_state.is_producer() and late_state.is_producer():
    early_oil = early_rates.get('oil', 0)
    late_oil = late_rates.get('oil', 0)
    
    if early_oil > 0:
        decline_rate = (early_oil - late_oil) / early_oil * 100
        print(f"\nOil rate decline: {decline_rate:.1f}%")
        
    early_wcut = early_rates.get('water', 0) / (early_rates.get('oil', 0) + early_rates.get('water', 0) + 1e-10)
    late_wcut = late_rates.get('water', 0) / (late_rates.get('oil', 0) + late_rates.get('water', 0) + 1e-10)
    
    print(f"Water cut increase: {early_wcut*100:.1f}% → {late_wcut*100:.1f}%")

Types

# Well types
WellType = Literal[
    "UNDOCUMENTED_ZERO",  # Unknown/undefined type
    "PRODUCER",           # Production well
    "WATER_INJECTOR",     # Water injection well
    "GAS_INJECTOR",       # Gas injection well
    "OIL_INJECTOR"        # Oil injection well
]

# Connection directions
WellConnectionDirection = Literal[
    "DIR_X",      # X-direction perforation
    "DIR_Y",      # Y-direction perforation
    "DIR_Z",      # Z-direction perforation
    "DIR_FRACX",  # X-direction fracture
    "DIR_FRACY"   # Y-direction fracture
]

# Production/injection rates dictionary
ProductionRates = dict[str, float]  # Keys: 'oil', 'gas', 'water', 'liquid'
InjectionRates = dict[str, float]   # Keys: 'water', 'gas'

# Grid cell indices for connections
CellIndices = tuple[int, int, int]  # (i, j, k)

Install with Tessl CLI