449 lines
16 KiB
Python
449 lines
16 KiB
Python
"""Versioned, solver-neutral APS problem and solution contract.
|
|
|
|
This module is deliberately isolated from ``scheduling_dto.py``. It models the
|
|
complete demand/supply/activity/resource graph needed by Round 65 while keeping
|
|
all records immutable and JSON serialisable. Runtime adapters are expected to
|
|
be added by the integration owner in a later slice.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from datetime import date, datetime
|
|
from enum import StrEnum
|
|
from typing import Any, Mapping
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
|
|
|
|
|
|
class _ContractModel(BaseModel):
|
|
"""Strict immutable base used by every V2 contract record."""
|
|
|
|
model_config = ConfigDict(extra="forbid", frozen=True, populate_by_name=True)
|
|
|
|
|
|
class SourcingType(StrEnum):
|
|
MAKE = "MAKE"
|
|
BUY = "BUY"
|
|
SUBCONTRACT = "SUBCONTRACT"
|
|
|
|
|
|
class SupplySource(StrEnum):
|
|
STOCK = "STOCK"
|
|
IN_TRANSIT = "IN_TRANSIT"
|
|
PURCHASE = "PURCHASE"
|
|
SUBCONTRACT = "SUBCONTRACT"
|
|
MAKE = "MAKE"
|
|
|
|
|
|
class RoutingStatus(StrEnum):
|
|
CONFIRMED = "CONFIRMED"
|
|
TEMPLATE = "TEMPLATE"
|
|
MISSING = "MISSING"
|
|
|
|
|
|
class ResourceKind(StrEnum):
|
|
FACTORY = "FACTORY"
|
|
WORKSHOP = "WORKSHOP"
|
|
LINE = "LINE"
|
|
WORKSTATION = "WORKSTATION"
|
|
EQUIPMENT = "EQUIPMENT"
|
|
TEAM = "TEAM"
|
|
TOOLING = "TOOLING"
|
|
|
|
|
|
class SolveStatus(StrEnum):
|
|
OPTIMAL = "OPTIMAL"
|
|
FEASIBLE = "FEASIBLE"
|
|
INFEASIBLE = "INFEASIBLE"
|
|
PARTIAL = "PARTIAL"
|
|
ERROR = "ERROR"
|
|
|
|
|
|
class ObjectiveMode(StrEnum):
|
|
LEXICOGRAPHIC = "LEXICOGRAPHIC"
|
|
WEIGHTED = "WEIGHTED"
|
|
|
|
|
|
class SourceFingerprint(_ContractModel):
|
|
sourceId: str = Field(min_length=1)
|
|
revision: str = Field(min_length=1)
|
|
sha256: str = Field(pattern=r"^[0-9a-fA-F]{64}$")
|
|
|
|
|
|
class TimeInterval(_ContractModel):
|
|
start: datetime
|
|
end: datetime
|
|
label: str | None = None
|
|
|
|
@field_validator("start", "end")
|
|
@classmethod
|
|
def _timezone_required(cls, value: datetime) -> datetime:
|
|
if value.tzinfo is None or value.utcoffset() is None:
|
|
raise ValueError("timezone-aware datetime required")
|
|
return value
|
|
|
|
@model_validator(mode="after")
|
|
def _positive_interval(self) -> "TimeInterval":
|
|
if self.end <= self.start:
|
|
raise ValueError("interval end must be after start")
|
|
return self
|
|
|
|
|
|
class Requirement(_ContractModel):
|
|
requirementId: str = Field(min_length=1)
|
|
orderId: str = Field(min_length=1)
|
|
orderLineId: str = Field(min_length=1)
|
|
materialId: str = Field(min_length=1)
|
|
quantity: float = Field(gt=0)
|
|
sourcingType: SourcingType
|
|
requiredAt: datetime
|
|
# W1 may aggregate one requirement into multiple parent demands. The
|
|
# singular fields remain as a compatibility bridge for existing callers.
|
|
parentRequirementId: str | None = None
|
|
parentRequirementIds: tuple[str, ...] = ()
|
|
requiredByActivityId: str | None = None
|
|
requiredByActivityIds: tuple[str, ...] = ()
|
|
routingStatus: RoutingStatus = RoutingStatus.CONFIRMED
|
|
routingId: str | None = None
|
|
routingVersion: str | None = None
|
|
activityIds: tuple[str, ...] = ()
|
|
sourceRef: str = Field(min_length=1)
|
|
sourceRevision: str = Field(min_length=1)
|
|
sourceHash: str = Field(pattern=r"^[0-9a-fA-F]{64}$")
|
|
|
|
@field_validator("requiredAt")
|
|
@classmethod
|
|
def _timezone_required(cls, value: datetime) -> datetime:
|
|
if value.tzinfo is None or value.utcoffset() is None:
|
|
raise ValueError("timezone-aware datetime required")
|
|
return value
|
|
|
|
|
|
class SupplyEvent(_ContractModel):
|
|
supplyEventId: str = Field(min_length=1)
|
|
requirementId: str = Field(min_length=1)
|
|
materialId: str = Field(min_length=1)
|
|
source: SupplySource
|
|
quantity: float = Field(gt=0)
|
|
availableAt: datetime
|
|
sourceRef: str = Field(min_length=1)
|
|
sourceRevision: str = Field(min_length=1)
|
|
sourceHash: str = Field(pattern=r"^[0-9a-fA-F]{64}$")
|
|
|
|
@field_validator("availableAt")
|
|
@classmethod
|
|
def _timezone_required(cls, value: datetime) -> datetime:
|
|
if value.tzinfo is None or value.utcoffset() is None:
|
|
raise ValueError("timezone-aware datetime required")
|
|
return value
|
|
|
|
|
|
class ActivityResourceRequirement(_ContractModel):
|
|
"""One concurrent resource dimension required by an operation activity."""
|
|
|
|
kind: ResourceKind
|
|
eligibleResourceIds: tuple[str, ...] = ()
|
|
requiredCapabilities: tuple[str, ...] = ()
|
|
units: float = Field(default=1.0, gt=0)
|
|
lifeUnits: float = Field(default=0.0, ge=0)
|
|
|
|
|
|
class OperationActivity(_ContractModel):
|
|
activityId: str = Field(min_length=1)
|
|
activityIdentity: str = Field(pattern=r"^[0-9a-fA-F]{64}$")
|
|
requirementId: str = Field(min_length=1)
|
|
routingId: str = Field(min_length=1)
|
|
routingVersion: str = Field(min_length=1)
|
|
operationId: str = Field(min_length=1)
|
|
operationCode: str = Field(min_length=1)
|
|
sequence: int = Field(ge=1)
|
|
durationMin: float = Field(gt=0)
|
|
predecessorActivityIds: tuple[str, ...] = ()
|
|
inputRequirementIds: tuple[str, ...] = ()
|
|
eligibleResourceIds: tuple[str, ...] = ()
|
|
requiredCapabilities: tuple[str, ...] = ()
|
|
requiredResourceUnits: float = Field(default=1.0, gt=0)
|
|
# Empty keeps the V1 single-equipment contract. Production adapters emit
|
|
# one requirement per concurrent dimension (equipment, team, tooling).
|
|
resourceRequirements: tuple[ActivityResourceRequirement, ...] = ()
|
|
materialReleaseAt: datetime | None = None
|
|
continuous: bool = False
|
|
sourceRef: str = Field(min_length=1)
|
|
sourceRevision: str = Field(min_length=1)
|
|
sourceHash: str = Field(pattern=r"^[0-9a-fA-F]{64}$")
|
|
|
|
@field_validator("materialReleaseAt")
|
|
@classmethod
|
|
def _timezone_required(cls, value: datetime | None) -> datetime | None:
|
|
if value is not None and (value.tzinfo is None or value.utcoffset() is None):
|
|
raise ValueError("timezone-aware datetime required")
|
|
return value
|
|
|
|
|
|
class Resource(_ContractModel):
|
|
resourceId: str = Field(min_length=1)
|
|
code: str = Field(min_length=1)
|
|
name: str = ""
|
|
kind: ResourceKind
|
|
parentResourceId: str | None = None
|
|
capabilities: tuple[str, ...] = ()
|
|
compatibleResourceIds: tuple[str, ...] = ()
|
|
calendarIntervals: tuple[TimeInterval, ...]
|
|
maintenanceIntervals: tuple[TimeInterval, ...] = ()
|
|
cumulativeCapacity: float = Field(default=1.0, gt=0)
|
|
dailyCapacityMinutes: float | None = Field(default=None, gt=0)
|
|
lifeTotal: float | None = Field(default=None, gt=0)
|
|
lifeUsed: float = Field(default=0.0, ge=0)
|
|
sourceRef: str = Field(min_length=1)
|
|
sourceRevision: str = Field(min_length=1)
|
|
sourceHash: str = Field(pattern=r"^[0-9a-fA-F]{64}$")
|
|
|
|
|
|
class ObjectivePolicy(_ContractModel):
|
|
mode: ObjectiveMode = ObjectiveMode.WEIGHTED
|
|
priorities: tuple[str, ...] = ("weightedTardiness",)
|
|
weights: Mapping[str, float] = Field(default_factory=lambda: {"weightedTardiness": 1.0})
|
|
version: str = Field(default="1", min_length=1)
|
|
|
|
@field_validator("weights")
|
|
@classmethod
|
|
def _non_negative_weights(cls, value: Mapping[str, float]) -> Mapping[str, float]:
|
|
if any(float(weight) < 0 for weight in value.values()):
|
|
raise ValueError("objective weights must be non-negative")
|
|
return dict(value)
|
|
|
|
|
|
class ConstraintPolicy(_ContractModel):
|
|
enabled: Mapping[str, bool] = Field(default_factory=dict)
|
|
hardConstraints: tuple[str, ...] = ()
|
|
penalties: Mapping[str, float] = Field(default_factory=dict)
|
|
version: str = Field(default="1", min_length=1)
|
|
|
|
@field_validator("penalties")
|
|
@classmethod
|
|
def _non_negative_penalties(cls, value: Mapping[str, float]) -> Mapping[str, float]:
|
|
if any(float(penalty) < 0 for penalty in value.values()):
|
|
raise ValueError("constraint penalties must be non-negative")
|
|
return dict(value)
|
|
|
|
|
|
class SchedulingProblemV2(_ContractModel):
|
|
schemaVersion: str = Field(default="2.0", pattern=r"^2(?:\.[0-9]+)?$")
|
|
problemId: str = Field(min_length=1)
|
|
businessDate: date
|
|
planningStart: datetime
|
|
planningEnd: datetime
|
|
requirements: tuple[Requirement, ...]
|
|
supplyEvents: tuple[SupplyEvent, ...] = ()
|
|
activities: tuple[OperationActivity, ...] = ()
|
|
resources: tuple[Resource, ...]
|
|
objectivePolicy: ObjectivePolicy = Field(default_factory=ObjectivePolicy)
|
|
constraintPolicy: ConstraintPolicy = Field(default_factory=ConstraintPolicy)
|
|
sourceRevision: str = Field(min_length=1)
|
|
sourceFingerprints: tuple[SourceFingerprint, ...]
|
|
metadata: Mapping[str, Any] = Field(default_factory=dict)
|
|
|
|
@field_validator("planningStart", "planningEnd")
|
|
@classmethod
|
|
def _timezone_required(cls, value: datetime) -> datetime:
|
|
if value.tzinfo is None or value.utcoffset() is None:
|
|
raise ValueError("timezone-aware datetime required")
|
|
return value
|
|
|
|
@model_validator(mode="after")
|
|
def _positive_horizon(self) -> "SchedulingProblemV2":
|
|
if self.planningEnd <= self.planningStart:
|
|
raise ValueError("planningEnd must be after planningStart")
|
|
return self
|
|
|
|
|
|
class ScheduledResourceAllocation(_ContractModel):
|
|
"""Concrete resource occupied by a scheduled activity."""
|
|
|
|
resourceId: str = Field(min_length=1)
|
|
kind: ResourceKind
|
|
units: float = Field(default=1.0, gt=0)
|
|
lifeUnits: float = Field(default=0.0, ge=0)
|
|
|
|
|
|
class ScheduledActivity(_ContractModel):
|
|
activityId: str = Field(min_length=1)
|
|
activityIdentity: str = Field(pattern=r"^[0-9a-fA-F]{64}$")
|
|
requirementId: str = Field(min_length=1)
|
|
operationId: str = Field(min_length=1)
|
|
sequence: int = Field(ge=1)
|
|
resourceId: str = Field(min_length=1)
|
|
start: datetime
|
|
end: datetime
|
|
resourceUnits: float = Field(default=1.0, gt=0)
|
|
# Empty means the legacy resourceId/resourceUnits pair is the equipment
|
|
# allocation. New production candidates must enumerate every dimension.
|
|
resourceAllocations: tuple[ScheduledResourceAllocation, ...] = ()
|
|
|
|
@field_validator("start", "end")
|
|
@classmethod
|
|
def _timezone_required(cls, value: datetime) -> datetime:
|
|
if value.tzinfo is None or value.utcoffset() is None:
|
|
raise ValueError("timezone-aware datetime required")
|
|
return value
|
|
|
|
|
|
class SupplyDecision(_ContractModel):
|
|
decisionId: str = Field(min_length=1)
|
|
supplyEventId: str = Field(min_length=1)
|
|
requirementId: str = Field(min_length=1)
|
|
materialId: str = Field(min_length=1)
|
|
source: SupplySource
|
|
quantity: float = Field(gt=0)
|
|
availableAt: datetime
|
|
consumedByActivityId: str | None = None
|
|
|
|
@field_validator("availableAt")
|
|
@classmethod
|
|
def _timezone_required(cls, value: datetime) -> datetime:
|
|
if value.tzinfo is None or value.utcoffset() is None:
|
|
raise ValueError("timezone-aware datetime required")
|
|
return value
|
|
|
|
|
|
class PeggingAllocation(_ContractModel):
|
|
peggingId: str = Field(min_length=1)
|
|
childRequirementId: str = Field(min_length=1)
|
|
parentRequirementId: str = Field(min_length=1)
|
|
quantity: float = Field(gt=0)
|
|
|
|
|
|
class UnscheduledRequirement(_ContractModel):
|
|
requirementId: str = Field(min_length=1)
|
|
quantity: float = Field(gt=0)
|
|
reasonCode: str = Field(min_length=1)
|
|
details: str = ""
|
|
|
|
|
|
class HardViolation(_ContractModel):
|
|
code: str = Field(min_length=1)
|
|
message: str = Field(min_length=1)
|
|
entityRefs: tuple[str, ...] = ()
|
|
|
|
|
|
class Assumption(_ContractModel):
|
|
code: str = Field(min_length=1)
|
|
message: str = Field(min_length=1)
|
|
sourceRef: str = Field(min_length=1)
|
|
confidence: float = Field(ge=0, le=1)
|
|
|
|
|
|
class SolutionProvenance(_ContractModel):
|
|
runId: str = Field(min_length=1)
|
|
solverId: str = Field(min_length=1)
|
|
solverVersion: str = Field(min_length=1)
|
|
generatedAt: datetime
|
|
businessDate: date
|
|
problemHash: str = Field(pattern=r"^[0-9a-fA-F]{64}$")
|
|
sourceRevision: str = Field(min_length=1)
|
|
sourceFingerprints: tuple[SourceFingerprint, ...]
|
|
|
|
@field_validator("generatedAt")
|
|
@classmethod
|
|
def _timezone_required(cls, value: datetime) -> datetime:
|
|
if value.tzinfo is None or value.utcoffset() is None:
|
|
raise ValueError("timezone-aware datetime required")
|
|
return value
|
|
|
|
|
|
class SchedulingSolutionV2(_ContractModel):
|
|
schemaVersion: str = Field(default="2.0", pattern=r"^2(?:\.[0-9]+)?$")
|
|
problemId: str = Field(min_length=1)
|
|
solveStatus: SolveStatus
|
|
objectiveValues: Mapping[str, float] = Field(default_factory=dict)
|
|
bestBound: float | None = None
|
|
gap: float | None = Field(default=None, ge=0)
|
|
activities: tuple[ScheduledActivity, ...] = ()
|
|
supplyDecisions: tuple[SupplyDecision, ...] = ()
|
|
pegging: tuple[PeggingAllocation, ...] = ()
|
|
unscheduledRequirements: tuple[UnscheduledRequirement, ...] = ()
|
|
hardViolations: tuple[HardViolation, ...] = ()
|
|
assumptions: tuple[Assumption, ...] = ()
|
|
provenance: SolutionProvenance
|
|
|
|
|
|
def canonical_sha256(value: Any) -> str:
|
|
"""Return a deterministic SHA-256 for JSON-compatible contract content."""
|
|
|
|
if isinstance(value, BaseModel):
|
|
value = value.model_dump(mode="json")
|
|
payload = json.dumps(
|
|
value,
|
|
ensure_ascii=False,
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
default=str,
|
|
).encode("utf-8")
|
|
return hashlib.sha256(payload).hexdigest()
|
|
|
|
|
|
def operation_activity_identity(activity: OperationActivity | Mapping[str, Any]) -> str:
|
|
"""Hash immutable operation identity, excluding the self-referential hash field."""
|
|
|
|
raw = activity.model_dump(mode="json") if isinstance(activity, OperationActivity) else dict(activity)
|
|
payload = {
|
|
"activityId": raw.get("activityId"),
|
|
"requirementId": raw.get("requirementId"),
|
|
"routingId": raw.get("routingId"),
|
|
"routingVersion": raw.get("routingVersion"),
|
|
"operationId": raw.get("operationId"),
|
|
"operationCode": raw.get("operationCode"),
|
|
"sequence": raw.get("sequence"),
|
|
"durationMin": raw.get("durationMin"),
|
|
"predecessorActivityIds": sorted(raw.get("predecessorActivityIds") or []),
|
|
"inputRequirementIds": sorted(raw.get("inputRequirementIds") or []),
|
|
"eligibleResourceIds": sorted(raw.get("eligibleResourceIds") or []),
|
|
"requiredCapabilities": sorted(raw.get("requiredCapabilities") or []),
|
|
"requiredResourceUnits": raw.get("requiredResourceUnits", 1.0),
|
|
"resourceRequirements": sorted(
|
|
[
|
|
{
|
|
"kind": str(
|
|
requirement.kind.value
|
|
if isinstance(requirement, ActivityResourceRequirement)
|
|
else requirement.get("kind")
|
|
),
|
|
"eligibleResourceIds": sorted(
|
|
requirement.eligibleResourceIds
|
|
if isinstance(requirement, ActivityResourceRequirement)
|
|
else requirement.get("eligibleResourceIds") or []
|
|
),
|
|
"requiredCapabilities": sorted(
|
|
requirement.requiredCapabilities
|
|
if isinstance(requirement, ActivityResourceRequirement)
|
|
else requirement.get("requiredCapabilities") or []
|
|
),
|
|
"units": (
|
|
requirement.units
|
|
if isinstance(requirement, ActivityResourceRequirement)
|
|
else requirement.get("units", 1.0)
|
|
),
|
|
"lifeUnits": (
|
|
requirement.lifeUnits
|
|
if isinstance(requirement, ActivityResourceRequirement)
|
|
else requirement.get("lifeUnits", 0.0)
|
|
),
|
|
}
|
|
for requirement in raw.get("resourceRequirements") or []
|
|
],
|
|
key=lambda item: (item["kind"], item["eligibleResourceIds"]),
|
|
),
|
|
"continuous": bool(raw.get("continuous", False)),
|
|
"sourceRef": raw.get("sourceRef"),
|
|
"sourceRevision": raw.get("sourceRevision"),
|
|
"sourceHash": raw.get("sourceHash"),
|
|
}
|
|
return canonical_sha256(payload)
|
|
|
|
|
|
def scheduling_problem_hash(problem: SchedulingProblemV2) -> str:
|
|
"""Hash the full immutable V2 input, including source fingerprints and policies."""
|
|
|
|
return canonical_sha256(problem)
|