analysis/advanced_analysis.py

"""
Advanced Analysis Module - adapted from clearskies_agent_v1.1
Peak pressure detection, plateau detection, downsampling
"""
import pandas as pd
import numpy as np
from typing import Dict, List, Optional


class AdvancedAnalyzer:
    """Advanced analysis functions for cycle detail views"""

    def find_peak_pressure_blocks(
        self,
        df: pd.DataFrame,
        pressure_tag: str = 'PT130',
        threshold_bar: float = 900.0,
    ) -> Dict:
        """Find time blocks where pressure exceeded a threshold"""
        if pressure_tag not in df.columns:
            # Try long format
            pressure_data = df[df.get('tag_name', pd.Series()) == pressure_tag].copy()
            if pressure_data.empty:
                return {'error': f'No data for {pressure_tag}', 'num_blocks': 0}
            pressure_data = pressure_data.sort_values('timestamp')
            values = pressure_data['value']
            timestamps = pressure_data['timestamp']
        else:
            # Wide format
            pressure_data = df[['timestamp', pressure_tag]].dropna(subset=[pressure_tag]).copy()
            pressure_data = pressure_data.sort_values('timestamp')
            values = pressure_data[pressure_tag]
            timestamps = pressure_data['timestamp']

        above = values > threshold_bar
        if not above.any():
            return {
                'threshold': threshold_bar,
                'sensor': pressure_tag,
                'num_blocks': 0,
                'total_points_above': 0,
                'message': f'No data points exceeded {threshold_bar} bar',
            }

        # Group consecutive above-threshold points
        above_df = pd.DataFrame({'timestamp': timestamps[above].values, 'value': values[above].values})
        blocks = self._group_consecutive_timestamps(above_df)

        block_stats = []
        for block in blocks:
            mask = (above_df['timestamp'] >= block['start']) & (above_df['timestamp'] <= block['end'])
            block_data = above_df[mask]
            block_stats.append({
                'start_time': block['start'],
                'end_time': block['end'],
                'duration_minutes': (block['end'] - block['start']).total_seconds() / 60,
                'num_points': len(block_data),
                'peak_pressure': block_data['value'].max(),
                'avg_pressure': block_data['value'].mean(),
            })

        return {
            'threshold': threshold_bar,
            'sensor': pressure_tag,
            'num_blocks': len(block_stats),
            'total_points_above': int(above.sum()),
            'percentage_above': (above.sum() / len(values)) * 100,
            'blocks': block_stats,
            'overall_peak': values.max(),
        }

    def detect_constant_periods(
        self,
        df: pd.DataFrame,
        tag: str,
        tolerance: float = 0.01,
        min_duration_minutes: float = 2.0,
    ) -> List[Dict]:
        """
        Detect periods where a sensor value is constant within tolerance.

        Works with both wide-format (tag as column) and long-format data.
        """
        if tag in df.columns:
            # Wide format
            sensor_data = df[['timestamp', tag]].dropna(subset=[tag]).sort_values('timestamp')
            timestamps = sensor_data['timestamp'].values
            values = sensor_data[tag].values
        else:
            # Long format
            sensor_data = df[df.get('tag_name', pd.Series()) == tag].sort_values('timestamp')
            if sensor_data.empty:
                return []
            timestamps = sensor_data['timestamp'].values
            values = sensor_data['value'].values

        if len(values) < 2:
            return []

        constant_periods = []
        period_start = 0
        period_value = values[0]

        for i in range(1, len(values)):
            # Check if value is constant within tolerance
            denom = abs(period_value) + 1e-10
            if abs(values[i] - period_value) / denom <= tolerance:
                continue

            # Period ended — check duration
            duration_sec = (pd.Timestamp(timestamps[i - 1]) - pd.Timestamp(timestamps[period_start])).total_seconds()
            duration_min = duration_sec / 60

            if duration_min >= min_duration_minutes:
                constant_periods.append({
                    'start': pd.Timestamp(timestamps[period_start]),
                    'end': pd.Timestamp(timestamps[i - 1]),
                    'duration_minutes': duration_min,
                    'value': float(period_value),
                })

            period_start = i
            period_value = values[i]

        # Check last period
        duration_sec = (pd.Timestamp(timestamps[-1]) - pd.Timestamp(timestamps[period_start])).total_seconds()
        duration_min = duration_sec / 60
        if duration_min >= min_duration_minutes:
            constant_periods.append({
                'start': pd.Timestamp(timestamps[period_start]),
                'end': pd.Timestamp(timestamps[-1]),
                'duration_minutes': duration_min,
                'value': float(period_value),
            })

        return constant_periods

    def _group_consecutive_timestamps(self, df: pd.DataFrame, max_gap_seconds: float = 2.0) -> List[Dict]:
        """Group consecutive timestamps into blocks"""
        if df.empty:
            return []

        blocks = []
        current_start = df.iloc[0]['timestamp']
        current_end = df.iloc[0]['timestamp']

        for i in range(1, len(df)):
            t = df.iloc[i]['timestamp']
            gap = (t - current_end).total_seconds() if hasattr(t - current_end, 'total_seconds') else 0

            if gap <= max_gap_seconds:
                current_end = t
            else:
                blocks.append({'start': current_start, 'end': current_end})
                current_start = t
                current_end = t

        blocks.append({'start': current_start, 'end': current_end})
        return blocks