From 93528b2df340c47b7c070fa8f652745dc27309ff Mon Sep 17 00:00:00 2001 From: Nameru Thapa Date: Mon, 10 Aug 2026 18:08:43 +0200 Subject: [PATCH 1/2] feat: implement product unit and job completion --- app/services.py | 46 ++++++- tests/test_job_completion.py | 228 +++++++++++++++++++++++++++++++++++ 2 files changed, 272 insertions(+), 2 deletions(-) create mode 100644 tests/test_job_completion.py diff --git a/app/services.py b/app/services.py index dc23177..5c9c4e7 100644 --- a/app/services.py +++ b/app/services.py @@ -196,6 +196,9 @@ def scan_job_barcode(barcode, station_id): if job is None: return None, "Barcode not recognized." + if job.status == "Completed": + return None, "This production job is complete." + if job.status != "Active": return None, "This production job is not active." @@ -352,10 +355,42 @@ def complete_current_step(job, station_id): progress.status = "Completed" progress.completed_at = datetime.now(timezone.utc) - current_unit.status = "In Progress" db.session.flush() + completed_step_ids = { + unit_progress.job_step_id + for unit_progress in UnitStepProgress.query.filter_by( + product_unit_id=current_unit.id, + status="Completed", + ).all() + } + + unit_completed = all( + step.id in completed_step_ids + for step in job.steps + ) + + current_unit.status = ( + "Completed" + if unit_completed + else "In Progress" + ) + + db.session.flush() + + job_units = ProductUnit.query.filter_by( + job_id=job.id, + ).all() + + job_completed = bool(job_units) and all( + unit.status == "Completed" + for unit in job_units + ) + + if job_completed: + job.status = "Completed" + transfer_requests = create_remaining_material_requests( job, current_step.station_id, @@ -370,11 +405,18 @@ def complete_current_step(job, station_id): next_unit, next_step = get_current_unit_and_step(job) return ( - { + { "completed_unit": current_unit, "completed_step": current_step, "next_unit": next_unit, "next_step": next_step, + "next_station_id": ( + next_step.station_id + if next_step is not None + else None + ), + "unit_completed": unit_completed, + "job_completed": job_completed, "transfer_requests": transfer_requests, }, None, diff --git a/tests/test_job_completion.py b/tests/test_job_completion.py new file mode 100644 index 0000000..37478e9 --- /dev/null +++ b/tests/test_job_completion.py @@ -0,0 +1,228 @@ +import pytest + +from app import create_app, db +from app.models import ( + Job, + JobStep, + Part, + ProductUnit, + StationInventory, + StepPart, + UnitStepProgress, +) +from app.services import ( + complete_current_step, + scan_job_barcode, +) + + +@pytest.fixture +def app(): + test_app = create_app( + { + "TESTING": True, + "SQLALCHEMY_DATABASE_URI": "sqlite:///:memory:", + } + ) + + yield test_app + + +def create_multi_station_job(): + part_a = Part(name="Part A", unit="pcs") + part_b = Part(name="Part B", unit="pcs") + + job = Job( + name="Day 5 Completion Test", + quantity=2, + threshold_percent=20.0, + status="Active", + barcode="MES-DAY5-0001", + ) + + db.session.add_all([part_a, part_b, job]) + db.session.flush() + + unit_1 = ProductUnit( + job_id=job.id, + unit_number=1, + status="Pending", + ) + + unit_2 = ProductUnit( + job_id=job.id, + unit_number=2, + status="Pending", + ) + + step_1 = JobStep( + job_id=job.id, + step_number=1, + station_id="ST-01", + instruction="Assemble Part A", + photo_url="day5-step-1.jpg", + ) + + step_2 = JobStep( + job_id=job.id, + step_number=2, + station_id="ST-02", + instruction="Install Part B", + photo_url="day5-step-2.jpg", + ) + + db.session.add_all([unit_1, unit_2, step_1, step_2]) + db.session.flush() + + db.session.add_all( + [ + StepPart( + step_id=step_1.id, + part_id=part_a.id, + quantity=1, + ), + StepPart( + step_id=step_2.id, + part_id=part_b.id, + quantity=1, + ), + StationInventory( + station_id="ST-01", + part_id=part_a.id, + quantity=2, + ), + StationInventory( + station_id="ST-02", + part_id=part_b.id, + quantity=2, + ), + ] + ) + + db.session.commit() + + return { + "job_id": job.id, + "unit_1_id": unit_1.id, + "unit_2_id": unit_2.id, + "step_1_id": step_1.id, + "step_2_id": step_2.id, + } + + +def test_tc13_pass_product_to_next_station(app): + with app.app_context(): + ids = create_multi_station_job() + job = db.session.get(Job, ids["job_id"]) + + result, error = complete_current_step( + job, + "ST-01", + ) + + assert error is None + assert result["unit_completed"] is False + assert result["job_completed"] is False + assert result["next_unit"].id == ids["unit_1_id"] + assert result["next_step"].id == ids["step_2_id"] + assert result["next_station_id"] == "ST-02" + + next_step, scan_error = scan_job_barcode( + "MES-DAY5-0001", + "ST-02", + ) + + assert scan_error is None + assert next_step.id == ids["step_2_id"] + assert next_step.instruction == "Install Part B" + + +def test_tc14_final_step_completes_one_product_unit(app): + with app.app_context(): + ids = create_multi_station_job() + job = db.session.get(Job, ids["job_id"]) + + first_result, first_error = complete_current_step( + job, + "ST-01", + ) + + assert first_error is None + assert first_result is not None + + result, error = complete_current_step( + job, + "ST-02", + ) + + assert error is None + assert result["unit_completed"] is True + assert result["job_completed"] is False + + unit_1 = db.session.get( + ProductUnit, + ids["unit_1_id"], + ) + + assert unit_1.status == "Completed" + + completed_units = ProductUnit.query.filter_by( + job_id=job.id, + status="Completed", + ).count() + + assert completed_units == 1 + assert job.status == "Active" + assert result["next_unit"].id == ids["unit_2_id"] + assert result["next_step"].id == ids["step_1_id"] + + +def test_tc15_all_units_complete_the_production_job(app): + with app.app_context(): + ids = create_multi_station_job() + job = db.session.get(Job, ids["job_id"]) + + completion_sequence = [ + "ST-01", + "ST-02", + "ST-01", + "ST-02", + ] + + result = None + + for station_id in completion_sequence: + result, error = complete_current_step( + job, + station_id, + ) + assert error is None + + assert result["unit_completed"] is True + assert result["job_completed"] is True + assert result["next_unit"] is None + assert result["next_step"] is None + assert result["next_station_id"] is None + + assert job.status == "Completed" + + completed_units = ProductUnit.query.filter_by( + job_id=job.id, + status="Completed", + ).count() + + assert completed_units == 2 + + completed_progress = UnitStepProgress.query.filter_by( + status="Completed", + ).count() + + assert completed_progress == 4 + + step, scan_error = scan_job_barcode( + "MES-DAY5-0001", + "ST-02", + ) + + assert step is None + assert scan_error == "This production job is complete." \ No newline at end of file -- GitLab From a81ab49746a4f0819059a0bebe9a812053ac6e6b Mon Sep 17 00:00:00 2001 From: Nameru Thapa Date: Mon, 10 Aug 2026 18:38:57 +0200 Subject: [PATCH 2/2] feat: add production completion guidance --- app/routes.py | 52 ++++++++++++++++++++++++++++++--- app/templates/worker_kiosk.html | 9 ++++++ 2 files changed, 57 insertions(+), 4 deletions(-) diff --git a/app/routes.py b/app/routes.py index 455b470..fa63baf 100644 --- a/app/routes.py +++ b/app/routes.py @@ -223,6 +223,8 @@ def worker_kiosk(): error = None message = None stock_message = None + completion_message = None + transition_message = None if request.method == "POST": selected_station_id = request.form.get( @@ -271,14 +273,54 @@ def worker_kiosk(): next_unit = result["next_unit"] next_step = result["next_step"] + completed_count = ProductUnit.query.filter_by( + job_id=completed_unit.job_id, + status="Completed", + ).count() + + if result["job_completed"]: + completion_message = ( + "Product Complete: Unit " + f"{completed_unit.unit_number} " + "is completed. Move this unit to " + "Finished Goods. Production Complete: " + f"all {completed_step.job.quantity} " + "product units are completed." + ) + elif result["unit_completed"]: + completion_message = ( + "Product Complete: Unit " + f"{completed_unit.unit_number} " + "is completed. Move this unit to " + "Finished Goods. Production Progress: " + f"{completed_count} of " + f"{completed_step.job.quantity} " + "units completed." + ) + if ( next_unit is not None and next_step is not None - and next_step.station_id - == selected_station_id ): - current_unit = next_unit - step = next_step + if ( + next_step.station_id + == selected_station_id + ): + current_unit = next_unit + step = next_step + elif result["unit_completed"]: + transition_message = ( + "Next Unit " + f"{next_unit.unit_number} starts at " + f"{next_step.station.name}." + ) + else: + transition_message = ( + "Pass Unit " + f"{completed_unit.unit_number} " + "to Next Station: " + f"{next_step.station.name}." + ) else: step, error = scan_job_barcode( @@ -301,4 +343,6 @@ def worker_kiosk(): error=error, message=message, stock_message=stock_message, + completion_message=completion_message, + transition_message=transition_message, ) \ No newline at end of file diff --git a/app/templates/worker_kiosk.html b/app/templates/worker_kiosk.html index 469684b..1eaed70 100644 --- a/app/templates/worker_kiosk.html +++ b/app/templates/worker_kiosk.html @@ -65,6 +65,15 @@ {{ stock_message }}

{% endif %} + {% if completion_message %} +

+ {{ completion_message }} +

+ {% endif %} + + {% if transition_message %} +

{{ transition_message }}

+ {% endif %} {% if step and current_unit %}
-- GitLab