Spaces:
Sleeping
Sleeping
File size: 6,032 Bytes
199bfa3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 | """
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
|