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

RFT and PLT Data Processing

Name: tessl/pypi-resdata
Author: tessl

Comprehensive processing of Repeat Formation Tester (RFT) and Production Logging Tool (PLT) data for pressure analysis, saturation profiles, and flow rate evaluation from reservoir simulation results.

Capabilities

RFT File Operations

Main interface for loading and processing RFT/PLT files containing well test and logging data.

class ResdataRFTFile:
    """RFT/PLT file operations and data access."""
    
    def __init__(self, filename: str):
        """
        Load RFT/PLT file.
        
        Args:
            filename (str): Path to RFT file (.RFT)
        """
    
    def get_header(self) -> dict:
        """Get file header information."""
    
    def get_rft(self, well_name: str, date: datetime) -> ResdataRFT:
        """
        Get RFT data for specific well and date.
        
        Args:
            well_name (str): Name of the well
            date (datetime): Date of the RFT measurement
            
        Returns:
            ResdataRFT: RFT data container
        """
    
    def get_rft_node(self, well_name: str, date: datetime):
        """Get RFT node for well and date."""
    
    def get_well_time_rft(self, well_name: str, time: datetime) -> ResdataRFT:
        """Get RFT data for well at specific time."""
    
    def get_dates(self, well_name: str) -> list:
        """
        Get list of RFT dates for well.
        
        Args:
            well_name (str): Name of the well
            
        Returns:
            list: List of datetime objects
        """
    
    def num_wells(self) -> int:
        """Get number of wells with RFT data."""
    
    def size(self) -> int:
        """Get total number of RFT records."""

RFT Data Container

Individual RFT measurement data with pressure, saturation, and depth information.

class ResdataRFT:
    """Individual RFT/PLT data container and operations."""
    
    def get_well_name(self) -> str:
        """Get well name."""
    
    def get_date(self) -> datetime:
        """Get measurement date."""
    
    def size(self) -> int:
        """Get number of measurement points."""
    
    def get_pressure(self) -> numpy.ndarray:
        """
        Get pressure measurements.
        
        Returns:
            numpy.ndarray: Pressure values in bar or psi
        """
    
    def get_depth(self) -> numpy.ndarray:
        """
        Get depth measurements.
        
        Returns:
            numpy.ndarray: Depth values in meters or feet
        """
    
    def get_swat(self) -> numpy.ndarray:
        """
        Get water saturation profile.
        
        Returns:
            numpy.ndarray: Water saturation values (0-1)
        """
    
    def get_sgas(self) -> numpy.ndarray:
        """
        Get gas saturation profile.
        
        Returns:
            numpy.ndarray: Gas saturation values (0-1)
        """
    
    def get_ijk(self) -> list:
        """
        Get grid indices for measurement points.
        
        Returns:
            list: List of (i, j, k) tuples
        """
    
    def sort(self):
        """Sort data by depth."""

RFT Cell Data

Individual cell-level RFT measurements with detailed pressure and saturation data.

class ResdataRFTCell:
    """RFT cell data representation."""
    
    def get_i(self) -> int:
        """Get I-index of cell."""
    
    def get_j(self) -> int:
        """Get J-index of cell."""
    
    def get_k(self) -> int:
        """Get K-index of cell."""
    
    def get_depth(self) -> float:
        """Get cell depth."""
    
    def get_pressure(self) -> float:
        """Get pressure measurement."""
    
    def get_swat(self) -> float:
        """Get water saturation."""
    
    def get_sgas(self) -> float:
        """Get gas saturation."""
    
    def get_soil(self) -> float:
        """Get oil saturation."""

PLT Cell Data

Production Logging Tool cell data extending RFT data with flow rate measurements.

class ResdataPLTCell(ResdataRFTCell):
    """PLT cell data with flow rate measurements."""
    
    def get_orat(self) -> float:
        """Get oil flow rate."""
    
    def get_grat(self) -> float:
        """Get gas flow rate."""
    
    def get_wrat(self) -> float:
        """Get water flow rate."""
    
    def get_connection_start(self) -> float:
        """Get connection start depth."""
    
    def get_flowrate(self) -> float:
        """Get total flow rate."""
    
    def get_oil_flowrate(self) -> float:
        """Get oil flow rate (alias for get_orat)."""
    
    def get_gas_flowrate(self) -> float:
        """Get gas flow rate (alias for get_grat)."""
    
    def get_water_flowrate(self) -> float:
        """Get water flow rate (alias for get_wrat)."""

Well Trajectory

Well trajectory information for spatial analysis and grid intersection calculations.

class WellTrajectory:
    """Well trajectory data and spatial operations."""
    
    def get_ijk(self) -> list:
        """Get grid indices along trajectory."""
    
    def get_xyz(self) -> list:
        """Get world coordinates along trajectory."""
    
    def intersect_grid(self, grid: Grid) -> list:
        """Get grid intersections for trajectory."""
    
    def branch_range(self) -> tuple:
        """Get branch range for multi-lateral wells."""

Usage Examples

Basic RFT Analysis

from resdata.rft import ResdataRFTFile
import matplotlib.pyplot as plt
import numpy as np

# Load RFT file
rft_file = ResdataRFTFile("SIMULATION.RFT")

print(f"Number of wells with RFT data: {rft_file.num_wells()}")
print(f"Total RFT records: {rft_file.size()}")

# Get RFT dates for a specific well
well_name = "PROD01"
rft_dates = rft_file.get_dates(well_name)
print(f"RFT dates for {well_name}: {len(rft_dates)} measurements")

# Analyze first RFT measurement
if rft_dates:
    first_rft = rft_file.get_rft(well_name, rft_dates[0])
    
    print(f"First RFT: {first_rft.get_date()}")
    print(f"Measurement points: {first_rft.size()}")
    
    # Get measurement data
    depths = first_rft.get_depth()
    pressures = first_rft.get_pressure()
    water_sat = first_rft.get_swat()
    
    print(f"Depth range: {depths.min():.1f} - {depths.max():.1f} m")
    print(f"Pressure range: {pressures.min():.1f} - {pressures.max():.1f} bar")
    print(f"Water saturation range: {water_sat.min():.3f} - {water_sat.max():.3f}")

Pressure Profile Analysis

from resdata.rft import ResdataRFTFile
import matplotlib.pyplot as plt
import numpy as np

# Load RFT data
rft_file = ResdataRFTFile("SIMULATION.RFT")
well_name = "PROD01"
rft_dates = rft_file.get_dates(well_name)

# Plot pressure profiles over time
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(15, 8))

colors = plt.cm.viridis(np.linspace(0, 1, len(rft_dates)))

for i, date in enumerate(rft_dates[:5]):  # First 5 measurements
    rft = rft_file.get_rft(well_name, date)
    
    depths = rft.get_depth()
    pressures = rft.get_pressure()
    water_sat = rft.get_swat()
    
    # Sort by depth
    sort_idx = np.argsort(depths)
    depths_sorted = depths[sort_idx]
    pressures_sorted = pressures[sort_idx]
    water_sat_sorted = water_sat[sort_idx]
    
    # Pressure profile
    ax1.plot(pressures_sorted, depths_sorted, 'o-', 
             color=colors[i], label=f"{date.strftime('%Y-%m-%d')}")
    
    # Water saturation profile
    ax2.plot(water_sat_sorted, depths_sorted, 'o-',
             color=colors[i], label=f"{date.strftime('%Y-%m-%d')}")

ax1.set_xlabel('Pressure (bar)')
ax1.set_ylabel('Depth (m)')
ax1.set_title(f'{well_name} - Pressure Profiles')
ax1.legend()
ax1.grid(True)
ax1.invert_yaxis()  # Depth increases downward

ax2.set_xlabel('Water Saturation')
ax2.set_ylabel('Depth (m)')
ax2.set_title(f'{well_name} - Water Saturation Profiles')
ax2.legend()
ax2.grid(True)
ax2.invert_yaxis()

plt.tight_layout()
plt.show()

RFT Time Series Analysis

from resdata.rft import ResdataRFTFile
import numpy as np
import matplotlib.pyplot as plt

# Load RFT data
rft_file = ResdataRFTFile("SIMULATION.RFT")
well_name = "PROD01"
rft_dates = rft_file.get_dates(well_name)

# Analyze pressure evolution at a specific depth
target_depth = 2500.0  # Target depth in meters
tolerance = 50.0       # Depth tolerance

pressure_evolution = []
dates_with_data = []
water_sat_evolution = []

for date in rft_dates:
    rft = rft_file.get_rft(well_name, date)
    
    depths = rft.get_depth()
    pressures = rft.get_pressure()
    water_sat = rft.get_swat()
    
    # Find measurements near target depth
    depth_mask = np.abs(depths - target_depth) <= tolerance
    
    if np.any(depth_mask):
        # Use average if multiple points found
        avg_pressure = np.mean(pressures[depth_mask])
        avg_water_sat = np.mean(water_sat[depth_mask])
        
        pressure_evolution.append(avg_pressure)
        water_sat_evolution.append(avg_water_sat)
        dates_with_data.append(date)

if pressure_evolution:
    # Plot time evolution
    fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 10))
    
    # Pressure evolution
    ax1.plot(dates_with_data, pressure_evolution, 'bo-', linewidth=2, markersize=6)
    ax1.set_ylabel('Pressure (bar)')
    ax1.set_title(f'{well_name} - Pressure Evolution at ~{target_depth:.0f}m depth')
    ax1.grid(True)
    
    # Water saturation evolution
    ax2.plot(dates_with_data, water_sat_evolution, 'ro-', linewidth=2, markersize=6)
    ax2.set_ylabel('Water Saturation')
    ax2.set_xlabel('Date')
    ax2.set_title(f'{well_name} - Water Saturation Evolution at ~{target_depth:.0f}m depth')
    ax2.grid(True)
    
    plt.tight_layout()
    plt.show()
    
    # Statistics
    pressure_decline = pressure_evolution[0] - pressure_evolution[-1]
    water_sat_increase = water_sat_evolution[-1] - water_sat_evolution[0]
    
    print(f"Pressure decline: {pressure_decline:.1f} bar")
    print(f"Water saturation increase: {water_sat_increase:.3f}")

PLT Flow Analysis

from resdata.rft import ResdataRFTFile, ResdataPLTCell

# Load RFT file (assuming it contains PLT data)
rft_file = ResdataRFTFile("SIMULATION.RFT")
well_name = "PROD01"
rft_dates = rft_file.get_dates(well_name)

# Analyze PLT data if available
for date in rft_dates:
    rft = rft_file.get_rft(well_name, date)
    
    print(f"\nPLT Analysis for {well_name} on {date.strftime('%Y-%m-%d')}:")
    
    # Get cell-level data
    for i in range(rft.size()):
        # Note: This is a simplified example
        # In practice, you'd need to access individual cell data
        # through the RFT interface
        depths = rft.get_depth()
        
        if i < len(depths):
            depth = depths[i]
            print(f"  Depth {depth:.1f}m:")
            
            # In a real implementation, you'd access PLT-specific data
            # This example shows the expected interface

Grid Integration

from resdata.rft import ResdataRFTFile, WellTrajectory
from resdata.grid import Grid

# Load grid and RFT data
grid = Grid("SIMULATION.EGRID")
rft_file = ResdataRFTFile("SIMULATION.RFT")

well_name = "PROD01"
rft_dates = rft_file.get_dates(well_name)

if rft_dates:
    rft = rft_file.get_rft(well_name, rft_dates[0])
    
    # Get grid indices for RFT measurements
    ijk_indices = rft.get_ijk()
    depths = rft.get_depth()
    pressures = rft.get_pressure()
    
    print(f"RFT measurements in grid cells:")
    for idx, (i, j, k) in enumerate(ijk_indices):
        if idx < len(depths) and idx < len(pressures):
            # Get cell properties from grid
            if grid.cell_valid(i, j, k):
                cell_depth = grid.depth(i, j, k)
                cell_volume = grid.get_cell_volume(i, j, k)
                
                print(f"  Cell ({i},{j},{k}): RFT depth={depths[idx]:.1f}m, "
                      f"grid depth={cell_depth:.1f}m, pressure={pressures[idx]:.1f}bar")

Types

# RFT measurement data arrays
PressureProfile = numpy.ndarray  # Pressure measurements
DepthProfile = numpy.ndarray     # Depth measurements  
SaturationProfile = numpy.ndarray # Saturation measurements
FlowRateProfile = numpy.ndarray   # Flow rate measurements

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

# Measurement metadata
RFTMetadata = dict[str, any]  # RFT header information

Install with Tessl CLI