208 lines
7.5 KiB
Python
208 lines
7.5 KiB
Python
from __future__ import annotations
|
|
|
|
from copy import deepcopy
|
|
|
|
import pytest
|
|
|
|
from server.aps_domain.closed_loop_problem import build_closed_loop_problem
|
|
from server.aps_domain.closed_loop_runtime import (
|
|
closed_loop_to_problem_v2,
|
|
flex_version_to_solution_v2,
|
|
)
|
|
from server.aps_domain.scheduling_problem_v2 import (
|
|
ResourceKind,
|
|
SchedulingProblemV2,
|
|
SchedulingSolutionV2,
|
|
scheduling_problem_hash,
|
|
)
|
|
from server.aps_domain.scheduling_validator import validate_scheduling_solution_v2
|
|
from tests.golden.test_closed_loop_runtime import _ready_world
|
|
|
|
BUSINESS_DATE = "2026-08-02"
|
|
|
|
|
|
def _production_resource_world() -> dict:
|
|
world = _ready_world()
|
|
world.update({
|
|
"factories": [{"id": 1, "code": "F1", "name": "Factory", "status": "ACTIVE"}],
|
|
"workshops": [{"id": 1, "factoryId": 1, "code": "W1", "name": "Workshop", "status": "ACTIVE"}],
|
|
"lines": [{
|
|
"id": 1,
|
|
"workshopId": 1,
|
|
"code": "L1",
|
|
"name": "Line",
|
|
"status": "ACTIVE",
|
|
"parallelCapacity": 2,
|
|
"capacityPerDay": 1_000,
|
|
}],
|
|
"workstations": [{
|
|
"id": 1,
|
|
"lineId": 1,
|
|
"code": "S1",
|
|
"name": "Workstation",
|
|
"status": "ACTIVE",
|
|
"parallelCapacity": 2,
|
|
}],
|
|
"flexTeams": [{
|
|
"code": "T-CUT",
|
|
"name": "Cutting team",
|
|
"memberCount": 2,
|
|
"supportOps": ["CUT"],
|
|
"workstationId": 1,
|
|
}],
|
|
"flexMolds": [{
|
|
"id": 1,
|
|
"code": "M-CUT",
|
|
"name": "Cutting mold",
|
|
"operationCode": "CUT",
|
|
"adaptableEquipment": ["EQ-CUT"],
|
|
"lifeTotal": 10,
|
|
"lifeUsed": 2,
|
|
"status": "AVAILABLE",
|
|
"workstationId": 1,
|
|
}],
|
|
})
|
|
world["flexEquipment"][0].update({
|
|
"workstationId": 1,
|
|
"adaptableMolds": ["M-CUT"],
|
|
})
|
|
world["routingSteps"][0].update({
|
|
"requireMold": True,
|
|
"requiredTeamMembers": 1,
|
|
"moldLifePerUnit": 1,
|
|
})
|
|
return world
|
|
|
|
|
|
def _problem_and_solution(*, include_team: bool = True, include_tooling: bool = True):
|
|
world = _production_resource_world()
|
|
closed_loop = build_closed_loop_problem(world, business_date=BUSINESS_DATE)
|
|
problem = closed_loop_to_problem_v2(world, closed_loop)
|
|
requirement_id = closed_loop.manufacturing_demands[0].requirement_id
|
|
world["flexOrders"] = [{
|
|
"id": 1,
|
|
"orderNo": "SO-READY",
|
|
"closedLoopRequirementId": requirement_id,
|
|
}]
|
|
world["flexScheduleVersions"] = [{
|
|
"id": 1,
|
|
"createdAt": f"{BUSINESS_DATE} 00:00",
|
|
"totalTardiness": 0,
|
|
"avgUtilization": 0.5,
|
|
}]
|
|
world["flexWorkOrders"] = [{
|
|
"id": 1,
|
|
"versionId": 1,
|
|
"flexOrderNo": "SO-READY",
|
|
"closedLoopRequirementId": requirement_id,
|
|
"seq": 1,
|
|
"operationCode": "CUT",
|
|
"equipmentId": 1,
|
|
"equipmentCode": "EQ-CUT",
|
|
"teamCode": "T-CUT" if include_team else None,
|
|
"moldCode": "M-CUT" if include_tooling else None,
|
|
"plannedStartTime": "2026-08-03 08:00",
|
|
"plannedEndTime": "2026-08-03 08:10",
|
|
}]
|
|
world["flexConflicts"] = []
|
|
solution = flex_version_to_solution_v2(world, closed_loop, problem, 1)
|
|
return world, problem, solution
|
|
|
|
|
|
def _rebind(problem: SchedulingProblemV2, solution: SchedulingSolutionV2) -> SchedulingSolutionV2:
|
|
provenance = solution.provenance.model_copy(update={"problemHash": scheduling_problem_hash(problem)})
|
|
return solution.model_copy(update={"provenance": provenance})
|
|
|
|
|
|
def _replace_resource(problem: SchedulingProblemV2, resource_id: str, **updates) -> SchedulingProblemV2:
|
|
resources = tuple(
|
|
resource.model_copy(update=updates) if resource.resourceId == resource_id else resource
|
|
for resource in problem.resources
|
|
)
|
|
return problem.model_copy(update={"resources": resources})
|
|
|
|
|
|
def test_production_adapter_reads_organization_team_and_tooling_resources():
|
|
world = _production_resource_world()
|
|
before = deepcopy(world)
|
|
closed_loop = build_closed_loop_problem(world, business_date=BUSINESS_DATE)
|
|
|
|
problem = closed_loop_to_problem_v2(world, closed_loop)
|
|
|
|
assert world == before
|
|
resources = {resource.resourceId: resource for resource in problem.resources}
|
|
assert resources["workshop:1"].parentResourceId == "factory:1"
|
|
assert resources["line:1"].parentResourceId == "workshop:1"
|
|
assert resources["workstation:1"].parentResourceId == "line:1"
|
|
assert resources["equipment:1"].parentResourceId == "workstation:1"
|
|
assert resources["team:T-CUT"].kind is ResourceKind.TEAM
|
|
assert resources["team:T-CUT"].cumulativeCapacity == 2
|
|
assert resources["tooling:M-CUT"].kind is ResourceKind.TOOLING
|
|
assert resources["tooling:M-CUT"].compatibleResourceIds == ("equipment:1",)
|
|
assert resources["tooling:M-CUT"].lifeTotal == 10
|
|
requirements = {item.kind: item for item in problem.activities[0].resourceRequirements}
|
|
assert set(requirements) == {ResourceKind.EQUIPMENT, ResourceKind.TEAM, ResourceKind.TOOLING}
|
|
assert requirements[ResourceKind.TEAM].eligibleResourceIds == ("team:T-CUT",)
|
|
assert requirements[ResourceKind.TOOLING].eligibleResourceIds == ("tooling:M-CUT",)
|
|
assert requirements[ResourceKind.TOOLING].lifeUnits == 1
|
|
|
|
|
|
def test_complete_multi_resource_candidate_passes_v2_validator():
|
|
world, problem, solution = _problem_and_solution()
|
|
|
|
report = validate_scheduling_solution_v2(problem, solution, world=world)
|
|
|
|
assert report.valid is True
|
|
assert report.hardViolations == ()
|
|
assert {allocation.kind for allocation in solution.activities[0].resourceAllocations} == {
|
|
ResourceKind.EQUIPMENT,
|
|
ResourceKind.TEAM,
|
|
ResourceKind.TOOLING,
|
|
}
|
|
|
|
|
|
def test_external_style_candidate_missing_team_and_tooling_fails_closed():
|
|
world, problem, solution = _problem_and_solution(include_team=False, include_tooling=False)
|
|
|
|
report = validate_scheduling_solution_v2(problem, solution, world=world)
|
|
codes = {issue.code for issue in report.hardViolations}
|
|
|
|
assert report.valid is False
|
|
assert "REQUIRED_RESOURCE_ASSIGNMENT_MISSING" in codes
|
|
missing_kinds = {
|
|
issue.entityRefs[-1]
|
|
for issue in report.hardViolations
|
|
if issue.code == "REQUIRED_RESOURCE_ASSIGNMENT_MISSING"
|
|
}
|
|
assert missing_kinds == {"TEAM", "TOOLING"}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("resource_id", "expected_ref"),
|
|
[("team:T-CUT", "team:T-CUT"), ("line:1", "line:1")],
|
|
)
|
|
def test_team_and_hierarchy_cumulative_capacity_fail_closed(resource_id: str, expected_ref: str):
|
|
world, problem, solution = _problem_and_solution()
|
|
constrained = _replace_resource(problem, resource_id, cumulativeCapacity=0.5)
|
|
|
|
report = validate_scheduling_solution_v2(constrained, _rebind(constrained, solution), world=world)
|
|
|
|
assert report.valid is False
|
|
assert any(
|
|
issue.code == "RESOURCE_CAPACITY_OVERLAP" and expected_ref in issue.entityRefs
|
|
for issue in report.hardViolations
|
|
)
|
|
|
|
|
|
def test_tooling_life_exhaustion_fails_closed():
|
|
world, problem, solution = _problem_and_solution()
|
|
exhausted = _replace_resource(problem, "tooling:M-CUT", lifeTotal=2, lifeUsed=2)
|
|
|
|
report = validate_scheduling_solution_v2(exhausted, _rebind(exhausted, solution), world=world)
|
|
|
|
assert report.valid is False
|
|
assert any(
|
|
issue.code == "TOOLING_LIFE_EXCEEDED" and "tooling:M-CUT" in issue.entityRefs
|
|
for issue in report.hardViolations
|
|
)
|