Fix path message truncation for emoji (double-width) characters

Use UTF-8 byte length instead of character count or display width for message
truncation. This fixes an issue where emoji characters like 😀 (4 UTF-8 bytes,
1 character, 2 display units) caused messages to exceed the RF packet size.

Changes:
- Replace _count_display_width with byte-based length calculation
- Add _count_byte_length and _truncate_to_byte_length helper methods
- Update get_max_message_length to use 127 bytes (channel limit) not 150
- Add type hints to new methods
This commit is contained in:
Chris Close
2026-03-29 13:33:58 -07:00
committed by agessaman
parent 83b23385a7
commit e2725686ef
5 changed files with 33 additions and 89 deletions
@@ -1143,35 +1143,8 @@ class GlobalWxCommand(BaseCommand):
return self.translate('commands.gwx.multiday_error', num_days=num_days)
def _count_display_width(self, text: str) -> int:
"""Count display width of text, accounting for emojis which may take 2 display units.
Args:
text: Text to measure.
Returns:
int: Estimated display width.
"""
import re
# Count regular characters
width = len(text)
# Emojis typically take 2 display units in terminals/clients
# Count emoji characters (basic emoji pattern)
emoji_pattern = re.compile(
"["
"\U0001F600-\U0001F64F" # emoticons
"\U0001F300-\U0001F5FF" # symbols & pictographs
"\U0001F680-\U0001F6FF" # transport & map symbols
"\U0001F1E0-\U0001F1FF" # flags
"\U00002702-\U000027B0" # dingbats
"\U000024C2-\U0001F251" # enclosed characters
"]+",
flags=re.UNICODE
)
emoji_matches = emoji_pattern.findall(text)
# Each emoji sequence adds 1 extra width unit (since len() already counts it as 1)
# So we add 1 for each emoji sequence to account for display width
width += len(emoji_matches)
return width
"""Count UTF-8 byte length of text. Matches RF packet byte limit from get_max_message_length()."""
return len(text.encode('utf-8'))
async def _send_multiday_forecast(self, message: MeshMessage, forecast_text: str) -> None:
"""Send multi-day forecast response, splitting into multiple messages if needed.
+5 -4
View File
@@ -584,12 +584,13 @@ class BaseCommand(ABC):
if not username:
username = self.bot.config.get('Bot', 'bot_name', fallback='Bot')
# Calculate max length: 150 - username_length - 2 (for ": ")
max_length = 150 - len(str(username)) - 2
# 127 bytes are available for channel messages
# Calculate max length: 127 - username_length - 2 (for ": ")
max_length = 127 - len(str(username).encode('utf-8')) - 2
# Ensure we don't return a negative or unreasonably small value
# Minimum of 130 characters to ensure some functionality
return max(130, max_length)
# Minimum of 110 bytes to ensure some functionality
return max(110, max_length)
def check_cooldown(self, user_id: Optional[str] = None) -> tuple[bool, float]:
"""Check if user is on cooldown.
+24 -13
View File
@@ -1629,12 +1629,10 @@ class PathCommand(BaseCommand):
# Geographic or graph-based selection
name = info.get('name', self.translate('commands.path.unknown_name'))
confidence = info.get('confidence', 0.0)
info.get('graph_guess', False)
# Truncate name if too long
truncation = self.translate('commands.path.truncation')
if len(name) > 20:
name = name[:17] + truncation
name = self._truncate_to_byte_length(name, 20, truncation)
# Add confidence indicator
if confidence >= 0.9:
@@ -1650,20 +1648,15 @@ class PathCommand(BaseCommand):
# Single repeater found
name = info.get('name', self.translate('commands.path.unknown_name'))
# Truncate name if too long
truncation = self.translate('commands.path.truncation')
if len(name) > 27:
name = name[:24] + truncation
name = self._truncate_to_byte_length(name, 27, truncation)
line = self.translate('commands.path.node_format', node_id=node_id, name=name)
else:
# Unknown repeater
line = self.translate('commands.path.node_unknown', node_id=node_id)
# Ensure line fits within 130 character limit
if len(line) > 130:
truncation = self.translate('commands.path.truncation')
line = line[:127] + truncation
line = self._truncate_to_byte_length(line, 130)
lines.append(line)
@@ -1680,7 +1673,7 @@ class PathCommand(BaseCommand):
# Get dynamic max message length based on message type and bot username
max_length = self.get_max_message_length(message)
if len(response) <= max_length:
if self._count_byte_length(response) <= max_length:
# Single message is fine
await self.send_response(message, response)
else:
@@ -1691,8 +1684,8 @@ class PathCommand(BaseCommand):
message_count = 0
for i, line in enumerate(lines):
# Check if adding this line would exceed max_length characters
if len(current_message) + len(line) + 1 > max_length: # +1 for newline
# Check if adding this line would exceed max_length display width
if self._count_byte_length(current_message) + self._count_byte_length(line) + 1 > max_length: # +1 for newline
# Send current message and start new one
if current_message:
# Add ellipsis on new line to end of continued message (if not the last message)
@@ -1766,6 +1759,24 @@ class PathCommand(BaseCommand):
self.logger.error(f"Error extracting path from current message: {e}")
return self.translate('commands.path.error_extracting', error=str(e))
def _count_byte_length(self, text: str) -> int:
"""Count UTF-8 byte length of text. This matches the RF packet byte limit."""
return len(text.encode('utf-8'))
def _truncate_to_byte_length(self, text: str, max_bytes: int, ellipsis: str = "...") -> str:
"""Truncate text to fit within max UTF-8 byte length, never splitting multi-byte chars."""
text_bytes: bytes = text.encode('utf-8')
if len(text_bytes) <= max_bytes:
return text
ellipsis_bytes: bytes = ellipsis.encode('utf-8')
available: int = max_bytes - len(ellipsis_bytes)
if available <= 0:
return ellipsis
truncated: str = text_bytes[:available].decode('utf-8', errors='ignore')
return truncated + ellipsis
def get_help(self) -> str:
"""Get help text for the path command"""
return self.translate('commands.path.help')
-22
View File
@@ -26,10 +26,6 @@ class RepeaterCommand(BaseCommand):
category = "management"
requires_internet = True # Requires internet access for geocoding (Nominatim)
# LoRa message size limit (conservative to avoid overload)
# DM limit is 150 chars, public channel is 237 chars
MAX_LORA_MESSAGE_SIZE = 140 # characters, leaves buffer for protocol overhead
def __init__(self, bot):
super().__init__(bot)
self.repeater_enabled = self.get_config_value('Repeater_Command', 'enabled', fallback=True, value_type='bool')
@@ -47,24 +43,6 @@ class RepeaterCommand(BaseCommand):
return False
return super().can_execute(message)
def _truncate_for_lora(self, message: str, max_size: int = None) -> str:
"""Truncate message to fit within LoRa size limits.
Args:
message: The message to truncate
max_size: Maximum size (defaults to MAX_LORA_MESSAGE_SIZE)
Returns:
Truncated message with indicator if truncated
"""
max_size = max_size or self.MAX_LORA_MESSAGE_SIZE
if len(message) <= max_size:
return message
truncate_suffix = "...(use web viewer)"
available_size = max_size - len(truncate_suffix)
return message[:available_size] + truncate_suffix
def _get_deprecation_warning(self, web_viewer_url: str = None) -> str:
"""Get deprecation warning message for commands replaced by web viewer.
+2 -21
View File
@@ -1761,27 +1761,8 @@ class WxCommand(BaseCommand):
return result
def _count_display_width(self, text: str) -> int:
"""Count display width of text, accounting for emojis which may take 2 display units"""
# Count regular characters
width = len(text)
# Emojis typically take 2 display units in terminals/clients
# Count emoji characters (basic emoji pattern)
emoji_pattern = re.compile(
"["
"\U0001F600-\U0001F64F" # emoticons
"\U0001F300-\U0001F5FF" # symbols & pictographs
"\U0001F680-\U0001F6FF" # transport & map symbols
"\U0001F1E0-\U0001F1FF" # flags
"\U00002702-\U000027B0" # dingbats
"\U000024C2-\U0001F251" # enclosed characters
"]+",
flags=re.UNICODE
)
emoji_matches = emoji_pattern.findall(text)
# Each emoji sequence adds 1 extra width unit (since len() already counts it as 1)
# So we add 1 for each emoji sequence to account for display width
width += len(emoji_matches)
return width
"""Count UTF-8 byte length of text. Matches RF packet byte limit from get_max_message_length()."""
return len(text.encode('utf-8'))
async def _send_multiday_forecast(self, message: MeshMessage, forecast_text: str):
"""Send multi-day forecast response, splitting into multiple messages if needed"""