"""固定取数与预处理的少量回归：类型、分页、窗口和首次Agent调用前的数据。""" from datetime import datetime, timedelta import json from types import SimpleNamespace import pytest from db import main_fan_store from extensions.main_fan import discovery, gateway_queries, history, history_processing from extensions.main_fan.batch import BatchRunner from extensions.main_fan.batch_context import current_scope from runtime.python_sandbox import workspace_file from test_main_fan_batch import batch_workspace, bound def realtime_rows(mine): identity = {'deptId': mine['dept_id'], 'mineId': mine['mine_id']} safety = [{**identity, 'nodeId': f'state{fan}{index}', 'nodePlace': f'主通{fan}{stage}级', 'sensorTypeId': '1010', 'sensorName': '主通风机'} for fan in (1, 2) for index, stage in enumerate(('一', '二'), 1)] equipment = [{**identity, 'nodeId': f'flow{fan}', 'nodeTypeNo': '30101', 'nodeTypeName': name + '号风量', 'sensorUnit': 'm3/min'} for fan, name in ((1, '一'), (2, '二'))] equipment += [{**identity, 'nodeId': 'speed2', 'nodeTypeNo': '30102', 'nodeTypeName': '二号风速'}, {**identity, 'nodeId': 'equipment-state2', 'nodeTypeNo': '30007', 'nodeTypeName': '二号风机运行状态'}] return safety, equipment def test_fixed_selection_keeps_stages_and_excludes_equipment_state_and_speed(): mine = {'dept_id': '1237', 'mine_id': '610831018590'} safety, equipment = realtime_rows(mine) safety.append({'nodeId': 'vibration', 'sensorTypeId': '1022', 'nodePlace': '主通1振动'}) plan = discovery.plan_from_rows(mine, safety, equipment) assert len(plan.shafts) == 1 and plan.shafts[0].shaft_id == 'unresolved-main' points = plan.shafts[0].points assert len(points) == 6 assert {p.fan_id for p in points} == {'1号风机', '2号风机'} assert {p.node_id for p in points} == {'state11', 'state12', 'state21', 'state22', 'flow1', 'flow2'} assert all(not p.mapping_confirmed for p in points) @pytest.mark.parametrize('label,fan', [ ('风机房1号主扇一级电机&其它&其它', '1号风机'), ('风机房1号主扇二级电机&其它&其它', '1号风机'), ('风机房2号主扇一级电机&其它&其它', '2号风机'), ('风机房2号主扇二级电机&其它&其它', '2号风机'), ('主通1一级', '1号风机'), ('主通1二级', '1号风机'), ('主通2一级', '2号风机'), ('主通12二级', '12号风机'), ('主扇一级电机', ''), ('主扇二级电机', ''), ]) def test_fan_number_is_not_motor_stage(label, fan): assert discovery.fan_name({'nodePlace': label}) == fan def test_fixed_selection_rejects_other_mine_and_conflicting_duplicate(): mine = {'dept_id': '1237', 'mine_id': '610831018590'} safety, equipment = realtime_rows(mine) safety[0]['deptId'] = '1045' with pytest.raises(ValueError, match='deptId'): discovery.plan_from_rows(mine, safety, equipment) safety[0]['deptId'] = '1237' safety.append({**safety[0], 'nodePlace': '主通2一级'}) with pytest.raises(ValueError, match='冲突'): discovery.plan_from_rows(mine, safety, equipment) class FixedGateway: def __init__(self, mine): self.safety, self.equipment = realtime_rows(mine) # 测点ID本身不含中文，名称保留一级/二级。 for index, row in enumerate(self.safety): row['nodeId'] = 'state' + str(index + 1) self.calls = [] async def proxy(self, agent, path, *, params): params = dict(params) self.calls.append((path, params)) if path.endswith('getNodeRealtimeDetails') or path.endswith('selectSbNodeList'): rows = self.safety if path.endswith('getNodeRealtimeDetails') else self.equipment page, size = int(params['pageNum']), int(params['pageSize']) return {'code': 200, 'total': len(rows), 'rows': rows[(page - 1) * size:page * size]} if 'nodeIds' in params: return {'code': 200, 'data': [[{'NodeID': params['nodeIds'], 'DataTime': params['startTime'], 'RealData': 1, 'KgDesc': '开', 'StateID': 0, 'stateName': '正常'}], []]} return {'code': 200, 'data': [{'nodeId': params['nodeId'], 'sbNodeCurve': [ {'dataTime': params['startTime'], 'realData': '0.00'}]}]} @pytest.mark.asyncio async def test_fixed_discovery_gets_all_pages_without_model(batch_workspace, monkeypatch): gateway = FixedGateway(batch_workspace.mine) monkeypatch.setenv('SHANXI_GATEWAY_ENABLED', '1') monkeypatch.setenv('MAIN_FAN_REALTIME_MAX_PAGE_SIZE', '2') monkeypatch.setattr(gateway_queries, 'get_service', lambda _: gateway) with bound(batch_workspace, days=1) as (scope, identifier): plan = await discovery.discover(identifier, scope.user, scope.session) assert len(plan.shafts[0].points) == 6 and len(gateway.calls) == 4 assert [params['sensorTypeList[0]'] for path, params in gateway.calls if '/aqjk/' in path] == ['1010', '1010'] assert [params['devOfSystem'] for path, params in gateway.calls if '/zdsb/' in path] == ['30', '30'] def test_explicit_two_shaft_names_are_grouped_without_splitting_by_fan_number(): mine = {'dept_id': '1237', 'mine_id': '610831018590'} safety, equipment = realtime_rows(mine) for row in safety[:2] + equipment[:1]: row['shaftName'] = '北风井' for row in safety[2:] + equipment[1:2]: row['shaftName'] = '南风井' plan = discovery.plan_from_rows(mine, safety, equipment) assert {shaft.shaft_name for shaft in plan.shafts} == {'北风井', '南风井'} def test_preprocessing_keeps_zero_and_sampling_bounds_for_single_point_comparison(): def state(fan, old, new, by): return {'node_id': 'state' + fan, 'system': 'safety', 'role': 'fan_state', 'fan_id': fan + '号风机', 'state_intervals': [], 'state_value_semantics': '接口标签', 'state_changes': [ {'from_state': old, 'to_state': new, 'after': '2026-10-03T15:00:00+08:00', 'by': by, 'intervening_unknown_or_gap': False, 'evidence': []}]} points = [state('2', '开', '停', '2026-10-03T15:00:10+08:00'), state('1', '停', '开', '2026-10-03T15:01:00+08:00')] points += [{'node_id': 'flow' + fan, 'system': 'equipment', 'role': 'flow', 'fan_id': fan + '号风机', 'unit': 'm3/min', 'values': {}} for fan in ('1', '2')] details = {'points': points} segments = [{'key': point['node_id'], 'system': 'equipment', 'node_id': point['node_id'], 'status': 'completed', 'start_time': '2026-10-03T00:00:00+08:00', 'end_time': '2026-10-04T00:00:00+08:00'} for point in points[2:]] class Reader: def data(self, record): if record['node_id'] == 'flow1': samples = [{'time': '2026-10-03T15:10:00+08:00', 'value': 0}] else: start = datetime.fromisoformat('2026-10-03T14:40:00+08:00') samples = [{'time': (start + timedelta(seconds=i)).isoformat(), 'value': 5000} for i in range(30)] return {'series': [{'samples': samples}]} rules = {'switch': {'maximum_switch_interval_minutes': 10, 'transition_minutes': 5, 'window_minutes': 30, 'sample_interval_seconds': 60, 'minimum_completeness': .9, 'maximum_cv': .05}} compact = history_processing.prepare(details, {'segments': segments}, Reader(), rules, '/workspace/details.json') event = compact['preprocessing']['switch_candidates'][0] assert event['from_fan'] == '2号风机' and event['to_fan'] == '1号风机' assert event['sampling_localization_seconds'] == 60 and event['interval_seconds'] == 50 assert event['interval_minutes'] == pytest.approx(50 / 60) assert event['interval_status'] == 'estimated' assert event['interval_bounds_minutes'] == {'lower': 0, 'upper': 1} old = next(row for row in event['flow_context'] if row['fan_id'] == '2号风机') assert old['before_reference']['value'] == 5000 and old['before_reference']['valid'] new = next(row for row in event['flow_context'] if row['fan_id'] == '1号风机') assert new['after_reference']['value'] == 0 and new['after_reference']['valid'] assert event['flow_comparison']['relative_change_percent'] == 100 # 默认输入省略缺口字段，原观测信息保留；零值、查询失败和比较口径不变。 compact.update(cutoff='2026-10-21T09:42:02+08:00', lookback_days=40, transport={'coverage_complete': False}) compact['points'][0]['state_periods'] = [{'state': '开', 'first_observed_at': '2026-10-03T14:00:00+08:00', 'last_observed_at': '2026-10-03T15:00:00+08:00', 'observed_parts': 2}] slim = history_processing.agent_input(compact) assert 'gaps' not in slim['states'][0]['periods'][0] assert compact['points'][0]['state_periods'][0]['observed_parts'] == 2 slim_event = slim['switch_candidates'][0] assert slim_event['comparison']['change_percent'] == 100 assert slim_event['duration_minutes'] == pytest.approx(50 / 60) assert slim_event['duration_status'] == 'estimated' assert slim_event['from_stopped_at'] == '2026-10-03 15:00:10' assert slim_event['to_started_at'] == '2026-10-03 15:01:00' assert slim_event['comparison']['before_value'] == 5000 and slim_event['comparison']['after_value'] == 0 assert not {'before_mean', 'after_mean', 'before_samples', 'after_samples'} & slim_event['comparison'].keys() assert next(row for row in slim_event['flow'] if row['fan'] == '1号风机')['after'][1] == 0 assert any('日段取数失败' in note for note in slim['notes']) def timing_point(node, fan, changes): return {'node_id': node, 'label': node, 'system': 'safety', 'role': 'fan_state', 'fan_id': fan, 'state_changes': [{'from_state': old, 'to_state': new, 'after': '2026-10-03T' + after + '+08:00', 'by': '2026-10-03T' + by + '+08:00', 'intervening_unknown_or_gap': False} for old, new, after, by in changes]} def test_switch_interval_longer_than_normal_limit_is_not_lost_by_grouping(): points = [timing_point('old', '2号风机', [('开', '停', '14:59:55', '15:00:00')]), timing_point('new', '1号风机', [('停', '开', '15:14:55', '15:15:00')])] events = history_processing.switch_candidates(points, 10) assert len(events) == 1 event = events[0] assert event['from_fan'] == '2号风机' and event['to_fan'] == '1号风机' assert event['interval_minutes'] == 15 and event['interval_status'] == 'estimated' assert event['interval_bounds_minutes']['lower'] == pytest.approx(895 / 60) assert event['interval_bounds_minutes']['upper'] == pytest.approx(905 / 60) assert not event['interval_has_gaps'] def test_switch_interval_uses_final_start_after_jitter_and_last_participating_stage(): points = [timing_point('old1', '2号风机', [('开', '停', '15:00:10', '15:00:15')]), timing_point('old2', '2号风机', [('开', '停', '15:00:15', '15:00:20')]), timing_point('new1', '1号风机', [('停', '开', '15:00:00', '15:00:05')]), timing_point('new2', '1号风机', [('停', '开', '15:00:05', '15:00:10'), ('开', '停', '15:00:10', '15:00:12'), ('停', '开', '15:00:45', '15:00:50')])] event = history_processing.switch_candidates(points, 10)[0] assert event['from_stopped_at'] == '2026-10-03T15:00:20+08:00' assert event['to_started_at'] == '2026-10-03T15:00:50+08:00' assert event['interval_seconds'] == 30 and event['overlap_seconds'] == 0 assert event['interval_bounds_minutes'] == {'lower': 25 / 60, 'upper': 35 / 60} assert len(event['interval_sources']) == 4 def test_start_before_stop_retains_overlap_instead_of_negative_gap(): points = [timing_point('old', '2号风机', [('开', '停', '15:00:25', '15:00:30')]), timing_point('new', '1号风机', [('停', '开', '15:00:05', '15:00:10')])] event = history_processing.switch_candidates(points, 10)[0] assert event['interval_seconds'] == 0 and event['interval_minutes'] == 0 assert event['overlap_seconds'] == 20 assert event['interval_bounds_minutes'] == {'lower': 0, 'upper': 0} def test_original_fan_restart_cancels_pending_handover(): points = [timing_point('old', '2号风机', [('开', '停', '14:59:55', '15:00:00'), ('停', '开', '15:14:55', '15:15:00')]), timing_point('new', '1号风机', [('停', '开', '15:29:55', '15:30:00')])] assert not history_processing.switch_candidates(points, 10) def test_timing_across_unknown_samples_is_computed_but_marked_as_limited(): points = [timing_point('old', '2号风机', [('开', '停', '14:59:55', '15:00:00')]), timing_point('new', '1号风机', [('停', '开', '15:14:55', '15:15:00')])] points[1]['state_changes'][0]['intervening_unknown_or_gap'] = True event = history_processing.switch_candidates(points, 10)[0] assert event['interval_minutes'] == 15 and event['interval_has_gaps'] def test_nearest_point_uses_one_hour_target_and_never_crosses_switch(): boundary = datetime.fromisoformat('2026-10-03T15:00:00+08:00') samples = [(boundary + timedelta(minutes=offset), value) for offset, value in [(-120, 100), (-70, 200), (-61, 300), (-45, 400), (-1, 500), (1, 9999)]] result = history_processing.reference_sample(samples, boundary, 'before') assert result['value'] == 300 assert result['time'] == '2026-10-03T13:59:00+08:00' assert result['target_time'] == '2026-10-03T14:00:00+08:00' after = history_processing.reference_sample( [(boundary + timedelta(minutes=offset), 0) for offset in (-1, 2, 28, 59, 80)], boundary, 'after') assert after['valid'] and after['value'] == 0 assert after['time'] == (boundary + timedelta(minutes=59)).isoformat() missing = history_processing.reference_sample([(boundary + timedelta(minutes=1), 9000)], boundary, 'before') assert not missing['valid'] and missing['time'] is None and missing['value'] is None @pytest.mark.asyncio async def test_runner_supplies_processed_data_before_workflow_call(batch_workspace, monkeypatch): gateway = FixedGateway(batch_workspace.mine) monkeypatch.setenv('SHANXI_GATEWAY_ENABLED', '1') monkeypatch.setattr(gateway_queries, 'get_service', lambda _: gateway) monkeypatch.setattr(history, 'get_service', lambda _: gateway) calls = [] async def executor(**kwargs): scope = current_scope.get() workflow = main_fan_store.workflow(scope.active_workflow_id, scope.user, scope.session) prepared = workflow['context']['prepared_data'] data = json.loads(workspace_file(scope.user, scope.session, prepared['agent_data_path']).read_text()) assert prepared['mode'] == 'fixed_code' assert len(data['states']) == 4 and len(data['flows']) == 2 assert not {'points', 'preprocessing', 'transport', 'source_policy', 'data_preparation'} & data.keys() assert data['reports'][0]['mode'] == 'analyze' assert kwargs['workflow_id'] == scope.active_workflow_id assert 'message' not in kwargs and 'graph_config' not in kwargs calls.append(workflow['context']['task_input']['measurements']['batch']['phase']) return SimpleNamespace(status='success', run_id='test', error='', output='', usage={}, structured_output={ 'result': '# 测试报告', 'analysis_id': 'test', 'workflow_id': scope.active_workflow_id, 'platform_payload': {'taskId': 'test-task', 'judgment': {'resultCode': 'UNDETERMINED'}}}) output = batch_workspace.root / 'prepared-output' result = await BatchRunner(output, batch_workspace.pdf, executor=executor).run( {'dept_ids': ['1219'], 'cutoff': '2026-10-20 15:00:00', 'lookback_days': 1}, snapshot={'snapshot_id': 'c' * 32, 'mines': [batch_workspace.mine]}) assert result['status'] == 'completed', result assert calls == ['assessment'] assert len(gateway.calls) == 8, '两个实时列表和六个日段全部先由固定代码获取' assert (output / '1219/unresolved-main/给Agent的数据.json').is_file()