bsee-sodir-extraction

name	bsee-sodir-extraction
version	1.0.0
description	Extract and process energy data from BSEE (Gulf of Mexico) and SODIR (Norway) regulatory databases
author	workspace-hub
category	data-analysis
tags	["bsee","sodir","energy-data","oil-gas","offshore","web-scraping","api"]
platforms	["python"]

BSEE/SODIR Data Extraction Skill

Master data extraction from the Bureau of Safety and Environmental Enforcement (BSEE) and Norwegian Offshore Directorate (SODIR) for comprehensive offshore energy analysis.

When to Use This Skill

Use BSEE/SODIR data extraction when you need:

• Production data - Oil, gas, water production by field/well
• Well information - Directional surveys, completions, drilling data
• Field data - Reserves, operators, development status
• HSE data - Safety incidents, environmental compliance
• Economic analysis - NPV calculations using regulatory data
• Regulatory compliance - Track permits, violations, inspections

Data sources covered:

• BSEE (US Gulf of Mexico): Production, wells, platforms, safety
• SODIR (Norway): Fields, production, wells, discoveries
• NPD FactPages: Norwegian petroleum data (legacy)

Core Capabilities

1. BSEE Data Extraction

Available datasets:

• Production data (monthly oil/gas/water)
• Well data (API numbers, directional surveys)
• Platform/structure data
• Operator information
• Safety and incident data (OCS incidents)
• Environmental compliance

Base URLs:

BSEE_BASE_URLS = {
    "production": "https://www.data.bsee.gov/Production/",
    "well": "https://www.data.bsee.gov/Well/",
    "platform": "https://www.data.bsee.gov/Platform/",
    "company": "https://www.data.bsee.gov/Company/",
    "field": "https://www.data.bsee.gov/Field/",
    "incidents": "https://www.data.bsee.gov/Incidents/",
}

Production Data Extraction:

import pandas as pd
import requests
from pathlib import Path
from datetime import datetime
from typing import Optional


def fetch_bsee_production_data(
    year: int,
    output_dir: Path,
    area_code: Optional[str] = None
) -> pd.DataFrame:
    """
    Fetch BSEE production data for a given year.

    Args:
        year: Production year (e.g., 2024)
        output_dir: Directory to save downloaded data
        area_code: Optional area filter ('GC', 'MC', 'WR', etc.)

    Returns:
        DataFrame with production data
    """
    output_dir.mkdir(parents=True, exist_ok=True)

    # BSEE provides production data as downloadable files
    url = f"https://www.data.bsee.gov/Production/Files/ogoraan{year}.zip"

    # Download file
    response = requests.get(url, timeout=60)
    response.raise_for_status()

    zip_path = output_dir / f"production_{year}.zip"
    with open(zip_path, "wb") as f:
        f.write(response.content)

    # Extract and read
    import zipfile
    with zipfile.ZipFile(zip_path, "r") as z:
        z.extractall(output_dir)

    # Read the extracted CSV
    csv_files = list(output_dir.glob(f"*{year}*.csv"))
    if not csv_files:
         FileNotFoundError()

    df = pd.read_csv(csv_files[])

    
     area_code:
        df = df[df[] == area_code]

    
    df.columns = df.columns..strip()..upper()

    
    df[] = datetime.now().isoformat()
    df[] = 

    ()

     df


 () -> pd.DataFrame:
    
    
    group_cols = [, , ]

     time_period == :
        group_cols.extend([, ])
     time_period == :
        df[] = ((df[] - ) // ) + 
        group_cols.extend([, ])
    :  
        group_cols.append()

    
    agg_dict = {
        : ,
        : ,
        : ,
        :     df.columns  
    }

    
    agg_dict = {k: v  k, v  agg_dict.items()  k  df.columns}

    aggregated = df.groupby(group_cols).agg(agg_dict).reset_index()

     aggregated



production_2024 = fetch_bsee_production_data(
    year=,
    output_dir=Path(),
    area_code=  
)

field_production = aggregate_production_by_field(
    production_2024,
    time_period=
)

(field_production.head())

Well Data Extraction:

def fetch_bsee_well_data(
    api_number: Optional[str] = None,
    field_name: Optional[str] = None,
    output_dir: Path = Path("data/raw/bsee")
) -> pd.DataFrame:
    """
    Fetch BSEE well data.

    Args:
        api_number: Specific API number (14-digit)
        field_name: Filter by field name
        output_dir: Output directory

    Returns:
        DataFrame with well data
    """
    output_dir.mkdir(parents=True, exist_ok=True)

    # BSEE Well File download
    url = "https://www.data.bsee.gov/Well/Files/Well.zip"

    response = requests.get(url, timeout=120)
    response.raise_for_status()

    zip_path = output_dir / "well_data.zip"
    with open(zip_path, "wb") as f:
        f.write(response.content)

    import zipfile
    with zipfile.ZipFile(zip_path, "r") as z:
        z.extractall(output_dir)

    # Read well data
    well_file = output_dir / "Well.csv"
    df = pd.read_csv(well_file)

    # Filter if specified
    if api_number:
        df = df[df["API_WELL_NUMBER"] == api_number]
    if field_name:
        df = df[df["FIELD_NAME"].str.contains(field_name, case=False, na=False)]

    return df


def fetch_directional_surveys(
    api_number: str,
    output_dir: Path = Path()
) -> pd.DataFrame:
    
    
    url = 

    response = requests.get(url, timeout=)
    response.raise_for_status()

    zip_path = output_dir / 
     (zip_path, )  f:
        f.write(response.content)

     zipfile
     zipfile.ZipFile(zip_path, )  z:
        z.extractall(output_dir)

    survey_file = output_dir / 
    df = pd.read_csv(survey_file)

    
    df = df[df[] == api_number]

    
    df = df.sort_values()

     df



gom_wells = fetch_bsee_well_data(field_name=)
()

HSE Data Extraction:

def fetch_bsee_incident_data(
    start_year: int = 2020,
    end_year: int = 2024,
    output_dir: Path = Path("data/raw/bsee")
) -> pd.DataFrame:
    """
    Fetch BSEE incident/accident data.

    Args:
        start_year: Start year for data
        end_year: End year for data
        output_dir: Output directory

    Returns:
        DataFrame with incident records
    """
    output_dir.mkdir(parents=True, exist_ok=True)

    # Fetch incident data
    url = "https://www.data.bsee.gov/Incidents/Files/Accidents.zip"

    response = requests.get(url, timeout=120)
    response.raise_for_status()

    zip_path = output_dir / "incidents.zip"
    with open(zip_path, "wb") as f:
        f.write(response.content)

    import zipfile
    with zipfile.ZipFile(zip_path, "r") as z:
        z.extractall(output_dir)

    incident_file = output_dir / "Accidents.csv"
    df = pd.read_csv(incident_file)

    # Convert date columns
    df["INCIDENT_DATE"] = pd.to_datetime(df["INCIDENT_DATE"], errors="coerce")

    # Filter by year range
    df = df[
        (df["INCIDENT_DATE"].dt.year >= start_year) &
        (df["INCIDENT_DATE"].dt.year <= end_year)
    ]

    return df


def calculate_operator_safety_score(
    incidents_df: pd.DataFrame,
    production_df: pd.DataFrame
) -> pd.DataFrame:
    """
    Calculate safety score per operator based on incidents per production.

    Args:
        incidents_df: Incident data
        production_df: Production data

    Returns:
        DataFrame with operator safety metrics
    """
    
    incident_counts = incidents_df.groupby().agg({
        : ,
        : ,
        : 
    }).rename(columns={
        : ,
        : ,
        : 
    })

    
    production_totals = production_df.groupby().agg({
        : ,
        : 
    })

    
    safety_df = incident_counts.join(production_totals, how=).fillna()

    
    safety_df[] = safety_df[] + safety_df[] / 
    safety_df[] = (
        safety_df[] / safety_df[] * 
    )

    
    safety_df[] = (
        safety_df[] +
        safety_df[] *  +
        safety_df[] * 
    )

     safety_df.sort_values(, ascending=)



incidents = fetch_bsee_incident_data(start_year=, end_year=)
production = fetch_bsee_production_data(year=, output_dir=Path())

safety_scores = calculate_operator_safety_score(incidents, production)
()
(safety_scores.head())

2. SODIR/NPD Data Extraction (Norway)

Available datasets:

• Field production (oil, gas, NGL, condensate)
• Well data (exploration, development)
• Discoveries and prospects
• Company information
• Pipeline and infrastructure

FactPages API:

import requests
import pandas as pd
from typing import Dict, List, Optional


class SODIRDataFetcher:
    """Fetch data from SODIR (Norwegian Offshore Directorate) FactPages."""

    BASE_URL = "https://factpages.sodir.no/api/v1"

    ENDPOINTS = {
        "fields": "/fields",
        "field_production": "/field-production-yearly",
        "wells": "/wells",
        "discoveries": "/discoveries",
        "companies": "/companies",
        "pipelines": "/pipelines",
        "facilities": "/facilities",
    }

    def __init__(self):
        self.session = requests.Session()
        self.session.headers.update({
            "Accept": "application/json",
            "User-Agent": "EnergyDataAnalysis/1.0"
        })

    def _fetch(self, endpoint: str, params: Optional[Dict] = None) -> List[Dict]:
        """Fetch data from SODIR API."""
        url = f"{self.BASE_URL}{endpoint}"

        response = .session.get(url, params=params, timeout=)
        response.raise_for_status()

         response.json()

     () -> pd.DataFrame:
        
        data = ._fetch(.ENDPOINTS[])
        df = pd.DataFrame(data)
         df

     () -> pd.DataFrame:
        
        data = ._fetch(.ENDPOINTS[])
        df = pd.DataFrame(data)

        
         field_name:
            df = df[df[]..contains(field_name, =, na=)]
         start_year:
            df = df[df[] >= start_year]
         end_year:
            df = df[df[] <= end_year]

         df

     () -> pd.DataFrame:
        
        data = ._fetch(.ENDPOINTS[])
        df = pd.DataFrame(data)

         well_type:
            df = df[df[]..lower() == well_type.lower()]
         status:
            df = df[df[]..contains(status, =, na=)]

         df

     () -> pd.DataFrame:
        
        data = ._fetch(.ENDPOINTS[])
        df = pd.DataFrame(data)

         status:
            df = df[df[]..contains(status, =, na=)]

         df



sodir = SODIRDataFetcher()


fields = sodir.get_all_fields()
()


sverdrup_production = sodir.get_field_production(
    field_name=,
    start_year=
)
(sverdrup_production)


exploration_wells = sodir.get_wells(well_type=)
()

3. Combined Analysis

Cross-Basin Comparison:

import pandas as pd
import plotly.graph_objects as go
from plotly.subplots import make_subplots
from pathlib import Path


def compare_gom_norway_production(
    gom_data: pd.DataFrame,
    norway_data: pd.DataFrame,
    output_dir: Path = Path("reports")
) -> None:
    """
    Create comparative analysis of GOM vs Norway production.

    Args:
        gom_data: BSEE production data
        norway_data: SODIR production data
        output_dir: Report output directory
    """
    output_dir.mkdir(parents=True, exist_ok=True)

    # Aggregate by year
    gom_annual = gom_data.groupby("PRODUCTION_YEAR").agg({
        "OIL_BBL": "sum",
        "GAS_MCF": "sum"
    }).reset_index()
    gom_annual["REGION"] = "Gulf of Mexico"
    gom_annual["OIL_MM_BBL"] = gom_annual["OIL_BBL"] / 1e6
    gom_annual["GAS_BCF"] = gom_annual["GAS_MCF"] / 1e6

    norway_annual = norway_data.groupby("year").agg({
        "oilProduction": "sum",
        "gasProduction": "sum"
    }).reset_index()
    norway_annual.columns = ["PRODUCTION_YEAR", "OIL_MM_BBL", "GAS_BCF"]
    norway_annual["REGION"] = "Norway"

    # Create comparison chart
    fig = make_subplots(
        rows=, cols=,
        subplot_titles=[, ]
    )

    
    fig.add_trace(
        go.Bar(
            x=gom_annual[],
            y=gom_annual[],
            name=,
            marker_color=
        ),
        row=, col=
    )
    fig.add_trace(
        go.Bar(
            x=norway_annual[],
            y=norway_annual[],
            name=,
            marker_color=
        ),
        row=, col=
    )

    
    fig.add_trace(
        go.Bar(
            x=gom_annual[],
            y=gom_annual[],
            name=,
            marker_color=
        ),
        row=, col=
    )
    fig.add_trace(
        go.Bar(
            x=norway_annual[],
            y=norway_annual[],
            name=,
            marker_color=
        ),
        row=, col=
    )

    fig.update_layout(
        title=,
        barmode=,
        height=
    )

    fig.write_html(output_dir / )
    ()

4. NPV Analysis with Regulatory Data

import numpy as np
import numpy_financial as npf
import pandas as pd
from dataclasses import dataclass
from typing import List, Tuple


@dataclass
class EconomicAssumptions:
    """Economic assumptions for NPV calculation."""
    oil_price: float = 75.0      # $/bbl
    gas_price: float = 3.0       # $/mcf
    opex_per_boe: float = 15.0   # $/BOE
    capex_remaining: float = 0   # $ millions (for ongoing development)
    discount_rate: float = 0.10  # 10%
    royalty_rate: float = 0.125  # 12.5% federal royalty
    tax_rate: float = 0.21       # Corporate tax rate


def calculate_field_npv(
    production_df: pd.DataFrame,
    assumptions: EconomicAssumptions,
    forecast_years: int = 10
) -> Tuple[float, pd.DataFrame]:
    """
    Calculate NPV for a field based on BSEE production data.

    Args:
        production_df: Historical production data
        assumptions: Economic assumptions
        forecast_years: Years to forecast

    Returns:
        Tuple of (NPV, detailed cashflow DataFrame)
    """
    
    latest_year = production_df[].()
    baseline = production_df[production_df[] == latest_year]

    annual_oil = baseline[].()
    annual_gas = baseline[].()

    
    decline_rate =   

    cashflows = []

     year  (, forecast_years + ):
        
        oil_prod = annual_oil * (( - decline_rate) ** year)
        gas_prod = annual_gas * (( - decline_rate) ** year)

        
        oil_revenue = oil_prod * assumptions.oil_price
        gas_revenue = gas_prod * assumptions.gas_price
        gross_revenue = oil_revenue + gas_revenue

        
        royalties = gross_revenue * assumptions.royalty_rate
        net_revenue = gross_revenue - royalties

        
        boe_produced = oil_prod + gas_prod / 
        opex = boe_produced * assumptions.opex_per_boe

        
        ebitda = net_revenue - opex

        
        capex = assumptions.capex_remaining / forecast_years  year <=   

        
        pretax_income = ebitda - capex

        
        taxes = (, pretax_income * assumptions.tax_rate)

        
        ncf = pretax_income - taxes

        cashflows.append({
            : year,
            : oil_prod,
            : gas_prod,
            : gross_revenue / ,
            : royalties / ,
            : opex / ,
            : capex / ,
            : pretax_income / ,
            : taxes / ,
            : ncf / 
        })

    cashflow_df = pd.DataFrame(cashflows)

    
    ncf_series = [-assumptions.capex_remaining] + cashflow_df[].tolist()
    npv = npf.npv(assumptions.discount_rate, ncf_series)

     npv, cashflow_df



production = fetch_bsee_production_data(
    year=,
    output_dir=Path()
)


thunder_horse = production[
    production[]..contains(, =, na=)
]

assumptions = EconomicAssumptions(
    oil_price=,
    gas_price=,
    opex_per_boe=,
    discount_rate=
)

npv, cashflows = calculate_field_npv(thunder_horse, assumptions)

()
()
(cashflows.to_string(index=))

Complete Pipeline Example

"""
Complete BSEE/SODIR data extraction and analysis pipeline.
"""
import pandas as pd
from pathlib import Path
from datetime import datetime
import plotly.graph_objects as go


def run_extraction_pipeline(
    output_dir: Path = Path("data"),
    report_dir: Path = Path("reports")
) -> dict:
    """
    Run complete data extraction and analysis pipeline.

    Returns:
        Dictionary with extraction summary
    """
    output_dir.mkdir(parents=True, exist_ok=True)
    report_dir.mkdir(parents=True, exist_ok=True)

    results = {
        "extraction_date": datetime.now().isoformat(),
        "datasets": {}
    }

    # 1. Extract BSEE Production Data
    print("Fetching BSEE production data...")
    try:
        bsee_production = fetch_bsee_production_data(
            year=2024,
            output_dir=output_dir / "raw" / "bsee"
        )
        bsee_production.to_csv(
            output_dir / "processed" / "bsee_production.csv",
            index=False
        )
        results["datasets"]["bsee_production"] = len(bsee_production)
    except Exception as e:
        print(f"BSEE production error: {e}")
        results["datasets"][] = 

    
    ()
    :
        bsee_wells = fetch_bsee_well_data(
            output_dir=output_dir /  / 
        )
        results[][] = (bsee_wells)
     Exception  e:
        ()
        results[][] = 

    
    ()
    :
        incidents = fetch_bsee_incident_data(start_year=, end_year=)
        incidents.to_csv(
            output_dir /  / ,
            index=
        )
        results[][] = (incidents)
     Exception  e:
        ()
        results[][] = 

    
    ()
    :
        sodir = SODIRDataFetcher()
        norway_fields = sodir.get_all_fields()
        norway_production = sodir.get_field_production(start_year=)

        norway_fields.to_csv(
            output_dir /  / ,
            index=
        )
        norway_production.to_csv(
            output_dir /  / ,
            index=
        )

        results[][] = (norway_fields)
        results[][] = (norway_production)
     Exception  e:
        ()
        results[][] = 

    
    ()
    generate_summary_report(results, report_dir)

     results


 () -> :
    

    html_content = 

     dataset, count  results[].items():
        status_class =   count ==   
        status_text =   count ==   
        html_content += 

    html_content += 

    report_path = report_dir / 
     (report_path, )  f:
        f.write(html_content)

    ()



 __name__ == :
    results = run_extraction_pipeline()
    ()
    ()

Best Practices

1. Rate Limiting

import time
from functools import wraps

def rate_limit(calls_per_minute: int = 30):
    """Decorator to rate limit API calls."""
    min_interval = 60.0 / calls_per_minute
    last_call = [0.0]

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            elapsed = time.time() - last_call[0]
            if elapsed < min_interval:
                time.sleep(min_interval - elapsed)
            last_call[0] = time.time()
            return func(*args, **kwargs)
        return wrapper
    return decorator

@rate_limit(calls_per_minute=30)
def fetch_with_rate_limit(url: str) -> requests.Response:
    return requests.get(url)

2. Caching

from functools import lru_cache
from datetime import datetime, timedelta

@lru_cache(maxsize=100)
def cached_fetch(url: str, cache_hours: int = 24) -> pd.DataFrame:
    """Fetch with caching."""
    cache_file = Path(f".cache/{hash(url)}.parquet")

    if cache_file.exists():
        mtime = datetime.fromtimestamp(cache_file.stat().st_mtime)
        if datetime.now() - mtime < timedelta(hours=cache_hours):
            return pd.read_parquet(cache_file)

    # Fetch fresh data
    response = requests.get(url)
    df = pd.DataFrame(response.json())

    cache_file.parent.mkdir(exist_ok=True)
    df.to_parquet(cache_file)

    return df

3. Error Handling

import logging
from tenacity import retry, stop_after_attempt, wait_exponential

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def robust_fetch(url: str) -> requests.Response:
    """Fetch with automatic retry on failure."""
    try:
        response = requests.get(url, timeout=60)
        response.raise_for_status()
        return response
    except requests.exceptions.RequestException as e:
        logger.error(f"Fetch failed for {url}: {e}")
        raise

Resources

• BSEE Data Center: https://www.data.bsee.gov/
• SODIR FactPages: https://factpages.sodir.no/
• BSEE API Documentation: https://www.data.bsee.gov/api-documentation
• NPD (legacy): https://www.npd.no/en/facts/

---

Use this skill for all energy regulatory data extraction in worldenergydata!