feat: add Talker Alias text encoding (utf8/iso8/7bit)

Add TALKER_ALIAS_TEXT_FORMAT: single or comma-separated encodings for
embedded LC (e.g. utf8,iso8 for Motorola and Hytera). Example template
defaults to both formats; runtime default remains utf8 when omitted.

adn-server.example.yaml

@@ -21,6 +21,8 @@ GLOBAL:
   TALKER_ALIAS: false
   TALKER_ALIAS_MODE: both
   TALKER_ALIAS_FORMAT: "{callsign} {fname}"
+  # utf8 (Motorola), iso8 (Hytera), 7bit; both vendors by default
+  TALKER_ALIAS_TEXT_FORMAT: "utf8,iso8"
 
 REPORTS:
   REPORT: true

docs/en/server/user-guide/configuration.md

@@ -88,6 +88,7 @@ Optional DMR Talker Alias on HBP (`DMRA` packets). Full guide: [Talker Alias](ta
 | **TALKER_ALIAS** | Enable server TA inject/passthrough (`false` default). |
 | **TALKER_ALIAS_MODE** | `both` (default), `passthrough`, or `inject`. |
 | **TALKER_ALIAS_FORMAT** | Template, e.g. `{callsign} {fname}`. Max **29** chars (protocol limit, not YAML). |
+| **TALKER_ALIAS_TEXT_FORMAT** | `utf8`, `iso8`, `7bit`, or comma list (e.g. `utf8,iso8` for Motorola + Hytera). Default `utf8`. |
 
 ---
 

docs/en/server/user-guide/talker-alias.md

@@ -42,6 +42,7 @@ GLOBAL:
   TALKER_ALIAS: false
   TALKER_ALIAS_MODE: both
   TALKER_ALIAS_FORMAT: "{callsign} {fname}"
+  TALKER_ALIAS_TEXT_FORMAT: "utf8,iso8"
 ```
 
 | Key | Meaning |
@@ -49,6 +50,7 @@ GLOBAL:
 | **TALKER_ALIAS** | Master switch (`false` by default). |
 | **TALKER_ALIAS_MODE** | `both`, `passthrough`, or `inject`. Default **`both`** if omitted. |
 | **TALKER_ALIAS_FORMAT** | Python format string; fields: `{callsign}`, `{fname}`, `{surname}`, `{id}`. |
+| **TALKER_ALIAS_TEXT_FORMAT** | TA payload encoding: `utf8` (Motorola / MMDVMHost default), `iso8` (ISO-8859-1, many Hytera models), or `7bit` (oldest radios). Comma-separated list (e.g. `utf8,iso8`) emits **both** encodings back-to-back in embedded LC so each vendor can pick the format it displays; standalone `DMRA` UDP uses the **first** format only. Default **`utf8`**. |
 
 Maximum string length is **29 characters** (ETSI / MMDVMHost). This limit is fixed in code and is **not** configurable, to avoid incompatible payloads on radios and hotspots.
 
@@ -65,7 +67,7 @@ Subscriber JSON may include `fname`, `surname`, or a dedicated `talker_alias` fi
 | 7 | Block index 0–3 |
 | 8–14 | 7 payload bytes |
 
-Encoding uses **UTF-8 format** (format 2), matching MMDVMHost `DMRTA.cpp`.
+Encoding follows **TALKER_ALIAS_TEXT_FORMAT** (ETSI formats 0/1/2: 7-bit, ISO-8859-1, UTF-8), matching MMDVMHost `DMRTA.cpp` when set to `utf8`.
 
 Embedded TA in `DMRD` voice alternates superframes: one cycle (bursts B–E) with the normal group embedded LC, the next with a TA block (FLCO 4–7), repeating until the stream ends.
 

docs/es/server/user-guide/configuration.md

@@ -88,6 +88,7 @@ Talker Alias DMR opcional en HBP (paquetes `DMRA`). Guía completa: [Talker Alia
 | **TALKER_ALIAS** | Activa inyección/passthrough de TA (`false` por defecto). |
 | **TALKER_ALIAS_MODE** | `both` (por defecto), `passthrough` o `inject`. |
 | **TALKER_ALIAS_FORMAT** | Plantilla, p. ej. `{callsign} {fname}`. Máx. **29** caracteres (límite de protocolo, no YAML). |
+| **TALKER_ALIAS_TEXT_FORMAT** | `utf8`, `iso8`, `7bit` o lista con comas (p. ej. `utf8,iso8`). Por defecto `utf8`. |
 
 ---
 

docs/es/server/user-guide/talker-alias.md

@@ -37,6 +37,7 @@ GLOBAL:
   TALKER_ALIAS: false
   TALKER_ALIAS_MODE: both
   TALKER_ALIAS_FORMAT: "{callsign} {fname}"
+  TALKER_ALIAS_TEXT_FORMAT: "utf8,iso8"
 ```
 
 | Clave | Significado |
@@ -44,6 +45,7 @@ GLOBAL:
 | **TALKER_ALIAS** | Interruptor maestro (`false` por defecto). |
 | **TALKER_ALIAS_MODE** | `both`, `passthrough` o `inject`. Por defecto **`both`** si se omite. |
 | **TALKER_ALIAS_FORMAT** | Plantilla tipo Python; campos: `{callsign}`, `{fname}`, `{surname}`, `{id}`. |
+| **TALKER_ALIAS_TEXT_FORMAT** | Codificación del TA: `utf8` (Motorola), `iso8` (Hytera), `7bit`. Lista separada por comas (p. ej. `utf8,iso8`) emite ambas en LC embebido; `DMRA` UDP usa solo el **primer** formato. Por defecto **`utf8`**. |
 
 La longitud máxima es **29 caracteres** (ETSI / MMDVMHost). Este límite está fijado en código y **no** es configurable, para evitar payloads incompatibles en radios y hotspots.
 

src/adn_server/application/ports.py

@@ -26,6 +26,7 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
+from collections.abc import Sequence
 from typing import Any, Protocol
 
 
@@ -172,7 +173,12 @@ class DmrEmbeddedLcEncoder(Protocol):
 class TalkerAliasEmblcEncoder(Protocol):
     """Encode Talker Alias into embedded-LC burst dicts for DMRD overlay."""
 
-    def encode_text(self, text: str) -> tuple[list[dict[int, Any]], int]:
+    def encode_text(
+        self,
+        text: str,
+        *,
+        text_formats: Sequence[str] | None = None,
+    ) -> tuple[list[dict[int, Any]], int]:
         ...
 
     def encode_blocks(self, blocks: dict[int, bytes]) -> tuple[list[dict[int, Any]], int]:

src/adn_server/application/talker_alias_use_cases.py

@@ -18,6 +18,7 @@ from ..domain.talker_alias import (
     build_dmra_packets,
     decode_ta_from_blocks,
     is_ta_header_byte,
+    parse_ta_text_formats,
     required_ta_block_count,
     talker_alias_decode_complete,
     truncate_talker_alias,
@@ -44,10 +45,14 @@ def talker_alias_settings(config: dict[str, Any], system_name: str | None = None
     fmt = sys_cfg.get("TALKER_ALIAS_FORMAT")
     if fmt is None:
         fmt = global_cfg.get("TALKER_ALIAS_FORMAT", "{callsign} {fname}")
+    tfmt = sys_cfg.get("TALKER_ALIAS_TEXT_FORMAT")
+    if tfmt is None:
+        tfmt = global_cfg.get("TALKER_ALIAS_TEXT_FORMAT", "utf8")
     return {
         "enabled": bool(enabled),
         "mode": mode,
         "format": str(fmt),
+        "text_formats": parse_ta_text_formats(tfmt),
     }
 
 
@@ -174,7 +179,7 @@ class TalkerAliasUseCases:
                 "(%s) *TALKER ALIAS* inject '%s' via %s -> %s stream %s",
                 source_system, text, via, target, int_id(stream_id),
             )
-            return build_dmra_packets(rf_src, text)
+            return build_dmra_packets(rf_src, text, settings["text_formats"][0])
         # both: prefer the source's own TA. If a valid MMDVM DMRA buffer arrived,
         # relay it. Otherwise the source's embedded LC (e.g. MMDVM voice) is passed
         # through unchanged in the DMRD voice, so do NOT inject a template here.
@@ -191,7 +196,7 @@ class TalkerAliasUseCases:
                 "(%s) *TALKER ALIAS* inject '%s' (no source TA) via %s -> %s stream %s",
                 source_system, text, via, target, int_id(stream_id),
             )
-            return build_dmra_packets(rf_src, text)
+            return build_dmra_packets(rf_src, text, settings["text_formats"][0])
         logger.debug(
             "(%s) *TALKER ALIAS* passthrough (source embedded TA) via %s -> %s stream %s",
             source_system, via, target, int_id(stream_id),
@@ -241,6 +246,7 @@ class TalkerAliasUseCases:
             return self._ta_emblc.encode_blocks(blocks)
         if mode == "inject" or fallback_inject:
             suffix = "" if mode == "inject" else " (no source TA)"
-            _log(format_talker_alias_text(self._config, rf_src), suffix)
-            return self._ta_emblc.encode_text(format_talker_alias_text(self._config, rf_src))
+            text = format_talker_alias_text(self._config, rf_src)
+            _log(text, suffix)
+            return self._ta_emblc.encode_text(text, text_formats=settings["text_formats"])
         return None

src/adn_server/domain/talker_alias.py

@@ -5,7 +5,7 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 if TYPE_CHECKING:
     from bitarray import bitarray
@@ -66,6 +66,21 @@ def decode_ta(buf: bytes) -> str:
 decode_7bit = decode_ta  # backwards-compatible alias
 
 
+VALID_TA_TEXT_FORMATS = frozenset({"utf8", "iso8", "7bit"})
+
+
+def parse_ta_text_formats(raw: Any) -> list[str]:
+    """Parse ``TALKER_ALIAS_TEXT_FORMAT`` (comma list or YAML list) → ordered encodings."""
+    if raw is None:
+        return ["utf8"]
+    if isinstance(raw, list):
+        parts = [str(x).strip().lower() for x in raw]
+    else:
+        parts = [f.strip().lower() for f in str(raw).split(",")]
+    valid = [f for f in parts if f in VALID_TA_TEXT_FORMATS]
+    return valid or ["utf8"]
+
+
 def encode_utf8(text: str) -> bytes:
     """Encode text into 28-byte TA buffer (format 2 / UTF-8)."""
     text = truncate_talker_alias(text)
@@ -80,30 +95,46 @@ def encode_utf8(text: str) -> bytes:
     return bytes(buf)
 
 
+def encode_iso8(text: str) -> bytes:
+    """Encode text into 28-byte TA buffer (format 1 / ISO-8859-1, best for Hytera)."""
+    text = truncate_talker_alias(text)
+    raw = text.encode("latin-1", errors="replace")[:27]
+    size = len(raw)
+    header = (TA_FORMAT_ISO8 << 6) | (size << 1) | 0
+    buf = bytearray(DMRA_BUF_LEN)
+    buf[0] = header
+    buf[1 : 1 + size] = raw
+    return bytes(buf)
+
+
 def encode_7bit(text: str) -> bytes:
-    """Encode text into 28-byte TA buffer (format 0 / 7-bit)."""
+    """Encode text into 28-byte TA buffer (format 0 / 7-bit packed, inverse of ``decode_ta``)."""
     text = truncate_talker_alias(text)
-    size = len(text)
-    bits: list[int] = []
-    for b in (0, 0):
-        bits.append(b)
+    chars = text.encode("ascii", errors="replace")[:31]
+    size = len(chars)
+    bits = [(TA_FORMAT_7BIT >> 1) & 1, TA_FORMAT_7BIT & 1]
     for b in range(4, -1, -1):
         bits.append((size >> b) & 1)
-    for ch in text:
-        v = ord(ch) & 0x7F
+    for ch in chars:
         for b in range(6, -1, -1):
-            bits.append((v >> b) & 1)
-    while len(bits) < DMRA_BUF_LEN * 8:
-        bits.append(0)
+            bits.append((ch >> b) & 1)
+    total = DMRA_BUF_LEN * 8
+    bits = (bits + [0] * total)[:total]
     buf = bytearray(DMRA_BUF_LEN)
-    for i in range(DMRA_BUF_LEN):
-        byte = 0
-        for j in range(8):
-            byte = (byte << 1) | bits[i * 8 + j]
-        buf[i] = byte
+    for i, bit in enumerate(bits):
+        if bit:
+            buf[i // 8] |= 1 << (7 - (i % 8))
     return bytes(buf)
 
 
+_TA_ENCODERS = {"utf8": encode_utf8, "iso8": encode_iso8, "7bit": encode_7bit}
+
+
+def encode_ta_buffer(text: str, text_format: str = "utf8") -> bytes:
+    """Encode text into a 28-byte TA buffer in the requested format."""
+    return _TA_ENCODERS.get(text_format, encode_utf8)(text)
+
+
 def blocks_from_buffer(buf: bytes) -> list[bytes]:
     """Split 28-byte encoded buffer into four 7-byte DMRA payloads."""
     buf = buf.ljust(DMRA_BUF_LEN, b"\x00")[:DMRA_BUF_LEN]
@@ -170,10 +201,10 @@ def decode_ta_from_blocks(blocks: dict[int, bytes]) -> str:
     return ""
 
 
-def build_dmra_packets(rf_src: bytes, text: str) -> list[bytes]:
+def build_dmra_packets(rf_src: bytes, text: str, text_format: str = "utf8") -> list[bytes]:
     """Build HBP DMRA packets (1–4) for server injection."""
     rf = rf_src[:3] if len(rf_src) >= 3 else rf_src.ljust(3, b"\x00")[:3]
-    encoded = encode_utf8(text)
+    encoded = encode_ta_buffer(text, text_format)
     blocks = blocks_from_buffer(encoded)
     count = required_ta_block_count(encoded)
     packets: list[bytes] = []

src/adn_server/infrastructure/config_normalizer.py

@@ -89,6 +89,7 @@ def apply_talker_alias_defaults(config: dict) -> None:
     g.setdefault("TALKER_ALIAS", False)
     g.setdefault("TALKER_ALIAS_MODE", "both")
     g.setdefault("TALKER_ALIAS_FORMAT", "{callsign} {fname}")
+    g.setdefault("TALKER_ALIAS_TEXT_FORMAT", "utf8")
 
 
 def normalize_peer_config(config: dict) -> None:

src/adn_server/infrastructure/config_validator.py

@@ -48,6 +48,7 @@ GLOBAL_STRING_KEYS = frozenset(
         "HASH_ENCRYPT",
         "TALKER_ALIAS_MODE",
         "TALKER_ALIAS_FORMAT",
+        "TALKER_ALIAS_TEXT_FORMAT",
         *ACL_KEYS,
     }
 )

src/adn_server/infrastructure/talker_alias_emblc.py

@@ -5,13 +5,14 @@
 
 from __future__ import annotations
 
+from collections.abc import Sequence
 from typing import TYPE_CHECKING
 
 from ..domain.talker_alias import (
     blocks_from_buffer,
     buffer_from_blocks,
     buffer_from_wire_blocks,
-    encode_utf8,
+    encode_ta_buffer,
     is_ta_header_byte,
     required_ta_block_count,
     talker_alias_decode_complete,
@@ -23,13 +24,22 @@ if TYPE_CHECKING:
     from bitarray import bitarray
 
 
-def encode_talker_alias_emblc(text: str) -> tuple[list[dict[int, bitarray]], int]:
-    """Embedded-LC dicts for TA blocks 0..N-1 and block count N (1–4)."""
-    encoded = encode_utf8(text)
-    blocks = blocks_from_buffer(encoded)
-    count = required_ta_block_count(encoded)
-    emblcs = [encode_emblc(talker_alias_lc_bytes(i, blocks[i])) for i in range(count)]
-    return emblcs, count
+def encode_talker_alias_emblc(
+    text: str,
+    text_formats: Sequence[str] | str = ("utf8",),
+) -> tuple[list[dict[int, bitarray]], int]:
+    """Embedded-LC dicts for TA blocks; multiple encodings are emitted back-to-back."""
+    if isinstance(text_formats, str):
+        formats: list[str] = [text_formats]
+    else:
+        formats = list(text_formats) or ["utf8"]
+    emblcs: list[dict[int, bitarray]] = []
+    for tf in formats:
+        encoded = encode_ta_buffer(text, tf)
+        blocks = blocks_from_buffer(encoded)
+        count = required_ta_block_count(encoded)
+        emblcs += [encode_emblc(talker_alias_lc_bytes(i, blocks[i])) for i in range(count)]
+    return emblcs, len(emblcs)
 
 
 def encode_talker_alias_emblc_from_blocks(
@@ -52,8 +62,14 @@ def encode_talker_alias_emblc_from_blocks(
 class DefaultTalkerAliasEmblcEncoder:
     """Infrastructure adapter for ``TalkerAliasEmblcEncoder`` (wired from ``main``)."""
 
-    def encode_text(self, text: str) -> tuple[list[dict[int, bitarray]], int]:
-        return encode_talker_alias_emblc(text)
+    def encode_text(
+        self,
+        text: str,
+        *,
+        text_formats: Sequence[str] | None = None,
+    ) -> tuple[list[dict[int, bitarray]], int]:
+        fmts: Sequence[str] = text_formats if text_formats is not None else ("utf8",)
+        return encode_talker_alias_emblc(text, fmts)
 
     def encode_blocks(self, blocks: dict[int, bytes]) -> tuple[list[dict[int, bitarray]], int]:
         return encode_talker_alias_emblc_from_blocks(blocks)