metadata:
name: "bsee-production-analysis"
version: "2.0.0"
created: "2026-01-14"
author: "energy-analyst"
description: "BSEE production data analysis with NPV calculation"
input:
source:
type: "csv"
path: "data/raw/bsee_production.csv"
date_columns: ["production_date"]
parse_dates: true
filters:
- column: "field_name"
operator: "in"
values: ["JULIA", "ANCHOR", "JACK"]
- column: "production_date"
operator: ">="
value: "2020-01-01"
validation:
required_columns:
- "api_number"
- "field_name"
- "oil_bbl"
- "gas_mcf"
- "production_date"
numeric_columns: ["oil_bbl", "gas_mcf", "water_bbl"]
processing:
steps:
- name: "clean"
operation: "fillna"
columns: ["water_bbl"]
value: 0
- name: "calculate_boe"
operation: "add_column"
expression: "oil_bbl + gas_mcf / 6"
output_column: "boe"
- name: "monthly_aggregate"
operation: "resample"
date_column: "production_date"
frequency: "M"
aggregations:
oil_bbl: "sum"
gas_mcf: "sum"
boe: "sum"
- name: "npv_calculation"
operation: "npv"
cash_flow_column: "revenue"
discount_rates: [0.08, 0.10, 0.12]
periods: 20
output:
format: "html"
path: "reports/bsee_analysis_{timestamp}.html"
title: "BSEE Production Analysis"
summary:
include: true
metrics:
- "total_oil_bbl"
- "total_gas_mcf"
- "total_boe"
- "npv_results"
plots:
- type: "time_series"
title: "Monthly Production"
x: "production_date"
y: ["oil_bbl", "gas_mcf"]
- type: "bar"
title: "Production by Field"
x: "field_name"
y: "total_boe"
- type: "line"
title: "NPV Sensitivity"
x: "discount_rate"
y: "npv"
execution:
log_level: "INFO"
save_intermediate: true
intermediate_path: "data/processed/"
parallel: true
n_workers: 4
"""
ABOUTME: YAML workflow executor for standardized data pipelines
ABOUTME: Executes configuration-driven processing workflows
"""
import yaml
import logging
from pathlib import Path
from typing import Dict, Any, Optional, List
from dataclasses import dataclass
from datetime import datetime
import pandas as pd
from pydantic import BaseModel, validator
class InputConfig(BaseModel):
"""Input configuration model."""
type: str = "csv"
path: str
encoding: str = "utf-8"
date_columns: List[str] = []
parse_dates: bool = True
class ProcessingStep(BaseModel):
"""Processing step configuration."""
name: str
operation: str
columns: Optional[List[str]] = None
expression: Optional[str] = None
output_column: Optional[str] = None
class WorkflowConfig(BaseModel):
"""Complete workflow configuration."""
metadata: Dict[str, Any]
input: Dict[str, Any]
processing: Dict[str, Any]
output: Dict[str, Any]
execution: Dict[str, Any] = {}
class YAMLWorkflowExecutor:
"""Execute workflows defined in YAML configuration."""
def __init__(self, config_path: Path):
"""
Initialize executor with configuration file.
Args:
config_path: Path to YAML configuration
"""
self.config_path = Path(config_path)
self.config: Optional[WorkflowConfig] = None
self.data: Optional[pd.DataFrame] = None
self.results: Dict[str, Any] = {}
self._setup_logging()
self._load_config()
def _setup_logging(self):
"""Configure logging."""
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
self.logger = logging.getLogger(__name__)
def _load_config(self):
"""Load and validate configuration."""
if not self.config_path.exists():
raise FileNotFoundError(f"Config not found: {self.config_path}")
with open(self.config_path) as f:
raw_config = yaml.safe_load(f)
self.config = WorkflowConfig(**raw_config)
self.logger.info(f"Loaded config: {self.config.metadata.get('name')}")
def validate(self) -> bool:
"""
Validate configuration without executing.
Returns:
True if valid, False otherwise
"""
errors = []
input_path = Path(self.config.input['source']['path'])
if not input_path.exists():
errors.append(f"Input file not found: {input_path}")
output_path = Path(self.config.output['path']).parent
if not output_path.exists():
self.logger.warning(f"Output directory will be created: {output_path}")
for step in self.config.processing.get('steps', []):
if 'name' not in step:
errors.append("Processing step missing 'name'")
if 'operation' not in step:
errors.append(f"Step '{step.get('name')}' missing 'operation'")
if errors:
for error in errors:
self.logger.error(error)
return False
self.logger.info("Configuration validated successfully")
return True
def execute(self, dry_run: bool = False) -> Dict[str, Any]:
"""
Execute the workflow.
Args:
dry_run: If True, validate without executing
Returns:
Results dictionary
"""
if dry_run:
self.validate()
return {"status": "dry_run", "valid": True}
start_time = datetime.now()
self.logger.info(f"Starting workflow: {self.config.metadata.get('name')}")
try:
self._load_data()
self._execute_processing()
self._generate_output()
execution_time = (datetime.now() - start_time).total_seconds()
self.results['status'] = 'success'
self.results['execution_time'] = execution_time
self.logger.info(f"Workflow completed in {execution_time:.2f}s")
except Exception as e:
self.logger.error(f"Workflow failed: {e}")
self.results['status'] = 'failed'
self.results['error'] = str(e)
raise
return self.results
def _load_data(self):
"""Load input data based on configuration."""
source = self.config.input['source']
path = Path(source['path'])
file_type = source.get('type', 'csv')
self.logger.info(f"Loading data from: {path}")
if file_type == 'csv':
self.data = pd.read_csv(
path,
encoding=source.get('encoding', 'utf-8'),
parse_dates=source.get('date_columns', [])
)
elif file_type == 'excel':
self.data = pd.read_excel(path)
elif file_type == 'parquet':
self.data = pd.read_parquet(path)
elif file_type == 'json':
self.data = pd.read_json(path)
else:
raise ValueError(f"Unsupported file type: {file_type}")
self.logger.info(f"Loaded {len(self.data)} rows")
self.results['input_rows'] = len(self.data)
def _execute_processing(self):
"""Execute processing steps."""
steps = self.config.processing.get('steps', [])
for i, step in enumerate(steps):
self.logger.info(f"Step {i+1}/{len(steps)}: {step['name']}")
self._execute_step(step)
self.results['output_rows'] = len(self.data)
def _execute_step(self, step: Dict[str, Any]):
"""Execute a single processing step."""
operation = step['operation']
if operation == 'remove_nulls':
columns = step.get('columns', self.data.columns.tolist())
self.data = self.data.dropna(subset=columns)
elif operation == 'fillna':
columns = step.get('columns', self.data.columns.tolist())
value = step.get('value', 0)
self.data[columns] = self.data[columns].fillna(value)
elif operation == 'calculate':
expression = step['expression']
output_col = step['output_column']
self.data[output_col] = self.data.eval(expression)
elif operation == 'add_column':
expression = step['expression']
output_col = step['output_column']
self.data[output_col] = self.data.eval(expression)
elif operation == 'group_by':
by = step['by']
aggs = step['aggregations']
self.data = self.data.groupby(by).agg(aggs).reset_index()
elif operation == 'filter':
column = step['column']
op = step['operator']
value = step['value']
if op == '==':
self.data = self.data[self.data[column] == value]
elif op == '>':
self.data = self.data[self.data[column] > value]
elif op == '>=':
self.data = self.data[self.data[column] >= value]
elif op == 'in':
self.data = self.data[self.data[column].isin(value)]
elif operation == 'resample':
date_col = step['date_column']
freq = step['frequency']
aggs = step['aggregations']
self.data = self.data.set_index(date_col).resample(freq).agg(aggs).reset_index()
else:
self.logger.warning(f"Unknown operation: {operation}")
def _generate_output(self):
"""Generate output based on configuration."""
output = self.config.output
format_type = output.get('format', 'csv')
path_str = output['path']
if '{timestamp}' in path_str:
path_str = path_str.replace(
'{timestamp}',
datetime.now().strftime('%Y%m%d_%H%M%S')
)
output_path = Path(path_str)
output_path.parent.mkdir(parents=True, exist_ok=True)
if format_type == 'csv':
self.data.to_csv(output_path, index=False)
elif format_type == 'excel':
self.data.to_excel(output_path, index=False)
elif format_type == 'parquet':
self.data.to_parquet(output_path, index=False)
elif format_type == 'html':
self._generate_html_report(output_path, output)
elif format_type == 'json':
self.data.to_json(output_path, orient='records', indent=2)
self.logger.info(f"Output saved: {output_path}")
self.results['output_path'] = str(output_path)
def _generate_html_report(self, output_path: Path, config: Dict[str, Any]):
"""Generate interactive HTML report."""
import plotly.express as px
import plotly.graph_objects as go
from plotly.subplots import make_subplots
title = config.get('title', 'Analysis Report')
plots = config.get('plots', [])
if plots:
fig = make_subplots(
rows=len(plots),
cols=1,
subplot_titles=[p.get('title', f'Plot {i+1}') for i, p in enumerate(plots)]
)
for i, plot_config in enumerate(plots):
plot_type = plot_config['type']
if plot_type == 'time_series' or plot_type == 'line':
for y_col in plot_config['y']:
fig.add_trace(
go.Scatter(
x=self.data[plot_config['x']],
y=self.data[y_col],
name=y_col,
mode='lines'
),
row=i+1, col=1
)
elif plot_type == 'bar':
fig.add_trace(
go.Bar(
x=self.data[plot_config['x']],
y=self.data[plot_config['y']],
name=plot_config['y']
),
row=i+1, col=1
)
fig.update_layout(height=400 * len(plots), title_text=title)
fig.write_html(output_path, include_plotlyjs='cdn')
else:
self.data.to_html(output_path)
def run_workflow(config_path: str, dry_run: bool = False) -> Dict[str, Any]:
"""
Run workflow from configuration file.
Args:
config_path: Path to YAML configuration
dry_run: If True, validate only
Returns:
Results dictionary
"""
executor = YAMLWorkflowExecutor(Path(config_path))
return executor.execute(dry_run=dry_run)
