autocad-mcp-pro/engineering/pid/insert.py

"""Placing catalogue symbols: define the block on first use, insert with
attributes, write the ACADMCP_PID payload, return the ports in WCS."""

from __future__ import annotations

from typing import TYPE_CHECKING

from engineering.layers import PID_LAYERS

from .symbols import FAMILY_LAYER, SymbolSpec, resolve, transform_port
from .tags import parse_tag, split_instrument_tag
from .xdata import symbol_payload, write_payload

if TYPE_CHECKING:
    from backends.base import AutoCADBackend

FAIL_POSITIONS = ("", "FO", "FC", "FL")
_PID_LAYER_DEFS = {name: (color, linetype, lw) for name, color, linetype, lw, _desc in PID_LAYERS}


async def ensure_layer(backend: AutoCADBackend, layer_name: str) -> bool:
    """Create ``layer_name`` if missing — from PID_LAYERS when it is one of ours."""
    existing = {lyr.name.lower() for lyr in await backend.layer_list()}
    if layer_name.lower() in existing:
        return False
    color, linetype, lineweight = _PID_LAYER_DEFS.get(layer_name, (7, "Continuous", 0.25))
    await backend.layer_create(
        name=layer_name, color=color, linetype=linetype, lineweight=lineweight
    )
    return True


async def ensure_block(backend: AutoCADBackend, spec: SymbolSpec) -> bool:
    """Define ``spec``'s block if the drawing does not have it yet."""
    names = {blk.name for blk in await backend.block_list()}
    if spec.name in names:
        return False
    await backend.block_define(
        spec.name, list(spec.primitives), list(spec.attdefs), 0.0, 0.0, False
    )
    return True


async def insert_symbol(
    backend: AutoCADBackend,
    spec: SymbolSpec,
    x: float,
    y: float,
    rotation: float = 0.0,
    scale: float = 1.0,
    attributes: dict | None = None,
    layer: str | None = None,
    payload: dict | None = None,
) -> dict:
    layer_name = layer or FAMILY_LAYER[spec.family]
    if layer_name == "line":
        raise ValueError(f"{spec.name}: markers need an explicit layer (the line's)")
    created = await ensure_layer(backend, layer_name)
    defined = await ensure_block(backend, spec)
    values = {att["tag"]: att["default"] for att in spec.attdefs}
    values.update({k: str(v) for k, v in (attributes or {}).items()})
    ref = await backend.block_insert(
        spec.name, x, y, scale, scale, rotation, values or None, layer_name
    )
    await write_payload(backend, ref.handle, payload or symbol_payload(spec))
    ports = {p.name: transform_port(p, x, y, rotation, scale) for p in spec.ports}
    return {
        "handle": ref.handle,
        "block_name": spec.name,
        "defined": defined,
        "layers_created": [layer_name] if created else [],
        "ports": ports,
        "layer": layer_name,
    }


async def place_symbol(
    backend: AutoCADBackend,
    symbol: str,
    x: float,
    y: float,
    rotation: float = 0.0,
    scale: float = 1.0,
    tag: str | None = None,
    desc: str | None = None,
    fail: str | None = None,
    link: str | None = None,
    layer: str | None = None,
    **options,
) -> dict:
    """The ``pid_symbol_insert`` tool body."""
    if scale <= 0:
        raise ValueError("scale must be > 0")
    spec = resolve(symbol, **options)
    if spec.family == "marker":
        raise ValueError(f"{symbol} is a line marker; pid_line_draw places markers and arrows")
    tags_present = {att["tag"] for att in spec.attdefs}
    attributes: dict[str, str] = {}
    warnings: list[str] = []
    tag_text: str | None = None
    if tag:
        tag_text = tag.strip()
        if spec.family == "instrument":
            parsed = parse_tag(tag_text, kind="instrument")
            func, loop = split_instrument_tag(tag_text)
            attributes.update(FUNC=func, LOOP=loop)
        else:
            parsed = parse_tag(tag_text, kind="auto")
            attributes["TAG"] = tag_text
        if not parsed["valid"]:
            warnings.extend(parsed["errors"])
    if desc is not None:
        if "DESC" not in tags_present:
            warnings.append(f"{symbol} has no DESC attribute; description ignored")
        else:
            attributes["DESC"] = desc
    if fail is not None:
        if "FAIL" not in tags_present:
            raise ValueError(f"{symbol}: 'fail' needs an actuator (no FAIL attribute)")
        if fail.upper() not in FAIL_POSITIONS:
            raise ValueError("fail must be one of FO, FC, FL (or empty)")
        attributes["FAIL"] = fail.upper()
    if link is not None:
        if "LINK" not in tags_present:
            raise ValueError(f"{symbol}: 'link' is only valid for offpage connectors")
        attributes["LINK"] = link
    result = await insert_symbol(backend, spec, x, y, rotation, scale, attributes, layer)
    result.update(tag=tag_text, tag_warnings=warnings, backend=backend.name)
    return result