1062 lines
76 KiB
Python
1062 lines
76 KiB
Python
from __future__ import annotations
|
|
|
|
from collections import Counter, defaultdict
|
|
from collections.abc import Iterable
|
|
from dataclasses import asdict
|
|
from datetime import date, datetime, time, timedelta
|
|
from itertools import pairwise
|
|
from math import isclose
|
|
from typing import Any
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from .alternatives import ALTERNATIVE_MODES
|
|
from .config import (
|
|
DATA_DISCLAIMER,
|
|
DATASET_TYPE,
|
|
GENERATED_FOR,
|
|
ORGANIZATION_SCENARIO,
|
|
TIMEZONE,
|
|
GeneratorConfig,
|
|
)
|
|
from .models import DatasetBundle, stable_hash
|
|
from .rag_skills import RAG_CATEGORIES, SKILL_IDS
|
|
from .registry import TABLE_SPECS
|
|
from .scenarios import CANONICAL_FILES, EVENT_TYPES
|
|
|
|
__all__ = ["validate_bundle"]
|
|
|
|
_ALLOWED_SOURCING = {"MAKE", "BUY", "OUTSOURCE", "OWNER_SUPPLIED"}
|
|
_ALLOWED_FULFILLMENT = {"STOCK", "TRANSFER", "PLANNED_RECEIPT", "NEW_SUPPLY", "DESIGN_PENDING"}
|
|
_EXECUTION_TABLES = ("mes-orders", "operation-reports", "material-issues", "quality-inspections", "nonconformities", "rework-orders")
|
|
_SKILL_FIELDS = ("skillId", "name", "description", "inputSchema", "outputSchema", "requiredData", "algorithmCandidates", "hardConstraints", "softConstraints", "fallbackAlgorithm", "timeoutSeconds", "validationRules", "evidenceFields", "version")
|
|
_KNOWLEDGE_FIELDS = ("knowledgeId", "category", "title", "content", "sourceType", "version", "effectiveDate", "confidence", "tags", "evidenceRef")
|
|
_ALTERNATIVES = {"DELIVERY_FIRST", "RESOURCE_BALANCED", "COST_FIRST", "MIN_OVERTIME", "DOCK_UTILIZATION", "MIN_RISK", "RECOMMENDED"}
|
|
_METRICS = (
|
|
"projectCount", "sectionCount", "materialCount", "bomLineCount", "productionOrderCount",
|
|
"purchaseSuggestionCount", "outsourceSuggestionCount", "operationCount", "resourceCount", "scheduleSlotCount",
|
|
"foreignKeyErrorCount", "requiredFieldErrorCount", "duplicatePrimaryKeyCount", "missingTableCount",
|
|
"scaleCountMismatchCount", "syntheticBoundaryViolationCount", "missingMakeRoutingCount",
|
|
"makeDesignReleaseViolationCount", "missingBuyOutsourceSupplierCount", "sourcingModeViolationCount",
|
|
"fulfillmentModeViolationCount", "nettingEquationViolationCount", "purchaseTraceViolationCount",
|
|
"productionRoutingCoverageViolationCount", "productionResourceCoverageViolationCount",
|
|
"activeOperationCoverageViolationCount", "duplicateOperationScheduleCount", "resourceOverlapCount",
|
|
"teamOverlapCount", "precedenceViolationCount", "materialReadinessViolationCount", "illegalFrozenChangeCount",
|
|
"maintenanceCalendarViolationCount", "calendarViolationCount", "resourceTypeViolationCount",
|
|
"resourceCapabilityViolationCount", "resourceWeightDimensionViolationCount", "teamSkillViolationCount",
|
|
"teamCrewViolationCount", "teamQualificationViolationCount", "dockConflictCount", "berthConflictCount",
|
|
"fixtureConflictCount", "paintBoothConflictCount", "liftingCapacityViolationCount",
|
|
"transportConstraintViolationCount", "weatherWindowViolationCount", "hotWorkDensityViolationCount",
|
|
"qualityHoldPointViolationCount", "reworkClosureViolationCount", "reinspectionViolationCount",
|
|
"scheduleVersionEvidenceViolationCount", "unexplainedScheduleDecisionCount", "unresolvedEvidenceRefCount",
|
|
"skillRegistryViolationCount", "skillSchemaViolationCount", "ragKnowledgeViolationCount",
|
|
"alternativeViolationCount", "scenarioViolationCount", "infeasibleCoreViolationCount",
|
|
"executionClosureViolationCount", "supplierCapacityViolationCount", "supplierBlackoutViolationCount",
|
|
"outsourceTimeChainViolationCount", "outsourceSupplierEligibilityViolationCount",
|
|
"constraintCoverageViolationCount", "alternativeScheduleViolationCount", "scenarioDiffViolationCount",
|
|
"unmarkedHardConstraintViolationCount",
|
|
)
|
|
|
|
|
|
def _empty(value: Any) -> bool:
|
|
return value is None or value == "" or value == [] or value == {}
|
|
|
|
|
|
def _num(value: Any, default: float = 0.0) -> float:
|
|
try:
|
|
return float(value)
|
|
except (TypeError, ValueError):
|
|
return default
|
|
|
|
|
|
def _integer(value: Any, default: int = 0) -> int:
|
|
try:
|
|
return int(value)
|
|
except (TypeError, ValueError):
|
|
return default
|
|
|
|
|
|
def _date(value: Any) -> date | None:
|
|
if isinstance(value, datetime):
|
|
return value.date()
|
|
if isinstance(value, date):
|
|
return value
|
|
try:
|
|
return date.fromisoformat(str(value or "")[:10])
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def _datetime(value: Any, timezone: ZoneInfo) -> datetime | None:
|
|
try:
|
|
parsed = value if isinstance(value, datetime) else datetime.fromisoformat(str(value or ""))
|
|
except ValueError:
|
|
return None
|
|
return parsed.replace(tzinfo=timezone) if parsed.tzinfo is None else parsed.astimezone(timezone)
|
|
|
|
|
|
def _clock(value: Any, fallback: time) -> time:
|
|
try:
|
|
return time.fromisoformat(str(value or ""))
|
|
except ValueError:
|
|
return fallback
|
|
|
|
|
|
def _refs(value: Any) -> list[str]:
|
|
if value is None:
|
|
return []
|
|
if isinstance(value, str):
|
|
return [part.strip() for part in value.split("|") if part.strip()]
|
|
if isinstance(value, Iterable) and not isinstance(value, (bytes, bytearray, dict)):
|
|
return [str(part).strip() for part in value if str(part).strip()]
|
|
return [str(value).strip()] if str(value).strip() else []
|
|
|
|
|
|
def _calendar_span(calendar: dict[str, Any] | None, shifts: list[dict[str, Any]], day: date, timezone: ZoneInfo) -> tuple[datetime, datetime] | None:
|
|
if calendar is None or calendar.get("status", "ACTIVE") != "ACTIVE":
|
|
return None
|
|
start_day, end_day = _date(calendar.get("effectiveFrom")), _date(calendar.get("effectiveTo"))
|
|
if (start_day and day < start_day) or (end_day and day > end_day):
|
|
return None
|
|
if day.isoformat() in {str(value)[:10] for value in calendar.get("blackoutDates") or []}:
|
|
return None
|
|
working_days = {_integer(value) for value in calendar.get("workingDays") or []}
|
|
if working_days and day.isoweekday() not in working_days:
|
|
return None
|
|
active = [row for row in shifts if row.get("status", "ACTIVE") == "ACTIVE"]
|
|
if not active:
|
|
return None
|
|
spans: list[tuple[datetime, datetime]] = []
|
|
for shift in active:
|
|
start = datetime.combine(day, _clock(shift.get("startTime"), time(8)), tzinfo=timezone)
|
|
end = datetime.combine(day, _clock(shift.get("endTime"), time(16)), tzinfo=timezone)
|
|
if end <= start:
|
|
end += timedelta(days=1)
|
|
spans.append((start, end))
|
|
return min(value[0] for value in spans), max(value[1] for value in spans)
|
|
|
|
|
|
def _expected_counts(config: GeneratorConfig) -> dict[str, int]:
|
|
p = config.profile
|
|
return {
|
|
"ship-projects": p.project_count, "contracts": p.project_count, "milestones": p.milestone_count,
|
|
"grand-blocks": p.grand_block_count, "sections": p.section_count, "work-packages": p.work_package_count,
|
|
"wbs-tasks": p.wbs_task_count, "materials": p.material_count, "bom-lines": p.bom_relation_count,
|
|
"routings": p.routing_count, "production-orders": p.production_order_count,
|
|
"work-orders": p.production_order_count, "operations": p.operation_count, "schedule-slots": p.schedule_slot_count,
|
|
"resources": p.equipment_resource_count, "teams": p.team_count, "suppliers": p.supplier_count,
|
|
"purchase-suggestions": p.purchase_suggestion_count, "outsource-suggestions": p.outsource_suggestion_count,
|
|
"conflicts": p.risk_conflict_count, "knowledge-assets": p.knowledge_asset_count,
|
|
}
|
|
|
|
|
|
def _schedule_input_digest(bundle: DatasetBundle, config: GeneratorConfig) -> str:
|
|
operations = sorted(
|
|
[
|
|
{key: value for key, value in row.items() if key not in {"timeFence", "frozenBaselineStart", "frozenChangeAuthorized"}}
|
|
for row in bundle.rows("operations")
|
|
if row.get("active", True)
|
|
],
|
|
key=lambda row: (str(row.get("materialReadyAt") or ""), str(row.get("needDate") or ""), str(row.get("productionOrderId") or ""), _integer(row.get("sequence")), str(row.get("operationId") or "")),
|
|
)
|
|
resources = sorted([row for row in bundle.rows("resources") if row.get("status", "ACTIVE") == "ACTIVE"], key=lambda row: str(row.get("resourceCode") or row.get("resourceId") or ""))
|
|
teams = sorted([row for row in bundle.rows("teams") if row.get("status", "ACTIVE") == "ACTIVE"], key=lambda row: str(row.get("teamCode") or row.get("teamId") or ""))
|
|
return stable_hash({"operations": operations, "resources": resources, "teams": teams, "calendars": bundle.rows("calendars"), "shifts": bundle.rows("shifts"), "outsourceSuggestions": bundle.rows("outsource-suggestions"), "planningBaseDate": config.planning_base_date.isoformat(), "planningHorizonEnd": config.planning_horizon_end.isoformat()})
|
|
|
|
|
|
def validate_bundle(bundle: DatasetBundle, config: GeneratorConfig) -> dict:
|
|
"""Validate a complete synthetic shipyard APS bundle without mutating it."""
|
|
metrics = {name: 0 for name in _METRICS}
|
|
violations: list[dict[str, Any]] = []
|
|
checks: Counter[str] = Counter()
|
|
|
|
def fail(metric: str, code: str, entity: Any, message: str) -> None:
|
|
metrics[metric] += 1
|
|
violations.append({"code": code, "metric": metric, "entityId": str(entity or "dataset"), "message": message, "severity": "BLOCKING"})
|
|
|
|
def rows(name: str) -> list[dict[str, Any]]:
|
|
value = bundle.tables.get(name)
|
|
return value if isinstance(value, list) else []
|
|
|
|
table_counts = {name: len(rows(name)) for name in TABLE_SPECS}
|
|
indexes: dict[str, dict[str, dict[Any, dict[str, Any]]]] = {}
|
|
for table_name, spec in TABLE_SPECS.items():
|
|
table_rows = rows(table_name)
|
|
checks["tableSpecCount"] += 1
|
|
if table_name not in bundle.tables:
|
|
fail("missingTableCount", "MISSING_TABLE", table_name, "TABLE_SPECS table is absent")
|
|
seen: set[Any] = set()
|
|
for row_no, row in enumerate(table_rows, 1):
|
|
checks["rowSchemaCheckCount"] += 1
|
|
for field in spec.required:
|
|
if field not in row or _empty(row.get(field)):
|
|
fail("requiredFieldErrorCount", "REQUIRED_FIELD", f"{table_name}[{row_no}]", f"required field {field} is missing or empty")
|
|
primary_key = row.get(spec.primary_key)
|
|
if _empty(primary_key) or primary_key in seen:
|
|
fail("duplicatePrimaryKeyCount", "PRIMARY_KEY", f"{table_name}[{row_no}]", f"primary key {spec.primary_key} is empty or duplicated: {primary_key!r}")
|
|
seen.add(primary_key)
|
|
target_fields = {spec.primary_key, *(fk.target_field for fk in spec.foreign_keys)}
|
|
indexes[table_name] = {field: {row.get(field): row for row in table_rows if not _empty(row.get(field))} for field in target_fields}
|
|
indexes["ship-projects"]["projectId"].update(
|
|
{row["projectCode"]: row for row in rows("ship-projects") if not _empty(row.get("projectCode"))}
|
|
)
|
|
|
|
for table_name, spec in TABLE_SPECS.items():
|
|
for row_no, row in enumerate(rows(table_name), 1):
|
|
for foreign_key in spec.foreign_keys:
|
|
checks["foreignKeyCheckCount"] += 1
|
|
value = row.get(foreign_key.field)
|
|
if _empty(value) and foreign_key.nullable:
|
|
continue
|
|
target = indexes.get(foreign_key.target_table, {}).get(foreign_key.target_field, {})
|
|
if _empty(value) or value not in target:
|
|
fail("foreignKeyErrorCount", "FOREIGN_KEY", f"{table_name}[{row_no}]", f"{foreign_key.field}={value!r} does not resolve to {foreign_key.target_table}.{foreign_key.target_field}")
|
|
|
|
metadata_expected = {
|
|
"datasetType": DATASET_TYPE, "organizationScenario": ORGANIZATION_SCENARIO, "timezone": TIMEZONE,
|
|
"generatedFor": GENERATED_FOR, "dataDisclaimer": DATA_DISCLAIMER, "randomSeed": config.seed,
|
|
"planningBaseDate": config.planning_base_date.isoformat(), "planningHorizonEnd": config.planning_horizon_end.isoformat(),
|
|
}
|
|
for field, expected in metadata_expected.items():
|
|
checks["syntheticBoundaryCheckCount"] += 1
|
|
if bundle.metadata.get(field) != expected:
|
|
fail("syntheticBoundaryViolationCount", "SYNTHETIC_METADATA", field, f"expected {expected!r}, found {bundle.metadata.get(field)!r}")
|
|
projects = rows("ship-projects")
|
|
project_codes = [str(row.get("projectCode") or "") for row in projects]
|
|
if sorted(project_codes) != sorted(config.project_codes):
|
|
fail("syntheticBoundaryViolationCount", "PROJECT_CODES", "ship-projects", f"project codes must exactly match {list(config.project_codes)!r}")
|
|
for row in projects:
|
|
if not str(row.get("projectCode") or "").startswith("BH-SYN-"):
|
|
fail("syntheticBoundaryViolationCount", "PROJECT_CODE", row.get("projectId"), "projectCode must start with BH-SYN-")
|
|
for row in rows("contracts"):
|
|
if not str(row.get("shipOwner") or "").startswith("船东-SYN-"):
|
|
fail("syntheticBoundaryViolationCount", "SHIP_OWNER_NAME", row.get("contractId"), "ship owner must use 船东-SYN-* naming")
|
|
for row in rows("suppliers"):
|
|
if not str(row.get("name") or row.get("supplierName") or "").startswith("供应商-SYN-"):
|
|
fail("syntheticBoundaryViolationCount", "SUPPLIER_NAME", row.get("supplierId"), "supplier must use 供应商-SYN-* naming")
|
|
for row in rows("employees"):
|
|
if "模拟" not in str(row.get("name") or ""):
|
|
fail("syntheticBoundaryViolationCount", "EMPLOYEE_NAME", row.get("employeeId"), "employee name must identify a simulated worker")
|
|
|
|
blocks = rows("blocks")
|
|
actual_counts = {
|
|
"ship-projects": len(projects), "contracts": len(rows("contracts")), "milestones": len(rows("milestones")),
|
|
"grand-blocks": sum(row.get("blockType") == "GRAND_BLOCK" for row in blocks),
|
|
"sections": sum(row.get("blockType") == "SECTION" for row in blocks),
|
|
"work-packages": len(rows("work-packages")), "wbs-tasks": sum(row.get("wbsType") == "TASK" for row in rows("wbs")),
|
|
"materials": len(rows("materials")), "bom-lines": sum(len(rows(name)) for name in ("ebom", "pbom", "mbom")),
|
|
"routings": len(rows("routings")), "production-orders": len(rows("production-orders")),
|
|
"work-orders": len(rows("work-orders")), "operations": len(rows("operations")),
|
|
"schedule-slots": len(rows("schedule-slots")), "resources": len(rows("resources")),
|
|
"teams": len(rows("teams")), "suppliers": len(rows("suppliers")),
|
|
"purchase-suggestions": len(rows("purchase-suggestions")), "outsource-suggestions": len(rows("outsource-suggestions")),
|
|
"conflicts": len(rows("conflicts")), "knowledge-assets": len(bundle.artifacts.get("knowledge-assets") or []),
|
|
}
|
|
expected_counts = _expected_counts(config)
|
|
for name, expected in expected_counts.items():
|
|
checks["scaleCountCheckCount"] += 1
|
|
if actual_counts.get(name) != expected:
|
|
fail("scaleCountMismatchCount", "EXACT_SCALE_COUNT", name, f"expected {expected}, found {actual_counts.get(name, 0)}")
|
|
metrics.update({
|
|
"projectCount": actual_counts["ship-projects"], "sectionCount": actual_counts["sections"],
|
|
"materialCount": actual_counts["materials"], "bomLineCount": actual_counts["bom-lines"],
|
|
"productionOrderCount": actual_counts["production-orders"], "purchaseSuggestionCount": actual_counts["purchase-suggestions"],
|
|
"outsourceSuggestionCount": actual_counts["outsource-suggestions"], "operationCount": actual_counts["operations"],
|
|
"resourceCount": actual_counts["resources"], "scheduleSlotCount": actual_counts["schedule-slots"],
|
|
})
|
|
|
|
materials = {str(row.get("materialId")): row for row in rows("materials")}
|
|
routings = {str(row.get("routingId")): row for row in rows("routings")}
|
|
suppliers = {str(row.get("supplierId")): row for row in rows("suppliers")}
|
|
routing_operations: dict[str, list[dict[str, Any]]] = defaultdict(list)
|
|
for row in rows("routing-operations"):
|
|
routing_operations[str(row.get("routingId"))].append(row)
|
|
for material_id, material in materials.items():
|
|
sourcing, fulfillment = str(material.get("sourcingMode") or ""), str(material.get("fulfillmentMode") or "")
|
|
checks["sourcingCheckCount"] += 1
|
|
if sourcing not in _ALLOWED_SOURCING:
|
|
fail("sourcingModeViolationCount", "SOURCING_MODE", material_id, f"unsupported sourcingMode {sourcing!r}")
|
|
if fulfillment not in _ALLOWED_FULFILLMENT:
|
|
fail("fulfillmentModeViolationCount", "FULFILLMENT_MODE", material_id, f"unsupported fulfillmentMode {fulfillment!r}")
|
|
if sourcing == "MAKE":
|
|
routing_id = str(material.get("routingId") or "")
|
|
if routing_id not in routings or not routing_operations.get(routing_id):
|
|
fail("missingMakeRoutingCount", "MAKE_ROUTING", material_id, "MAKE material lacks an effective routing and operations")
|
|
if material.get("designStatus") != "RELEASED" and fulfillment != "DESIGN_PENDING":
|
|
fail(
|
|
"makeDesignReleaseViolationCount",
|
|
"MAKE_DESIGN_RELEASE",
|
|
material_id,
|
|
"unreleased MAKE material must remain explicit DESIGN_PENDING",
|
|
)
|
|
if sourcing in {"BUY", "OUTSOURCE"}:
|
|
supplier_id = str(material.get("primarySupplierId") or "")
|
|
explicit_risk = any(not _empty(material.get(field)) for field in ("riskLevel", "supplyRisk", "supplyRiskId", "dataQualityIssue"))
|
|
if supplier_id not in suppliers and not explicit_risk:
|
|
fail("missingBuyOutsourceSupplierCount", "SUPPLIER_OR_RISK", material_id, f"{sourcing} material lacks a valid supplier and explicit risk")
|
|
|
|
requirements = {str(row.get("requirementId")): row for row in rows("material-requirements")}
|
|
for requirement_id, requirement in requirements.items():
|
|
checks["nettingEquationCheckCount"] += 1
|
|
expected_net = max(0.0, _num(requirement.get("grossRequirement")) + _num(requirement.get("safetyStockRequirement")) + _num(requirement.get("scrapRequirement")) - _num(requirement.get("stockUsed")) - _num(requirement.get("releasedAllocationUsed")) - _num(requirement.get("plannedReceiptUsed")) - _num(requirement.get("substituteUsed")))
|
|
if not isclose(expected_net, _num(requirement.get("netRequirement")), abs_tol=0.002):
|
|
fail("nettingEquationViolationCount", "NETTING_EQUATION", requirement_id, f"netRequirement must be {expected_net:.3f}")
|
|
if _num(requirement.get("netRequirement")) < -0.0001:
|
|
fail("nettingEquationViolationCount", "NEGATIVE_NET_REQUIREMENT", requirement_id, "net requirement cannot be negative")
|
|
|
|
for suggestion in rows("purchase-suggestions"):
|
|
suggestion_id = suggestion.get("purchaseSuggestionId")
|
|
requirement = requirements.get(str(suggestion.get("requirementId") or ""))
|
|
checks["purchaseTraceCheckCount"] += 1
|
|
if requirement is None:
|
|
fail("purchaseTraceViolationCount", "PURCHASE_REQUIREMENT", suggestion_id, "purchase suggestion has no source requirement")
|
|
continue
|
|
for field in ("projectId", "wbsId", "materialId", "needDate"):
|
|
if suggestion.get(field) != requirement.get(field):
|
|
fail("purchaseTraceViolationCount", "PURCHASE_TRACE", suggestion_id, f"{field} does not match source requirement")
|
|
for field in ("grossRequirement", "stockUsed", "releasedAllocationUsed", "plannedReceiptUsed", "safetyStockRequirement", "scrapRequirement", "netRequirement"):
|
|
if not isclose(_num(suggestion.get(field)), _num(requirement.get(field)), abs_tol=0.002):
|
|
fail("purchaseTraceViolationCount", "PURCHASE_NETTING_DETAIL", suggestion_id, f"{field} does not match source requirement")
|
|
material = materials.get(str(requirement.get("materialId"))) or {}
|
|
net, quantity = _num(requirement.get("netRequirement")), _num(suggestion.get("suggestedQuantity"))
|
|
moq, multiple = max(0.0, _num(material.get("moq"))), max(0.0, _num(material.get("orderMultiple")))
|
|
if net <= 0 or material.get("sourcingMode") != "BUY":
|
|
fail("purchaseTraceViolationCount", "PURCHASE_ELIGIBILITY", suggestion_id, "purchase requires BUY sourcing and positive net shortage")
|
|
if quantity + 0.002 < max(net, moq):
|
|
fail("purchaseTraceViolationCount", "PURCHASE_QUANTITY", suggestion_id, "suggested quantity is below net requirement or MOQ")
|
|
if multiple > 0 and not isclose(quantity / multiple, round(quantity / multiple), abs_tol=0.002):
|
|
fail("purchaseTraceViolationCount", "PURCHASE_MULTIPLE", suggestion_id, "suggested quantity does not respect order multiple")
|
|
if str(suggestion.get("supplierId") or "") not in suppliers or not _refs(suggestion.get("evidenceRefs")):
|
|
fail("purchaseTraceViolationCount", "PURCHASE_SUPPLIER_EVIDENCE", suggestion_id, "supplier/evidence trace is incomplete")
|
|
|
|
production_orders = {str(row.get("productionOrderId")): row for row in rows("production-orders")}
|
|
work_orders = {str(row.get("workOrderId")): row for row in rows("work-orders")}
|
|
operations = {str(row.get("operationId")): row for row in rows("operations")}
|
|
operations_by_order: dict[str, list[dict[str, Any]]] = defaultdict(list)
|
|
for operation in operations.values():
|
|
operations_by_order[str(operation.get("productionOrderId"))].append(operation)
|
|
demands_by_operation = {str(row.get("operationId")) for row in rows("capacity-demands")}
|
|
mbom_by_order = Counter(str(row.get("productionOrderId")) for row in rows("mbom"))
|
|
for order_id, order in production_orders.items():
|
|
checks["productionCoverageCheckCount"] += 1
|
|
routing_id = str(order.get("routingId") or "")
|
|
order_operations = operations_by_order.get(order_id, [])
|
|
if routing_id not in routings or not routing_operations.get(routing_id) or not order_operations or not mbom_by_order.get(order_id):
|
|
fail("productionRoutingCoverageViolationCount", "PRODUCTION_ROUTING_COVERAGE", order_id, "production order must have MBOM, routing and operations")
|
|
if order.get("designReleaseStatus") != "RELEASED" or order.get("status") not in {"RELEASED", "ACTIVE", "IN_PROGRESS", "COMPLETED"}:
|
|
fail("productionRoutingCoverageViolationCount", "PRODUCTION_DESIGN_RELEASE", order_id, "formal production order lacks released design")
|
|
if any(str(operation.get("operationId")) not in demands_by_operation for operation in order_operations):
|
|
fail("productionResourceCoverageViolationCount", "PRODUCTION_RESOURCE_COVERAGE", order_id, "one or more operations lack capacity demand")
|
|
|
|
outsource_by_operation: dict[str, dict[str, Any]] = {}
|
|
for suggestion in rows("outsource-suggestions"):
|
|
suggestion_id = suggestion.get("outsourceSuggestionId")
|
|
operation_id = str(suggestion.get("operationId") or "")
|
|
operation = operations.get(operation_id)
|
|
checks["outsourceTraceCheckCount"] += 1
|
|
if operation is None or operation.get("sourcingMode") != "OUTSOURCE":
|
|
fail("purchaseTraceViolationCount", "OUTSOURCE_OPERATION", suggestion_id, "outsource suggestion lacks an OUTSOURCE operation")
|
|
continue
|
|
if operation_id in outsource_by_operation:
|
|
fail("purchaseTraceViolationCount", "OUTSOURCE_DUPLICATE_OPERATION", suggestion_id, "outsource operation has multiple suggestions")
|
|
outsource_by_operation[operation_id] = suggestion
|
|
supplier_id = str(suggestion.get("supplierId") or "")
|
|
supplier = suppliers.get(supplier_id)
|
|
if supplier is None:
|
|
fail("missingBuyOutsourceSupplierCount", "OUTSOURCE_SUPPLIER", suggestion_id, "outsource supplier does not resolve")
|
|
continue
|
|
monthly_capacity = _num(supplier.get("monthlyCapacity"))
|
|
allocated_capacity = _num(suggestion.get("supplierMonthlyAllocated"))
|
|
suggestion["supplierCapacityCheck"] = (
|
|
"CALCULATED_PASS"
|
|
if monthly_capacity > 0
|
|
and allocated_capacity <= monthly_capacity + 0.001
|
|
else "CAPACITY_EXCEEDED"
|
|
)
|
|
suggestion["supplierCapacityCheckBasis"] = (
|
|
"CALCULATED_MONTHLY_CAPACITY"
|
|
)
|
|
operation_code = str(operation.get("operationCode") or "")
|
|
allowed_codes = {str(value) for value in supplier.get("outsourceOperationCodes") or []}
|
|
if supplier.get("approved") is not True or operation_code not in allowed_codes:
|
|
fail("outsourceSupplierEligibilityViolationCount", "OUTSOURCE_SUPPLIER_ELIGIBILITY", suggestion_id, "supplier must be approved for the exact outsource operation code")
|
|
send, returned = _date(suggestion.get("sendDate")), _date(suggestion.get("returnDate"))
|
|
if send is None or returned is None or returned <= send:
|
|
fail("outsourceTimeChainViolationCount", "OUTSOURCE_DATES", suggestion_id, "outsource send/return dates are incomplete or reversed")
|
|
outbound_days = _integer(suggestion.get("transportOutboundDays") or supplier.get("transportOutboundDays"))
|
|
processing_days = _integer(suggestion.get("processingDays"))
|
|
inspection_days = _integer(suggestion.get("inspectionDays"))
|
|
inbound_days = _integer(suggestion.get("transportInboundDays") or supplier.get("transportInboundDays"))
|
|
expected_cycle_days = outbound_days + processing_days + inspection_days + inbound_days
|
|
if expected_cycle_days <= 0 or send is None or returned is None or (returned - send).days != expected_cycle_days:
|
|
fail("outsourceTimeChainViolationCount", "OUTSOURCE_CYCLE", suggestion_id, "returnDate must equal outbound transport + processing + inspection + inbound transport")
|
|
predecessor_id = str(suggestion.get("predecessorOperationId") or suggestion.get("previousOperationId") or "")
|
|
next_operation_id = str(suggestion.get("nextOperationId") or "")
|
|
if not predecessor_id or not next_operation_id or predecessor_id not in operations or next_operation_id not in operations:
|
|
fail("purchaseTraceViolationCount", "OUTSOURCE_ROUTE_NEIGHBORS", suggestion_id, "outsource suggestion requires resolvable predecessor and successor operations")
|
|
operation_predecessor = str(operation.get("predecessorOperationId") or operation.get("previousOperationId") or "")
|
|
if predecessor_id != operation_predecessor or next_operation_id != str(operation.get("nextOperationId") or ""):
|
|
fail("purchaseTraceViolationCount", "OUTSOURCE_TRACE", suggestion_id, "outsource predecessor/successor trace is inconsistent")
|
|
for field, expected in (
|
|
("outsourceSuggestionId", suggestion_id),
|
|
("supplierId", supplier_id),
|
|
("sendDate", suggestion.get("sendDate")),
|
|
("returnDate", suggestion.get("returnDate")),
|
|
("outsourceReturnAt", suggestion.get("returnDate")),
|
|
):
|
|
actual = operation.get(field)
|
|
if field == "supplierId":
|
|
actual = actual or operation.get("outsourceSupplierId")
|
|
elif field == "sendDate":
|
|
actual = actual or operation.get("outsourceSendDate")
|
|
if str(actual or "")[:10] != str(expected or "")[:10]:
|
|
fail("purchaseTraceViolationCount", "OUTSOURCE_OPERATION_BINDING", suggestion_id, f"operation {field} does not match suggestion")
|
|
blackout = {str(value)[:10] for value in supplier.get("blackoutDates") or []}
|
|
checks["outsourceBlackoutIntervalCheckCount"] += 1
|
|
blackout_overlap: list[str] = []
|
|
if send is not None and returned is not None and send <= returned:
|
|
current_date = send
|
|
while current_date <= returned:
|
|
if current_date.isoformat() in blackout:
|
|
blackout_overlap.append(current_date.isoformat())
|
|
current_date += timedelta(days=1)
|
|
if blackout_overlap:
|
|
fail(
|
|
"supplierBlackoutViolationCount",
|
|
"OUTSOURCE_SUPPLIER_BLACKOUT_INTERVAL",
|
|
suggestion_id,
|
|
"outsource closed interval sendDate..returnDate overlaps supplier "
|
|
f"blackout dates: {','.join(blackout_overlap)}",
|
|
)
|
|
if (
|
|
suggestion.get("supplierBlackoutCheck") != "CALCULATED_INTERVAL_PASS"
|
|
or suggestion.get("supplierBlackoutCheckBasis")
|
|
!= "CLOSED_INTERVAL_SEND_TO_RETURN"
|
|
or list(suggestion.get("supplierBlackoutOverlapDates") or [])
|
|
!= blackout_overlap
|
|
):
|
|
fail(
|
|
"supplierBlackoutViolationCount",
|
|
"OUTSOURCE_SUPPLIER_BLACKOUT_EVIDENCE",
|
|
suggestion_id,
|
|
"outsource supplier blackout evidence must be independently "
|
|
"recomputable for the complete closed interval",
|
|
)
|
|
capacity_unit = str(supplier.get("monthlyCapacityUnit") or "")
|
|
suggestion_unit = str(suggestion.get("capacityUnit") or suggestion.get("unit") or "")
|
|
if not capacity_unit or suggestion_unit != capacity_unit or _num(suggestion.get("capacityQuantity") or suggestion.get("quantity")) <= 0:
|
|
fail("outsourceSupplierEligibilityViolationCount", "OUTSOURCE_CAPACITY_UNIT", suggestion_id, "outsource capacity quantity/unit must be explicit and match supplier capacity")
|
|
|
|
timezone = ZoneInfo(config.timezone)
|
|
resources = {str(row.get("resourceId")): row for row in rows("resources")}
|
|
teams = {str(row.get("teamId")): row for row in rows("teams")}
|
|
employees_by_team: dict[str, list[dict[str, Any]]] = defaultdict(list)
|
|
for employee in rows("employees"):
|
|
employees_by_team[str(employee.get("teamId"))].append(employee)
|
|
calendars = {str(row.get("calendarId")): row for row in rows("calendars")}
|
|
shifts_by_calendar: dict[str, list[dict[str, Any]]] = defaultdict(list)
|
|
for shift in rows("shifts"):
|
|
shifts_by_calendar[str(shift.get("calendarId"))].append(shift)
|
|
packages = {str(row.get("workPackageId")): row for row in rows("work-packages")}
|
|
block_index = {str(row.get("blockId")): row for row in blocks}
|
|
zones = {str(row.get("zoneId")): row for row in rows("zones")}
|
|
kit_by_package = {str(row.get("workPackageId")): row for row in rows("kit-readiness")}
|
|
|
|
slots = rows("schedule-slots")
|
|
slot_by_operation: dict[str, dict[str, Any]] = {}
|
|
parsed_slots: list[tuple[dict[str, Any], datetime, datetime]] = []
|
|
for slot in slots:
|
|
operation_id = str(slot.get("operationId") or "")
|
|
if operation_id in slot_by_operation:
|
|
fail("duplicateOperationScheduleCount", "DUPLICATE_OPERATION_SLOT", operation_id, "operation appears in more than one baseline slot")
|
|
slot_by_operation[operation_id] = slot
|
|
start, end = _datetime(slot.get("start"), timezone), _datetime(slot.get("end"), timezone)
|
|
if start is None or end is None or end <= start:
|
|
fail("activeOperationCoverageViolationCount", "SLOT_INTERVAL", slot.get("scheduleSlotId"), "slot interval is invalid")
|
|
continue
|
|
parsed_slots.append((slot, start, end))
|
|
active_operations = {key: value for key, value in operations.items() if value.get("active", True)}
|
|
for operation_id in sorted(active_operations):
|
|
if operation_id not in slot_by_operation:
|
|
fail("activeOperationCoverageViolationCount", "ACTIVE_OPERATION_COVERAGE", operation_id, "active operation has no schedule slot")
|
|
for operation_id in sorted(set(slot_by_operation) - set(active_operations)):
|
|
fail("activeOperationCoverageViolationCount", "ORPHAN_OR_INACTIVE_SLOT", operation_id, "slot references an absent/inactive operation")
|
|
|
|
for operation_id, suggestion in outsource_by_operation.items():
|
|
suggestion_id = suggestion.get("outsourceSuggestionId")
|
|
operation = operations[operation_id]
|
|
predecessor_id = str(suggestion.get("predecessorOperationId") or suggestion.get("previousOperationId") or "")
|
|
next_operation_id = str(suggestion.get("nextOperationId") or "")
|
|
predecessor_slot = slot_by_operation.get(predecessor_id)
|
|
outsource_slot = slot_by_operation.get(operation_id)
|
|
next_slot = slot_by_operation.get(next_operation_id)
|
|
if predecessor_slot is None or outsource_slot is None or next_slot is None:
|
|
fail("outsourceTimeChainViolationCount", "OUTSOURCE_SLOT_CHAIN", suggestion_id, "predecessor, outsource, and successor must all have baseline slots")
|
|
continue
|
|
predecessor_end = _datetime(predecessor_slot.get("end"), timezone)
|
|
outsource_start = _datetime(outsource_slot.get("start"), timezone)
|
|
next_start = _datetime(next_slot.get("start"), timezone)
|
|
send = _date(suggestion.get("sendDate"))
|
|
returned = _date(suggestion.get("returnDate"))
|
|
if predecessor_end is None or send is None or predecessor_end.date() > send:
|
|
fail("outsourceTimeChainViolationCount", "OUTSOURCE_SEND_BEFORE_PREDECESSOR", suggestion_id, "sendDate must not precede predecessor completion")
|
|
if outsource_start is None or returned is None or outsource_start.date() < returned:
|
|
fail("outsourceTimeChainViolationCount", "OUTSOURCE_START_BEFORE_RETURN", suggestion_id, "outsource operation slot must not start before supplier return")
|
|
if next_start is None or returned is None or next_start.date() < returned:
|
|
fail("outsourceTimeChainViolationCount", "OUTSOURCE_SUCCESSOR_BEFORE_RETURN", suggestion_id, "successor operation must not start before supplier return")
|
|
if str(operation.get("outsourceReturnAt") or "")[:10] != str(suggestion.get("returnDate") or "")[:10]:
|
|
fail("outsourceTimeChainViolationCount", "OUTSOURCE_RETURN_SINGLE_SOURCE", suggestion_id, "operation and suggestion must share the same return date")
|
|
|
|
owner_intervals: dict[tuple[str, str], list[tuple[datetime, datetime, str]]] = defaultdict(list)
|
|
hazardous_by_zone: dict[str, list[tuple[datetime, datetime, str, str]]] = defaultdict(list)
|
|
for slot, start, end in parsed_slots:
|
|
slot_id, operation_id = str(slot.get("scheduleSlotId") or ""), str(slot.get("operationId") or "")
|
|
operation = active_operations.get(operation_id)
|
|
resource = resources.get(str(slot.get("resourceId") or ""))
|
|
team = teams.get(str(slot.get("teamId") or ""))
|
|
if operation is None or resource is None or team is None:
|
|
continue
|
|
if resource.get("exclusive") is True:
|
|
owner_intervals[("resource", str(resource.get("resourceId")))].append((start, end, slot_id))
|
|
owner_intervals[("team", str(team.get("teamId")))].append((start, end, slot_id))
|
|
required_type = str(operation.get("requiredResourceType") or "")
|
|
if str(resource.get("resourceType") or "") != required_type:
|
|
fail("resourceTypeViolationCount", "RESOURCE_TYPE", slot_id, f"resource does not match required type {required_type}")
|
|
required_tags = {str(value) for value in operation.get("requiredCapabilityTags") or []}
|
|
resource_tags = {str(value) for value in resource.get("capabilityTags") or []}
|
|
if not required_tags.issubset(resource_tags):
|
|
fail("resourceCapabilityViolationCount", "RESOURCE_CAPABILITY", slot_id, "resource lacks required capability tags")
|
|
required_skills = {str(value) for value in operation.get("requiredSkillCodes") or []}
|
|
team_skills = {str(value) for value in team.get("skillCodes") or []}
|
|
if not required_skills.issubset(team_skills):
|
|
fail("teamSkillViolationCount", "TEAM_SKILL", slot_id, "team lacks required skills")
|
|
if _integer(team.get("crewSize")) < _integer(operation.get("crewSize"), 1):
|
|
fail("teamCrewViolationCount", "TEAM_CREW", slot_id, "team crew size is below operation demand")
|
|
required_qualifications = {str(value) for value in operation.get("requiredQualificationCodes") or []}
|
|
for qualification_code in required_qualifications:
|
|
qualified = any(
|
|
employee.get("status", "ACTIVE") == "ACTIVE"
|
|
and employee.get("qualificationCode") == qualification_code
|
|
and (_date(employee.get("qualificationValidTo")) or date.min) >= start.date()
|
|
for employee in employees_by_team.get(str(team.get("teamId")), [])
|
|
)
|
|
if not qualified:
|
|
fail(
|
|
"teamQualificationViolationCount",
|
|
"TEAM_QUALIFICATION",
|
|
slot_id,
|
|
f"no assigned-team employee holds {qualification_code}",
|
|
)
|
|
|
|
resource_cal_id, team_cal_id = str(resource.get("shiftCalendarId") or ""), str(team.get("shiftCalendarId") or "")
|
|
resource_span = _calendar_span(calendars.get(resource_cal_id), shifts_by_calendar.get(resource_cal_id, []), start.date(), timezone)
|
|
team_span = _calendar_span(calendars.get(team_cal_id), shifts_by_calendar.get(team_cal_id, []), start.date(), timezone)
|
|
if resource_span is None or not (resource_span[0] <= start and end <= resource_span[1]):
|
|
fail("calendarViolationCount", "RESOURCE_CALENDAR", slot_id, "slot is outside resource shift calendar")
|
|
if team_span is None or not (team_span[0] <= start and end <= team_span[1]):
|
|
fail("calendarViolationCount", "TEAM_CALENDAR", slot_id, "slot is outside team shift calendar")
|
|
maintenance = calendars.get(str(resource.get("maintenanceCalendarId") or ""))
|
|
if maintenance is None or maintenance.get("calendarType") != "MAINTENANCE":
|
|
fail("maintenanceCalendarViolationCount", "MAINTENANCE_CALENDAR", slot_id, "resource lacks a maintenance calendar")
|
|
elif start.date().isoformat() in {str(value)[:10] for value in maintenance.get("blackoutDates") or []}:
|
|
fail("maintenanceCalendarViolationCount", "MAINTENANCE_WINDOW", slot_id, "slot overlaps maintenance")
|
|
|
|
ready_dates = [_date(operation.get("materialReadyAt")), _date(operation.get("outsourceReturnAt")), _date((work_orders.get(str(operation.get("workOrderId"))) or {}).get("materialReadyAt")), _date((kit_by_package.get(str(operation.get("workPackageId"))) or {}).get("readyDate"))]
|
|
material_ready = max((value for value in ready_dates if value is not None), default=config.planning_base_date)
|
|
if start.date() < material_ready:
|
|
fail("materialReadinessViolationCount", "MATERIAL_READY", slot_id, f"slot starts before {material_ready.isoformat()}")
|
|
if operation.get("holdPoint"):
|
|
hold_release = _date(operation.get("holdReleaseAt"))
|
|
if operation.get("holdStatus") != "RELEASED" or hold_release is None or start.date() < hold_release:
|
|
fail("qualityHoldPointViolationCount", "QUALITY_HOLD", slot_id, "hold point is not released before execution")
|
|
if operation.get("weatherSensitive") and operation.get("weatherWindowStatus") != "OPEN":
|
|
fail("weatherWindowViolationCount", "WEATHER_WINDOW", slot_id, "weather-sensitive operation lacks open window")
|
|
|
|
start_day = start.date()
|
|
expected_fence = "FROZEN" if start_day <= date(2026, 9, 14) else "SEMI_FROZEN" if start_day <= date(2026, 10, 12) else "STRATEGIC" if start_day >= date(2027, 3, 1) else "FREE"
|
|
if slot.get("timeFence") != expected_fence:
|
|
fail("illegalFrozenChangeCount", "TIME_FENCE", slot_id, f"expected timeFence {expected_fence}")
|
|
if (
|
|
expected_fence == "FROZEN"
|
|
and str(slot.get("baselineStart") or "") != str(slot.get("start") or "")
|
|
and not (slot.get("changeAuthorized") and slot.get("changeReason") and slot.get("impactReport"))
|
|
):
|
|
fail("illegalFrozenChangeCount", "FROZEN_CHANGE", slot_id, "frozen change lacks authorization/reason/impact")
|
|
|
|
package = packages.get(str(operation.get("workPackageId") or "")) or {}
|
|
block = block_index.get(str(package.get("blockId") or "")) or {}
|
|
load_basis_is_block = operation.get("loadBasis") == "BLOCK"
|
|
for operation_field, block_field, capacity_field in (
|
|
("requiredWeightT", "weightT", "maximumWeight"),
|
|
("requiredLengthM", "lengthM", "maximumLength"),
|
|
("requiredWidthM", "widthM", "maximumWidth"),
|
|
("requiredHeightM", "heightM", "maximumHeight"),
|
|
):
|
|
demand = operation.get(operation_field)
|
|
if demand is None and load_basis_is_block:
|
|
demand = block.get(block_field)
|
|
capacity = resource.get(capacity_field)
|
|
if demand is not None and capacity is not None and _num(demand) > _num(capacity) + 1e-9:
|
|
fail("resourceWeightDimensionViolationCount", "RESOURCE_DIMENSION", slot_id, f"{operation_field} exceeds {capacity_field}")
|
|
resource_type = str(resource.get("resourceType") or "")
|
|
if "CRANE" in resource_type:
|
|
lift_weight = operation.get("requiredLiftWeightT")
|
|
if lift_weight is None and load_basis_is_block:
|
|
lift_weight = block.get("weightT")
|
|
maximum_weight = resource.get("maximumWeight")
|
|
if lift_weight is not None and maximum_weight is not None and _num(lift_weight) > _num(maximum_weight):
|
|
fail("liftingCapacityViolationCount", "CRANE_WEIGHT", slot_id, "lift weight exceeds crane capacity")
|
|
if operation.get("requiredLiftRadiusM") and resource.get("maximumRadiusM") is not None and _num(operation.get("requiredLiftRadiusM")) > _num(resource.get("maximumRadiusM")):
|
|
fail("liftingCapacityViolationCount", "CRANE_RADIUS", slot_id, "lift radius exceeds crane capacity")
|
|
if resource_type == "TRANSPORTER":
|
|
if operation.get("transportWindowStatus", "OPEN") != "OPEN":
|
|
fail("transportConstraintViolationCount", "TRANSPORT_WINDOW", slot_id, "transport window is closed")
|
|
if operation.get("transportZone") and operation.get("transportZone") != resource.get("transportZone"):
|
|
fail("transportConstraintViolationCount", "TRANSPORT_ZONE", slot_id, "transporter route zone mismatch")
|
|
zone_id = str(package.get("zoneId") or "")
|
|
hazard = str(operation.get("hazardClass") or "")
|
|
if hazard:
|
|
hazardous_by_zone[zone_id].append((start, end, hazard, slot_id))
|
|
|
|
special_metrics = {"DOCK": "dockConflictCount", "BERTH": "berthConflictCount", "JIG": "fixtureConflictCount", "PAINT_BOOTH": "paintBoothConflictCount"}
|
|
for (kind, owner_id), timeline in sorted(owner_intervals.items()):
|
|
for previous, current in pairwise(sorted(timeline)):
|
|
if previous[1] <= current[0]:
|
|
continue
|
|
if kind == "team":
|
|
fail("teamOverlapCount", "TEAM_OVERLAP", owner_id, f"{previous[2]} overlaps {current[2]}")
|
|
else:
|
|
fail("resourceOverlapCount", "RESOURCE_OVERLAP", owner_id, f"{previous[2]} overlaps {current[2]}")
|
|
resource_type = str((resources.get(owner_id) or {}).get("resourceType") or "")
|
|
if resource_type in special_metrics:
|
|
fail(special_metrics[resource_type], f"{resource_type}_OVERLAP", owner_id, f"{previous[2]} overlaps {current[2]}")
|
|
|
|
slot_intervals = {str(slot.get("operationId")): (start, end) for slot, start, end in parsed_slots if str(slot.get("operationId")) in active_operations}
|
|
for operation_id, operation in active_operations.items():
|
|
predecessor_id = str(operation.get("predecessorOperationId") or "")
|
|
if not predecessor_id:
|
|
continue
|
|
current, predecessor = slot_intervals.get(operation_id), slot_intervals.get(predecessor_id)
|
|
if current is None or predecessor is None:
|
|
continue
|
|
lag, relation = timedelta(hours=_num(operation.get("lagHours"))), str(operation.get("relationType") or "FS")
|
|
satisfied = {"FS": current[0] >= predecessor[1] + lag, "SS": current[0] >= predecessor[0] + lag, "FF": current[1] >= predecessor[1] + lag, "SF": current[1] >= predecessor[0] + lag}.get(relation, False)
|
|
if not satisfied:
|
|
fail("precedenceViolationCount", "PRECEDENCE", operation_id, f"{relation}+lag precedence is violated")
|
|
|
|
incompatible = {frozenset(("HOT_WORK", "PAINT")), frozenset(("HOT_WORK", "CONFINED_SPACE"))}
|
|
for zone_id, timeline in hazardous_by_zone.items():
|
|
maximum_density = max(1, _integer((zones.get(zone_id) or {}).get("maxConcurrentHazardOperations"), 3))
|
|
for index, current in enumerate(timeline):
|
|
simultaneous = 1
|
|
for other in timeline[index + 1 :]:
|
|
if current[0] < other[1] and other[0] < current[1]:
|
|
simultaneous += 1
|
|
if frozenset((current[2], other[2])) in incompatible:
|
|
fail("hotWorkDensityViolationCount", "HAZARD_INCOMPATIBILITY", zone_id, f"{current[3]} conflicts with {other[3]}")
|
|
if simultaneous > maximum_density:
|
|
fail("hotWorkDensityViolationCount", "AREA_DENSITY", zone_id, f"hazard density {simultaneous} exceeds {maximum_density}")
|
|
|
|
schedule_versions = rows("schedule-versions")
|
|
expected_digest = _schedule_input_digest(bundle, config)
|
|
for version in schedule_versions:
|
|
version_id = version.get("scheduleVersionId")
|
|
checks["scheduleVersionEvidenceCheckCount"] += 1
|
|
if version.get("inputDigest") != expected_digest:
|
|
fail("scheduleVersionEvidenceViolationCount", "INPUT_DIGEST", version_id, "inputDigest does not match canonical scheduling inputs")
|
|
if any(_empty(version.get(field)) for field in ("algorithm", "algorithmVersion", "inputDigest", "solveStatus", "evidenceRefs")):
|
|
fail("scheduleVersionEvidenceViolationCount", "VERSION_EVIDENCE", version_id, "schedule version lacks input/algorithm/evidence fields")
|
|
if version.get("solveStatus") == "FEASIBLE" and _integer(version.get("hardViolationCount")) != 0:
|
|
fail("unmarkedHardConstraintViolationCount", "FALSE_FEASIBLE", version_id, "FEASIBLE version reports hard violations")
|
|
if "HEURISTIC" in str(version.get("algorithm") or "") and not (version.get("optimalityGap") is None and version.get("gapType") == "NOT_APPLICABLE"):
|
|
fail("scheduleVersionEvidenceViolationCount", "HEURISTIC_GAP", version_id, "heuristic may not claim numeric optimality gap")
|
|
if not version.get("immutable"):
|
|
fail("scheduleVersionEvidenceViolationCount", "IMMUTABLE_VERSION", version_id, "baseline version must be immutable")
|
|
for slot in slots:
|
|
if not str(slot.get("explanation") or "").strip():
|
|
fail("unexplainedScheduleDecisionCount", "SCHEDULE_EXPLANATION", slot.get("scheduleSlotId"), "schedule decision lacks explanation")
|
|
if not _refs(slot.get("evidenceRefs")):
|
|
fail("unresolvedEvidenceRefCount", "SCHEDULE_EVIDENCE", slot.get("scheduleSlotId"), "schedule decision lacks evidenceRefs")
|
|
|
|
reference_ids: set[str] = set()
|
|
aliases: dict[str, set[str]] = defaultdict(set)
|
|
alias_map = {
|
|
"operation": "operations", "resource": "resources", "team": "teams", "routing": "routings",
|
|
"routing-operation": "routing-operations", "production-order": "production-orders", "work-order": "work-orders",
|
|
"work-package": "work-packages", "schedule-slot": "schedule-slots", "schedule-version": "schedule-versions",
|
|
"material": "materials", "material-requirement": "material-requirements", "requirement": "material-requirements",
|
|
"supplier": "suppliers", "inspection": "quality-inspections", "nonconformity": "nonconformities", "rework-order": "rework-orders",
|
|
}
|
|
for alias, table_name in alias_map.items():
|
|
primary_key = TABLE_SPECS[table_name].primary_key
|
|
aliases[alias].update(str(row.get(primary_key)) for row in rows(table_name) if not _empty(row.get(primary_key)))
|
|
knowledge_assets = bundle.artifacts.get("knowledge-assets") or []
|
|
skill_registry = bundle.artifacts.get("skill-registry") or {}
|
|
registered_skills = skill_registry.get("skills") or []
|
|
for asset in knowledge_assets:
|
|
reference_ids.update((str(asset.get("knowledgeId") or ""), str(asset.get("evidenceRef") or "")))
|
|
for skill in registered_skills:
|
|
reference_ids.add(str(skill.get("evidenceRef") or f"skill:{skill.get('skillId')}"))
|
|
for version in schedule_versions:
|
|
reference_ids.update((f"input-digest:{version.get('inputDigest')}", f"algorithm:{version.get('algorithm')}@{version.get('algorithmVersion')}"))
|
|
|
|
def resolves(reference: str) -> bool:
|
|
if reference in reference_ids:
|
|
return True
|
|
if ":" not in reference:
|
|
return False
|
|
prefix, value = reference.split(":", 1)
|
|
if prefix == "material-ready":
|
|
return _date(value) is not None
|
|
if prefix == "relation":
|
|
parts = value.split(":", 1)
|
|
if len(parts) != 2 or parts[0] not in {"FS", "SS", "FF", "SF"} or not parts[1].endswith("h"):
|
|
return False
|
|
try:
|
|
float(parts[1][:-1])
|
|
except ValueError:
|
|
return False
|
|
return True
|
|
return value in aliases.get(prefix, set())
|
|
|
|
evidence_holders: list[tuple[str, Any, Any]] = []
|
|
for table_name in ("schedule-slots", "purchase-suggestions", "outsource-suggestions", *_EXECUTION_TABLES):
|
|
primary_key = TABLE_SPECS[table_name].primary_key
|
|
evidence_holders.extend((table_name, row.get(primary_key), row.get("evidenceRefs")) for row in rows(table_name))
|
|
alternatives_artifact = bundle.artifacts.get("schedule-alternatives") or {}
|
|
evidence_holders.extend(("alternative", row.get("alternativeId"), row.get("evidenceRefs")) for row in alternatives_artifact.get("alternatives") or [])
|
|
evidence_holders.extend(("scenario", row.get("scenarioId"), row.get("evidenceRefs")) for row in bundle.artifacts.get("scenario-events") or [])
|
|
for holder, entity_id, evidence in evidence_holders:
|
|
for reference in _refs(evidence):
|
|
checks["evidenceRefCheckCount"] += 1
|
|
if not resolves(reference):
|
|
fail("unresolvedEvidenceRefCount", "UNRESOLVED_EVIDENCE", entity_id, f"{holder} evidence does not resolve: {reference}")
|
|
|
|
skill_ids = [str(row.get("skillId") or "") for row in registered_skills]
|
|
if len(skill_ids) != 20 or set(skill_ids) != set(SKILL_IDS) or len(skill_ids) != len(set(skill_ids)):
|
|
fail("skillRegistryViolationCount", "SKILL_IDS", "skill-registry", "registry must contain exact 20 unique skill IDs")
|
|
skill_schemas = bundle.artifacts.get("skill-schemas") or {}
|
|
for skill in registered_skills:
|
|
skill_id = str(skill.get("skillId") or "")
|
|
for field in _SKILL_FIELDS:
|
|
if _empty(skill.get(field)):
|
|
fail("skillRegistryViolationCount", "SKILL_FIELD", skill_id, f"required skill field {field} is empty")
|
|
if _integer(skill.get("timeoutSeconds")) <= 0:
|
|
fail("skillRegistryViolationCount", "SKILL_TIMEOUT", skill_id, "timeoutSeconds must be positive")
|
|
schema = skill_schemas.get(skill_id)
|
|
if not isinstance(schema, dict):
|
|
fail("skillSchemaViolationCount", "SKILL_SCHEMA", skill_id, "skill schema is missing")
|
|
continue
|
|
definitions = schema.get("$defs") or {}
|
|
if not isinstance(definitions.get("input"), dict) or not isinstance(definitions.get("output"), dict):
|
|
fail("skillSchemaViolationCount", "SKILL_SCHEMA_DEFS", skill_id, "schema must define input/output")
|
|
expected_prefix = f"skills/skill-schemas/{skill_id}.schema.json#/$defs/"
|
|
if skill.get("inputSchema") != expected_prefix + "input" or skill.get("outputSchema") != expected_prefix + "output":
|
|
fail("skillSchemaViolationCount", "SKILL_SCHEMA_REF", skill_id, "schema references are not canonical")
|
|
|
|
category_counts = Counter(str(row.get("category") or "") for row in knowledge_assets)
|
|
required_per_category = 6 if config.scale == "full" else max(1, config.profile.knowledge_asset_count // 20)
|
|
if set(category_counts) != set(RAG_CATEGORIES):
|
|
fail("ragKnowledgeViolationCount", "RAG_CATEGORIES", "knowledge-assets", "knowledge must cover exact 20 categories")
|
|
for category in RAG_CATEGORIES:
|
|
if category_counts.get(category) != required_per_category:
|
|
fail("ragKnowledgeViolationCount", "RAG_CATEGORY_COUNT", category, f"expected {required_per_category}, found {category_counts.get(category, 0)}")
|
|
for asset in knowledge_assets:
|
|
asset_id = asset.get("knowledgeId")
|
|
for field in _KNOWLEDGE_FIELDS:
|
|
if _empty(asset.get(field)):
|
|
fail("ragKnowledgeViolationCount", "RAG_FIELD", asset_id, f"required knowledge field {field} is empty")
|
|
if asset.get("sourceType") != "SYNTHETIC_KNOWLEDGE":
|
|
fail("ragKnowledgeViolationCount", "RAG_SOURCE_TYPE", asset_id, "sourceType must be SYNTHETIC_KNOWLEDGE")
|
|
if not 0.0 <= _num(asset.get("confidence"), -1.0) <= 1.0:
|
|
fail("ragKnowledgeViolationCount", "RAG_CONFIDENCE", asset_id, "confidence must be in [0,1]")
|
|
if asset.get("relatedSkillId") and asset.get("relatedSkillId") not in set(SKILL_IDS):
|
|
fail("ragKnowledgeViolationCount", "RAG_SKILL_REF", asset_id, "relatedSkillId does not resolve")
|
|
|
|
alternatives = alternatives_artifact.get("alternatives") or []
|
|
modes = [str(row.get("mode") or "") for row in alternatives]
|
|
if len(alternatives) != 7 or set(modes) != _ALTERNATIVES or {row[0] for row in ALTERNATIVE_MODES} != _ALTERNATIVES:
|
|
fail("alternativeViolationCount", "ALTERNATIVE_MODES", "schedule-alternatives", "exactly seven canonical alternatives are required")
|
|
recommended = [row for row in alternatives if row.get("mode") == "RECOMMENDED"]
|
|
if len(recommended) != 1 or alternatives_artifact.get("recommendedAlternativeId") != (recommended[0].get("alternativeId") if recommended else None):
|
|
fail("alternativeViolationCount", "RECOMMENDED_ALTERNATIVE", "schedule-alternatives", "RECOMMENDED alternative is missing/inconsistent")
|
|
alternative_version_ids: set[str] = set()
|
|
for alternative in alternatives:
|
|
alternative_id = alternative.get("alternativeId")
|
|
schedule_version_id = str(alternative.get("scheduleVersionId") or "")
|
|
if not schedule_version_id or schedule_version_id == str(alternative.get("baseScheduleVersionId") or "") or schedule_version_id in alternative_version_ids:
|
|
fail("alternativeScheduleViolationCount", "ALTERNATIVE_VERSION", alternative_id, "alternative requires a unique schedule version distinct from baseline")
|
|
alternative_version_ids.add(schedule_version_id)
|
|
if alternative.get("solveStatus") == "FEASIBLE" and (_integer(alternative.get("hardConstraintViolations")) != 0 or _integer(alternative.get("unmarkedHardViolationCount")) != 0):
|
|
fail("alternativeViolationCount", "ALTERNATIVE_FALSE_FEASIBLE", alternative_id, "feasible alternative has hard violations")
|
|
if alternative.get("optimalityGap") is not None or alternative.get("gapType") != "NOT_APPLICABLE":
|
|
fail("alternativeViolationCount", "ALTERNATIVE_GAP", alternative_id, "heuristic alternative must use NOT_APPLICABLE gap")
|
|
if not isclose(sum(_num(value) for value in (alternative.get("weights") or {}).values()), 1.0, abs_tol=1e-6):
|
|
fail("alternativeViolationCount", "ALTERNATIVE_WEIGHTS", alternative_id, "weights must sum to 1")
|
|
deltas = alternative.get("slotDeltas") or []
|
|
changed_count = _integer(alternative.get("changedSlotCount"))
|
|
if changed_count <= 0 or changed_count != len(deltas) or changed_count != len(alternative.get("alternativeSlots") or []):
|
|
fail("alternativeScheduleViolationCount", "ALTERNATIVE_SLOT_COUNT", alternative_id, "alternative changedSlotCount must equal real slot deltas and alternative slots")
|
|
actual_field_counts: Counter[str] = Counter()
|
|
for delta in deltas:
|
|
before, after = delta.get("before") or {}, delta.get("after") or {}
|
|
changed_fields = [field for field in ("start", "end", "resourceId", "teamId") if before.get(field) != after.get(field)]
|
|
actual_field_counts.update(changed_fields)
|
|
if not changed_fields or set(changed_fields) != set(delta.get("changedFields") or []):
|
|
fail("alternativeScheduleViolationCount", "ALTERNATIVE_SLOT_DELTA", alternative_id, "alternative delta must contain a real start/end/resource/team change")
|
|
if str(after.get("scheduleVersionId") or "") != schedule_version_id:
|
|
fail("alternativeScheduleViolationCount", "ALTERNATIVE_SLOT_VERSION", alternative_id, "alternative slot must bind the alternative schedule version")
|
|
if dict(actual_field_counts) != (alternative.get("changedFieldCounts") or {}):
|
|
fail("alternativeScheduleViolationCount", "ALTERNATIVE_FIELD_COUNTS", alternative_id, "changedFieldCounts does not match actual slot deltas")
|
|
constraint_validation = alternative.get("constraintValidation") or {}
|
|
if alternative.get("solveStatus") == "FEASIBLE" and (constraint_validation.get("valid") is not True or _integer(constraint_validation.get("hardViolationCount")) != 0):
|
|
fail("alternativeScheduleViolationCount", "ALTERNATIVE_REVALIDATION", alternative_id, "FEASIBLE alternative must include a passing finite-capacity revalidation")
|
|
|
|
events = bundle.artifacts.get("scenario-events") or []
|
|
event_types = [str(row.get("eventType") or "") for row in events]
|
|
if len(events) != 15 or set(event_types) != set(EVENT_TYPES) or len(event_types) != len(set(event_types)):
|
|
fail("scenarioViolationCount", "SCENARIO_TYPES", "scenario-events", "exactly 15 canonical event types are required")
|
|
for file_name, event_type in CANONICAL_FILES.items():
|
|
artifact = bundle.artifacts.get(file_name)
|
|
if not isinstance(artifact, dict) or artifact.get("eventType") != event_type:
|
|
fail("scenarioViolationCount", "CANONICAL_SCENARIO", file_name, f"canonical scenario must represent {event_type}")
|
|
scenario_version_ids: set[str] = set()
|
|
required_input_fields = (
|
|
"occurredAt", "affectedProjectIds", "affectedWbsIds", "affectedProductionOrderIds",
|
|
"affectedResourceIds", "parameters", "originalImpact", "recommendedActions",
|
|
"requiresReschedule", "frozenImpact",
|
|
)
|
|
for event in events:
|
|
scenario_id = event.get("scenarioId")
|
|
event_input = event.get("input") or {}
|
|
before = event.get("before") or []
|
|
after = event.get("after") or []
|
|
diff = event.get("diff") or {}
|
|
if not isinstance(event.get("input"), dict) or not isinstance(event.get("before"), list) or not isinstance(event.get("after"), list) or not isinstance(event.get("diff"), dict):
|
|
fail("scenarioViolationCount", "SCENARIO_ENVELOPE", scenario_id, "scenario must include input/before/after/diff")
|
|
continue
|
|
for field in required_input_fields:
|
|
if _empty(event_input.get(field)) and field != "affectedResourceIds":
|
|
fail("scenarioViolationCount", "SCENARIO_INPUT_FIELD", scenario_id, f"scenario input field {field} is empty")
|
|
if event_input.get("synthetic") is not True or event_input.get("datasetType") != DATASET_TYPE or event_input.get("requiresReschedule") is not True:
|
|
fail("scenarioViolationCount", "SCENARIO_INPUT_BOUNDARY", scenario_id, "scenario input must be synthetic and require explicit rescheduling")
|
|
after_version_id = str(event.get("afterScheduleVersionId") or "")
|
|
if not after_version_id or after_version_id == str(event.get("baseScheduleVersionId") or "") or after_version_id in scenario_version_ids:
|
|
fail("scenarioDiffViolationCount", "SCENARIO_VERSION", scenario_id, "scenario requires a unique repaired schedule version")
|
|
scenario_version_ids.add(after_version_id)
|
|
before_by_operation = {str(row.get("operationId") or row.get("scheduleSlotId") or ""): row for row in before}
|
|
after_by_operation = {str(row.get("operationId") or row.get("scheduleSlotId") or ""): row for row in after}
|
|
changed_field_counts: Counter[str] = Counter()
|
|
actual_changed = 0
|
|
for key in sorted(set(before_by_operation) & set(after_by_operation)):
|
|
left, right = before_by_operation[key], after_by_operation[key]
|
|
changed_fields = [field for field in ("start", "end", "resourceId", "teamId") if left.get(field) != right.get(field)]
|
|
if changed_fields:
|
|
actual_changed += 1
|
|
changed_field_counts.update(changed_fields)
|
|
if str(right.get("scheduleVersionId") or "") != after_version_id:
|
|
fail("scenarioDiffViolationCount", "SCENARIO_SLOT_VERSION", scenario_id, "after slot must bind the repaired schedule version")
|
|
declared_changed = _integer(diff.get("changedSlotCount"))
|
|
if event.get("solveStatus") == "FEASIBLE" and (declared_changed <= 0 or actual_changed != declared_changed or len(before) != declared_changed or len(after) != declared_changed):
|
|
fail("scenarioDiffViolationCount", "SCENARIO_REAL_DIFF", scenario_id, "FEASIBLE scenario changedSlotCount must match real before/after slot changes")
|
|
if actual_changed and dict(changed_field_counts) != (diff.get("changedFieldCounts") or {}):
|
|
fail("scenarioDiffViolationCount", "SCENARIO_FIELD_COUNTS", scenario_id, "scenario changedFieldCounts does not match actual changes")
|
|
frozen_impact = event_input.get("frozenImpact") or {}
|
|
if frozen_impact.get("timeFenceField") != "timeFence" or _integer(frozen_impact.get("unauthorizedFrozenChangeCount")) != 0:
|
|
fail("illegalFrozenChangeCount", "SCENARIO_FROZEN_INPUT", scenario_id, "scenario frozen impact must use timeFence and forbid unauthorized changes")
|
|
if _integer(diff.get("unauthorizedFrozenChangeCount")) != 0:
|
|
fail("illegalFrozenChangeCount", "SCENARIO_FROZEN_CHANGE", scenario_id, "scenario has unauthorized frozen change")
|
|
if event.get("solveStatus") == "INFEASIBLE":
|
|
if not event.get("conflictCore") or not event.get("reliefSuggestions"):
|
|
fail("infeasibleCoreViolationCount", "INFEASIBLE_CORE", scenario_id, "INFEASIBLE scenario requires conflict core and relief suggestions")
|
|
elif event.get("solveStatus") == "FEASIBLE":
|
|
constraint_validation = event.get("constraintValidation") or {}
|
|
if _integer(event.get("unmarkedHardViolationCount")) != 0 or constraint_validation.get("valid") is not True or _integer(constraint_validation.get("hardViolationCount")) != 0:
|
|
fail("unmarkedHardConstraintViolationCount", "SCENARIO_FALSE_FEASIBLE", scenario_id, "FEASIBLE scenario has hard violations or lacks passing revalidation")
|
|
elif event.get("solveStatus") not in {"INFEASIBLE", "UNKNOWN"}:
|
|
fail("scenarioViolationCount", "SCENARIO_STATUS", scenario_id, "invalid solveStatus")
|
|
|
|
for table_name in _EXECUTION_TABLES:
|
|
if not rows(table_name):
|
|
fail("executionClosureViolationCount", "EXECUTION_TABLE_EMPTY", table_name, "execution/quality table must be non-empty")
|
|
slot_ids = {str(row.get("scheduleSlotId")) for row in slots}
|
|
for mes_order in rows("mes-orders"):
|
|
if mes_order.get("autoDeleteAllowed") is not False or mes_order.get("dispatchStatus") != "DISPATCHED":
|
|
fail("executionClosureViolationCount", "MES_PROTECTION", mes_order.get("mesOrderId"), "dispatched MES order must be deletion-protected")
|
|
if mes_order.get("frozen") and mes_order.get("autoRescheduleAllowed") is not False:
|
|
fail("illegalFrozenChangeCount", "MES_FROZEN_PROTECTION", mes_order.get("mesOrderId"), "frozen MES order cannot auto-reschedule")
|
|
for material_issue in rows("material-issues"):
|
|
if _num(material_issue.get("quantity")) <= 0 or material_issue.get("negativeInventoryAllowed") is not False:
|
|
fail("executionClosureViolationCount", "MATERIAL_ISSUE", material_issue.get("materialIssueId"), "material issue must be positive and forbid negative inventory")
|
|
inspections = {str(row.get("inspectionId")): row for row in rows("quality-inspections")}
|
|
reworks = {str(row.get("reworkOrderId")): row for row in rows("rework-orders")}
|
|
rework_operations = {str(row.get("reworkOperationId")): row for row in rows("rework-operations")}
|
|
rework_slots = {str(row.get("reworkScheduleSlotId") or row.get("scheduleSlotId")): row for row in rows("rework-schedule-slots")}
|
|
for inspection_id, inspection in inspections.items():
|
|
if inspection.get("holdPoint") and not {"HOLD", "RELEASED"}.issubset({str(value) for value in inspection.get("stateHistory") or []}):
|
|
fail("qualityHoldPointViolationCount", "INSPECTION_HOLD_HISTORY", inspection_id, "hold inspection lacks HOLD/RELEASED history")
|
|
|
|
rework_owner_intervals: dict[tuple[str, str], list[tuple[datetime, datetime, str]]] = defaultdict(list)
|
|
for rework_slot_id, rework_slot in rework_slots.items():
|
|
start = _datetime(rework_slot.get("start"), timezone)
|
|
end = _datetime(rework_slot.get("end"), timezone)
|
|
resource_id = str(rework_slot.get("resourceId") or "")
|
|
team_id = str(rework_slot.get("teamId") or "")
|
|
if start is None or end is None or end <= start:
|
|
fail("reworkClosureViolationCount", "REWORK_SLOT_INTERVAL", rework_slot_id, "rework slot interval is invalid")
|
|
continue
|
|
if resource_id not in resources or team_id not in teams or not rework_slot.get("finiteCapacityReserved"):
|
|
fail("reworkClosureViolationCount", "REWORK_SLOT_CAPACITY", rework_slot_id, "rework slot must reserve a real finite resource and team")
|
|
rework_owner_intervals[("resource", resource_id)].append((start, end, rework_slot_id))
|
|
rework_owner_intervals[("team", team_id)].append((start, end, rework_slot_id))
|
|
for baseline_slot, baseline_start, baseline_end in parsed_slots:
|
|
same_resource = resource_id == str(baseline_slot.get("resourceId") or "")
|
|
same_team = team_id == str(baseline_slot.get("teamId") or "")
|
|
if (same_resource or same_team) and start < baseline_end and baseline_start < end:
|
|
fail("reworkClosureViolationCount", "REWORK_BASELINE_OVERLAP", rework_slot_id, "rework slot overlaps baseline finite capacity")
|
|
break
|
|
for owner, intervals in rework_owner_intervals.items():
|
|
for left, right in pairwise(sorted(intervals, key=lambda value: (value[0], value[1], value[2]))):
|
|
if right[0] < left[1]:
|
|
fail("reworkClosureViolationCount", "REWORK_SLOT_OVERLAP", f"{owner[0]}:{owner[1]}", f"{left[2]} overlaps {right[2]}")
|
|
|
|
for nonconformity in rows("nonconformities"):
|
|
ncr_id = str(nonconformity.get("nonconformityId") or "")
|
|
initial = inspections.get(str(nonconformity.get("inspectionId") or ""))
|
|
rework = reworks.get(str(nonconformity.get("reworkOrderId") or ""))
|
|
reinspection = inspections.get(str(nonconformity.get("reinspectionId") or ""))
|
|
if nonconformity.get("status") != "CLOSED" or rework is None or initial is None:
|
|
fail("reworkClosureViolationCount", "NCR_REWORK", ncr_id, "NCR must close through an initial inspection and rework")
|
|
continue
|
|
rework_slot_id = str(rework.get("reworkScheduleSlotId") or rework.get("scheduleSlotId") or "")
|
|
rework_slot = rework_slots.get(rework_slot_id)
|
|
rework_operation_id = str(rework.get("reworkOperationId") or "")
|
|
if rework.get("status") != "CLOSED" or not rework.get("reinspectionRequired"):
|
|
fail("reworkClosureViolationCount", "REWORK_STATUS", rework.get("reworkOrderId"), "rework must close and require reinspection")
|
|
if str(rework.get("resourceId") or "") not in resources or str(rework.get("teamId") or "") not in teams or str(rework.get("sourceScheduleSlotId") or "") not in slot_ids:
|
|
fail("reworkClosureViolationCount", "REWORK_SOURCE_CAPACITY", rework.get("reworkOrderId"), "rework must retain valid source resource/team/schedule evidence")
|
|
if rework_slot is None or rework_slot_id in slot_ids or rework_operation_id not in rework_operations:
|
|
fail("reworkClosureViolationCount", "REWORK_INDEPENDENT_SLOT", rework.get("reworkOrderId"), "rework must use a distinct operation and schedule slot")
|
|
elif str(rework_slot.get("resourceId") or "") != str(rework.get("resourceId") or "") or str(rework_slot.get("teamId") or "") != str(rework.get("teamId") or ""):
|
|
fail("reworkClosureViolationCount", "REWORK_SLOT_BINDING", rework.get("reworkOrderId"), "rework order and slot resource/team bindings differ")
|
|
if initial.get("nonconformityId") != ncr_id or initial.get("reworkOrderId") != rework.get("reworkOrderId"):
|
|
fail("reworkClosureViolationCount", "INITIAL_INSPECTION_BACKREF", initial.get("inspectionId"), "failed inspection must reference its NCR and rework order")
|
|
if reinspection is None or reinspection.get("inspectionType") != "REINSPECTION" or reinspection.get("status") not in {"PASSED", "CLOSED", "RELEASED"}:
|
|
fail("reinspectionViolationCount", "REINSPECTION", ncr_id, "NCR must have successful reinspection")
|
|
continue
|
|
timeline = (
|
|
_datetime(initial.get("heldAt"), timezone),
|
|
_datetime(initial.get("inspectedAt"), timezone),
|
|
_datetime(nonconformity.get("openedAt"), timezone),
|
|
_datetime((rework_slot or {}).get("start"), timezone),
|
|
_datetime((rework_slot or {}).get("end"), timezone),
|
|
_datetime(reinspection.get("inspectedAt"), timezone),
|
|
_datetime(reinspection.get("releasedAt"), timezone),
|
|
_datetime(nonconformity.get("closedAt"), timezone),
|
|
)
|
|
if any(value is None for value in timeline) or not all(left <= right for left, right in pairwise(timeline)):
|
|
fail("reinspectionViolationCount", "REWORK_TIMELINE", ncr_id, "FAILED -> NCR -> REWORK -> REINSPECTION -> RELEASE/CLOSE timeline is invalid")
|
|
planned = _num(rework.get("plannedHours"))
|
|
scheduled = _num((rework_slot or {}).get("durationHours"))
|
|
if planned <= 0 or not isclose(planned, scheduled, abs_tol=0.01):
|
|
fail("reworkClosureViolationCount", "REWORK_HOURS", rework.get("reworkOrderId"), "planned and independently scheduled rework hours must match")
|
|
|
|
supplier_monthly: dict[tuple[str, str, str], float] = defaultdict(float)
|
|
for suggestion in rows("purchase-suggestions"):
|
|
supplier_id, month = str(suggestion.get("supplierId") or ""), str(suggestion.get("needDate") or "")[:7]
|
|
supplier = suppliers.get(supplier_id) or {}
|
|
capacity_unit = str(supplier.get("monthlyCapacityUnit") or "")
|
|
suggestion_unit = str(suggestion.get("capacityUnit") or suggestion.get("unit") or "")
|
|
if capacity_unit and suggestion_unit == capacity_unit:
|
|
supplier_monthly[(supplier_id, month, capacity_unit)] += _num(suggestion.get("suggestedQuantity"))
|
|
else:
|
|
fail("supplierCapacityViolationCount", "PURCHASE_CAPACITY_UNIT", suggestion.get("purchaseSuggestionId"), "purchase quantity unit must match supplier monthly capacity unit")
|
|
blackout = {str(value)[:10] for value in supplier.get("blackoutDates") or []}
|
|
if str(suggestion.get("suggestedOrderDate") or "")[:10] in blackout:
|
|
fail("supplierBlackoutViolationCount", "SUPPLIER_BLACKOUT", suggestion.get("purchaseSuggestionId"), "purchase occurs in supplier blackout")
|
|
for suggestion in rows("outsource-suggestions"):
|
|
supplier_id, month = str(suggestion.get("supplierId") or ""), str(suggestion.get("sendDate") or "")[:7]
|
|
supplier = suppliers.get(supplier_id) or {}
|
|
capacity_unit = str(supplier.get("monthlyCapacityUnit") or "")
|
|
suggestion_unit = str(suggestion.get("capacityUnit") or suggestion.get("unit") or "")
|
|
quantity = _num(suggestion.get("capacityQuantity") or suggestion.get("quantity"))
|
|
if capacity_unit and suggestion_unit == capacity_unit and quantity > 0:
|
|
supplier_monthly[(supplier_id, month, capacity_unit)] += quantity
|
|
else:
|
|
fail("supplierCapacityViolationCount", "OUTSOURCE_MONTHLY_CAPACITY_UNIT", suggestion.get("outsourceSuggestionId"), "outsource quantity unit must match supplier monthly capacity unit")
|
|
for (supplier_id, month, capacity_unit), demand in supplier_monthly.items():
|
|
capacity = _num((suppliers.get(supplier_id) or {}).get("monthlyCapacity"))
|
|
if capacity > 0 and demand > capacity + 0.001:
|
|
fail("supplierCapacityViolationCount", "SUPPLIER_CAPACITY", f"{supplier_id}:{month}:{capacity_unit}", f"quantity {demand} exceeds capacity {capacity}")
|
|
|
|
relation_counts = Counter(str(row.get("relationType") or "") for row in operations.values() if row.get("predecessorOperationId"))
|
|
coverage = {
|
|
"precedenceFS": relation_counts["FS"],
|
|
"precedenceSS": relation_counts["SS"],
|
|
"precedenceFF": relation_counts["FF"],
|
|
"precedenceSF": relation_counts["SF"],
|
|
"positiveLag": sum(_num(row.get("lagHours")) > 0 for row in operations.values()),
|
|
"negativeLag": sum(_num(row.get("lagHours")) < 0 for row in operations.values()),
|
|
"qualification": sum(bool(row.get("requiredQualificationCodes")) for row in operations.values()),
|
|
"weightDimension": sum(any(_num(row.get(field)) > 0 for field in ("requiredWeightT", "requiredLengthM", "requiredWidthM", "requiredHeightM", "requiredLiftingRadiusM")) for row in operations.values()),
|
|
"transport": sum(bool(row.get("transportWindowStatus")) and bool(row.get("transportZone")) for row in operations.values()),
|
|
"hazard": sum(bool(row.get("hazardClass")) for row in operations.values()),
|
|
"areaDensity": sum(_num(row.get("maxAreaDensityUnits") or row.get("areaDensityLimit")) > 0 for row in operations.values()) + sum(_num(row.get("maxConcurrentOperations") or row.get("maxHotWorkOperations")) > 0 for row in zones.values()),
|
|
"maintenanceBlackout": sum(bool(row.get("blackoutDates")) for row in calendars.values() if row.get("calendarType") == "MAINTENANCE"),
|
|
"supplierBlackout": sum(bool(row.get("blackoutDates")) for row in suppliers.values()),
|
|
"supplierCapacityUnit": sum(bool(row.get("monthlyCapacityUnit")) and _num(row.get("monthlyCapacity")) > 0 for row in suppliers.values()),
|
|
"weatherWindow": sum(bool(row.get("weatherSensitive")) and bool(row.get("weatherWindowStatus")) and bool(row.get("weatherWindowId") or row.get("weatherWindowDate") or (row.get("weatherWindowStart") and row.get("weatherWindowEnd"))) for row in operations.values()),
|
|
}
|
|
for name, count in coverage.items():
|
|
checks[f"constraintCoverage.{name}"] = count
|
|
if count <= 0:
|
|
fail("constraintCoverageViolationCount", "CONSTRAINT_COVERAGE", name, f"constraint family {name} has no executable synthetic sample")
|
|
|
|
baseline = bundle.artifacts.get("baseline") or {}
|
|
if baseline.get("solveStatus") == "FEASIBLE" and (_integer(baseline.get("hardConstraintViolations")) != 0 or _integer(baseline.get("unmarkedConflictCount")) != 0):
|
|
fail("unmarkedHardConstraintViolationCount", "BASELINE_FALSE_FEASIBLE", "baseline", "baseline FEASIBLE status conflicts with hard violations")
|
|
|
|
hard_metrics = ("resourceOverlapCount", "teamOverlapCount", "precedenceViolationCount", "materialReadinessViolationCount", "illegalFrozenChangeCount", "qualityHoldPointViolationCount", "liftingCapacityViolationCount", "hotWorkDensityViolationCount")
|
|
issue_counts = Counter(row["code"] for row in violations)
|
|
return {
|
|
"valid": not violations,
|
|
"status": "PASS" if not violations else "FAIL",
|
|
"solveStatus": "FEASIBLE" if not violations else "INFEASIBLE",
|
|
"blockingIssues": [f"{row['code']}:{row['entityId']}:{row['message']}" for row in violations],
|
|
"violations": violations,
|
|
"metrics": metrics,
|
|
"checks": dict(sorted(checks.items())),
|
|
"issueCounts": dict(sorted(issue_counts.items())),
|
|
"tableCounts": table_counts,
|
|
"expectedCounts": expected_counts,
|
|
"scaleProfile": asdict(config.profile),
|
|
"datasetType": bundle.metadata.get("datasetType"),
|
|
"organizationScenario": bundle.metadata.get("organizationScenario"),
|
|
"planningBaseDate": config.planning_base_date.isoformat(),
|
|
"planningHorizonEnd": config.planning_horizon_end.isoformat(),
|
|
"referentialIntegrity": {
|
|
"valid": metrics["foreignKeyErrorCount"] == metrics["duplicatePrimaryKeyCount"] == metrics["requiredFieldErrorCount"] == 0,
|
|
"foreignKeyErrorCount": metrics["foreignKeyErrorCount"],
|
|
"duplicatePrimaryKeyCount": metrics["duplicatePrimaryKeyCount"],
|
|
"requiredFieldErrorCount": metrics["requiredFieldErrorCount"],
|
|
},
|
|
"constraintValidation": {
|
|
"valid": not any(metrics[name] for name in hard_metrics),
|
|
"hardViolationCount": sum(metrics[name] for name in hard_metrics),
|
|
},
|
|
}
|
|
|
|
|
|
def _invalid_selftest_fixture(config: GeneratorConfig) -> DatasetBundle:
|
|
return DatasetBundle(metadata={**config.metadata(), "datasetType": "REAL"}, tables={name: [] for name in TABLE_SPECS}, artifacts={})
|
|
|
|
|
|
def _selftest() -> None:
|
|
config = GeneratorConfig.for_scale("small")
|
|
report = validate_bundle(_invalid_selftest_fixture(config), config)
|
|
if report["valid"] or report["metrics"]["syntheticBoundaryViolationCount"] < 1:
|
|
raise AssertionError("validation selftest fixture did not fail predictably")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
_selftest()
|