aps-agent/server/shipyard_synthetic/config.py

from __future__ import annotations

from dataclasses import dataclass, replace
from datetime import date
from typing import ClassVar

DATASET_TYPE = "SYNTHETIC"
ORGANIZATION_SCENARIO = "BEIHAI_SHIPYARD_APS"
TIMEZONE = "Asia/Shanghai"
GENERATED_FOR = "APS development and validation"
DATA_DISCLAIMER = "模拟数据，不代表北海造船真实业务数据"
GENERATOR_VERSION = "1.0.0"
SCHEMA_VERSION = "1.0.0"
DEFAULT_RANDOM_SEED = 20260901
PLANNING_BASE_DATE = date(2026, 9, 1)
PLANNING_HORIZON_END = date(2027, 12, 31)


@dataclass(frozen=True, slots=True)
class ScaleProfile:
    project_count: int
    milestone_count: int
    grand_block_count: int
    section_count: int
    work_package_count: int
    wbs_task_count: int
    material_count: int
    bom_relation_count: int
    routing_count: int
    production_order_count: int
    operation_count: int
    schedule_slot_count: int
    equipment_resource_count: int
    team_count: int
    supplier_count: int
    purchase_suggestion_count: int
    outsource_suggestion_count: int
    risk_conflict_count: int
    knowledge_asset_count: int


PROFILES: dict[str, ScaleProfile] = {
    "small": ScaleProfile(1, 18, 4, 12, 40, 80, 160, 600, 20, 60, 300, 300, 24, 8, 8, 30, 8, 20, 20),
    "standard": ScaleProfile(4, 72, 36, 140, 500, 2000, 6000, 25000, 120, 1200, 6000, 6000, 100, 40, 30, 300, 80, 100, 60),
    "full": ScaleProfile(4, 72, 72, 280, 1000, 4000, 12000, 50000, 220, 2400, 12000, 12000, 150, 60, 40, 550, 140, 180, 120),
}


@dataclass(frozen=True, slots=True)
class GeneratorConfig:
    scale: str
    seed: int
    profile: ScaleProfile
    planning_base_date: date = PLANNING_BASE_DATE
    planning_horizon_end: date = PLANNING_HORIZON_END
    timezone: str = TIMEZONE
    scenario: str = "all"
    incremental: bool = False

    PROJECT_CODES: ClassVar[tuple[str, ...]] = (
        "BH-SYN-2601",
        "BH-SYN-2602",
        "BH-SYN-2603",
        "BH-SYN-2604",
    )

    @classmethod
    def for_scale(
        cls,
        scale: str = "full",
        *,
        project_count: int | None = None,
        seed: int = DEFAULT_RANDOM_SEED,
        scenario: str = "all",
        incremental: bool = False,
    ) -> GeneratorConfig:
        normalized = scale.strip().lower()
        if normalized not in PROFILES:
            raise ValueError(f"unknown scale: {scale}")
        source = PROFILES[normalized]
        count = source.project_count if project_count is None else int(project_count)
        if count < 1 or count > 4:
            raise ValueError("project_count must be between 1 and 4")
        if count == source.project_count:
            profile = source
        else:
            ratio = count / source.project_count
            def scaled(value: int, minimum: int = 1) -> int:
                return max(minimum, round(value * ratio))
            profile = replace(
                source,
                project_count=count,
                milestone_count=18 * count,
                grand_block_count=scaled(source.grand_block_count, count),
                section_count=scaled(source.section_count, count),
                work_package_count=scaled(source.work_package_count, count),
                wbs_task_count=scaled(source.wbs_task_count, count),
                material_count=scaled(source.material_count, 40),
                bom_relation_count=scaled(source.bom_relation_count, 100),
                routing_count=scaled(source.routing_count, 10),
                production_order_count=scaled(source.production_order_count, count),
                operation_count=scaled(source.operation_count, count * 5),
                schedule_slot_count=scaled(source.schedule_slot_count, count * 5),
                purchase_suggestion_count=scaled(source.purchase_suggestion_count, count),
                outsource_suggestion_count=scaled(source.outsource_suggestion_count, count),
                risk_conflict_count=scaled(source.risk_conflict_count, count),
                knowledge_asset_count=max(20, scaled(source.knowledge_asset_count, 20)),
            )
        if profile.operation_count != profile.schedule_slot_count:
            raise ValueError("active operation count must equal baseline schedule slot count")
        return cls(
            scale=normalized,
            seed=int(seed),
            profile=profile,
            scenario=scenario,
            incremental=bool(incremental),
        )

    @property
    def project_codes(self) -> tuple[str, ...]:
        return self.PROJECT_CODES[: self.profile.project_count]

    def metadata(self) -> dict:
        return {
            "datasetType": DATASET_TYPE,
            "organizationScenario": ORGANIZATION_SCENARIO,
            "timezone": self.timezone,
            "generatedFor": GENERATED_FOR,
            "dataDisclaimer": DATA_DISCLAIMER,
            "generatorVersion": GENERATOR_VERSION,
            "schemaVersion": SCHEMA_VERSION,
            "randomSeed": self.seed,
            "datasetProfile": self.scale.upper(),
            "planningBaseDate": self.planning_base_date.isoformat(),
            "planningHorizonEnd": self.planning_horizon_end.isoformat(),
            "projectCodes": list(self.project_codes),
            "scenarioSelection": self.scenario,
        }