# Bsee Data Extractor > Extract and process BSEE (Bureau of Safety and Environmental Enforcement) data including production, WAR (Well Activity Reports), and APD (Application for Permit to Drill) data. Use for querying production data, well activities, drilling permits, completions, and workovers by API number, block, lease, or field with automatic data normalization and caching. - Skill: `vamseeachanta/bsee-data-extractor-2` (Agent Skill) - Install (CLI): `npx skillmds@latest add vamseeachanta/bsee-data-extractor-2` - Raw SKILL.md: https://api.skillmd.com/api/skills/vamseeachanta/bsee-data-extractor-2/raw - Safety review: pending (external: skill-scanner PASS, skillspector PASS) - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Integrations & APIs - Author: vamseeachanta (https://skillmd.com/u/vamseeachanta) - Updated: 2026-09-21 - Page: https://skillmd.com/skills/vamseeachanta/bsee-data-extractor-2 --- # BSEE Data Extractor Extract and process production data from BSEE (Bureau of Safety and Environmental Enforcement) databases for oil & gas analysis in the Gulf of Mexico. ## When to Use - Querying BSEE production data by API number, block, or lease - Downloading and parsing BSEE ZIP file archives - Normalizing production data across different time periods - Building production timelines for specific wells or fields - Tracking well status changes over time - Preparing data for economic analysis (NPV, decline curves) - Analyzing Well Activity Reports (WAR) for drilling and completion history - Tracking drilling operations, workovers, and sidetracks - Reviewing APD (Application for Permit to Drill) records - Calculating drilling and completion durations - Building drilling timelines for rig scheduling analysis ## Core Pattern ``` Query Parameters → BSEE API/Files → Parse → Normalize → Cache → Output ``` ### Data Types Supported | Data Type | Source | Size | Update Frequency | |-----------|--------|------|------------------| | Production | ProductionRawData.zip | ~15-50 MB | Monthly | | WAR | eWellWARRawData.zip | ~120+ MB | Weekly | | APD | APDRawData.zip | ~5-10 MB | Weekly | ## Implementation ### Data Models ```python from dataclasses import dataclass, field from datetime import date from typing import Optional, List, Dict, Any from enum import Enum import pandas as pd class ProductType(Enum): """Production fluid types.""" OIL = "oil" GAS = "gas" WATER = "water" CONDENSATE = "condensate" class WellStatus(Enum): """Well operational status.""" PRODUCING = "producing" SHUT_IN = "shut_in" TEMPORARILY_ABANDONED = "ta" PERMANENTLY_ABANDONED = "pa" DRILLING = "drilling" COMPLETING = "completing" @dataclass class WellIdentifier: """Unique well identification.""" api_number: str well_name: Optional[str] = None lease_number: Optional[str] = None area_code: Optional[str] = None block_number: Optional[str] = None @property def api_10(self) -> str: """Return 10-digit API number.""" return self.api_number[:10] if len(self.api_number) >= 10 else self.api_number @property def api_14(self) -> str: """Return full 14-digit API number.""" return self.api_number.ljust(14, '0') @dataclass class ProductionRecord: """Single month production record.""" well_id: WellIdentifier production_date: date oil_bbls: float = 0.0 gas_mcf: float = 0.0 water_bbls: float = 0.0 condensate_bbls: float = 0.0 days_on_production: int = 0 status: WellStatus = WellStatus.PRODUCING @property def oil_bopd(self) -> float: """Oil production in barrels per day.""" if self.days_on_production > 0: return self.oil_bbls / self.days_on_production return 0.0 @property def gas_mcfd(self) -> float: """Gas production in MCF per day.""" if self.days_on_production > 0: return self.gas_mcf / self.days_on_production return 0.0 @property def boe(self) -> float: """Barrels of oil equivalent (6:1 gas conversion).""" return self.oil_bbls + self.condensate_bbls + (self.gas_mcf / 6.0) @dataclass class WellProduction: """Complete production history for a well.""" well_id: WellIdentifier records: List[ProductionRecord] = field(default_factory=list) first_production: Optional[date] = None last_production: Optional[date] = None def to_dataframe(self) -> pd.DataFrame: class ActivityType(Enum): """Well activity types from WAR data.""" DRILLING = "drilling" COMPLETION = "completion" WORKOVER = "workover" SIDETRACK = "sidetrack" PLUG_ABANDON = "plug_abandon" TEMPORARY_ABANDON = "temporary_abandon" RECOMPLETION = "recompletion" STIMULATION = "stimulation" LOGGING = "logging" TESTING = "testing" class APDStatus(Enum): """APD application status.""" PENDING = "pending" APPROVED = "approved" DENIED = "denied" WITHDRAWN = "withdrawn" EXPIRED = "expired" @dataclass class WARRecord: """Well Activity Report record.""" well_id: WellIdentifier activity_type: ActivityType start_date: date end_date: Optional[date] = None spud_date: Optional[date] = None rig_name: Optional[str] = None water_depth_ft: Optional[float] = None total_depth_md: Optional[float] = None total_depth_tvd: Optional[float] = None target_formation: Optional[str] = None operator_name: Optional[str] = None status: Optional[str] = None remarks: Optional[str] = None @property def duration_days(self) -> Optional[int]: """Calculate activity duration in days.""" if self.end_date and self.start_date: return (self.end_date - self.start_date).days return None @dataclass class APDRecord: """Application for Permit to Drill record.""" well_id: WellIdentifier application_date: date approval_date: Optional[date] = None status: APDStatus = APDStatus.PENDING permit_number: Optional[str] = None proposed_spud_date: Optional[date] = None proposed_total_depth: Optional[float] = None well_type: Optional[str] = None # exploration, development, injection operator_name: Optional[str] = None surface_location: Optional[str] = None bottom_hole_location: Optional[str] = None @dataclass class WellActivity: """Complete activity history for a well.""" well_id: WellIdentifier war_records: List[WARRecord] = field(default_factory=list) apd_records: List[APDRecord] = field(default_factory=list) first_activity: Optional[date] = None last_activity: Optional[date] = None def to_war_dataframe(self) -> pd.DataFrame: """Convert WAR records to DataFrame.""" data = [] for rec in self.war_records: data.append({ 'api_number': rec.well_id.api_number, 'activity_type': rec.activity_type.value, 'start_date': rec.start_date, 'end_date': rec.end_date, 'duration_days': rec.duration_days, 'spud_date': rec.spud_date, 'rig_name': rec.rig_name, 'water_depth_ft': rec.water_depth_ft, 'total_depth_md': rec.total_depth_md, 'total_depth_tvd': rec.total_depth_tvd, 'target_formation': rec.target_formation, 'operator_name': rec.operator_name, 'status': rec.status }) return pd.DataFrame(data).sort_values('start_date') if data else pd.DataFrame() def to_apd_dataframe(self) -> pd.DataFrame: """Convert APD records to DataFrame.""" data = [] for rec in self.apd_records: data.append({ 'api_number': rec.well_id.api_number, 'application_date': rec.application_date, 'approval_date': rec.approval_date, 'status': rec.status.value, 'permit_number': rec.permit_number, 'proposed_spud_date': rec.proposed_spud_date, 'proposed_total_depth': rec.proposed_total_depth, 'well_type': rec.well_type, 'operator_name': rec.operator_name }) return pd.DataFrame(data).sort_values('application_date') if data else pd.DataFrame() @property def total_drilling_days(self) -> int: """Total days spent drilling.""" return sum(r.duration_days or 0 for r in self.war_records if r.activity_type == ActivityType.DRILLING) @property def total_completion_days(self) -> int: """Total days spent on completions.""" return sum(r.duration_days or 0 for r in self.war_records if r.activity_type == ActivityType.COMPLETION) ``` ### WAR/APD Data Models (continued) ```python def to_dataframe(self) -> pd.DataFrame: """Convert production records to DataFrame.""" data = [] for rec in self.records: data.append({ 'date': rec.production_date, 'oil_bbls': rec.oil_bbls, 'gas_mcf': rec.gas_mcf, 'water_bbls': rec.water_bbls, 'condensate_bbls': rec.condensate_bbls, 'days': rec.days_on_production, 'oil_bopd': rec.oil_bopd, 'gas_mcfd': rec.gas_mcfd, 'boe': rec.boe, 'status': rec.status.value }) return pd.DataFrame(data).sort_values('date') @property def cumulative_oil(self) -> float: """Total cumulative oil production.""" return sum(r.oil_bbls for r in self.records) @property def cumulative_gas(self) -> float: """Total cumulative gas production.""" return sum(r.gas_mcf for r in self.records) @property def cumulative_boe(self) -> float: """Total cumulative BOE.""" return sum(r.boe for r in self.records) ``` ### BSEE Data Client ```python import requests import zipfile import io from pathlib import Path from typing import Optional, List, Dict, Generator import pandas as pd import logging logger = logging.getLogger(__name__) class BSEEDataClient: """ Client for accessing BSEE production data. Supports both API access and file-based data extraction. """ # BSEE data URLs BASE_URL = "https://www.data.bsee.gov" PRODUCTION_URL = f"{BASE_URL}/Production/Files" # Column mappings for BSEE data files PRODUCTION_COLUMNS = { 'API_WELL_NUMBER': 'api_number', 'PRODUCTION_DATE': 'production_date', 'OIL': 'oil_bbls', 'GAS': 'gas_mcf', 'WATER': 'water_bbls', 'CONDENSATE': 'condensate_bbls', 'DAYS_ON_PROD': 'days_on_production', 'WELL_STAT_CD': 'status_code' } # WAR (Well Activity Report) column mappings WAR_COLUMNS = { 'API_WELL_NUMBER': 'api_number', 'ACTIVITY_TYPE': 'activity_type', 'START_DATE': 'start_date', 'END_DATE': 'end_date', 'SPUD_DATE': 'spud_date', 'RIG_NAME': 'rig_name', 'WATER_DEPTH': 'water_depth_ft', 'TOTAL_DEPTH_MD': 'total_depth_md', 'TOTAL_DEPTH_TVD': 'total_depth_tvd', 'TARGET_FORMATION': 'target_formation', 'OPERATOR_NAME': 'operator_name', 'STATUS': 'status' } # APD (Application for Permit to Drill) column mappings APD_COLUMNS = { 'API_WELL_NUMBER': 'api_number', 'APPLICATION_DATE': 'application_date', 'APPROVAL_DATE': 'approval_date', 'STATUS': 'status', 'PERMIT_NUMBER': 'permit_number', 'PROPOSED_SPUD_DATE': 'proposed_spud_date', 'PROPOSED_TD': 'proposed_total_depth', 'WELL_TYPE': 'well_type', 'OPERATOR_NAME': 'operator_name', 'SURFACE_LOCATION': 'surface_location', 'BHL_LOCATION': 'bottom_hole_location' } # Data URLs WAR_URL = f"{BASE_URL}/Well/Files/eWellWARRawData.zip" APD_URL = f"{BASE_URL}/Well/Files/APDRawData.zip" def __init__(self, cache_dir: Path = None): """ Initialize BSEE data client. Args: cache_dir: Directory for caching downloaded data """ self.cache_dir = cache_dir or Path("data/bsee_cache") self.cache_dir.mkdir(parents=True, exist_ok=True) self.session = requests.Session() def download_production_file(self, year: int, month: int = None) -> Path: """ Download BSEE production data file. Args: year: Production year month: Optional month (downloads full year if not specified) Returns: Path to downloaded file """ if month: filename = f"ogoramon{year}{month:02d}.zip" else: filename = f"ogoramon{year}.zip" cache_path = self.cache_dir / filename if cache_path.exists(): logger.info(f"Using cached file: {cache_path}") return cache_path url = f"{self.PRODUCTION_URL}/{filename}" logger.info(f"Downloading: {url}") response = self.session.get(url) response.raise_for_status() cache_path.write_bytes(response.content) logger.info(f"Saved to: {cache_path}") return cache_path def extract_production_data(self, zip_path: Path) -> pd.DataFrame: """ Extract production data from BSEE ZIP file. Args: zip_path: Path to ZIP file Returns: DataFrame with production records """ with zipfile.ZipFile(zip_path, 'r') as zf: # Find the data file (usually .txt or .csv) data_files = [f for f in zf.namelist() if f.endswith(('.txt', '.csv', '.dat'))] if not data_files: raise ValueError(f"No data file found in {zip_path}") with zf.open(data_files[0]) as f: # Try different delimiters try: df = pd.read_csv(f, delimiter='\t', low_memory=False) except: f.seek(0) df = pd.read_csv(f, delimiter=',', low_memory=False) # Rename columns to standard names df = df.rename(columns={ k: v for k, v in self.PRODUCTION_COLUMNS.items() if k in df.columns }) return df def query_by_api(self, api_number: str, start_year: int = None, end_year: int = None) -> WellProduction: """ Query production data for specific API number. Args: api_number: Well API number (10 or 14 digit) start_year: Start year for query end_year: End year for query Returns: WellProduction object with complete history """ from datetime import datetime end_year = end_year or datetime.now().year start_year = start_year or (end_year - 20) # Default 20 years well_id = WellIdentifier(api_number=api_number) records = [] for year in range(start_year, end_year + 1): try: zip_path = self.download_production_file(year) df = self.extract_production_data(zip_path) # Filter for API number (match on first 10 digits) api_10 = well_id.api_10 df_well = df[df['api_number'].astype(str).str[:10] == api_10] for _, row in df_well.iterrows(): records.append(self._row_to_record(row, well_id)) except Exception as e: logger.warning(f"Error processing year {year}: {e}") continue production = WellProduction(well_id=well_id, records=records) if records: dates = [r.production_date for r in records] production.first_production = min(dates) production.last_production = max(dates) return production def query_by_block(self, area_code: str, block_number: str, year: int) -> List[WellProduction]: """ Query all wells in a specific OCS block. Args: area_code: OCS area code (e.g., 'GC', 'WR', 'MC') block_number: Block number year: Year to query Returns: List of WellProduction objects """ zip_path = self.download_production_file(year) df = self.extract_production_data(zip_path) # Filter by area and block (column names may vary) area_col = next((c for c in df.columns if 'AREA' in c.upper()), None) block_col = next((c for c in df.columns if 'BLOCK' in c.upper()), None) if area_col and block_col: df_block = df[ (df[area_col].astype(str).str.upper() == area_code.upper()) & (df[block_col].astype(str) == str(block_number)) ] else: logger.warning("Could not find area/block columns") return [] # Group by API number wells = {} for _, row in df_block.iterrows(): api = str(row.get('api_number', '')) if api not in wells: well_id = WellIdentifier( api_number=api, area_code=area_code, block_number=block_number ) wells[api] = WellProduction(well_id=well_id, records=[]) wells[api].records.append(self._row_to_record(row, wells[api].well_id)) return list(wells.values()) def _row_to_record(self, row: pd.Series, well_id: WellIdentifier) -> ProductionRecord: """Convert DataFrame row to ProductionRecord.""" prod_date = row.get('production_date') if isinstance(prod_date, str): # Parse various date formats for fmt in ['%Y%m', '%Y-%m', '%Y/%m', '%Y%m%d', '%Y-%m-%d']: try: prod_date = pd.to_datetime(prod_date, format=fmt).date() break except: continue elif hasattr(prod_date, 'date'): prod_date = prod_date.date() else: prod_date = date.today() return ProductionRecord( well_id=well_id, production_date=prod_date, oil_bbls=float(row.get('oil_bbls', 0) or 0), gas_mcf=float(row.get('gas_mcf', 0) or 0), water_bbls=float(row.get('water_bbls', 0) or 0), condensate_bbls=float(row.get('condensate_bbls', 0) or 0), days_on_production=int(row.get('days_on_production', 0) or 0), status=self._parse_status(row.get('status_code', '')) ) def _parse_status(self, code: str) -> WellStatus: """Parse BSEE status code to WellStatus enum.""" code = str(code).upper().strip() status_map = { 'P': WellStatus.PRODUCING, 'SI': WellStatus.SHUT_IN, 'TA': WellStatus.TEMPORARILY_ABANDONED, 'PA': WellStatus.PERMANENTLY_ABANDONED, 'DR': WellStatus.DRILLING, 'CO': WellStatus.COMPLETING } return status_map.get(code, WellStatus.PRODUCING) # ==================== WAR Data Methods ==================== def download_war_file(self, use_cache: bool = True) -> Path: """ Download BSEE Well Activity Report (WAR) data file. Args: use_cache: Use cached file if available Returns: Path to downloaded ZIP file """ cache_path = self.cache_dir / "eWellWARRawData.zip" if use_cache and cache_path.exists(): logger.info(f"Using cached WAR file: {cache_path}") return cache_path logger.info(f"Downloading WAR data from: {self.WAR_URL}") response = self.session.get(self.WAR_URL, timeout=2400) # 40 min timeout response.raise_for_status() cache_path.write_bytes(response.content) logger.info(f"Saved WAR data: {len(response.content) / (1024*1024):.1f} MB") return cache_path def extract_war_data(self, zip_path: Path) -> pd.DataFrame: """ Extract WAR data from BSEE ZIP file. Args: zip_path: Path to WAR ZIP file Returns: DataFrame with WAR records """ with zipfile.ZipFile(zip_path, 'r') as zf: data_files = [f for f in zf.namelist() if f.endswith(('.txt', '.csv', '.dat'))] if not data_files: raise ValueError(f"No data file found in {zip_path}") with zf.open(data_files[0]) as f: try: df = pd.read_csv(f, delimiter='\t', low_memory=False) except: f.seek(0) df = pd.read_csv(f, delimiter=',', low_memory=False) # Rename columns df = df.rename(columns={ k: v for k, v in self.WAR_COLUMNS.items() if k in df.columns }) return df def query_war_by_api(self, api_number: str) -> WellActivity: """ Query WAR data for specific API number. Args: api_number: Well API number (10 or 14 digit) Returns: WellActivity object with WAR records """ zip_path = self.download_war_file() df = self.extract_war_data(zip_path) well_id = WellIdentifier(api_number=api_number) api_10 = well_id.api_10 df_well = df[df['api_number'].astype(str).str[:10] == api_10] war_records = [] for _, row in df_well.iterrows(): war_records.append(self._row_to_war_record(row, well_id)) activity = WellActivity(well_id=well_id, war_records=war_records) if war_records: dates = [r.start_date for r in war_records if r.start_date] if dates: activity.first_activity = min(dates) activity.last_activity = max(dates) return activity def query_war_by_block(self, area_code: str, block_number: str) -> List[WellActivity]: """ Query WAR data for all wells in a block. Args: area_code: OCS area code (e.g., 'GC', 'WR', 'MC') block_number: Block number Returns: List of WellActivity objects """ zip_path = self.download_war_file() df = self.extract_war_data(zip_path) area_col = next((c for c in df.columns if 'AREA' in c.upper()), None) block_col = next((c for c in df.columns if 'BLOCK' in c.upper()), None) if area_col and block_col: df_block = df[ (df[area_col].astype(str).str.upper() == area_code.upper()) & (df[block_col].astype(str) == str(block_number)) ] else: logger.warning("Could not find area/block columns in WAR data") return [] wells = {} for _, row in df_block.iterrows(): api = str(row.get('api_number', '')) if api not in wells: well_id = WellIdentifier( api_number=api, area_code=area_code, block_number=block_number ) wells[api] = WellActivity(well_id=well_id) wells[api].war_records.append(self._row_to_war_record(row, wells[api].well_id)) return list(wells.values()) def _row_to_war_record(self, row: pd.Series, well_id: WellIdentifier) -> WARRecord: """Convert DataFrame row to WARRecord.""" def parse_date(val): if pd.isna(val): return None if isinstance(val, str): for fmt in ['%Y-%m-%d', '%Y%m%d', '%m/%d/%Y']: try: return pd.to_datetime(val, format=fmt).date() except: continue elif hasattr(val, 'date'): return val.date() return None activity_map = { 'DRILL': ActivityType.DRILLING, 'COMP': ActivityType.COMPLETION, 'WORK': ActivityType.WORKOVER, 'SIDE': ActivityType.SIDETRACK, 'PA': ActivityType.PLUG_ABANDON, 'TA': ActivityType.TEMPORARY_ABANDON, 'RECOMP': ActivityType.RECOMPLETION, 'STIM': ActivityType.STIMULATION, 'LOG': ActivityType.LOGGING, 'TEST': ActivityType.TESTING } activity_str = str(row.get('activity_type', '')).upper() activity_type = ActivityType.DRILLING for key, val in activity_map.items(): if key in activity_str: activity_type = val break return WARRecord( well_id=well_id, activity_type=activity_type, start_date=parse_date(row.get('start_date')) or date.today(), end_date=parse_date(row.get('end_date')), spud_date=parse_date(row.get('spud_date')), rig_name=row.get('rig_name'), water_depth_ft=float(row.get('water_depth_ft', 0) or 0) if pd.notna(row.get('water_depth_ft')) else None, total_depth_md=float(row.get('total_depth_md', 0) or 0) if pd.notna(row.get('total_depth_md')) else None, total_depth_tvd=float(row.get('total_depth_tvd', 0) or 0) if pd.notna(row.get('total_depth_tvd')) else None, target_formation=row.get('target_formation'), operator_name=row.get('operator_name'), status=row.get('status') ) # ==================== APD Data Methods ==================== def download_apd_file(self, use_cache: bool = True) -> Path: """ Download BSEE APD (Application for Permit to Drill) data file. Args: use_cache: Use cached file if available Returns: Path to downloaded ZIP file """ cache_path = self.cache_dir / "APDRawData.zip" if use_cache and cache_path.exists(): logger.info(f"Using cached APD file: {cache_path}") return cache_path logger.info(f"Downloading APD data from: {self.APD_URL}") response = self.session.get(self.APD_URL, timeout=600) # 10 min timeout response.raise_for_status() cache_path.write_bytes(response.content) logger.info(f"Saved APD data: {len(response.content) / (1024*1024):.1f} MB") return cache_path def extract_apd_data(self, zip_path: Path) -> pd.DataFrame: """ Extract APD data from BSEE ZIP file. Args: zip_path: Path to APD ZIP file Returns: DataFrame with APD records """ with zipfile.ZipFile(zip_path, 'r') as zf: data_files = [f for f in zf.namelist() if f.endswith(('.txt', '.csv', '.dat'))] if not data_files: raise ValueError(f"No data file found in {zip_path}") with zf.open(data_files[0]) as f: try: df = pd.read_csv(f, delimiter='\t', low_memory=False) except: f.seek(0) df = pd.read_csv(f, delimiter=',', low_memory=False) df = df.rename(columns={ k: v for k, v in self.APD_COLUMNS.items() if k in df.columns }) return df def query_apd_by_api(self, api_number: str) -> List[APDRecord]: """ Query APD records for specific API number. Args: api_number: Well API number Returns: List of APDRecord objects """ zip_path = self.download_apd_file() df = self.extract_apd_data(zip_path) well_id = WellIdentifier(api_number=api_number) api_10 = well_id.api_10 df_well = df[df['api_number'].astype(str).str[:10] == api_10] records = [] for _, row in df_well.iterrows(): records.append(self._row_to_apd_record(row, well_id)) return records def _row_to_apd_record(self, row: pd.Series, well_id: WellIdentifier) -> APDRecord: """Convert DataFrame row to APDRecord.""" def parse_date(val): if pd.isna(val): return None if isinstance(val, str): for fmt in ['%Y-%m-%d', '%Y%m%d', '%m/%d/%Y']: try: return pd.to_datetime(val, format=fmt).date() except: continue elif hasattr(val, 'date'): return val.date() return None status_map = { 'PEND': APDStatus.PENDING, 'APPR': APDStatus.APPROVED, 'DENY': APDStatus.DENIED, 'DENY': APDStatus.DENIED, 'WITH': APDStatus.WITHDRAWN, 'EXP': APDStatus.EXPIRED } status_str = str(row.get('status', '')).upper() status = APDStatus.PENDING for key, val in status_map.items(): if key in status_str: status = val break return APDRecord( well_id=well_id, application_date=parse_date(row.get('application_date')) or date.today(), approval_date=parse_date(row.get('approval_date')), status=status, permit_number=row.get('permit_number'), proposed_spud_date=parse_date(row.get('proposed_spud_date')), proposed_total_depth=float(row.get('proposed_total_depth', 0) or 0) if pd.notna(row.get('proposed_total_depth')) else None, well_type=row.get('well_type'), operator_name=row.get('operator_name'), surface_location=row.get('surface_location'), bottom_hole_location=row.get('bottom_hole_location') ) ``` ### Production Aggregator ```python from typing import Dict, List, Tuple import pandas as pd import numpy as np from datetime import date class ProductionAggregator: """ Aggregate production data across wells, fields, or time periods. """ def __init__(self, wells: List[WellProduction]): """ Initialize aggregator with well production data. Args: wells: List of WellProduction objects """ self.wells = wells self._combined_df = None @property def combined_dataframe(self) -> pd.DataFrame: """Get combined production data as DataFrame.""" if self._combined_df is None: dfs = [] for well in self.wells: df = well.to_dataframe() df['api_number'] = well.well_id.api_number df['well_name'] = well.well_id.well_name dfs.append(df) self._combined_df = pd.concat(dfs, ignore_index=True) return self._combined_df def monthly_totals(self) -> pd.DataFrame: """Aggregate production by month.""" df = self.combined_dataframe.copy() df['year_month'] = pd.to_datetime(df['date']).dt.to_period('M') return df.groupby('year_month').agg({ 'oil_bbls': 'sum', 'gas_mcf': 'sum', 'water_bbls': 'sum', 'boe': 'sum', 'days': 'sum', 'api_number': 'nunique' }).rename(columns={'api_number': 'well_count'}).reset_index() def annual_totals(self) -> pd.DataFrame: """Aggregate production by year.""" df = self.combined_dataframe.copy() df['year'] = pd.to_datetime(df['date']).dt.year return df.groupby('year').agg({ 'oil_bbls': 'sum', 'gas_mcf': 'sum', 'water_bbls': 'sum', 'boe': 'sum', 'days': 'sum', 'api_number': 'nunique' }).rename(columns={'api_number': 'well_count'}).reset_index() def well_summary(self) -> pd.DataFrame: """Get summary statistics per well.""" summaries = [] for well in self.wells: df = well.to_dataframe() if len(df) == 0: continue summary = { 'api_number': well.well_id.api_number, 'well_name': well.well_id.well_name, 'first_production': df['date'].min(), 'last_production': df['date'].max(), 'months_producing': len(df), 'cumulative_oil': df['oil_bbls'].sum(), 'cumulative_gas': df['gas_mcf'].sum(), 'cumulative_boe': df['boe'].sum(), 'peak_oil_bopd': df['oil_bopd'].max(), 'peak_gas_mcfd': df['gas_mcfd'].max(), 'avg_oil_bopd': df['oil_bopd'].mean(), 'avg_gas_mcfd': df['gas_mcfd'].mean() } summaries.append(summary) return pd.DataFrame(summaries) def decline_curve_data(self, well: WellProduction) -> Dict: """ Prepare data for decline curve analysis. Returns time on production and rate data. """ df = well.to_dataframe() df = df[df['oil_bopd'] > 0].sort_values('date') if len(df) == 0: return {'time': [], 'rate': []} # Calculate months from first production first_date = df['date'].iloc[0] df['months'] = ((pd.to_datetime(df['date']) - pd.to_datetime(first_date)) .dt.days / 30.44).astype(int) return { 'time': df['months'].tolist(), 'rate': df['oil_bopd'].tolist(), 'dates': df['date'].tolist() } ``` ### Activity Aggregator ```python from typing import Dict, List import pandas as pd class ActivityAggregator: """ Aggregate WAR and APD data across wells, fields, or time periods. """ def __init__(self, activities: List[WellActivity]): """ Initialize aggregator with well activity data. Args: activities: List of WellActivity objects """ self.activities = activities self._combined_war_df = None self._combined_apd_df = None @property def combined_war_dataframe(self) -> pd.DataFrame: """Get combined WAR data as DataFrame.""" if self._combined_war_df is None: dfs = [] for activity in self.activities: df = activity.to_war_dataframe() if not df.empty: df['area_code'] = activity.well_id.area_code df['block_number'] = activity.well_id.block_number dfs.append(df) self._combined_war_df = pd.concat(dfs, ignore_index=True) if dfs else pd.DataFrame() return self._combined_war_df def activity_summary(self) -> pd.DataFrame: """Summarize activities by type.""" df = self.combined_war_dataframe if df.empty: return pd.DataFrame() return df.groupby('activity_type').agg({ 'api_number': 'count', 'duration_days': ['sum', 'mean', 'min', 'max'] }).round(1) def drilling_timeline(self) -> pd.DataFrame: """Get drilling activity timeline.""" df = self.combined_war_dataframe if df.empty: return pd.DataFrame() drilling = df[df['activity_type'] == 'drilling'].copy() drilling = drilling.sort_values('start_date') return drilling[['api_number', 'start_date', 'end_date', 'duration_days', 'rig_name', 'water_depth_ft', 'total_depth_md', 'operator_name']] def rig_utilization(self) -> pd.DataFrame: """Analyze rig utilization from WAR data.""" df = self.combined_war_dataframe if df.empty: return pd.DataFrame() rig_summary = df.groupby('rig_name').agg({ 'api_number': 'nunique', 'duration_days': 'sum', 'start_date': 'min', 'end_date': 'max' }).rename(columns={ 'api_number': 'wells_drilled', 'duration_days': 'total_days' }) return rig_summary.sort_values('total_days', ascending=False) def operator_activity(self) -> pd.DataFrame: """Analyze activity by operator.""" df = self.combined_war_dataframe if df.empty: return pd.DataFrame() return df.groupby('operator_name').agg({ 'api_number': 'nunique', 'activity_type': 'count', 'duration_days': 'sum', 'water_depth_ft': 'mean', 'total_depth_md': 'mean' }).rename(columns={ 'api_number': 'unique_wells', 'activity_type': 'total_activities', 'duration_days': 'total_days', 'water_depth_ft': 'avg_water_depth', 'total_depth_md': 'avg_total_depth' }).round(0) def depth_statistics(self) -> Dict: """Calculate depth statistics from drilling data.""" df = self.combined_war_dataframe drilling = df[df['activity_type'] == 'drilling'] if drilling.empty: return {} return { 'water_depth': { 'min': drilling['water_depth_ft'].min(), 'max': drilling['water_depth_ft'].max(), 'mean': drilling['water_depth_ft'].mean(), 'median': drilling['water_depth_ft'].median() }, 'total_depth_md': { 'min': drilling['total_depth_md'].min(), 'max': drilling['total_depth_md'].max(), 'mean': drilling['total_depth_md'].mean(), 'median': drilling['total_depth_md'].median() } } ``` ### Report Generator ```python import plotly.graph_objects as go from plotly.subplots import make_subplots from pathlib import Path class BSEEReportGenerator: """Generate interactive HTML reports for BSEE data.""" def __init__(self, aggregator: ProductionAggregator): """ Initialize report generator. Args: aggregator: ProductionAggregator with well data """ self.aggregator = aggregator def generate_field_report(self, output_path: Path, field_name: str = "Field"): """ Generate comprehensive field production report. Args: output_path: Path for HTML output field_name: Name of the field for report title """ fig = make_subplots( rows=3, cols=2, subplot_titles=( 'Monthly Oil Production', 'Monthly Gas Production', 'Cumulative Production', 'Well Count Over Time', 'Water Cut Trend', 'GOR Trend' ), vertical_spacing=0.08 ) monthly = self.aggregator.monthly_totals() monthly['date'] = monthly['year_month'].dt.to_timestamp() # Monthly oil production fig.add_trace( go.Bar(x=monthly['date'], y=monthly['oil_bbls']/1000, name='Oil (Mbbls)', marker_color='green'), row=1, col=1 ) # Monthly gas production fig.add_trace( go.Bar(x=monthly['date'], y=monthly['gas_mcf']/1000, name='Gas (MMCF)', marker_color='red'), row=1, col=2 ) # Cumulative production monthly['cum_oil'] = monthly['oil_bbls'].cumsum() / 1e6 monthly['cum_gas'] = monthly['gas_mcf'].cumsum() / 1e6 fig.add_trace( go.Scatter(x=monthly['date'], y=monthly['cum_oil'], name='Cum Oil (MMbbls)', line=dict(color='green')), row=2, col=1 ) fig.add_trace( go.Scatter(x=monthly['date'], y=monthly['cum_gas'], name='Cum Gas (BCF)', line=dict(color='red')), row=2, col=1 ) # Well count fig.add_trace( go.Scatter(x=monthly['date'], y=monthly['well_count'], name='Active Wells', fill='tozeroy'), row=2, col=2 ) # Water cut df = self.aggregator.combined_dataframe.copy() df['date'] = pd.to_datetime(df['date']) monthly_avg = df.groupby(df['date'].dt.to_period('M')).agg({ 'oil_bbls': 'sum', 'water_bbls': 'sum', 'gas_mcf': 'sum' }).reset_index() monthly_avg['date'] = monthly_avg['date'].dt.to_timestamp() monthly_avg['water_cut'] = (monthly_avg['water_bbls'] / (monthly_avg['oil_bbls'] + monthly_avg['water_bbls'])) * 100 monthly_avg['gor'] = mon …(truncated)