Module adcp.decisioning.translation

Bidirectional, type-safe key translation between AdCP and upstream values.

Every translator-pattern adapter owes the framework two things: a thin HTTP client (see :mod:adcp.decisioning.upstream) and a set of small bidirectional key maps (AdCP "olv" ↔ upstream "video", AdCP "guaranteed" ↔ upstream "GUARANTEED_AT_PRICE", …). :func:create_translation_map() ships the map primitive so adapters declare the mapping once and the framework projects it both ways.

Mirrors the JS createTranslationMap helper from @adcp/sdk@6.7 (src/lib/server/upstream-helpers.ts).

Functions

def create_translation_map(adcp_to_upstream: Mapping[A, U],
*,
default_adcp: A | None = None,
default_upstream: U | None = None) ‑> TranslationMap[~A, ~U]

Expand source code

def create_translation_map(
    adcp_to_upstream: Mapping[A, U],
    *,
    default_adcp: A | None = None,
    default_upstream: U | None = None,
) -> TranslationMap[A, U]:
    """Build a bidirectional translation map from an AdCP→upstream record.

    Keys on the left side are AdCP wire values; values on the right
    are upstream platform values. Collisions in either direction (two
    AdCP keys mapping to the same upstream value) are detected at
    construction time and raise :class:`ValueError` — silent
    overwrite would produce wrong-tenant routing in production.

    :param adcp_to_upstream: ``{adcp_key: upstream_key}`` mapping.
    :param default_adcp: Returned by :meth:`TranslationMap.to_adcp`
        when the upstream key isn't in the map. ``None`` (default)
        raises ``KeyError``.
    :param default_upstream: Returned by
        :meth:`TranslationMap.to_upstream` when the AdCP key isn't in
        the map. ``None`` (default) raises ``KeyError``.

    Example::

        delivery_map = create_translation_map({
            "guaranteed": "GUARANTEED_AT_PRICE",
            "non_guaranteed": "STANDARD",
        })
        delivery_map.to_upstream("guaranteed")  # "GUARANTEED_AT_PRICE"
        delivery_map.to_adcp("STANDARD")        # "non_guaranteed"
    """
    return TranslationMap(
        adcp_to_upstream,
        default_adcp=default_adcp,
        default_upstream=default_upstream,
    )

Build a bidirectional translation map from an AdCP→upstream record.

Keys on the left side are AdCP wire values; values on the right are upstream platform values. Collisions in either direction (two AdCP keys mapping to the same upstream value) are detected at construction time and raise :class:ValueError — silent overwrite would produce wrong-tenant routing in production.

:param adcp_to_upstream: {adcp_key: upstream_key} mapping. :param default_adcp: Returned by :meth:TranslationMap.to_adcp() when the upstream key isn't in the map. None (default) raises KeyError. :param default_upstream: Returned by :meth:TranslationMap.to_upstream() when the AdCP key isn't in the map. None (default) raises KeyError.

Example::

delivery_map = create_translation_map({
    "guaranteed": "GUARANTEED_AT_PRICE",
    "non_guaranteed": "STANDARD",
})
delivery_map.to_upstream("guaranteed")  # "GUARANTEED_AT_PRICE"
delivery_map.to_adcp("STANDARD")        # "non_guaranteed"

Classes

class TranslationMap (adcp_to_upstream: Mapping[A, U],
*,
default_adcp: A | None = None,
default_upstream: U | None = None)

Expand source code

class TranslationMap(Generic[A, U]):
    """Reversible mapping between AdCP wire values and upstream platform values.

    Construct via :func:`create_translation_map`. Both lookup methods
    raise :class:`KeyError` for unknown keys unless a default is
    configured. Use the ``has_*`` predicates to guard a lookup, or
    pass ``default_adcp`` / ``default_upstream`` for graceful
    fallback.

    Example::

        channel_map = create_translation_map({
            "olv": "video",
            "ctv": "ctv",
            "display": "display",
            "streaming_audio": "audio",
        })
        channel_map.to_upstream("olv")     # "video"
        channel_map.to_adcp("video")       # "olv"
        channel_map.has_adcp("olv")        # True
        channel_map.has_upstream("audio")  # True
    """

    __slots__ = ("_forward", "_reverse", "_default_adcp", "_default_upstream")

    def __init__(
        self,
        adcp_to_upstream: Mapping[A, U],
        *,
        default_adcp: A | None = None,
        default_upstream: U | None = None,
    ) -> None:
        self._forward: dict[A, U] = dict(adcp_to_upstream)
        self._default_adcp = default_adcp
        self._default_upstream = default_upstream

        # Build reverse map and detect collisions in the same pass. Two
        # AdCP keys mapping to the same upstream value would silently
        # overwrite without this check — fail loud at construction so
        # the bug is caught at boot, not on a request.
        reverse: dict[U, A] = {}
        for adcp_key, upstream_key in self._forward.items():
            if upstream_key in reverse:
                raise ValueError(
                    f"translation collision: AdCP keys "
                    f"{reverse[upstream_key]!r} and {adcp_key!r} both map to "
                    f"upstream value {upstream_key!r}"
                )
            reverse[upstream_key] = adcp_key
        self._reverse: dict[U, A] = reverse

    def to_upstream(self, adcp_key: A) -> U:
        """Translate an AdCP wire value to the upstream platform value.

        :raises KeyError: when ``adcp_key`` isn't in the map and no
            ``default_upstream`` was provided.
        """
        if adcp_key in self._forward:
            return self._forward[adcp_key]
        if self._default_upstream is not None:
            return self._default_upstream
        known = sorted(self._forward.keys(), key=repr)
        raise KeyError(f"unknown AdCP key {adcp_key!r}; known keys: {known!r}")

    def to_adcp(self, upstream_key: U) -> A:
        """Translate an upstream platform value back to the AdCP wire value.

        :raises KeyError: when ``upstream_key`` isn't in the map and
            no ``default_adcp`` was provided.
        """
        if upstream_key in self._reverse:
            return self._reverse[upstream_key]
        if self._default_adcp is not None:
            return self._default_adcp
        known = sorted(self._reverse.keys(), key=repr)
        raise KeyError(f"unknown upstream key {upstream_key!r}; known keys: {known!r}")

    def has_adcp(self, value: object) -> bool:
        """True when ``value`` is a known AdCP-side key."""
        return value in self._forward

    def has_upstream(self, value: object) -> bool:
        """True when ``value`` is a known upstream-side key."""
        return value in self._reverse

Reversible mapping between AdCP wire values and upstream platform values.

Construct via :func:create_translation_map(). Both lookup methods raise :class:KeyError for unknown keys unless a default is configured. Use the has_* predicates to guard a lookup, or pass default_adcp / default_upstream for graceful fallback.

Example::

channel_map = create_translation_map({
    "olv": "video",
    "ctv": "ctv",
    "display": "display",
    "streaming_audio": "audio",
})
channel_map.to_upstream("olv")     # "video"
channel_map.to_adcp("video")       # "olv"
channel_map.has_adcp("olv")        # True
channel_map.has_upstream("audio")  # True

Ancestors

• typing.Generic

Methods

def has_adcp(self, value: object) ‑> bool

Expand source code

def has_adcp(self, value: object) -> bool:
    """True when ``value`` is a known AdCP-side key."""
    return value in self._forward

True when value is a known AdCP-side key.

def has_upstream(self, value: object) ‑> bool

Expand source code

def has_upstream(self, value: object) -> bool:
    """True when ``value`` is a known upstream-side key."""
    return value in self._reverse

True when value is a known upstream-side key.

def to_adcp(self, upstream_key: U) ‑> ~A

Expand source code

def to_adcp(self, upstream_key: U) -> A:
    """Translate an upstream platform value back to the AdCP wire value.

    :raises KeyError: when ``upstream_key`` isn't in the map and
        no ``default_adcp`` was provided.
    """
    if upstream_key in self._reverse:
        return self._reverse[upstream_key]
    if self._default_adcp is not None:
        return self._default_adcp
    known = sorted(self._reverse.keys(), key=repr)
    raise KeyError(f"unknown upstream key {upstream_key!r}; known keys: {known!r}")

Translate an upstream platform value back to the AdCP wire value.

:raises KeyError: when upstream_key isn't in the map and no default_adcp was provided.

def to_upstream(self, adcp_key: A) ‑> ~U

Expand source code

def to_upstream(self, adcp_key: A) -> U:
    """Translate an AdCP wire value to the upstream platform value.

    :raises KeyError: when ``adcp_key`` isn't in the map and no
        ``default_upstream`` was provided.
    """
    if adcp_key in self._forward:
        return self._forward[adcp_key]
    if self._default_upstream is not None:
        return self._default_upstream
    known = sorted(self._forward.keys(), key=repr)
    raise KeyError(f"unknown AdCP key {adcp_key!r}; known keys: {known!r}")

Translate an AdCP wire value to the upstream platform value.

:raises KeyError: when adcp_key isn't in the map and no default_upstream was provided.