"""固定整理切换候选及前后各单点参考风量；不替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