Part number variants: optional {variant} token, derived families

src/opal/api/routes/parts.py

@@ -16,7 +16,9 @@
 from opal.core.audit import get_model_dict, log_create, log_delete, log_update
 from opal.core.numbering import (
     PartNumberError,
+    format_has_variant,
     next_part_number,
+    next_variant_part_number,
     peek_next_part_number,
     pn_exists,
     register_part_number,
@@ -30,7 +32,7 @@
     ensure_identity_mutable,
     reference_counts,
 )
-from opal.db.models import InventoryRecord, Part, Supplier, SupplierPart
+from opal.db.models import BOMLine, InventoryRecord, Part, Supplier, SupplierPart
 
 router = APIRouter()
 
@@ -73,6 +75,26 @@ class PartUpdate(BaseModel):
     metadata: dict[str, Any] | None = None
 
 
+class PartVariantCreate(BaseModel):
+    """Overrides for a new variant; omitted fields copy from the source part.
+
+    Identity (tier, internal_pn) is derived from the source and never
+    overridable — a variant is a sibling, not a new sequence.
+    """
+
+    name: str | None = None
+    description: str | None = None
+    category: str | None = None
+    unit_of_measure: str | None = None
+    tracking_type: str | None = None  # "bulk" or "serialized"
+    external_pn: str | None = None
+    reorder_point: Decimal | None = None
+    is_tooling: bool | None = None
+    calibration_interval_days: int | None = None
+    parent_id: int | None = None
+    metadata: dict[str, Any] | None = None
+
+
 class PartResponse(BaseModel):
     """Schema for part response."""
 
@@ -310,6 +332,99 @@ def create_part(
     return get_part_with_quantity(db, part)
 
 
+@router.post(
+    "/{part_id}/variants", response_model=PartResponse, status_code=status.HTTP_201_CREATED
+)
+def create_part_variant(
+    db: DbSession,
+    part_id: int,
+    user_id: CurrentUserId,
+    variant_in: PartVariantCreate | None = None,
+) -> PartResponse:
+    """Mint the next variant of a part: a draft sibling copying attributes and BOM.
+
+    The variant shares the source's tier+sequence base with the next variant
+    code; the tier sequence counter does not advance. Body fields override
+    the copied attributes; omitted fields copy, explicit nulls clear.
+    """
+    part = db.query(Part).filter(Part.id == part_id, Part.deleted_at.is_(None)).first()
+    if not part:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail=f"Part {part_id} not found"
+        )
+    if not format_has_variant(get_active_project()):
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="Part numbering format has no {variant} placeholder",
+        )
+    if not part.internal_pn:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail="Part has no internal part number, so its variant family cannot be derived",
+        )
+
+    overrides = variant_in.model_dump(exclude_unset=True) if variant_in else {}
+    parent_id = overrides.get("parent_id", part.parent_id)
+    if parent_id is not None and parent_id != part.parent_id:
+        parent = db.query(Part).filter(Part.id == parent_id, Part.deleted_at.is_(None)).first()
+        if not parent:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail=f"Parent part {parent_id} not found",
+            )
+
+    try:
+        new_pn = next_variant_part_number(db, part.tier, part.internal_pn)
+    except PartNumberError as e:
+        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(e)) from e
+
+    if "metadata" in overrides:
+        metadata = overrides["metadata"]
+    else:
+        metadata = dict(part.metadata_) if part.metadata_ else None
+
+    variant = Part(
+        # name is NOT NULL: a null/empty override falls back to the source name
+        name=overrides.get("name") or part.name,
+        internal_pn=new_pn,
+        external_pn=overrides.get("external_pn", part.external_pn),
+        description=overrides.get("description", part.description),
+        category=overrides.get("category", part.category),
+        unit_of_measure=overrides.get("unit_of_measure") or part.unit_of_measure,
+        tracking_type=overrides.get("tracking_type") or part.tracking_type,
+        tier=part.tier,
+        parent_id=parent_id,
+        reorder_point=overrides.get("reorder_point", part.reorder_point),
+        # NOT NULL: a null override means "no opinion", not "clear"
+        is_tooling=(
+            part.is_tooling if overrides.get("is_tooling") is None else overrides["is_tooling"]
+        ),
+        calibration_interval_days=overrides.get(
+            "calibration_interval_days", part.calibration_interval_days
+        ),
+        metadata_=metadata,
+    )
+    db.add(variant)
+    db.flush()
+    log_create(db, variant, user_id)
+
+    for line in part.bom_lines:
+        bom_copy = BOMLine(
+            assembly_id=variant.id,
+            component_id=line.component_id,
+            quantity=line.quantity,
+            reference_designator=line.reference_designator,
+            notes=line.notes,
+        )
+        db.add(bom_copy)
+        db.flush()
+        log_create(db, bom_copy, user_id)
+
+    db.commit()
+    db.refresh(variant)
+    return get_part_with_quantity(db, variant)
+
+
 class NextPnResponse(BaseModel):
     """Preview of the next part number for a tier."""
 

src/opal/api/routes/project.py

@@ -35,6 +35,7 @@ class PartNumberingInput(BaseModel):
     prefix: str = ""
     separator: str = "-"
     sequence_digits: int = 4
+    variant_digits: int = 3
     format: str = "{prefix}{sep}{tier_code}{sep}{sequence}"
 
 
@@ -98,6 +99,7 @@ def from_config(cls, config: ProjectConfig) -> "ProjectConfigResponse":
                 prefix=config.part_numbering.prefix,
                 separator=config.part_numbering.separator,
                 sequence_digits=config.part_numbering.sequence_digits,
+                variant_digits=config.part_numbering.variant_digits,
                 format=config.part_numbering.format,
             ),
             categories=config.categories,
@@ -125,6 +127,7 @@ def _numbering_from_input(pn: PartNumberingInput) -> PartNumberingConfig:
         prefix=pn.prefix,
         separator=pn.separator,
         sequence_digits=pn.sequence_digits,
+        variant_digits=pn.variant_digits,
         format=pn.format,
     )
 

src/opal/core/numbering.py

@@ -42,7 +42,26 @@ def part_number_regex(project: ProjectConfig | None, tier_level: int) -> re.Patt
     """
     if project is None:
         return re.compile(rf"PN-{tier_level}-(?P<sequence>\d{{{_FALLBACK_SEQUENCE_DIGITS},}})$")
+    return _compile_format_regex(project, tier_level, project.part_numbering.format)
 
+
+def base_part_number_regex(project: ProjectConfig, tier_level: int) -> re.Pattern[str]:
+    """Regex for the format with its {variant} segment stripped.
+
+    Matches part numbers minted before {variant} entered the format; those
+    legacy bases anchor a variant family without being renumbered.
+    """
+    template = project.part_numbering.format
+    if "{sep}{variant}" in template:
+        template = template.replace("{sep}{variant}", "")
+    else:
+        template = template.replace("{variant}", "")
+    return _compile_format_regex(project, tier_level, template)
+
+
+def _compile_format_regex(
+    project: ProjectConfig, tier_level: int, template: str
+) -> re.Pattern[str]:
     tier = project.get_tier(tier_level)
     if tier is None:
         raise PartNumberError(f"Unknown tier level: {tier_level}")
@@ -54,14 +73,15 @@ def part_number_regex(project: ProjectConfig | None, tier_level: int) -> re.Patt
         "tier_name": tier.name,
         "tier_level": str(tier.level),
     }
-    template = project.part_numbering.format
     pattern = ""
     pos = 0
     for match in re.finditer(r"\{(\w+)\}", template):
         pattern += re.escape(template[pos : match.start()])
         token = match.group(1)
         if token == "sequence":
             pattern += rf"(?P<sequence>\d{{{project.part_numbering.sequence_digits},}})"
+        elif token == "variant":
+            pattern += rf"(?P<variant>\d{{{project.part_numbering.variant_digits},}})"
         elif token in values:
             pattern += re.escape(values[token])
         else:
@@ -127,6 +147,65 @@ def next_part_number(db: Session, tier_level: int) -> str:
     return pn
 
 
+def format_has_variant(project: ProjectConfig | None) -> bool:
+    """True when the numbering format mints variant codes."""
+    return project is not None and "{variant}" in project.part_numbering.format
+
+
+def variant_family_key(
+    project: ProjectConfig | None, tier_level: int, pn: str
+) -> tuple[int, int] | None:
+    """(sequence, variant) placing a PN in its variant family, or None.
+
+    A PN matching the format with the {variant} segment stripped is a legacy
+    base — numbered before {variant} entered the format. It counts as
+    variant 1 of its family: 001-0001 IS configuration 1, so its first
+    explicit variant mints -002 and code 001 stays a permanent gap.
+    """
+    if not format_has_variant(project):
+        return None
+    match = part_number_regex(project, tier_level).match(pn)
+    if match:
+        return int(match.group("sequence")), int(match.group("variant"))
+    match = base_part_number_regex(project, tier_level).match(pn)
+    if match:
+        return int(match.group("sequence")), 1
+    return None
+
+
+def next_variant_part_number(db: Session, tier_level: int, source_pn: str) -> str:
+    """Mint the next variant of an existing part number.
+
+    Variants share the source's tier+sequence base; the next code is
+    max(existing)+1 across ALL rows including soft-deleted ones — variant
+    codes are never reused. The tier sequence counter is not touched:
+    a variant is a sibling, not a new sequence.
+    """
+    from opal.config import get_active_project
+
+    project = get_active_project()
+    if project is None or not format_has_variant(project):
+        raise PartNumberError("Part numbering format has no {variant} placeholder")
+
+    key = variant_family_key(project, tier_level, source_pn)
+    if key is None:
+        example = _format_part_number(project, tier_level, 1)
+        raise PartNumberError(
+            f"'{source_pn}' does not match the tier-{tier_level} part number format "
+            f"(e.g. {example}), so its variant family cannot be derived"
+        )
+    sequence = key[0]
+
+    # Full-column scan + regex match: correct for arbitrary token order and
+    # fine at single-instance SQLite scale.
+    max_variant = 0
+    for (pn,) in db.query(Part.internal_pn).filter(Part.internal_pn.isnot(None)).all():
+        k = variant_family_key(project, tier_level, pn)
+        if k and k[0] == sequence:
+            max_variant = max(max_variant, k[1])
+    return project.generate_part_number(tier_level, sequence, max_variant + 1)
+
+
 def peek_next_part_number(db: Session, tier_level: int) -> tuple[str, int]:
     """Preview the next part number for a tier without consuming it."""
     from opal.config import get_active_project

src/opal/project.py

@@ -22,6 +22,7 @@ class PartNumberingConfig(BaseModel):
     prefix: str = ""
     separator: str = "-"
     sequence_digits: int = 4
+    variant_digits: int = 3
     format: str = "{prefix}{sep}{tier_code}{sep}{sequence}"
 
 
@@ -140,12 +141,13 @@ def get_requirement(self, req_id: str) -> RequirementConfig | None:
                 return req
         return None
 
-    def generate_part_number(self, tier_level: int, sequence: int) -> str:
+    def generate_part_number(self, tier_level: int, sequence: int, variant: int = 1) -> str:
         """Generate a part number according to project config.
 
         Args:
             tier_level: The tier level (1, 2, 3, etc.)
             sequence: The sequence number for this part.
+            variant: The variant code, used only when the format includes {variant}.
 
         Returns:
             Formatted part number string.
@@ -164,6 +166,7 @@ def generate_part_number(self, tier_level: int, sequence: int) -> str:
             tier_name=tier.name,
             tier_level=tier.level,
             sequence=str(sequence).zfill(self.part_numbering.sequence_digits),
+            variant=str(variant).zfill(self.part_numbering.variant_digits),
         )
 
 
@@ -280,6 +283,7 @@ def create_project_config(
     prefix: str = "",
     separator: str = "-",
     sequence_digits: int = 4,
+    variant_digits: int = 3,
     part_number_format: str = "{prefix}{sep}{tier_code}{sep}{sequence}",
     tiers: list[TierConfig] | None = None,
     requirements: list[RequirementConfig] | None = None,
@@ -295,6 +299,7 @@ def create_project_config(
         prefix: Part number prefix.
         separator: Part number separator (default: "-").
         sequence_digits: Number of digits in sequence (default: 4).
+        variant_digits: Number of digits in the variant code (default: 3).
         part_number_format: Format string for part numbers.
         tiers: List of inventory tiers (defaults to Flight/Ground/Loose).
         requirements: List of project requirements.
@@ -315,6 +320,7 @@ def create_project_config(
             prefix=prefix,
             separator=separator,
             sequence_digits=sequence_digits,
+            variant_digits=variant_digits,
             format=part_number_format,
         ),
         requirements=requirements or [],
@@ -358,6 +364,7 @@ def save_project_config(config: ProjectConfig) -> None:
             "prefix": config.part_numbering.prefix,
             "separator": config.part_numbering.separator,
             "sequence_digits": config.part_numbering.sequence_digits,
+            "variant_digits": config.part_numbering.variant_digits,
             "format": config.part_numbering.format,
         },
         "requirements": [