eric
310c0244df
- Resolve the Hermes gateway API lazily via get_adapter_class() so the adapter
always inherits the real gateway.platforms.base.BasePlatformAdapter instead
of the ImportError fallback stub (root cause of the AttributeError:
'set_message_handler' crash at gateway startup).
- Pass Platform('meshtastic') enum to super().__init__ and align
connect(*, is_reconnect=False) -> bool with the gateway contract.
- Split pure formatting utils into formatting.py; remove the stub fallback.
- Add tests/contract (L0-L3): real-base binding, ABC instantiation, gateway
run.py wiring replay, signature and event-shape conformance, run against
hermes-agent pinned to the deployed image source (v2026.8.31).
- Add scripts/contract-test.sh and a contract-test CI job; release publish
now requires contract tests to pass.
- Untrack stray __pycache__ artifacts.
62 lines
2.0 KiB
Python
62 lines
2.0 KiB
Python
"""Pure formatting utilities for LoRa-friendly text.
|
|
|
|
These helpers have NO dependency on the Hermes gateway package so they can be
|
|
imported and unit-tested in isolation (CI runs them without hermes-agent).
|
|
"""
|
|
|
|
import re
|
|
|
|
MAX_LORA_MESSAGE_LENGTH = 180 # Characters per packet to prevent LoRa airtime congestion
|
|
|
|
_MARKDOWN_RULES = (
|
|
(r"\*\*(.+?)\*\*", r"\1"), # bold
|
|
(r"__(.+?)__", r"\1"),
|
|
(r"\*(.+?)\*", r"\1"), # italic
|
|
(r"(?<!\w)_(.+?)_(?!\w)", r"\1"),
|
|
(r"`(.+?)`", r"\1"), # inline code
|
|
(r"```[\w]*\n?", ""), # code fences
|
|
(r"!\[([^\]]*)\]\(([^)]+)\)", r"\1"),
|
|
(r"\[([^\]]+)\]\(([^)]+)\)", r"\1 (\2)"),
|
|
(r"^#+\s*", ""), # headings
|
|
(r"^\s*[-*+]\s+", "- "), # normalize lists
|
|
)
|
|
|
|
|
|
def strip_markdown(text: str) -> str:
|
|
"""Convert markdown formatting into plain readable text suitable for radio."""
|
|
if not text:
|
|
return ""
|
|
result = text
|
|
for pattern, replacement in _MARKDOWN_RULES:
|
|
result = re.sub(pattern, replacement, result, flags=re.MULTILINE)
|
|
# Collapse multiple blank lines
|
|
result = re.sub(r"\n{3,}", "\n\n", result)
|
|
return result.strip()
|
|
|
|
|
|
def chunk_text(text: str, limit: int = MAX_LORA_MESSAGE_LENGTH) -> list:
|
|
"""Split text into smaller chunks for LoRa transmission, breaking at whitespace/newlines."""
|
|
if not text:
|
|
return []
|
|
if len(text) <= limit:
|
|
return [text]
|
|
|
|
chunks = []
|
|
remaining = text
|
|
while remaining:
|
|
if len(remaining) <= limit:
|
|
chunks.append(remaining.strip())
|
|
break
|
|
split_at = remaining.rfind(" ", 0, limit)
|
|
if split_at <= 0 or split_at < (limit // 3):
|
|
newline_at = remaining.rfind("\n", 0, limit)
|
|
if newline_at > 0:
|
|
split_at = newline_at
|
|
else:
|
|
split_at = limit
|
|
chunk = remaining[:split_at].strip()
|
|
if chunk:
|
|
chunks.append(chunk)
|
|
remaining = remaining[split_at:].lstrip()
|
|
return chunks
|