adam/meshcore-gui
1
0 Fork 0
mirror of https://github.com/pe1hvh/meshcore-gui.git synced 2026-11-04 19:42:33 +01:00
Code Issues 2 Packages Projects Releases Wiki Activity
Files
fd3d0d2f3515ebc31ddfa1fa079348101ac4403d
meshcore-gui/meshcore_gui/core/models.py
T
pe1hvh af74caf766 fix(route): default reply channel to message channel, add date to timestamps 
WHAT
- route_page: _default_reply_channel() derives the reply select default
  from msg.channel instead of the first entry in the channel list
- models: new Message.date field, now_date() and display_timestamp();
  format_line() and incoming()/outgoing() use them
- message_archive: persist "date" alongside "time" and "timestamp_utc"
- room_server_panel + route_page info header use display_timestamp()
- VERSION 1.22.2 -> 1.22.3

WHY
The route page exists to show one specific message, so that message —
not the ordering of the channel list — is the correct source for the
reply default. The date could not be fixed at the render site because
Message had no date at all: now_timestamp() yields HH:MM:SS and the
date lived only as timestamp_utc inside the archive layer, unread by
from_dict(). A first-class field plus a single formatting helper fixes
the cause and keeps the format in one place.

NOTES
No breaking changes. The archive schema gains one additive field;
pre-1.22.3 entries have their date derived from timestamp_utc
(converted to local time) and fall back to time-only when absent.
Message.incoming() gained an optional keyword-only date parameter.
Message display lines are ~11 characters wider.
2026-08-16 08:32:24 +02:00

414 lines
14 KiB
Python
Raw Blame History

"""
Domain model for MeshCore GUI.
Typed dataclasses that replace untyped Dict objects throughout the
codebase. Each class represents a core domain concept. All classes
are immutable-friendly (frozen is not used because SharedData mutates
collections, but fields are not reassigned after construction).
Migration note
~~~~~~~~~~~~~~
``SharedData.get_snapshot()`` still returns a plain dict for backward
compatibility with the NiceGUI timer loop. Inside that dict, however,
``messages`` and ``rx_log`` are now lists of dataclass instances.
UI code can access attributes directly (``msg.sender``) or fall back
to ``dataclasses.asdict(msg)`` if a plain dict is needed.
"""
from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, List, Optional
# ---------------------------------------------------------------------------
# Message
# ---------------------------------------------------------------------------
@dataclass
class Message:
"""A channel message or direct message (DM).
Attributes:
time: Formatted local time (HH:MM:SS).
sender: Display name of the sender.
text: Message body.
channel: Channel index, or ``None`` for a DM.
direction: ``'in'`` for received, ``'out'`` for sent.
snr: Signal-to-noise ratio (dB), if available.
path_len: Hop count from the LoRa frame header.
sender_pubkey: Full public key of the sender (hex string).
path_hashes: List of 2-char hex strings, one per repeater.
path_names: List of resolved display names for each path hash,
captured at receive time so the archive is self-contained.
message_hash: Deterministic packet identifier (hex string).
channel_name: Human-readable channel name (resolved at add time).
date: Formatted local date (YYYY-MM-DD). Added in 1.22.3;
empty for messages archived by older versions, in which
case :meth:`from_dict` derives it from ``timestamp_utc``.
Declared last so positional construction stays
backward-compatible.
"""
time: str
sender: str
text: str
channel: Optional[int]
direction: str
snr: Optional[float] = None
path_len: int = 0
sender_pubkey: str = ""
path_hashes: List[str] = field(default_factory=list)
path_names: List[str] = field(default_factory=list)
message_hash: str = ""
channel_name: str = ""
date: str = ""
@staticmethod
def from_dict(d: dict) -> "Message":
"""Create a Message from an archive dictionary.
Args:
d: Dictionary as stored by MessageArchive.
Returns:
Message dataclass instance.
"""
return Message(
time=d.get("time", ""),
sender=d.get("sender", ""),
text=d.get("text", ""),
channel=d.get("channel"),
direction=d.get("direction", "in"),
snr=d.get("snr"),
path_len=d.get("path_len", 0),
sender_pubkey=d.get("sender_pubkey", ""),
path_hashes=d.get("path_hashes", []),
path_names=d.get("path_names", []),
message_hash=d.get("message_hash", ""),
channel_name=d.get("channel_name", ""),
date=d.get("date") or Message._date_from_timestamp(
d.get("timestamp_utc")
),
)
# -- Timestamp helpers -----------------------------------------------
@staticmethod
def now_timestamp() -> str:
"""Current local time formatted as ``HH:MM:SS``."""
return datetime.now().strftime('%H:%M:%S')
@staticmethod
def now_date() -> str:
"""Current local date formatted as ``YYYY-MM-DD``."""
return datetime.now().strftime('%Y-%m-%d')
@staticmethod
def _date_from_timestamp(timestamp_utc: Optional[str]) -> str:
"""Derive a local ``YYYY-MM-DD`` date from an ISO UTC timestamp.
Backward-compatibility helper for archive entries written before
the ``date`` field existed. The archive stores ``timestamp_utc``
for every message, so the date can be recovered; it is converted
to local time first so it matches the local ``time`` field around
midnight.
Args:
timestamp_utc: ISO-8601 timestamp string, or ``None``.
Returns:
Local date as ``YYYY-MM-DD``, or an empty string when the
timestamp is missing or unparseable.
"""
if not timestamp_utc:
return ""
try:
parsed = datetime.fromisoformat(timestamp_utc)
except (ValueError, TypeError):
return ""
if parsed.tzinfo is not None:
parsed = parsed.astimezone()
return parsed.strftime('%Y-%m-%d')
def display_timestamp(self) -> str:
"""Return the display timestamp for this message.
Produces ``YYYY-MM-DD HH:MM:SS`` when the date is known, and
falls back to the bare ``HH:MM:SS`` time for legacy messages
that carry no date information.
Returns:
Formatted timestamp string.
"""
if self.date:
return f"{self.date} {self.time}"
return self.time
# -- Factory methods -------------------------------------------------
@classmethod
def incoming(
cls,
sender: str,
text: str,
channel: Optional[int],
*,
time: str = "",
date: str = "",
snr: Optional[float] = None,
path_len: int = 0,
sender_pubkey: str = "",
path_hashes: Optional[List[str]] = None,
path_names: Optional[List[str]] = None,
message_hash: str = "",
) -> "Message":
"""Create an incoming message with auto-generated timestamp.
Args:
sender: Display name of the sender.
text: Message body.
channel: Channel index, or ``None`` for a DM.
time: Optional pre-generated time (default: now).
date: Optional pre-generated date (default: today).
snr: Signal-to-noise ratio (dB).
path_len: Hop count from the LoRa frame header.
sender_pubkey: Full public key of the sender (hex string).
path_hashes: List of 2-char hex strings per repeater.
path_names: Resolved display names for each path hash.
message_hash: Deterministic packet identifier (hex string).
"""
return cls(
time=time or cls.now_timestamp(),
date=date or cls.now_date(),
sender=sender,
text=text,
channel=channel,
direction='in',
snr=snr,
path_len=path_len,
sender_pubkey=sender_pubkey,
path_hashes=path_hashes or [],
path_names=path_names or [],
message_hash=message_hash,
)
@classmethod
def outgoing(
cls,
text: str,
channel: Optional[int],
*,
sender_pubkey: str = "",
) -> "Message":
"""Create an outgoing message (sender ``'Me'``, auto-timestamp).
Args:
text: Message body.
channel: Channel index, or ``None`` for a DM.
sender_pubkey: Recipient public key (hex string).
"""
return cls(
time=cls.now_timestamp(),
date=cls.now_date(),
sender='Me',
text=text,
channel=channel,
direction='out',
sender_pubkey=sender_pubkey,
)
# -- Display formatting ----------------------------------------------
def format_line(
self,
channel_names: Optional[Dict[int, str]] = None,
show_channel: bool = True,
sender_prefix: str = '',
) -> str:
"""Format as a single display line for the messages panel.
The line opens with :meth:`display_timestamp`, which prefixes the
local date when it is known (legacy archive entries without a
date fall back to time only), e.g.::
2026-08-16 12:34:56 ← [Public] [2h✓] PE1ABC: Hello mesh!
When *show_channel* is ``False`` the ``[channel]`` / ``[DM]``
tag is omitted (useful when the panel header already indicates
the active channel).
Args:
channel_names: Optional ``{channel_idx: name}`` lookup.
Falls back to ``self.channel_name``, then ``'ch<idx>'``.
show_channel: Include ``[channel]`` / ``[DM]`` prefix.
Defaults to ``True`` for backward compatibility.
sender_prefix: Optional prefix placed before the sender name,
e.g. a node-type icon from the map/contact view.
Returns:
Formatted single-line string.
"""
direction = '→' if self.direction == 'out' else '←'
ch_label = ''
if show_channel:
if self.channel is not None:
if channel_names and self.channel in channel_names:
ch_name = channel_names[self.channel]
elif self.channel_name:
ch_name = self.channel_name
else:
ch_name = f'ch{self.channel}'
ch_label = f'[{ch_name}] '
else:
ch_label = '[DM] '
if self.direction == 'in' and self.path_len > 0:
hop_tag = f'[{self.path_len}h{"✓" if self.path_hashes else ""}] '
else:
hop_tag = ''
sender_display = f"{sender_prefix}{self.sender}" if self.sender else ''
stamp = self.display_timestamp()
if self.sender:
return f"{stamp} {direction} {ch_label}{hop_tag}{sender_display}: {self.text}"
return f"{stamp} {direction} {ch_label}{hop_tag}{self.text}"
# ---------------------------------------------------------------------------
# Contact
# ---------------------------------------------------------------------------
@dataclass
class Contact:
"""A known mesh network node.
Attributes:
pubkey: Full public key (hex string).
adv_name: Advertised display name.
type: Node type (0=unknown, 1=CLI, 2=REP, 3=ROOM).
adv_lat: Advertised latitude (0.0 if unknown).
adv_lon: Advertised longitude (0.0 if unknown).
out_path: Hex string of stored route (2 hex chars per hop).
out_path_len: Number of hops in ``out_path``.
"""
pubkey: str
adv_name: str = ""
type: int = 0
adv_lat: float = 0.0
adv_lon: float = 0.0
out_path: str = ""
out_path_len: int = 0
@staticmethod
def from_dict(pubkey: str, d: dict) -> "Contact":
"""Create a Contact from a meshcore contacts dict entry."""
return Contact(
pubkey=pubkey,
adv_name=d.get("adv_name", ""),
type=d.get("type", 0),
adv_lat=d.get("adv_lat", 0.0),
adv_lon=d.get("adv_lon", 0.0),
out_path=d.get("out_path", ""),
out_path_len=d.get("out_path_len", 0),
)
# ---------------------------------------------------------------------------
# DeviceInfo
# ---------------------------------------------------------------------------
@dataclass
class DeviceInfo:
"""Radio device identification and configuration.
Attributes:
name: Device display name.
public_key: Device public key (hex string).
radio_freq: Radio frequency in MHz.
radio_sf: LoRa spreading factor.
radio_bw: Bandwidth in kHz.
tx_power: Transmit power in dBm.
adv_lat: Advertised latitude.
adv_lon: Advertised longitude.
firmware_version: Firmware version string.
"""
name: str = ""
public_key: str = ""
radio_freq: float = 0.0
radio_sf: int = 0
radio_bw: float = 0.0
tx_power: int = 0
adv_lat: float = 0.0
adv_lon: float = 0.0
firmware_version: str = ""
# ---------------------------------------------------------------------------
# RxLogEntry
# ---------------------------------------------------------------------------
@dataclass
class RxLogEntry:
"""A single RX log entry from the radio.
Attributes:
time: Formatted timestamp (HH:MM:SS).
snr: Signal-to-noise ratio (dB).
rssi: Received signal strength (dBm).
payload_type: Packet type identifier.
hops: Number of hops (path_len from frame header).
message_hash: Optional message hash for correlation with messages.
path_hashes: 2-char hex repeater hashes from decoded packet.
path_names: Resolved display names for each path hash.
"""
time: str
snr: float = 0.0
rssi: float = 0.0
payload_type: str = "?"
hops: int = 0
message_hash: str = ""
path_hashes: List[str] = field(default_factory=list)
path_names: List[str] = field(default_factory=list)
sender: str = ""
receiver: str = ""
# ── Fase 1 Observer fields (raw packet metadata) ──
raw_payload: str = "" # Raw hex packet data
packet_len: int = 0 # Total packet length (bytes)
payload_len: int = 0 # Payload length (bytes)
route_type: str = "" # "F" (flood) or "D" (direct)
packet_type_num: int = -1 # Numeric packet type (0-15)
# ---------------------------------------------------------------------------
# RouteNode
# ---------------------------------------------------------------------------
@dataclass
class RouteNode:
"""A node in a message route (sender, repeater or receiver).
Attributes:
name: Display name (or ``'-'`` if unknown).
lat: Latitude (0.0 if unknown).
lon: Longitude (0.0 if unknown).
type: Node type (0=unknown, 1=CLI, 2=REP, 3=ROOM).
pubkey: Public key or 2-char hash (hex string).
"""
name: str
lat: float = 0.0
lon: float = 0.0
type: int = 0
pubkey: str = ""
@property
def has_location(self) -> bool:
"""True if the node has GPS coordinates."""
return self.lat != 0 or self.lon != 0
Reference in new issue View Git Blame Copy Permalink