Add feedback-aligned material efficiency KPI
This commit is contained in:
@@ -0,0 +1,184 @@
|
||||
from dataclasses import FrozenInstanceError, replace
|
||||
from datetime import UTC, datetime
|
||||
from unittest.mock import Mock
|
||||
from zoneinfo import ZoneInfo
|
||||
|
||||
import pytest
|
||||
|
||||
from production_analytics.erp import CurrentWorkplaceStatus
|
||||
from production_analytics.service.material_efficiency import MaterialEfficiencyService
|
||||
from production_analytics.service.postgres_material import MaterialConsumptionSnapshot
|
||||
|
||||
CALC = 'k7-fiber-consumption'
|
||||
MACHINE = 'c220f95c-a65e-4cb7-99b7-0626d6c7508c'
|
||||
START = datetime(2026, 9, 4, 10, tzinfo=UTC)
|
||||
FEEDBACK = START.replace(minute=5)
|
||||
STATUS = CurrentWorkplaceStatus(
|
||||
'K7', '12026000815', '212520', 'Stex R 1501 C (PR) 5,80 x 50 m',
|
||||
FEEDBACK, None, 800, None, None, None,
|
||||
)
|
||||
MATERIAL = MaterialConsumptionSnapshot(START, CALC, MACHINE, 'K 7-12026000815', 'run', 1000)
|
||||
|
||||
|
||||
def service(repository=None, **config):
|
||||
return MaterialEfficiencyService(
|
||||
Mock(get_current_workplace_status=Mock(return_value=STATUS)),
|
||||
repository if repository is not None else Mock(latest_at_or_before=Mock(
|
||||
return_value=MATERIAL,
|
||||
)),
|
||||
**dict(workplace='K7', machine_id=MACHINE, calculation_id=CALC,
|
||||
format_template='K 7-{production_order}', **config),
|
||||
)
|
||||
|
||||
|
||||
def test_valid_current_feedback():
|
||||
subject = service()
|
||||
result = subject.evaluate_current()
|
||||
assert result.material_consumption_kg_per_m2 == 1.25
|
||||
assert result.material_consumption_g_per_m2 == 1250
|
||||
assert result.nominal_width_m == 5.8
|
||||
assert result.good_quantity_m2 == 800
|
||||
assert result.material_consumption_kg == 1000
|
||||
assert result.erp_feedback_timestamp == FEEDBACK
|
||||
assert result.material_snapshot_timestamp == START
|
||||
assert (result.workplace, result.machine_id, result.calculation_id) == ('K7', MACHINE, CALC)
|
||||
assert (result.erp_production_order, result.enlyze_production_order) == (
|
||||
'12026000815', 'K 7-12026000815',
|
||||
)
|
||||
assert (result.article_number, result.article_description) == (
|
||||
STATUS.article_number, STATUS.article_description,
|
||||
)
|
||||
subject.erp.get_current_workplace_status.assert_called_once_with('K7')
|
||||
subject.materials.latest_at_or_before.assert_called_once_with(
|
||||
calculation_id=CALC, machine_id=MACHINE, production_order='K 7-12026000815',
|
||||
timestamp=FEEDBACK,
|
||||
)
|
||||
with pytest.raises(FrozenInstanceError):
|
||||
result.good_quantity_m2 = 5
|
||||
|
||||
|
||||
@pytest.mark.parametrize('quantity', [None, 0, -1, float('nan'), float('inf'), -float('inf')])
|
||||
def test_invalid_quantity(quantity):
|
||||
subject = service()
|
||||
assert subject.evaluate(replace(STATUS, good_quantity_m2=quantity)) is None
|
||||
subject.materials.latest_at_or_before.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.parametrize('description', [None, 'unstructured article'])
|
||||
def test_width_is_optional_context(description):
|
||||
result = service().evaluate(replace(STATUS, article_description=description))
|
||||
assert result.nominal_width_m is None
|
||||
assert result.material_consumption_kg_per_m2 == 1.25
|
||||
|
||||
|
||||
def test_unavailable_sources():
|
||||
subject = service()
|
||||
subject.erp.get_current_workplace_status.return_value = None
|
||||
assert subject.evaluate_current() is None
|
||||
subject.materials.latest_at_or_before.return_value = None
|
||||
assert subject.evaluate(STATUS) is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize('consumption, quantity', [
|
||||
(float('nan'), 800), (float('inf'), 800), (-1, 800),
|
||||
(1e308, 1e-308), (1e308, 1),
|
||||
])
|
||||
def test_invalid_consumption_and_overflow(consumption, quantity):
|
||||
subject = service()
|
||||
subject.materials.latest_at_or_before.return_value = replace(
|
||||
MATERIAL, consumption_kg=consumption,
|
||||
)
|
||||
assert subject.evaluate(replace(STATUS, good_quantity_m2=quantity)) is None
|
||||
|
||||
|
||||
def test_zero_consumption_is_valid():
|
||||
subject = service()
|
||||
subject.materials.latest_at_or_before.return_value = replace(MATERIAL, consumption_kg=0)
|
||||
assert subject.evaluate(STATUS).material_consumption_kg_per_m2 == 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize('changes', [
|
||||
dict(timestamp=START.replace(minute=10)), dict(timestamp=START.replace(tzinfo=None)),
|
||||
dict(production_order='K 7-12026000815-K 7-12026000816'),
|
||||
dict(machine_id='other'), dict(calculation_id='other'),
|
||||
])
|
||||
def test_repository_contract_is_checked(changes):
|
||||
subject = service()
|
||||
subject.materials.latest_at_or_before.return_value = replace(MATERIAL, **changes)
|
||||
with pytest.raises(ValueError, match='unaligned'):
|
||||
subject.evaluate(STATUS)
|
||||
|
||||
|
||||
def test_other_explicit_configuration():
|
||||
repository = Mock()
|
||||
repository.latest_at_or_before.return_value = replace(
|
||||
MATERIAL, machine_id='example-machine', production_order='ORDER/00123',
|
||||
)
|
||||
subject = MaterialEfficiencyService(
|
||||
Mock(), repository, workplace='example', machine_id='example-machine',
|
||||
calculation_id=CALC, format_template='ORDER/{production_order}',
|
||||
)
|
||||
result = subject.evaluate(replace(STATUS, workplace='example', production_order=' 00123 '))
|
||||
assert result.enlyze_production_order == 'ORDER/00123'
|
||||
|
||||
|
||||
def test_wrong_workplace_and_combined_order_rejected():
|
||||
with pytest.raises(ValueError, match='workplace'):
|
||||
service().evaluate(replace(STATUS, workplace='other'))
|
||||
with pytest.raises(ValueError, match='ERP production order'):
|
||||
service().evaluate(replace(STATUS, production_order='K 7-123-K 7-456'))
|
||||
|
||||
|
||||
def test_temporal_regression_and_successive_feedback():
|
||||
later = replace(
|
||||
MATERIAL, timestamp=START.replace(minute=10), consumption_kg=1100, run_id='run2',
|
||||
)
|
||||
repository = Mock()
|
||||
repository.latest_at_or_before.side_effect = lambda **kw: max(
|
||||
(row for row in [MATERIAL, later] if row.timestamp <= kw['timestamp']),
|
||||
key=lambda row: row.timestamp, default=None,
|
||||
)
|
||||
subject = service(repository)
|
||||
first = subject.evaluate(STATUS)
|
||||
second = subject.evaluate(replace(
|
||||
STATUS, feedback_timestamp=later.timestamp, good_quantity_m2=1000,
|
||||
))
|
||||
assert first.material_snapshot_timestamp == START
|
||||
assert first.material_consumption_kg == 1000
|
||||
assert first.material_consumption_kg_per_m2 == 1.25
|
||||
assert second.material_snapshot_timestamp == later.timestamp
|
||||
assert second.material_consumption_kg == 1100
|
||||
assert second.material_consumption_kg_per_m2 == 1.1
|
||||
assert first.good_quantity_m2 == 800
|
||||
|
||||
|
||||
def test_naive_feedback_requires_explicit_timezone_and_preserves_source():
|
||||
naive = datetime(2026, 9, 4, 12, 5)
|
||||
status = replace(STATUS, feedback_timestamp=naive)
|
||||
with pytest.raises(ValueError, match='explicit erp_timezone'):
|
||||
service().evaluate(status)
|
||||
subject = service(erp_timezone=ZoneInfo('Europe/Berlin'))
|
||||
result = subject.evaluate(status)
|
||||
assert result.erp_feedback_timestamp == naive
|
||||
assert subject.materials.latest_at_or_before.call_args.kwargs['timestamp'] == FEEDBACK
|
||||
|
||||
|
||||
@pytest.mark.parametrize('timestamp', [datetime(2026, 3, 29, 2, 30), datetime(2026, 10, 25, 2, 30)])
|
||||
def test_dst_gap_and_overlap_rejected(timestamp):
|
||||
with pytest.raises(ValueError, match='ambiguous or nonexistent'):
|
||||
service(erp_timezone=ZoneInfo('Europe/Berlin')).evaluate(
|
||||
replace(STATUS, feedback_timestamp=timestamp),
|
||||
)
|
||||
|
||||
|
||||
def test_aware_feedback_preserves_offset():
|
||||
timestamp = FEEDBACK.astimezone(ZoneInfo('Europe/Berlin'))
|
||||
result = service().evaluate(replace(STATUS, feedback_timestamp=timestamp))
|
||||
assert result.erp_feedback_timestamp is timestamp
|
||||
|
||||
|
||||
def test_database_failure_propagates():
|
||||
subject = service()
|
||||
subject.materials.latest_at_or_before.side_effect = RuntimeError('unavailable')
|
||||
with pytest.raises(RuntimeError, match='unavailable'):
|
||||
subject.evaluate(STATUS)
|
||||
@@ -0,0 +1,101 @@
|
||||
import sqlite3
|
||||
import sys
|
||||
from datetime import UTC, datetime
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from production_analytics.service.postgres_material import (
|
||||
PostgresMaterialSnapshotRepository,
|
||||
PostgresSettings,
|
||||
)
|
||||
|
||||
START = datetime(2026, 9, 4, 10, tzinfo=UTC)
|
||||
ORDER = " K 7-123'; -- "
|
||||
SETTINGS = PostgresSettings('localhost', 5432, 'analytics', 'reader', 'secret')
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def database():
|
||||
# Execute the actual portable SELECT in SQLite; only adapt driver placeholders
|
||||
# and datetime transport. This tests SQL semantics without a live PostgreSQL server.
|
||||
with sqlite3.connect(':memory:') as database:
|
||||
database.execute('''CREATE TABLE material_consumption_snapshots (
|
||||
timestamp TEXT, calculation_id TEXT, machine_id TEXT, production_order TEXT,
|
||||
run_id TEXT, consumption_kg REAL
|
||||
)''')
|
||||
rows = [
|
||||
(START.isoformat(), 'calc', 'machine', ORDER, 'run1', 1000),
|
||||
(START.replace(minute=10).isoformat(), 'calc', 'machine', ORDER, 'run2', 1100),
|
||||
(START.replace(minute=4).isoformat(), 'other', 'machine', ORDER, 'run', 9999),
|
||||
(START.replace(minute=4).isoformat(), 'calc', 'other', ORDER, 'run', 9999),
|
||||
(START.replace(minute=4).isoformat(), 'calc', 'machine', ORDER.strip(), 'run', 9999),
|
||||
(START.replace(minute=4).isoformat(), 'calc', 'machine', ORDER + '-other', 'run', 9999),
|
||||
]
|
||||
database.executemany(
|
||||
'INSERT INTO material_consumption_snapshots VALUES (?,?,?,?,?,?)', rows,
|
||||
)
|
||||
yield database
|
||||
|
||||
|
||||
@pytest.mark.parametrize('minute, expected_minute, consumption', [
|
||||
(-1, None, None), (0, 0, 1000), (5, 0, 1000), (10, 10, 1100), (15, 10, 1100),
|
||||
])
|
||||
def test_aligned_lookup_sql(database, minute, expected_minute, consumption):
|
||||
cutoff = START.replace(minute=minute) if minute >= 0 else START.replace(hour=9, minute=59)
|
||||
driver = MagicMock()
|
||||
connection = driver.connect.return_value.__enter__.return_value
|
||||
|
||||
def execute(sql, parameters):
|
||||
assert parameters == ('calc', 'machine', ORDER, cutoff)
|
||||
assert sql.count('%s') == 4
|
||||
assert ORDER not in sql
|
||||
assert 'ORDER BY timestamp DESC' in sql
|
||||
assert 'LIMIT 1' in sql
|
||||
row = database.execute(
|
||||
sql.replace('%s', '?'), (*parameters[:3], parameters[3].isoformat()),
|
||||
).fetchone()
|
||||
if row is not None:
|
||||
row = (datetime.fromisoformat(row[0]), *row[1:])
|
||||
return MagicMock(fetchone=MagicMock(return_value=row))
|
||||
|
||||
connection.execute.side_effect = execute
|
||||
with patch.dict(sys.modules, psycopg=driver):
|
||||
result = PostgresMaterialSnapshotRepository(SETTINGS).latest_at_or_before(
|
||||
calculation_id='calc', machine_id='machine', production_order=ORDER, timestamp=cutoff,
|
||||
)
|
||||
if expected_minute is None:
|
||||
assert result is None
|
||||
else:
|
||||
assert result.timestamp == START.replace(minute=expected_minute)
|
||||
assert result.timestamp <= cutoff
|
||||
assert result.consumption_kg == consumption
|
||||
assert (result.calculation_id, result.machine_id, result.production_order) == (
|
||||
'calc', 'machine', ORDER,
|
||||
)
|
||||
assert result.run_id == ('run1' if expected_minute == 0 else 'run2')
|
||||
connection.execute.assert_called_once()
|
||||
driver.connect.assert_called_once_with(
|
||||
host='localhost', port=5432, dbname='analytics', user='reader', password='secret',
|
||||
connect_timeout=10, options='-c statement_timeout=10000',
|
||||
)
|
||||
driver.connect.return_value.__exit__.assert_called_once_with(None, None, None)
|
||||
|
||||
|
||||
def test_naive_cutoff_rejected_before_connection():
|
||||
driver = MagicMock()
|
||||
with patch.dict(sys.modules, psycopg=driver), pytest.raises(ValueError, match='timezone-aware'):
|
||||
PostgresMaterialSnapshotRepository(SETTINGS).latest_at_or_before(
|
||||
calculation_id='calc', machine_id='machine', production_order=ORDER,
|
||||
timestamp=START.replace(tzinfo=None),
|
||||
)
|
||||
driver.connect.assert_not_called()
|
||||
|
||||
|
||||
def test_database_failure_propagates():
|
||||
driver = MagicMock()
|
||||
driver.connect.side_effect = RuntimeError('connection unavailable')
|
||||
with patch.dict(sys.modules, psycopg=driver), pytest.raises(RuntimeError):
|
||||
PostgresMaterialSnapshotRepository(SETTINGS).latest_at_or_before(
|
||||
calculation_id='calc', machine_id='machine', production_order=ORDER, timestamp=START,
|
||||
)
|
||||
Reference in New Issue
Block a user