mirror of
https://github.com/l5yth/potato-mesh.git
synced 2026-07-05 09:21:42 +02:00
l5y
d33ebd8f4c
* data: provide frequency and modem preset for meshcore * data: provide frequency and modem preset for meshcore * ingestor: address review comments * fix: remove duplicate _mark_packet_seen entry from handlers __all__ * ci: install meshcore in Python workflow protocols/meshcore.py now imports meshcore at module level (required to fix a self-referential import failure after the providers/ → protocols/ rename). test_provider_unit.py imports that module unconditionally, so meshcore must be present in the test environment. * data: run black
58 lines
1.8 KiB
Python
58 lines
1.8 KiB
Python
# Copyright © 2025-26 l5yth & contributors
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
"""MeshProtocol interface for ingestion sources.
|
|
|
|
This module defines the seam so future protocols (MeshCore, Reticulum, ...) can
|
|
be added without changing the web app ingest contract.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Iterable
|
|
from typing import Protocol, runtime_checkable
|
|
|
|
|
|
@runtime_checkable
|
|
class MeshProtocol(Protocol):
|
|
"""Abstract mesh protocol source."""
|
|
|
|
name: str
|
|
|
|
def subscribe(self) -> list[str]:
|
|
"""Subscribe to any async receive callbacks and return topic names."""
|
|
|
|
def connect(
|
|
self, *, active_candidate: str | None
|
|
) -> tuple[object, str | None, str | None]:
|
|
"""Create an interface connection.
|
|
|
|
Returns:
|
|
(iface, resolved_target, next_active_candidate)
|
|
"""
|
|
|
|
def extract_host_node_id(self, iface: object) -> str | None:
|
|
"""Best-effort extraction of the connected host node id."""
|
|
|
|
def node_snapshot_items(self, iface: object) -> Iterable[tuple[str, object]]:
|
|
"""Return iterable of (node_id, node_obj) for initial snapshot."""
|
|
|
|
|
|
__all__ = [
|
|
"MeshProtocol",
|
|
]
|
|
|
|
# Backwards-compatibility alias — import Provider from here during transition.
|
|
Provider = MeshProtocol
|