Enhance text handling by adding emoji support and sanitization for curses rendering

Succumb to wcwidth

contact/ui/contact_ui.py

@@ -12,10 +12,9 @@ from contact.ui.colors import get_color
 from contact.utilities.db_handler import get_name_from_database, update_node_info_in_db, is_chat_archived
 from contact.utilities.input_handlers import get_list_input
 from contact.utilities.i18n import t
-from contact.utilities.emoji_utils import normalize_message_text
 import contact.ui.default_config as config
 import contact.ui.dialog
-from contact.ui.nav_utils import move_main_highlight, draw_main_arrows, get_msg_window_lines, wrap_text
+from contact.ui.nav_utils import draw_main_arrows, fit_text, get_msg_window_lines, move_main_highlight, wrap_text
 from contact.utilities.singleton import ui_state, interface_state, menu_state
 
 
@@ -829,9 +828,7 @@ def draw_channel_list() -> None:
         notification = " " + config.notification_symbol if idx in ui_state.notifications else ""
 
         # Truncate the channel name if it's too long to fit in the window
-        truncated_channel = (
-            (channel[: win_width - 5] + "-" if len(channel) > win_width - 5 else channel) + notification
-        ).ljust(win_width - 3)
+        truncated_channel = fit_text(f"{channel}{notification}", win_width - 3, suffix="-")
 
         color = get_color("channel_list")
         if idx == ui_state.selected_channel:
@@ -866,7 +863,7 @@ def draw_messages_window(scroll_to_bottom: bool = False) -> None:
 
         row = 0
         for prefix, message in messages:
-            full_message = normalize_message_text(f"{prefix}{message}")
+            full_message = f"{prefix}{message}"
             wrapped_lines = wrap_text(full_message, messages_win.getmaxyx()[1] - 2)
             msg_line_count += len(wrapped_lines)
             messages_pad.resize(msg_line_count, messages_win.getmaxyx()[1])
@@ -937,8 +934,7 @@ def draw_node_list() -> None:
         snr_str = f"  ■  SNR: {node['snr']}dB" if node.get("hopsAway") == 0 and "snr" in node else ""
 
         # Future node name custom formatting possible
-        node_str = f"{status_icon} {node_name}"
-        node_str = node_str.ljust(box_width - 4)[: box_width - 2]
+        node_str = fit_text(f"{status_icon} {node_name}", box_width - 2)
         color = "node_list"
         if "isFavorite" in node and node["isFavorite"]:
             color = "node_favorite"
@@ -1067,9 +1063,16 @@ def draw_packetlog_win() -> None:
             span += column
 
         # Add headers
-        headers = f"{'From':<{columns[0]}} {'To':<{columns[1]}} {'Port':<{columns[2]}} {'Payload':<{width-span}}"
+        headers = " ".join(
+            [
+                fit_text("From", columns[0]),
+                fit_text("To", columns[1]),
+                fit_text("Port", columns[2]),
+                fit_text("Payload", max(1, width - span - 3)),
+            ]
+        )
         packetlog_win.addstr(
-            1, 1, headers[: width - 2], get_color("log_header", underline=True)
+            1, 1, fit_text(headers, width - 2), get_color("log_header", underline=True)
         )  # Truncate headers if they exceed window width
 
         for i, packet in enumerate(reversed(ui_state.packet_buffer)):
@@ -1077,22 +1080,22 @@ def draw_packetlog_win() -> None:
                 break
 
             # Format each field
-            from_id = get_name_from_database(packet["from"], "short").ljust(columns[0])
+            from_id = fit_text(get_name_from_database(packet["from"], "short"), columns[0])
             to_id = (
-                "BROADCAST".ljust(columns[1])
+                fit_text("BROADCAST", columns[1])
                 if str(packet["to"]) == "4294967295"
-                else get_name_from_database(packet["to"], "short").ljust(columns[1])
+                else fit_text(get_name_from_database(packet["to"], "short"), columns[1])
             )
             if "decoded" in packet:
-                port = str(packet["decoded"].get("portnum", "")).ljust(columns[2])
+                port = fit_text(str(packet["decoded"].get("portnum", "")), columns[2])
                 parsed_payload = parse_protobuf(packet)
             else:
-                port = "NO KEY".ljust(columns[2])
+                port = fit_text("NO KEY", columns[2])
                 parsed_payload = "NO KEY"
 
             # Combine and truncate if necessary
             logString = f"{from_id} {to_id} {port} {parsed_payload}"
-            logString = logString[: width - 3]
+            logString = fit_text(logString, width - 3)
 
             # Add to the window
             packetlog_win.addstr(i + 2, 1, logString, get_color("log"))

contact/ui/nav_utils.py

@@ -1,13 +1,18 @@
 import curses
 import re
-from unicodedata import east_asian_width
+import unicodedata
+from typing import Any, Optional, List, Dict
+
+from wcwidth import wcwidth, wcswidth
 
 from contact.ui.colors import get_color
 from contact.utilities.i18n import t
 from contact.utilities.control_utils import transform_menu_path
-from typing import Any, Optional, List, Dict
 from contact.utilities.singleton import interface_state, ui_state
 
+ZWJ = "\u200d"
+KEYCAP = "\u20e3"
+
 
 def get_node_color(node_index: int, reverse: bool = False):
     node_num = ui_state.node_list[node_index]
@@ -327,12 +332,158 @@ def get_wrapped_help_text(
     return wrapped_help
 
 
+def _is_regional_indicator(char: str) -> bool:
+    codepoint = ord(char)
+    return 0x1F1E6 <= codepoint <= 0x1F1FF
+
+
+def _is_variation_selector(char: str) -> bool:
+    codepoint = ord(char)
+    return 0xFE00 <= codepoint <= 0xFE0F or 0xE0100 <= codepoint <= 0xE01EF
+
+
+def _is_emoji_modifier(char: str) -> bool:
+    codepoint = ord(char)
+    return 0x1F3FB <= codepoint <= 0x1F3FF
+
+
+def _is_display_modifier(char: str) -> bool:
+    return unicodedata.category(char) in {"Mn", "Mc", "Me"} or _is_emoji_modifier(char)
+
+
+def iter_display_units(text: str) -> List[str]:
+    """Split text into display units so emoji sequences stay intact."""
+    units: List[str] = []
+    index = 0
+
+    while index < len(text):
+        unit = text[index]
+        index += 1
+
+        if _is_regional_indicator(unit) and index < len(text) and _is_regional_indicator(text[index]):
+            unit += text[index]
+            index += 1
+
+        while index < len(text) and _is_display_modifier(text[index]):
+            unit += text[index]
+            index += 1
+
+        while index < len(text) and text[index] == ZWJ and index + 1 < len(text):
+            unit += text[index]
+            index += 1
+            unit += text[index]
+            index += 1
+
+            while index < len(text) and _is_display_modifier(text[index]):
+                unit += text[index]
+                index += 1
+
+        units.append(unit)
+
+    return units
+
+
+def sanitize_for_curses(text: str) -> str:
+    """Collapse complex emoji sequences to stable fallbacks before rendering."""
+    sanitized: List[str] = []
+
+    for unit in iter_display_units(text):
+        if ZWJ not in unit and KEYCAP not in unit and not any(
+            _is_variation_selector(char) or _is_emoji_modifier(char) for char in unit
+        ):
+            sanitized.append(unit)
+            continue
+
+        visible = [
+            char
+            for char in unit
+            if char != ZWJ and char != KEYCAP and not _is_variation_selector(char) and not _is_emoji_modifier(char)
+        ]
+
+        if KEYCAP in unit and visible:
+            sanitized.append(visible[0])
+        elif ZWJ in unit and visible:
+            sanitized.append(visible[0])
+        elif any(_is_emoji_modifier(char) for char in unit) and visible:
+            sanitized.append(visible[0])
+        elif visible:
+            sanitized.append("".join(visible))
+        else:
+            sanitized.append(unit)
+
+    return "".join(sanitized)
+
+
 def text_width(text: str) -> int:
-    return sum(2 if east_asian_width(c) in "FW" else 1 for c in text)
+    text = sanitize_for_curses(text)
+    width = wcswidth(text)
+    if width >= 0:
+        return width
+    return sum(max(wcwidth(char), 0) for char in text)
+
+
+def slice_text_to_width(text: str, max_width: int) -> str:
+    """Return the longest prefix that fits within max_width terminal cells."""
+    text = sanitize_for_curses(text)
+    if max_width <= 0:
+        return ""
+
+    chunk = ""
+    for unit in iter_display_units(text):
+        candidate = chunk + unit
+        if text_width(candidate) > max_width:
+            break
+        chunk = candidate
+
+    return chunk
+
+
+def fit_text(text: str, width: int, suffix: str = "") -> str:
+    """Trim and pad text so its terminal display width fits exactly."""
+    text = sanitize_for_curses(text)
+    suffix = sanitize_for_curses(suffix)
+    if width <= 0:
+        return ""
+
+    if text_width(text) > width:
+        suffix = slice_text_to_width(suffix, width)
+        available = max(0, width - text_width(suffix))
+        text = slice_text_to_width(text, available).rstrip() + suffix
+
+    padding = max(0, width - text_width(text))
+    return text + (" " * padding)
+
+
+def split_text_to_width(text: str, max_width: int) -> List[str]:
+    """Split text into chunks that each fit within max_width terminal cells."""
+    text = sanitize_for_curses(text)
+    if max_width <= 0:
+        return [""]
+
+    chunks: List[str] = []
+    chunk = ""
+
+    for unit in iter_display_units(text):
+        candidate = chunk + unit
+        if chunk and text_width(candidate) > max_width:
+            chunks.append(chunk)
+            chunk = unit
+        else:
+            chunk = candidate
+
+        if text_width(chunk) > max_width:
+            chunks.append(slice_text_to_width(chunk, max_width))
+            chunk = ""
+
+    if chunk:
+        chunks.append(chunk)
+
+    return chunks or [""]
 
 
 def wrap_text(text: str, wrap_width: int) -> List[str]:
     """Wraps text while preserving spaces and breaking long words."""
+    text = sanitize_for_curses(text)
 
     whitespace = "\t\n\x0b\x0c\r "
     whitespace_trans = dict.fromkeys(map(ord, whitespace), ord(" "))
@@ -346,24 +497,18 @@ def wrap_text(text: str, wrap_width: int) -> List[str]:
     wrap_width -= margin
 
     for word in words:
-        word_length = text_width(word)
+        word_chunks = split_text_to_width(word, wrap_width) if text_width(word) > wrap_width else [word]
 
-        if word_length > wrap_width:  # Break long words
-            if line_buffer:
+        for chunk in word_chunks:
+            chunk_length = text_width(chunk)
+
+            if line_length + chunk_length > wrap_width and chunk.strip():
                 wrapped_lines.append(line_buffer.strip())
                 line_buffer = ""
                 line_length = 0
-            for i in range(0, word_length, wrap_width):
-                wrapped_lines.append(word[i : i + wrap_width])
-            continue
 
-        if line_length + word_length > wrap_width and word.strip():
-            wrapped_lines.append(line_buffer.strip())
-            line_buffer = ""
-            line_length = 0
-
-        line_buffer += word
-        line_length += word_length
+            line_buffer += chunk
+            line_length += chunk_length
 
     if line_buffer:
         wrapped_lines.append(line_buffer.strip())

contact/utilities/emoji_utils.py

@@ -1,54 +0,0 @@
-"""Helpers for normalizing emoji sequences in width-sensitive message rendering."""
-
-# Strip zero-width and presentation modifiers that make terminal cell width inconsistent.
-EMOJI_MODIFIER_REPLACEMENTS = {
-    "\u200d": "",
-    "\u20e3": "",
-    "\ufe0e": "",
-    "\ufe0f": "",
-    "\U0001F3FB": "",
-    "\U0001F3FC": "",
-    "\U0001F3FD": "",
-    "\U0001F3FE": "",
-    "\U0001F3FF": "",
-}
-
-_EMOJI_MODIFIER_TRANSLATION = str.maketrans(EMOJI_MODIFIER_REPLACEMENTS)
-_REGIONAL_INDICATOR_START = ord("\U0001F1E6")
-_REGIONAL_INDICATOR_END = ord("\U0001F1FF")
-
-
-def _regional_indicator_to_letter(char: str) -> str:
-    return chr(ord("A") + ord(char) - _REGIONAL_INDICATOR_START)
-
-
-def _normalize_flag_emoji(text: str) -> str:
-    """Convert flag emoji built from regional indicators into ASCII country codes."""
-    normalized = []
-    index = 0
-
-    while index < len(text):
-        current = text[index]
-        current_ord = ord(current)
-
-        if _REGIONAL_INDICATOR_START <= current_ord <= _REGIONAL_INDICATOR_END and index + 1 < len(text):
-            next_char = text[index + 1]
-            next_ord = ord(next_char)
-            if _REGIONAL_INDICATOR_START <= next_ord <= _REGIONAL_INDICATOR_END:
-                normalized.append(_regional_indicator_to_letter(current))
-                normalized.append(_regional_indicator_to_letter(next_char))
-                index += 2
-                continue
-
-        normalized.append(current)
-        index += 1
-
-    return "".join(normalized)
-
-
-def normalize_message_text(text: str) -> str:
-    """Strip modifiers and rewrite flag emoji into stable terminal-friendly text."""
-    if not text:
-        return text
-
-    return _normalize_flag_emoji(text.translate(_EMOJI_MODIFIER_TRANSLATION))

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "contact"
-version = "1.4.23"
+version = "1.4.20"
 description = "This Python curses client for Meshtastic is a terminal-based client designed to manage device settings, enable mesh chat communication, and handle configuration backups and restores."
 authors = [
     {name = "Ben Lipsey",email = "ben@pdxlocations.com"}
@@ -9,7 +9,8 @@ license = "GPL-3.0-only"
 readme = "README.md"
 requires-python = ">=3.9,<3.15"
 dependencies = [
-    "meshtastic (>=2.7.5,<3.0.0)"
+    "meshtastic (>=2.7.5,<3.0.0)",
+    "wcwidth (>=0.2.13,<1.0.0)"
 ]
 
 [project.urls]

requirements.txt

@@ -1 +1,2 @@
 meshtastic
+wcwidth>=0.2.13,<1.0.0