"""历史数据的无损证据压缩与观测状态区间；不判定整机切换或合规性。""" from collections import Counter, defaultdict from contextlib import ExitStack from copy import deepcopy import hashlib import json import math from statistics import median from zipfile import ZIP_DEFLATED, ZipFile from extensions.main_fan import equipment, gateway_queries VERSION = 'main-fan-observed-intervals-v2-open-close-labels' class RecordReader: """兼容原逐段JSON和单个ZIP中的原件，不解压出数百个文件。""" def __init__(self, resolve): self.resolve = resolve self.stack = ExitStack() self.archives = {} def __enter__(self): return self def __exit__(self, *args): return self.stack.__exit__(*args) def bytes(self, record, kind='data'): if record.get('archive'): archive = record['archive'] if archive['path'] not in self.archives: self.archives[archive['path']] = self.stack.enter_context(ZipFile(self.resolve(archive['path']))) item = archive['members'][kind] content = self.archives[archive['path']].read(item['name']) expected = item['sha256'] else: virtual = next(path for path in record['files'] if path.endswith('/' + kind + '.json')) content = self.resolve(virtual).read_bytes() expected = record['files'][virtual] if hashlib.sha256(content).hexdigest() != expected: raise ValueError('历史证据指纹不一致：' + record['key'] + '/' + kind) return content def data(self, record): return json.loads(self.bytes(record)) def compact_evidence(records, archive_path, resolve): """原件逐项校验后归入一个ZIP；调用者先持久化新索引，再清理本轮暂存文件。""" completed = [row for row in records if row['status'] == 'completed'] if not completed: return records, {}, [], {} if all(row.get('archive') for row in completed) and len({row['archive']['path'] for row in completed}) == 1: files = {path: sha for row in completed for path, sha in row['files'].items()} with ZipFile(resolve(completed[0]['archive']['path'])) as archive: original_bytes = sum(item.file_size for item in archive.infolist()) # 同一批次重试不改写已经完成的压缩证据。 return records, files, [], {'archive_path': completed[0]['archive']['path'], 'compressed_bytes': resolve(completed[0]['archive']['path']).stat().st_size, 'original_file_count': len(completed) * 3, 'original_bytes': original_bytes} destination = resolve(archive_path) destination.parent.mkdir(parents=True, exist_ok=True) temporary = destination.with_name(destination.name + '.writing') result, cleanup, size = deepcopy(records), [], 0 try: with RecordReader(resolve) as reader, ZipFile(temporary, 'w', ZIP_DEFLATED, compresslevel=6) as archive: for record in result: if record['status'] != 'completed': continue members = {} for kind in ('request', 'response', 'data'): content = reader.bytes(record, kind) name = record['key'] + '/' + kind + '.json' archive.writestr(name, content) size += len(content) members[kind] = {'name': name, 'sha256': hashlib.sha256(content).hexdigest()} cleanup.extend(record['files'] if not record.get('archive') else []) record['archive'] = {'path': archive_path, 'members': members} record.pop('data_path', None) record.pop('python_path', None) with ZipFile(temporary) as archive: if archive.testzip() is not None: raise ValueError('历史压缩证据校验失败') temporary.replace(destination) finally: temporary.unlink(missing_ok=True) sha = hashlib.sha256(destination.read_bytes()).hexdigest() for record in result: if record['status'] == 'completed': record['files'] = {archive_path: sha} return result, {archive_path: sha}, cleanup, {'archive_path': archive_path, 'original_file_count': len(completed) * 3, 'original_bytes': size, 'compressed_bytes': destination.stat().st_size} def _state(sample, system): if system == 'safety': if sample.get('status_label') not in (None, '正常') or str(sample.get('status_code')) not in ('None', '', '0'): return None, 'abnormal_status' label = sample.get('value_label') if label in ('停', '关'): return '停', '' return ('开', '') if label == '开' else (None, 'unrecognized_state') try: value = float(sample['value']) if value in (0, 1): return '原始值' + str(int(value)), '' except (KeyError, TypeError, ValueError): pass return None, 'unrecognized_state' def summarize_point(point, records, reader): counts, timestamps = Counter(), defaultdict(list) for record in records: if record['status'] != 'completed': counts['failed_windows'] += 1 continue document = reader.data(record) if point['system'] == 'safety': samples = document['samples'] else: document, shape = equipment._curve_data(document['payload'], point['node_id']) if shape['sample_count'] is None: counts['unrecognized_windows'] += 1 continue samples = [sample for series in document['series'] for sample in series['samples']] counts['nonempty_windows' if samples else 'empty_windows'] += 1 for index, sample in enumerate(samples): counts['raw_sample_count'] += 1 try: at = gateway_queries.parse_time(sample['time']) except (KeyError, TypeError, ValueError): counts['invalid_time_count'] += 1 continue if not gateway_queries.parse_time(record['start_time']) <= at <= gateway_queries.parse_time(record['end_time']): counts['out_of_window_count'] += 1 continue if point['role'] == 'fan_state': value, reason = _state(sample, point['system']) else: try: value = float(sample['value']) if not math.isfinite(value): raise ValueError('非有限数值') reason = '' except (KeyError, ValueError, TypeError, OverflowError): value, reason = None, 'invalid_numeric_value' if reason: counts['invalid_value_count'] += 1 timestamps[at].append((value, reason, {'segment': record['key'], 'sample_index': index})) timeline = [] for at, items in sorted(timestamps.items()): counts['duplicate_time_count'] += len(items) - 1 variants = {(value, reason) for value, reason, _ in items} if len(variants) > 1: counts['conflicting_timestamp_count'] += 1 value, reason = None, 'conflicting_samples' else: value, reason = next(iter(variants)) timeline.append((at, value, reason, items[0][2])) deltas = [(after[0] - before[0]).total_seconds() for before, after in zip(timeline, timeline[1:])] cadence = median(deltas) if deltas else None gap_limit = 5 * (cadence if len(deltas) >= 3 else (60 if point['system'] == 'safety' else 3000)) gaps = [{'after': before[0].isoformat(), 'before': after[0].isoformat(), 'seconds': delta} for before, after, delta in zip(timeline, timeline[1:], deltas) if gap_limit and delta > gap_limit] quality = {key: counts[key] for key in ('raw_sample_count', 'nonempty_windows', 'empty_windows', 'failed_windows', 'unrecognized_windows', 'invalid_time_count', 'out_of_window_count', 'invalid_value_count', 'duplicate_time_count', 'conflicting_timestamp_count')} quality.update(unique_sample_count=len(timeline), median_sample_interval_seconds=cadence, gap_threshold_seconds=gap_limit, largest_sample_interval_seconds=max(deltas) if deltas else None, gap_count=len(gaps)) result = {**point, 'quality': quality, 'first_observed_at': timeline[0][0].isoformat() if timeline else None, 'last_observed_at': timeline[-1][0].isoformat() if timeline else None, 'sampling_gaps': gaps} if point['role'] == 'flow': numeric = [value for _, value, reason, _ in timeline if not reason] result['values'] = {'numeric_sample_count': len(numeric), 'zero_count': numeric.count(0), 'positive_count': sum(value > 0 for value in numeric), 'negative_count': sum(value < 0 for value in numeric), 'minimum': min(numeric) if numeric else None, 'maximum': max(numeric) if numeric else None, 'stable_window_comparison': 'not_computed'} return result intervals, changes = [], [] previous, last_valid, interrupted = None, None, False for at, value, reason, source in timeline: gap = bool(previous and gap_limit and (at - previous[0]).total_seconds() > gap_limit) if (intervals and previous and (value, reason) == (previous[1], previous[2]) and not gap): intervals[-1].update(observed_until=at.isoformat(), last_source=source) intervals[-1]['sample_count'] += 1 else: intervals.append({'state': value or '未知', 'observed_from': at.isoformat(), 'observed_until': at.isoformat(), 'sample_count': 1, 'reason': reason, 'first_source': source, 'last_source': source}) if not reason: if last_valid and value != last_valid[1]: changes.append({'from_state': last_valid[1], 'to_state': value, 'after': last_valid[0].isoformat(), 'by': at.isoformat(), 'uncertainty_seconds': (at - last_valid[0]).total_seconds(), 'intervening_unknown_or_gap': interrupted or gap, 'machine_switch_confirmed': False, 'evidence': [last_valid[3], source]}) last_valid, interrupted = (at, value, reason, source), False else: interrupted = True previous = (at, value, reason, source) result.update(state_intervals=intervals, state_changes=changes, state_value_semantics='开停标签来自接口；“关”归一为“停”，原标签保留在采样证据中' if point['system'] == 'safety' else '0/1保留原值，开停含义待核对') return result def summarize(manifest, reader): points = [summarize_point(point, [row for row in manifest['segments'] if (row['system'], row['node_id']) == (point['system'], point['node_id'])], reader) for point in manifest['points']] return {'schema_version': VERSION, 'cutoff': manifest['cutoff'], 'lookback_days': manifest['lookback_days'], 'transport': {key: manifest[key] for key in ('expected_segments', 'completed_segments', 'failed_segments', 'coverage_complete')}, 'interval_semantics': 'observed_from/until是首末采样时刻；不外推到查询边界或长缺测段。after/by表示变化发生在(after,by]内，不是精确切换时刻。', 'gap_policy': '大于各测点中位采样间隔5倍时分段；此阈值仅用于缺测提示，不是切换合规标准。', 'machine_mapping': '按测点和系统保留，不把一级/二级自动合成为整机，不合并跨系统状态。', 'points': points} def agent_view(details, details_path, *, full=False): """整理状态观测区间；full=True 为单次工作流展开全部已有区间和变化。""" result = {key: value for key, value in details.items() if key != 'points'} result['details_path'] = details_path result['state_period_semantics'] = ('state_periods仅归并同状态的有效观测。observed_parts>1表示中间含未知状态或长缺测，' '不是整段持续开/停；具体分段和未知区间见details_path。state_changes只给变化的采样边界。') result['points'] = [] for index, point in enumerate(details['points']): view = {key: point[key] for key in ('system', 'node_id', 'label', 'fan_id', 'role', 'unit', 'mapping_confirmed', 'quality', 'first_observed_at', 'last_observed_at') if key in point} view['details_pointer'] = '/points/' + str(index) if point['role'] == 'flow': view['values'] = point['values'] else: periods = [] for interval in point['state_intervals']: if interval['state'] == '未知': continue if periods and periods[-1]['state'] == interval['state']: periods[-1]['last_observed_at'] = interval['observed_until'] periods[-1]['sample_count'] += interval['sample_count'] periods[-1]['observed_parts'] += 1 else: periods.append({'state': interval['state'], 'first_observed_at': interval['observed_from'], 'last_observed_at': interval['observed_until'], 'sample_count': interval['sample_count'], 'observed_parts': 1}) view['state_period_count'] = len(periods) view['state_periods'] = periods if full or len(periods) <= 20 else periods[:10] + periods[-10:] view['state_periods_truncated'] = not full and len(periods) > 20 view['state_interval_count'] = len(point['state_intervals']) view['unknown_interval_count'] = sum(row['state'] == '未知' for row in point['state_intervals']) view['state_change_count'] = len(point['state_changes']) changes = point['state_changes'] view['state_changes'] = [{key: value for key, value in change.items() if key != 'evidence'} for change in (changes if full or len(changes) <= 20 else changes[:10] + changes[-10:])] view['state_changes_truncated'] = not full and len(changes) > 20 view['state_value_semantics'] = point['state_value_semantics'] result['points'].append(view) return result