144 lines
6.1 KiB
Python
144 lines
6.1 KiB
Python
from dataclasses import replace
|
|
from datetime import UTC, date, datetime, timedelta
|
|
from pathlib import Path
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
from production_analytics.calculations.calibrations import load_calibrations, resolve_calibration
|
|
from production_analytics.calculations.config import (
|
|
CalculationConfigError,
|
|
load_material_calculation,
|
|
)
|
|
from production_analytics.calculations.material_application import rotational_application_g_m2
|
|
from production_analytics.calculations.material_consumption import (
|
|
MaterialConsumptionIntegrator,
|
|
MaterialIntegratorConfig,
|
|
)
|
|
from production_analytics.enlyze.gateway import EnlyzeApiGateway
|
|
|
|
CALIBRATIONS = Path('config/material-calibrations.yaml')
|
|
BENTO = Path('config/bento1-material-consumption.yaml')
|
|
ID = 'bento1-spreader-1-2'
|
|
|
|
|
|
def resolve(entries, identifier=ID):
|
|
return resolve_calibration(entries, identifier, expected_type='rotational_discharge',
|
|
expected_unit='kg_per_rev_m')
|
|
|
|
|
|
def test_load_and_resolve():
|
|
calibration = resolve(load_calibrations(CALIBRATIONS))
|
|
assert calibration.value == 2.75
|
|
assert calibration.calibrated_at == date(2026, 9, 8)
|
|
assert calibration.method == 'gravimetric_tray'
|
|
assert calibration.reference == {
|
|
'measured_application_g_m2': 4068, 'line_speed_m_min': 2.3,
|
|
'signal_values': {'left': 1.65, 'right': 1.75},
|
|
}
|
|
assert load_material_calculation(BENTO).specific_discharge_kg_per_rev_m == 2.75
|
|
|
|
|
|
@pytest.mark.parametrize('field', ['type', 'value', 'unit', 'calibrated_at', 'method',
|
|
'description', 'reference'])
|
|
def test_missing_metadata(tmp_path, field):
|
|
document = yaml.safe_load(CALIBRATIONS.read_text())
|
|
del document['calibrations'][ID][field]
|
|
path = tmp_path / 'material-calibrations.yaml'
|
|
path.write_text(yaml.safe_dump(document))
|
|
with pytest.raises(CalculationConfigError, match='required fields'):
|
|
load_calibrations(path)
|
|
|
|
|
|
@pytest.mark.parametrize('field,value', [
|
|
('type', ''), ('unit', None), ('method', 12), ('description', ' '),
|
|
('calibrated_at', '2026-02-30'), ('calibrated_at', 123),
|
|
('reference', []), ('reference', {}), ('value', True), ('value', float('nan')),
|
|
])
|
|
def test_invalid_metadata(tmp_path, field, value):
|
|
document = yaml.safe_load(CALIBRATIONS.read_text())
|
|
document['calibrations'][ID][field] = value
|
|
path = tmp_path / 'material-calibrations.yaml'
|
|
path.write_text(yaml.safe_dump(document))
|
|
with pytest.raises(CalculationConfigError, match=field):
|
|
load_calibrations(path)
|
|
|
|
|
|
@pytest.mark.parametrize('field,value,error', [
|
|
('type', 'other', 'incompatible type'), ('unit', 'kg_per_rev', 'incompatible unit'),
|
|
('value', 0, 'greater than zero'),
|
|
])
|
|
def test_incompatible_at_config_load(tmp_path, field, value, error):
|
|
document = yaml.safe_load(CALIBRATIONS.read_text())
|
|
document['calibrations'][ID][field] = value
|
|
(tmp_path / CALIBRATIONS.name).write_text(yaml.safe_dump(document))
|
|
path = tmp_path / BENTO.name
|
|
path.write_text(BENTO.read_text())
|
|
with pytest.raises(CalculationConfigError, match=error):
|
|
load_material_calculation(path)
|
|
|
|
|
|
def test_missing_id_and_file(tmp_path):
|
|
with pytest.raises(CalculationConfigError, match='Missing calibration ID'):
|
|
resolve(load_calibrations(CALIBRATIONS), 'unknown')
|
|
path = tmp_path / BENTO.name
|
|
path.write_text(BENTO.read_text())
|
|
with pytest.raises(CalculationConfigError, match='Cannot read calibration'):
|
|
load_material_calculation(path)
|
|
(tmp_path / CALIBRATIONS.name).write_text(CALIBRATIONS.read_text())
|
|
path.write_text(BENTO.read_text().replace(ID, 'unknown'))
|
|
with pytest.raises(CalculationConfigError, match='Missing calibration ID'):
|
|
load_material_calculation(path)
|
|
|
|
|
|
def test_exact_direct_value_equivalence(tmp_path):
|
|
referenced = load_material_calculation(BENTO)
|
|
document = yaml.safe_load(BENTO.read_text())
|
|
entry = document['calculations'][0]
|
|
del entry['calibration_ref']
|
|
entry['specific_discharge_kg_per_rev_m'] = 2.75
|
|
path = tmp_path / 'direct.yaml'
|
|
path.write_text(yaml.safe_dump(document))
|
|
direct = load_material_calculation(path)
|
|
assert replace(referenced, calibration_ref=None) == direct
|
|
start = datetime(2026, 9, 8, tzinfo=UTC)
|
|
outputs = []
|
|
for config in (referenced, direct):
|
|
factor = config.specific_discharge_kg_per_rev_m
|
|
application = rotational_application_g_m2([1.65, 1.75], factor, 2.3, 0.3)
|
|
client = Mock()
|
|
client.post_json.return_value.body = {'data': {
|
|
'columns': ['time', 'left', 'right', 'speed'],
|
|
'records': [[(start + timedelta(seconds=t)).isoformat(), 1.65, 1.75, speed]
|
|
for t, speed in [(0, 2.3), (10, 0), (20, 2.3), (50, 2.3), (60, 2.3)]],
|
|
}}
|
|
samples = EnlyzeApiGateway(client).get_material_samples(
|
|
machine_id='m', rate_variable_id='unused', gate_variable_id='speed',
|
|
rotational_speed_variable_ids=('left', 'right'), nominal_width_m=5,
|
|
specific_discharge_kg_per_rev_m=factor, start=start,
|
|
end=start + timedelta(seconds=60),
|
|
)
|
|
integrator = MaterialConsumptionIntegrator(MaterialIntegratorConfig(0.3, 20))
|
|
state = integrator.process_many(samples)
|
|
assert samples[0].material_rate_kg_per_hour == 2805.0
|
|
assert state.integrated_running_seconds == 20
|
|
outputs.append((application, samples, state))
|
|
assert outputs[0] == outputs[1]
|
|
assert outputs[0][0] == pytest.approx(4065.217391304348)
|
|
|
|
|
|
def test_update_only_central_file(tmp_path):
|
|
path = tmp_path / BENTO.name
|
|
path.write_text(BENTO.read_text())
|
|
(tmp_path / CALIBRATIONS.name).write_text(CALIBRATIONS.read_text().replace('2.75', '2.8'))
|
|
assert load_material_calculation(path).specific_discharge_kg_per_rev_m == 2.8
|
|
|
|
|
|
def test_k7_without_calibration_file(tmp_path):
|
|
original = Path('config/k7-material-consumption.yaml')
|
|
path = tmp_path / original.name
|
|
path.write_text(original.read_text())
|
|
assert load_material_calculation(path) == load_material_calculation(original)
|
|
assert load_material_calculation(path).calibration_ref is None
|