Source code for semi_cr.core.lab.station.graphing.parsing

from dataclasses import dataclass

from typing import Any
import re

_ENDPOINT_RE = re.compile(r"^(?P<instrument>[^\[]+)\[(?P<channel>.+)\]$")



@dataclass(frozen=True)
class Endpoint:
    owner: str
    terminal: str




def is_endpoint_ref(value: str) -> bool:
    return "[" in value and value.endswith("]")




def parse_endpoint_ref(endpoint: str) -> tuple[str, str]:
    owner, rest = endpoint.split("[", 1)
    name = rest.removesuffix("]")
    return owner, name


def _strip_matching_quotes(value: str) -> str:
    value = value.strip()

    if (
        len(value) >= 2
        and value[0] == value[-1]
        and value[0] in {"'", '"'}
    ):
        return value[1:-1]

    return value



def parse_endpoint(
    endpoint: str,
) -> tuple[str, str]:
    if not isinstance(endpoint, str):
        raise TypeError(
            f"Endpoint must be a string, got "
            f"{type(endpoint).__name__}."
        )

    endpoint = endpoint.strip()

    if not endpoint.endswith("]") or "[" not in endpoint:
        raise ValueError(
            f"Invalid endpoint {endpoint!r}. Expected "
            "'instrument[channel]'."
        )

    instrument_name, channel = (
        endpoint[:-1].split("[", maxsplit=1)
    )

    instrument_name = instrument_name.strip()
    channel = _strip_matching_quotes(channel)

    if not instrument_name or not channel:
        raise ValueError(
            f"Invalid endpoint {endpoint!r}: instrument "
            "and channel must be non-empty."
        )

    return instrument_name, channel


# def parse_endpoint(endpoint: str) -> Endpoint:
#     ...



def normalize_channel(
    channel: str | int,
) -> str:
    """
    Normalize a terminal channel path.

    Examples
    --------
    1              -> "1"
    "ch01"         -> "ch01"
    "LF_1[out1]"   -> "LF_1/out1"
    """

    if isinstance(channel, bool):
        raise TypeError(
            "Channel must be a string or integer, got bool."
        )

    if isinstance(channel, int):
        channel = str(channel)
    elif not isinstance(channel, str):
        raise TypeError(
            f"Channel must be a string or integer, got "
            f"{type(channel).__name__}."
        )

    channel = channel.strip()

    if not channel:
        raise ValueError("Terminal path cannot be empty.")

    has_opening = "[" in channel
    has_closing = "]" in channel

    if has_opening != has_closing:
        raise ValueError(
            f"Invalid terminal path {channel!r}: "
            "unmatched brackets."
        )

    if not has_opening:
        return channel

    if channel.count("[") != 1 or channel.count("]") != 1:
        raise ValueError(
            f"Invalid terminal path {channel!r}: "
            "expected one bracketed channel."
        )

    if not channel.endswith("]"):
        raise ValueError(
            f"Invalid terminal path {channel!r}: "
            "closing bracket must be at the end."
        )

    module_name, terminal_name = (
        channel[:-1].split("[", maxsplit=1)
    )

    if not module_name or not terminal_name:
        raise ValueError(
            f"Invalid terminal path {channel!r}: "
            "module and terminal names must be non-empty."
        )

    return f"{module_name}/{terminal_name}"




def parse_channel_number(name: str) -> int | None:
    match = re.search(r"(\d+)$", name)
    return int(match.group(1)) if match else None