vent_agent_2.0/extensions/main_fan/history_processing.py

"""固定整理切换候选及前后各单点参考风量；不替Agent作最终业务结论。"""
from collections import defaultdict
from datetime import timedelta
from math import isfinite

from extensions.main_fan import equipment, history_compaction
from extensions.main_fan.gateway_queries import parse_time

VERSION = 'main-fan-fixed-preprocessing-v7-open-close-timing-1h'
REFERENCE_OFFSET_MINUTES = 60
INTERVAL_METHOD = 'last_changed_stage_observation'


def net_changes(changes):
    result = {}
    for change in changes:
        node = change['node_id']
        if node not in result:
            result[node] = {key: change[key] for key in
                ('node_id', 'fan_id', 'label', 'from_state', 'to_state', 'after', 'by')}
        else:
            result[node].update(to_state=change['to_state'], by=change['by'])
    return list(result.values())


def switch_timing(event):
    """按参与净切换的各级最后完成停/启的观测估算，保留采样边界。"""
    timing = {'from_stopped_at': None, 'to_started_at': None, 'interval_minutes': None,
        'interval_seconds': None, 'interval_status': 'unknown', 'interval_method': INTERVAL_METHOD,
        'interval_bounds_minutes': None, 'interval_has_gaps': False, 'overlap_seconds': None,
        'interval_sources': []}
    changes = event.get('state_changes', [])
    if not changes:
        return timing
    changes = sorted(changes, key=lambda row: parse_time(row['by']))
    net = net_changes(changes)
    stops = {row['node_id'] for row in net if row['fan_id'] == event['from_fan']
             and (row['from_state'], row['to_state']) == ('开', '停')}
    starts = {row['node_id'] for row in net if row['fan_id'] == event['to_fan']
              and (row['from_state'], row['to_state']) == ('停', '开')}
    if not stops or not starts:
        return timing
    # 净变化用于配对，真正的时间边界取该点最后一次实际变化，避免启动抖动污染时间。
    last = {row['node_id']: row for row in changes if row['node_id'] in stops | starts}
    stop_rows, start_rows = [last[node] for node in sorted(stops)], [last[node] for node in sorted(starts)]
    if any(parse_time(row['after']) > parse_time(row['by']) for row in stop_rows + start_rows):
        return timing
    stop_lower = max(parse_time(row['after']) for row in stop_rows)
    stop_at = max(parse_time(row['by']) for row in stop_rows)
    start_lower = max(parse_time(row['after']) for row in start_rows)
    start_at = max(parse_time(row['by']) for row in start_rows)
    signed_seconds = (start_at - stop_at).total_seconds()
    seconds = max(0.0, signed_seconds)
    sources = [{'action': action, **{key: row[key] for key in
        ('node_id', 'label', 'fan_id', 'from_state', 'to_state', 'after', 'by')},
        'intervening_unknown_or_gap': bool(row.get('intervening_unknown_or_gap'))}
        for action, rows in (('stop', stop_rows), ('start', start_rows)) for row in rows]
    timing.update(from_stopped_at=stop_at.isoformat(), to_started_at=start_at.isoformat(),
        interval_minutes=seconds / 60, interval_seconds=seconds, interval_status='estimated',
        interval_bounds_minutes={'lower': max(0.0, (start_lower - stop_at).total_seconds()) / 60,
                                'upper': max(0.0, (start_at - stop_lower).total_seconds()) / 60},
        interval_has_gaps=any(row['intervening_unknown_or_gap'] for row in sources),
        overlap_seconds=max(0.0, -signed_seconds), interval_sources=sources)
    return timing


def switch_candidates(points, maximum_minutes):
    """按净变化定位候选；跨组配对停启，判定阈值不限制可发现的间隔。"""
    changes = sorted([{'node_id': point['node_id'], 'fan_id': point.get('fan_id', ''),
        'label': point.get('label', ''), **change}
        for point in points if point['system'] == 'safety' and point['role'] == 'fan_state'
        for change in point.get('state_changes', [])], key=lambda row: parse_time(row['by']))
    groups = []
    for change in changes:
        if not groups or parse_time(change['by']) - parse_time(groups[-1][0]['by']) > timedelta(minutes=maximum_minutes):
            groups.append([])
        groups[-1].append(change)
    events, pending = [], None

    def directions(net):
        stopped = {row['fan_id'] for row in net if row['fan_id'] and row['from_state'] == '开' and row['to_state'] == '停'}
        started = {row['fan_id'] for row in net if row['fan_id'] and row['from_state'] == '停' and row['to_state'] == '开'}
        return stopped, started

    def append(group, net, stopped, started):
        lower, upper = min(row['after'] for row in group), max(row['by'] for row in group)
        event = {'event_id': 'S' + str(len(events) + 1), 'status': 'candidate',
            'from_fan': next(iter(stopped)), 'to_fan': next(iter(started)),
            'after': lower, 'by': upper,
            'sampling_localization_seconds': (parse_time(upper) - parse_time(lower)).total_seconds(),
            'machine_switch_confirmed': False, 'state_changes': group, 'net_state_changes': net,
            'limitations': ['按安监同一组相反开停变化整理候选，一级/二级仍分别保留。',
                '候选方向按各测点首末状态整理，组内短暂开停反复保留在原始变化中。',
                'after/by是候选采样定位范围，停启间隔单独按参与净切换的各级最后完成停/启的观测估算。',
                '估算保留采样边界；先启后停记重叠时间，停启空档记0，不宣称已确认整机动作时刻。']}
        event.update(switch_timing(event))
        events.append(event)

    for group in groups:
        net = net_changes(group)
        stopped, started = directions(net)
        if len(stopped) == len(started) == 1 and stopped.isdisjoint(started):
            append(group, net, stopped, started)
            pending = None
        elif len(stopped) == 1 and not started:
            if pending and pending['fan'] in stopped:
                pending['changes'].extend(group)
            else:
                pending = {'fan': next(iter(stopped)), 'changes': list(group)}
        elif len(started) == 1 and not stopped:
            if pending and pending['fan'] not in started:
                combined = pending['changes'] + group
                combined_net = net_changes(combined)
                old, new = directions(combined_net)
                if len(old) == len(new) == 1 and old.isdisjoint(new):
                    append(combined, combined_net, old, new)
            # 原机重新启动会取消之前未完成的交接，不跨过它拼接更晚的切换。
            pending = None
        elif not stopped and not started:
            if pending:
                pending['changes'].extend(group)
        else:
            pending = None
    return events


def flow_samples(point, records, reader):
    timestamps = defaultdict(set)
    for record in records:
        if record['status'] != 'completed' or (record['system'], record['node_id']) != ('equipment', point['node_id']):
            continue
        document = reader.data(record)
        series = document.get('series')
        if series is None:
            normalized, _ = equipment._curve_data(document.get('payload'), point['node_id'])
            series = normalized['series']
        for lane in series or []:
            for sample in lane['samples']:
                try:
                    at, value = parse_time(sample['time']), float(sample['value'])
                    if isfinite(value) and parse_time(record['start_time']) <= at <= parse_time(record['end_time']):
                        timestamps[at].add(value)
                except (KeyError, ValueError, TypeError, OverflowError):
                    continue
    # 同时刻冲突不择一；零值和负值均保留，参考点另行标记有效性。
    return sorted((at, next(iter(values))) for at, values in timestamps.items() if len(values) == 1)


def reference_sample(samples, boundary, side):
    """以切换前/后的指定偏移时间为目标，限对应一侧取最近的一个采样，不求均值。"""
    if side not in {'before', 'after'}:
        raise ValueError('参考风量只支持before或after')
    offset = timedelta(minutes=REFERENCE_OFFSET_MINUTES)
    target = boundary - offset if side == 'before' else boundary + offset
    eligible = [(at, value) for at, value in samples
                if (at <= boundary if side == 'before' else at >= boundary)]
    selected = min(eligible, key=lambda item: (abs(item[0] - target), item[0]), default=None)
    reasons = []
    if selected is None:
        reasons.append('missing_reference_sample')
    elif selected[1] < 0:
        reasons.append('negative_flow_values')
    return {'target_time': target.isoformat(), 'side': side,
        'time': selected[0].isoformat() if selected else None,
        'value': selected[1] if selected else None, 'valid': not reasons, 'invalid_reasons': reasons}


def prepare(details, manifest, reader, rules, details_path):
    """返回小型Agent输入，完整候选同步写入details；原始采样仅留在证据ZIP。"""
    compact = history_compaction.agent_view(details, details_path)
    switch_rules = rules['switch']
    events = switch_candidates(details['points'], switch_rules['maximum_switch_interval_minutes'])
    flows = [point for point in details['points'] if point['system'] == 'equipment' and point['role'] == 'flow']
    series = {point['node_id']: flow_samples(point, manifest['segments'], reader) for point in flows} if events else {}
    for event in events:
        lower, upper = parse_time(event['after']), parse_time(event['by'])
        contexts = []
        for point in flows:
            samples = series[point['node_id']]
            contexts.append({'node_id': point['node_id'], 'fan_id': point.get('fan_id', ''),
                'unit': point.get('unit', ''), 'mapping_confirmed': point.get('mapping_confirmed', False),
                'before_reference': reference_sample(samples, lower, 'before'),
                'after_reference': reference_sample(samples, upper, 'after')})
        event['flow_context'] = contexts
        old = [row for row in contexts if row['fan_id'] == event['from_fan']]
        new = [row for row in contexts if row['fan_id'] == event['to_fan']]
        reasons = []
        if len(old) != 1 or len(new) != 1:
            reasons.append('flow_pair_not_unique')
        else:
            before, after = old[0]['before_reference'], new[0]['after_reference']
            if not before['valid'] or not after['valid']:
                reasons.append('invalid_reference_samples')
            if before['value'] is None or before['value'] <= 0:
                reasons.append('before_value_not_positive')
            units = [row['unit'].replace('³', '3').replace('^', '').replace(' ', '').lower() for row in (old[0], new[0])]
            if units[0] not in {'m3/min', 'm3/s'} or units[0] != units[1]:
                reasons.append('flow_units_not_comparable')
        event['flow_comparison'] = {'method': f'切换前后{REFERENCE_OFFSET_MINUTES}分钟附近单点对比', 'valid': not reasons,
            'relative_change_percent': abs(after['value'] - before['value']) / before['value'] * 100 if not reasons else None,
            'invalid_reasons': reasons, 'judgment_scope': 'same_number_reference',
            'number_mapping': '按接口原编号作参考点对照，未自动互换或确认物理绑定'}
    method = {'version': VERSION, 'mode': 'fixed_code', 'switch_candidate_count': len(events),
        'switch_grouping_minutes': switch_rules['maximum_switch_interval_minutes'],
        'interval_policy': {'method': INTERVAL_METHOD,
            'pairing': '相邻可配对的停启允许跨组连接，不以正常阈值截断间隔。',
            'stage_policy': '参与本次净切换的各级取最后完成停/启的新状态观测时刻。',
            'uncertainty': '保留原变化采样边界形成间隔上下界；跨未知或缺测不作为间隔告警依据。',
            'overlap': '先启后停时空档为0，另保留观测重叠秒数。'},
        'flow_policy': {'method': 'nearest_point', 'offset_minutes': REFERENCE_OFFSET_MINUTES,
            'same_side_only': True,
            'note': '用户指定的单点对比；保留零值，不求均值，不使用分钟完整率、过渡区间或CV门槛。'},
        'note': '固定程序已完成取数、开停区间及前后参考风量整理；保留目标时间与实际采样时间。单点参考不等同于连续稳定工况。'}
    details['preprocessing'] = {**method, 'switch_candidates': events}
    compact['preprocessing'] = {**method, 'switch_candidates': events[:20],
        'switch_candidates_truncated': len(events) > 20, 'all_candidates_pointer': '/preprocessing/switch_candidates'}
    return compact


def agent_input(summary):
    """默认输入只保留业务数据；类型、指纹、质量计数和计算过程留在details_path。"""
    local = lambda value: parse_time(value).strftime('%Y-%m-%d %H:%M:%S')
    cutoff = parse_time(summary['cutoff'])
    mine = summary.get('mine', {})
    result = {'mine': mine,
        'range': [local((cutoff - timedelta(days=summary['lookback_days'])).isoformat()), local(summary['cutoff'])],
        'states': [], 'flows': [], 'switch_candidates': []}
    notes = []
    shaft = summary.get('shaft', {})
    if shaft.get('identity_status') == 'confirmed':
        result['shaft'] = shaft['shaft_name']
    elif shaft or summary.get('mine'):
        notes.append('通风井归属待确认。')
    if not summary.get('transport', {}).get('coverage_complete', True):
        notes.append('部分历史日段取数失败或未完成，不能按全时段正常处理。')
    flows = [point for point in summary['points'] if point['role'] == 'flow']
    flow_names = {}
    used = set()
    for point in flows:
        name = point.get('fan_id') or point.get('label') or point['node_id']
        if name in used:
            name += ' / ' + (point.get('label') or point['node_id'])
        if name in used:
            name += ' / ' + point['node_id']
        used.add(name)
        flow_names[point['node_id']] = name
        values = point['values']
        row = {'fan': name, 'unit': point.get('unit', ''), 'samples': values.get('numeric_sample_count', 0)}
        if values.get('minimum') is not None:
            row['range'] = [values['minimum'], values['maximum']]
        if values.get('zero_count'):
            row['zero_samples'] = values['zero_count']
        result['flows'].append(row)
        if not point.get('mapping_confirmed'):
            notes.append('两系统风机号按原编号对照，物理对应尚未确认。')
    for point in summary['points']:
        quality = point.get('quality', {})
        if quality.get('unrecognized_windows'):
            notes.append((point.get('label') or point['node_id']) + '：部分日段结构未识别，不能当作无数据。')
        if point['role'] != 'fan_state':
            continue
        periods = []
        for period in point.get('state_periods', []):
            start, end = parse_time(period['first_observed_at']), parse_time(period['last_observed_at'])
            # 给模型的区间只保留超过5秒的观测；原始状态和切换候选保留。
            if end - start <= timedelta(seconds=5):
                continue
            periods.append({'state': period['state'], 'start': start.strftime('%Y-%m-%d %H:%M:%S'),
                            'end': end.strftime('%Y-%m-%d %H:%M:%S')})
        row = {'point': point.get('label') or point['node_id'], 'periods': periods}
        if point.get('state_periods_truncated'):
            row['truncated'] = True
        if not periods:
            notes.append(row['point'] + '：没有超过5秒的有效开停区间。')
        result['states'].append(row)
    preprocessing = summary['preprocessing']
    for event in preprocessing['switch_candidates']:
        row = {'id': event['event_id'], 'from_fan': event['from_fan'], 'to_fan': event['to_fan'],
            'time_range': [local(event['after']), local(event['by'])], 'flow': []}
        nearest = lambda sample: [local(sample['time']), sample['value']] if sample.get('time') else None
        for flow in event['flow_context']:
            row['flow'].append({'fan': flow_names[flow['node_id']],
                'before': nearest(flow['before_reference']), 'after': nearest(flow['after_reference'])})
        old = [flow for flow in event['flow_context'] if flow['fan_id'] == event['from_fan']]
        new = [flow for flow in event['flow_context'] if flow['fan_id'] == event['to_fan']]
        comparison = {'method': event['flow_comparison']['method'],
            'change_percent': event['flow_comparison']['relative_change_percent']}
        if len(old) == len(new) == 1:
            before, after = old[0]['before_reference'], new[0]['after_reference']
            comparison.update(before_value=before['value'], after_value=after['value'])
        reasons = event['flow_comparison']['invalid_reasons']
        labels = {'flow_pair_not_unique': '风量测点无法唯一配对',
            'invalid_reference_samples': '前后参考点缺失或为负风量',
            'before_value_not_positive': '切换前风量不大于0或缺失，无法计算变化比例',
            'flow_units_not_comparable': '风量单位不可直接比较',
            }
        if reasons:
            comparison['reason'] = '；'.join(labels[reason] for reason in reasons)
        row['comparison'] = comparison
        # 兼容已有原始历史详情：即使原事件曾把间隔留空，也按保留的实际变化重新派生。
        timing = switch_timing(event)
        row.update(duration_minutes=timing['interval_minutes'], duration_seconds=timing['interval_seconds'],
            duration_status=timing['interval_status'], duration_method=timing['interval_method'],
            duration_bounds_minutes=timing['interval_bounds_minutes'],
            duration_has_gaps=timing['interval_has_gaps'], overlap_seconds=timing['overlap_seconds'],
            from_stopped_at=local(timing['from_stopped_at']) if timing['from_stopped_at'] else None,
            to_started_at=local(timing['to_started_at']) if timing['to_started_at'] else None,
            duration_sources=[{'action': source['action'], 'point': source['label'] or source['node_id'],
                'node_id': source['node_id'], 'previous_state_at': local(source['after']),
                'new_state_at': local(source['by']),
                'intervening_unknown_or_gap': source['intervening_unknown_or_gap']}
                for source in timing['interval_sources']])
        result['switch_candidates'].append(row)
    if result['switch_candidates']:
        notes.append('停启间隔已按各级最后完成停/启的观测估算，duration_bounds_minutes保留采样范围；'
            'time_range仍只是整个候选的定位范围，不用它相减计算间隔。先启后停的空档为0，重叠另列。')
        notes.append(f'comparison按原编号比较切换前后{REFERENCE_OFFSET_MINUTES}分钟附近最近的单点；属于参考对照，未验证连续稳定工况。')
    if preprocessing.get('switch_candidates_truncated'):
        result['switch_candidates_truncated'] = True
    if notes:
        result['notes'] = list(dict.fromkeys(notes))
    result['details_path'] = summary['details_path']
    return result