from dataclasses import replace from datetime import datetime, timedelta from types import SimpleNamespace from unittest.mock import Mock import pytest import yaml from production_analytics.calculations.config import ( CalculationConfigError, load_material_calculation, ) from production_analytics.calculations.material_consumption import ( MaterialConsumptionIntegrator, MaterialIntegratorConfig, area_application_rate_kg_per_hour, ) from production_analytics.enlyze.gateway import EnlyzeApiGateway, EnlyzeProductionRun from production_analytics.service.material_context import ErpNominalWidthProvider from production_analytics.service.material_polling import MaterialPollingService from production_analytics.service.material_state_store import JsonMaterialStateStore def timestamp(value): return datetime.fromisoformat(value.replace('Z', '+00:00')) def test_area_units_and_strict_gate(): client = Mock() start = timestamp('2026-08-25T05:27:18Z') client.post_json.return_value.body = {'data': { 'columns': ['speed', 's2', 'time', 's1'], 'records': [[speed, 2000, (start + timedelta(minutes=i)).isoformat(), 3000] for i, speed in enumerate([2, 0.3, 0, 2, 2])], }} samples = EnlyzeApiGateway(client).get_material_samples( machine_id='machine', rate_variable_id='unused', gate_variable_id='speed', application_variable_ids=('s1', 's2'), nominal_width_m=5, start=start, end=start + timedelta(minutes=4), ) state = MaterialConsumptionIntegrator(MaterialIntegratorConfig(0.3, 60)).process_many(samples) assert state.cumulative_consumption_kg == 100 assert state.integrated_running_seconds == 120 assert client.post_json.call_args.args[1]['variables'] == [ {'uuid': 's1'}, {'uuid': 's2'}, {'uuid': 'speed'}, ] @pytest.mark.parametrize('sources,width,speed', [ ([], 5, 2), ([float('nan')], 5, 2), ([1], 0, 2), ([1], 5, float('inf')), ]) def test_invalid_area_inputs(sources, width, speed): with pytest.raises(ValueError): area_application_rate_kg_per_hour(sources, width, speed) @pytest.mark.parametrize('bad', [None, True, float('nan')]) def test_missing_or_invalid_spreader_fails(bad): client = Mock() now = timestamp('2026-08-25T05:27:18Z') client.post_json.return_value.body = {'data': { 'columns': ['time', 's1', 's2', 'speed'], 'records': [[now.isoformat(), 3000, bad, 2]], }} with pytest.raises(ValueError): EnlyzeApiGateway(client).get_material_samples( machine_id='machine', rate_variable_id='unused', gate_variable_id='speed', application_variable_ids=('s1', 's2'), nominal_width_m=5, start=now, end=now, ) def width_provider(): erp = Mock() erp.get_current_workplace_status.return_value = SimpleNamespace( workplace='Bento 1', production_order='12026000814', article_number='180305', article_description='Bfix NSP 5300, 5,00 x 40 m', ) return ErpNominalWidthProvider( erp, workplace='Bento 1', format_template='Bento 1-{production_order}', ) @pytest.mark.parametrize('configured,returned', [ ('Bento 1', 'BENTO 1'), ('Bento 1', ' \tBENTO 1\n'), (' \tBento 1\n', 'BENTO 1'), ('K7', ' k7 '), ]) def test_width_context_normalizes_workplace(configured, returned): provider = width_provider() provider.workplace = configured provider.gateway.get_current_workplace_status.return_value.workplace = returned assert provider('Bento 1-12026000814') == 5.0 @pytest.mark.parametrize('production_order', [ 'Bento 1-12026000815', 'BENTO 1-12026000814', ' Bento 1-12026000814 ', ]) def test_width_context_keeps_production_order_strict(production_order): provider = width_provider() provider.gateway.get_current_workplace_status.return_value.workplace = 'BENTO 1' with pytest.raises(ValueError, match='ERP context does not match the production order'): provider(production_order) @pytest.mark.parametrize('field,value', [ ('production_order', '12026000815'), ('workplace', 'K7'), ('article_description', None), ('article_description', '180305'), ]) def test_width_context_fails_closed(field, value): provider = width_provider() setattr(provider.gateway.get_current_workplace_status.return_value, field, value) with pytest.raises(ValueError): provider('Bento 1-12026000814') @pytest.mark.parametrize('inter_run_gap_seconds', [5, 3761]) def test_bento_rpm_bootstrap_and_persistent_resume(tmp_path, inter_run_gap_seconds): """Synthetic aggregate fixture, not recorded ENLYZE samples. Uses realistic rpm values and synthetic active time within actual run boundaries. """ config = load_material_calculation('config/bento1-material-consumption.yaml') order = 'Bento 1-12026000814' runs = [EnlyzeProductionRun( uuid, config.machine_ref, 'external-product', order, timestamp(start), timestamp(end), ) for uuid, start, end in [ ('b675818c-4636-4976-be6c-3e0e24da9e8a', '2026-08-25T05:27:18Z', '2026-08-26T04:40:58Z'), ('25d67424-8346-42d4-b844-75b4097382bd', '2026-08-26T05:43:39Z', '2026-08-26T05:54:05Z'), ]] runs[1] = replace( runs[1], start=runs[0].end + timedelta(seconds=inter_run_gap_seconds), end=runs[0].end + timedelta(seconds=inter_run_gap_seconds + 626), ) speed = 2.0 client = Mock() def response(path, body): start, end = timestamp(body['start']), timestamp(body['end']) run = next(run for run in runs if run.start <= start <= run.end) active_seconds = 1256 * 60 - 626 if run == runs[0] else 626 stop = run.start + timedelta(seconds=active_seconds) times = {start, end} t = start while t < end: times.add(t) t += timedelta(seconds=10) if start <= stop <= end: times.add(stop) return SimpleNamespace(body={'data': { 'columns': ['time', *config.rotational_speed_signal_refs, config.gate_signal_ref], 'records': [[t.isoformat(), 3.739, 3.956, speed if t < stop else 0] for t in sorted(times)], }}) client.post_json.side_effect = response gateway = EnlyzeApiGateway(client) gateway.get_open_production_run = Mock(return_value=replace(runs[1], end=None)) gateway.get_production_runs = Mock(return_value=runs) store = JsonMaterialStateStore(tmp_path) def service(): return MaterialPollingService( gateway=gateway, state_store=store, machine_id=config.machine_ref, rate_variable_id=config.rate_signal_ref, gate_variable_id=config.gate_signal_ref, gate_threshold=config.gate_threshold, max_sample_gap_seconds=config.max_sample_gap_seconds, rotational_speed_variable_ids=config.rotational_speed_signal_refs, specific_discharge_kg_per_rev_m=config.specific_discharge_kg_per_rev_m, nominal_width_provider=width_provider(), ) partial = service().poll_once(now=runs[1].start + timedelta(seconds=300)) result = service().poll_once(now=runs[1].end) assert result.state.cumulative_consumption_kg - partial.state.cumulative_consumption_kg == ( pytest.approx((3.739 + 3.956) * 5 * 2.75 * 326 / 60) ) assert result.state.integrated_running_seconds / 60 == pytest.approx(1256) assert result.state.cumulative_consumption_kg == pytest.approx( (3.739 + 3.956) * 5 * 2.75 * 1256, ) assert service().poll_once(now=runs[1].end).state == result.state assert gateway.get_production_runs.call_count == 1 assert [(timestamp(call.args[1]['start']), timestamp(call.args[1]['end'])) for call in client.post_json.call_args_list[:2]] == [ (runs[0].start, runs[0].end), (runs[1].start, runs[1].start + timedelta(seconds=300)), ] def test_k7_config_defaults_remain_direct(): config = load_material_calculation('config/k7-material-consumption.yaml') assert config.source_mode == 'direct_mass_rate' assert config.rotational_speed_signal_refs == () assert config.application_signal_refs == () assert config.gate_threshold == 0.5 assert config.rate_signal_ref == 'c9d06af5-f6d6-4ede-b6c4-5a98bac77129' @pytest.mark.parametrize('field,value', [ ('source_mode', 'unknown'), ('rotational_speed_signal_refs', []), ('rotational_speed_signal_refs', ['same', 'same']), ('erp_workplace', ''), ('production_order_format', '{wrong}'), ('rate_signal_ref', 'direct'), ]) def test_invalid_rotational_configuration(tmp_path, field, value): with open('config/bento1-material-consumption.yaml') as stream: document = yaml.safe_load(stream) document['calculations'][0][field] = value path = tmp_path / 'config.yaml' path.write_text(yaml.safe_dump(document)) with pytest.raises(CalculationConfigError): load_material_calculation(path) def test_rotational_context_failure_does_not_touch_persistent_state(tmp_path): config = load_material_calculation('config/bento1-material-consumption.yaml') gateway = Mock() now = timestamp('2026-08-25T05:27:18Z') gateway.get_open_production_run.return_value = EnlyzeProductionRun( 'run', config.machine_ref, None, 'Bento 1-12026000814', now, None, ) provider = width_provider() provider.gateway.get_current_workplace_status.return_value = None store = Mock() service = MaterialPollingService( gateway=gateway, state_store=store, machine_id=config.machine_ref, rate_variable_id=config.rate_signal_ref, gate_variable_id=config.gate_signal_ref, gate_threshold=config.gate_threshold, max_sample_gap_seconds=20, rotational_speed_variable_ids=config.rotational_speed_signal_refs, specific_discharge_kg_per_rev_m=config.specific_discharge_kg_per_rev_m, nominal_width_provider=provider, ) with pytest.raises(ValueError, match='context'): service.poll_once(now=now) gateway.get_material_samples.assert_not_called() store.save.assert_not_called() def test_area_gateway_pagination(): client = Mock() now = timestamp('2026-08-25T05:27:18Z') client.post_json.side_effect = [SimpleNamespace(body={ 'metadata': {'next_cursor': cursor}, 'data': {'columns': ['time', 's1', 's2', 'speed'], 'records': [[(now + timedelta(seconds=i * 10)).isoformat(), 3000, 2000, 2]]}, }) for i, cursor in enumerate(['next', None])] samples = EnlyzeApiGateway(client).get_material_samples( machine_id='machine', rate_variable_id='unused', gate_variable_id='speed', application_variable_ids=('s1', 's2'), nominal_width_m=5, start=now, end=now + timedelta(seconds=10), ) assert [sample.material_rate_kg_per_hour for sample in samples] == [3000, 3000] assert client.post_json.call_args.args[1]['cursor'] == 'next'