Add node tag import functionality to collector

- Add tag_import.py module with JSON file parsing and database upsert
- Convert collector CLI to group with subcommands for extensibility
- Add 'import-tags' command to import tags from JSON file
- Update docker-compose.yml.example with separated data directories:
  - data/collector for tags.json
  - data/web for members.json
- Add import-tags Docker service for easy containerized imports
- Add example data files in example/data/collector and example/data/web
- Add comprehensive test coverage (20 tests) for tag import
This commit is contained in:
Claude committed 2025-12-03 18:16:03 +00:00
1 parent df6f6b779c
commit fe1fd69904
7 files changed
+730 -5

No files matched your search

+2 -1
View File
@@ -1,5 +1,6 @@
# Project # Project
data/ /data/
!example/data/
# Byte-compiled / optimized / DLL files # Byte-compiled / optimized / DLL files
__pycache__/ __pycache__/
+37 -2
View File
@@ -14,6 +14,20 @@
# - web: Web dashboard # - web: Web dashboard
# - mock: All components with mock device (for testing) # - mock: All components with mock device (for testing)
# - all: All production components (requires real device) # - all: All production components (requires real device)
# - migrate: Run database migrations
# - import-tags: Import node tags from JSON file
#
# Data Directory Structure:
# ./data/
# ├── collector/
# │ └── tags.json # Node tags for import
# └── web/
# └── members.json # Network members list
#
# Example data files are provided in ./example/data/
#
# To import tags:
# docker compose --profile import-tags run --rm import-tags
services: services:
# ========================================================================== # ==========================================================================
@@ -162,6 +176,8 @@ services:
condition: service_healthy condition: service_healthy
volumes: volumes:
- meshcore_data:/data - meshcore_data:/data
# Mount collector data directory for tags.json (optional)
- ${COLLECTOR_DATA_PATH:-./data/collector}:/app/collector-data:ro
environment: environment:
- LOG_LEVEL=${LOG_LEVEL:-INFO} - LOG_LEVEL=${LOG_LEVEL:-INFO}
- MQTT_HOST=${MQTT_HOST:-mqtt} - MQTT_HOST=${MQTT_HOST:-mqtt}
@@ -263,9 +279,10 @@ services:
- NETWORK_RADIO_CONFIG=${NETWORK_RADIO_CONFIG:-} - NETWORK_RADIO_CONFIG=${NETWORK_RADIO_CONFIG:-}
- NETWORK_CONTACT_EMAIL=${NETWORK_CONTACT_EMAIL:-} - NETWORK_CONTACT_EMAIL=${NETWORK_CONTACT_EMAIL:-}
- NETWORK_CONTACT_DISCORD=${NETWORK_CONTACT_DISCORD:-} - NETWORK_CONTACT_DISCORD=${NETWORK_CONTACT_DISCORD:-}
- MEMBERS_FILE=${MEMBERS_FILE:-} - MEMBERS_FILE=${MEMBERS_FILE:-/app/web-data/members.json}
volumes: volumes:
- ${MEMBERS_FILE_PATH:-./example/data/members.json}:/app/members.json:ro # Mount web data directory for members.json (optional)
- ${WEB_DATA_PATH:-./data/web}:/app/web-data:ro
command: ["web"] command: ["web"]
healthcheck: healthcheck:
test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8080/health')"] test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8080/health')"]
@@ -290,6 +307,24 @@ services:
- DATABASE_URL=sqlite:////data/meshcore.db - DATABASE_URL=sqlite:////data/meshcore.db
command: ["db", "upgrade"] command: ["db", "upgrade"]
# ==========================================================================
# Import Tags - Import node tags from JSON file
# ==========================================================================
import-tags:
build:
context: .
dockerfile: Dockerfile
container_name: meshcore-import-tags
profiles:
- import-tags
volumes:
- meshcore_data:/data
- ${COLLECTOR_DATA_PATH:-./data/collector}:/app/collector-data:ro
environment:
- DATABASE_URL=sqlite:////data/meshcore.db
- LOG_LEVEL=${LOG_LEVEL:-INFO}
command: ["collector", "import-tags", "/app/collector-data/tags.json"]
# ========================================================================== # ==========================================================================
# Volumes # Volumes
# ========================================================================== # ==========================================================================
+28
View File
@@ -0,0 +1,28 @@
{
"tags": [
{
"public_key": "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
"key": "location",
"value": "San Francisco, CA",
"value_type": "string"
},
{
"public_key": "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
"key": "role",
"value": "gateway",
"value_type": "string"
},
{
"public_key": "fedcba9876543210fedcba9876543210fedcba9876543210fedcba9876543210",
"key": "location",
"value": "Oakland, CA",
"value_type": "string"
},
{
"public_key": "fedcba9876543210fedcba9876543210fedcba9876543210fedcba9876543210",
"key": "altitude",
"value": "150",
"value_type": "number"
}
]
}
+10
View File
@@ -0,0 +1,10 @@
{
"members": [
{
"name": "Example Member",
"callsign": "N0CALL",
"role": "Network Operator",
"description": "Example member entry"
}
]
}
+130 -2
View File
@@ -5,7 +5,8 @@ import click
from meshcore_hub.common.logging import configure_logging from meshcore_hub.common.logging import configure_logging
@click.command("collector") @click.group(invoke_without_command=True)
@click.pass_context
@click.option( @click.option(
"--mqtt-host", "--mqtt-host",
type=str, type=str,
@@ -56,6 +57,7 @@ from meshcore_hub.common.logging import configure_logging
help="Log level", help="Log level",
) )
def collector( def collector(
ctx: click.Context,
mqtt_host: str, mqtt_host: str,
mqtt_port: int, mqtt_port: int,
mqtt_username: str | None, mqtt_username: str | None,
@@ -64,7 +66,7 @@ def collector(
database_url: str, database_url: str,
log_level: str, log_level: str,
) -> None: ) -> None:
"""Run the collector component. """Collector component for storing MeshCore events.
The collector subscribes to MQTT broker and stores The collector subscribes to MQTT broker and stores
MeshCore events in the database for later retrieval. MeshCore events in the database for later retrieval.
@@ -76,6 +78,41 @@ def collector(
- Telemetry responses - Telemetry responses
- Informational events (battery, status, etc.) - Informational events (battery, status, etc.)
When invoked without a subcommand, runs the collector service.
"""
ctx.ensure_object(dict)
ctx.obj["mqtt_host"] = mqtt_host
ctx.obj["mqtt_port"] = mqtt_port
ctx.obj["mqtt_username"] = mqtt_username
ctx.obj["mqtt_password"] = mqtt_password
ctx.obj["prefix"] = prefix
ctx.obj["database_url"] = database_url
ctx.obj["log_level"] = log_level
# If no subcommand, run the collector service
if ctx.invoked_subcommand is None:
_run_collector_service(
mqtt_host=mqtt_host,
mqtt_port=mqtt_port,
mqtt_username=mqtt_username,
mqtt_password=mqtt_password,
prefix=prefix,
database_url=database_url,
log_level=log_level,
)
def _run_collector_service(
mqtt_host: str,
mqtt_port: int,
mqtt_username: str | None,
mqtt_password: str | None,
prefix: str,
database_url: str,
log_level: str,
) -> None:
"""Run the collector service.
Webhooks can be configured via environment variables: Webhooks can be configured via environment variables:
- WEBHOOK_ADVERTISEMENT_URL: Webhook for advertisement events - WEBHOOK_ADVERTISEMENT_URL: Webhook for advertisement events
- WEBHOOK_MESSAGE_URL: Webhook for all message events - WEBHOOK_MESSAGE_URL: Webhook for all message events
@@ -117,3 +154,94 @@ def collector(
database_url=database_url, database_url=database_url,
webhook_dispatcher=webhook_dispatcher, webhook_dispatcher=webhook_dispatcher,
) )
@collector.command("run")
@click.pass_context
def run_cmd(ctx: click.Context) -> None:
"""Run the collector service.
This is the default behavior when no subcommand is specified.
"""
_run_collector_service(
mqtt_host=ctx.obj["mqtt_host"],
mqtt_port=ctx.obj["mqtt_port"],
mqtt_username=ctx.obj["mqtt_username"],
mqtt_password=ctx.obj["mqtt_password"],
prefix=ctx.obj["prefix"],
database_url=ctx.obj["database_url"],
log_level=ctx.obj["log_level"],
)
@collector.command("import-tags")
@click.argument("file", type=click.Path(exists=True))
@click.option(
"--no-create-nodes",
is_flag=True,
default=False,
help="Skip tags for nodes that don't exist (default: create nodes)",
)
@click.pass_context
def import_tags_cmd(
ctx: click.Context,
file: str,
no_create_nodes: bool,
) -> None:
"""Import node tags from a JSON file.
Reads a JSON file containing tag definitions and upserts them
into the database. Existing tags are updated, new tags are created.
FILE is the path to the JSON file containing tags.
Expected JSON format:
\b
{
"tags": [
{
"public_key": "64-char-hex-string",
"key": "tag-name",
"value": "tag-value",
"value_type": "string"
}
]
}
Supported value_type: string, number, boolean, coordinate
"""
configure_logging(level=ctx.obj["log_level"])
click.echo(f"Importing tags from: {file}")
click.echo(f"Database: {ctx.obj['database_url']}")
from meshcore_hub.common.database import DatabaseManager
from meshcore_hub.collector.tag_import import import_tags
# Initialize database
db = DatabaseManager(ctx.obj["database_url"])
db.create_tables()
# Import tags
stats = import_tags(
file_path=file,
db=db,
create_nodes=not no_create_nodes,
)
# Report results
click.echo("")
click.echo("Import complete:")
click.echo(f" Total tags in file: {stats['total']}")
click.echo(f" Tags created: {stats['created']}")
click.echo(f" Tags updated: {stats['updated']}")
click.echo(f" Tags skipped: {stats['skipped']}")
click.echo(f" Nodes created: {stats['nodes_created']}")
if stats["errors"]:
click.echo("")
click.echo("Errors:")
for error in stats["errors"]:
click.echo(f" - {error}", err=True)
db.dispose()
+180
View File
@@ -0,0 +1,180 @@
"""Import node tags from JSON file."""
import json
import logging
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from pydantic import BaseModel, Field, field_validator
from sqlalchemy import select
from meshcore_hub.common.database import DatabaseManager
from meshcore_hub.common.models import Node, NodeTag
logger = logging.getLogger(__name__)
class TagEntry(BaseModel):
"""Schema for a tag entry in the import file."""
public_key: str = Field(..., min_length=64, max_length=64)
key: str = Field(..., min_length=1, max_length=100)
value: str | None = None
value_type: str = Field(
default="string", pattern=r"^(string|number|boolean|coordinate)$"
)
@field_validator("public_key")
@classmethod
def validate_public_key(cls, v: str) -> str:
"""Validate that public_key is a valid hex string."""
if not all(c in "0123456789abcdefABCDEF" for c in v):
raise ValueError("public_key must be a valid hex string")
return v.lower()
class TagsFile(BaseModel):
"""Schema for the tags JSON file."""
tags: list[TagEntry]
def load_tags_file(file_path: str | Path) -> TagsFile:
"""Load and validate tags from a JSON file.
Args:
file_path: Path to the tags JSON file
Returns:
Validated TagsFile instance
Raises:
FileNotFoundError: If file does not exist
json.JSONDecodeError: If file is not valid JSON
pydantic.ValidationError: If file content is invalid
"""
path = Path(file_path)
if not path.exists():
raise FileNotFoundError(f"Tags file not found: {file_path}")
with open(path, "r") as f:
data = json.load(f)
return TagsFile.model_validate(data)
def import_tags(
file_path: str | Path,
db: DatabaseManager,
create_nodes: bool = True,
) -> dict[str, Any]:
"""Import tags from a JSON file into the database.
Performs upsert operations - existing tags are updated, new tags are created.
Args:
file_path: Path to the tags JSON file
db: Database manager instance
create_nodes: If True, create nodes that don't exist. If False, skip tags
for non-existent nodes.
Returns:
Dictionary with import statistics:
- total: Total number of tags in file
- created: Number of new tags created
- updated: Number of existing tags updated
- skipped: Number of tags skipped (node not found and create_nodes=False)
- nodes_created: Number of new nodes created
- errors: List of error messages
"""
stats: dict[str, Any] = {
"total": 0,
"created": 0,
"updated": 0,
"skipped": 0,
"nodes_created": 0,
"errors": [],
}
# Load and validate file
try:
tags_file = load_tags_file(file_path)
except Exception as e:
stats["errors"].append(f"Failed to load tags file: {e}")
return stats
stats["total"] = len(tags_file.tags)
now = datetime.now(timezone.utc)
with db.session_scope() as session:
# Cache nodes by public_key to reduce queries
node_cache: dict[str, Node] = {}
for tag_entry in tags_file.tags:
try:
# Get or create node
node = node_cache.get(tag_entry.public_key)
if node is None:
query = select(Node).where(Node.public_key == tag_entry.public_key)
node = session.execute(query).scalar_one_or_none()
if node is None:
if create_nodes:
node = Node(
public_key=tag_entry.public_key,
first_seen=now,
last_seen=now,
)
session.add(node)
session.flush()
stats["nodes_created"] += 1
logger.debug(
f"Created node for {tag_entry.public_key[:12]}..."
)
else:
stats["skipped"] += 1
logger.debug(
f"Skipped tag for unknown node {tag_entry.public_key[:12]}..."
)
continue
node_cache[tag_entry.public_key] = node
# Find or create tag
tag_query = select(NodeTag).where(
NodeTag.node_id == node.id,
NodeTag.key == tag_entry.key,
)
existing_tag = session.execute(tag_query).scalar_one_or_none()
if existing_tag:
# Update existing tag
existing_tag.value = tag_entry.value
existing_tag.value_type = tag_entry.value_type
stats["updated"] += 1
logger.debug(
f"Updated tag {tag_entry.key}={tag_entry.value} "
f"for {tag_entry.public_key[:12]}..."
)
else:
# Create new tag
new_tag = NodeTag(
node_id=node.id,
key=tag_entry.key,
value=tag_entry.value,
value_type=tag_entry.value_type,
)
session.add(new_tag)
stats["created"] += 1
logger.debug(
f"Created tag {tag_entry.key}={tag_entry.value} "
f"for {tag_entry.public_key[:12]}..."
)
except Exception as e:
error_msg = f"Error processing tag {tag_entry.key} for {tag_entry.public_key[:12]}...: {e}"
stats["errors"].append(error_msg)
logger.error(error_msg)
return stats
+343
View File
@@ -0,0 +1,343 @@
"""Tests for tag import functionality."""
import json
import tempfile
from pathlib import Path
import pytest
from pydantic import ValidationError
from sqlalchemy import select
from meshcore_hub.collector.tag_import import (
TagEntry,
TagsFile,
import_tags,
load_tags_file,
)
from meshcore_hub.common.database import DatabaseManager
from meshcore_hub.common.models import Node, NodeTag
class TestTagEntry:
"""Tests for TagEntry model."""
def test_valid_tag_entry(self):
"""Test creating a valid tag entry."""
entry = TagEntry(
public_key="0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
key="location",
value="San Francisco",
value_type="string",
)
assert (
entry.public_key
== "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"
)
assert entry.key == "location"
assert entry.value == "San Francisco"
assert entry.value_type == "string"
def test_public_key_lowercase(self):
"""Test that public key is normalized to lowercase."""
entry = TagEntry(
public_key="0123456789ABCDEF0123456789ABCDEF0123456789ABCDEF0123456789ABCDEF",
key="test",
)
assert (
entry.public_key
== "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"
)
def test_default_value_type(self):
"""Test default value_type is string."""
entry = TagEntry(
public_key="0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
key="test",
)
assert entry.value_type == "string"
def test_invalid_public_key_length(self):
"""Test that short public key is rejected."""
with pytest.raises(ValidationError):
TagEntry(
public_key="0123456789abcdef",
key="test",
)
def test_invalid_public_key_chars(self):
"""Test that non-hex public key is rejected."""
with pytest.raises(ValidationError):
TagEntry(
public_key="zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz",
key="test",
)
def test_invalid_value_type(self):
"""Test that invalid value_type is rejected."""
with pytest.raises(ValidationError):
TagEntry(
public_key="0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
key="test",
value_type="invalid",
)
def test_valid_value_types(self):
"""Test all valid value types."""
for vt in ["string", "number", "boolean", "coordinate"]:
entry = TagEntry(
public_key="0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
key="test",
value_type=vt,
)
assert entry.value_type == vt
class TestTagsFile:
"""Tests for TagsFile model."""
def test_valid_tags_file(self):
"""Test creating a valid tags file."""
tags_file = TagsFile(
tags=[
TagEntry(
public_key="0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
key="location",
value="SF",
),
TagEntry(
public_key="fedcba9876543210fedcba9876543210fedcba9876543210fedcba9876543210",
key="role",
value="gateway",
),
]
)
assert len(tags_file.tags) == 2
def test_empty_tags(self):
"""Test tags file with empty tags list."""
tags_file = TagsFile(tags=[])
assert len(tags_file.tags) == 0
class TestLoadTagsFile:
"""Tests for load_tags_file function."""
def test_load_valid_file(self):
"""Test loading a valid tags file."""
data = {
"tags": [
{
"public_key": "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
"key": "location",
"value": "San Francisco",
"value_type": "string",
}
]
}
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
json.dump(data, f)
f.flush()
result = load_tags_file(f.name)
assert len(result.tags) == 1
assert result.tags[0].key == "location"
Path(f.name).unlink()
def test_file_not_found(self):
"""Test loading non-existent file."""
with pytest.raises(FileNotFoundError):
load_tags_file("/nonexistent/path/tags.json")
def test_invalid_json(self):
"""Test loading invalid JSON file."""
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
f.write("invalid json {{{")
f.flush()
with pytest.raises(json.JSONDecodeError):
load_tags_file(f.name)
Path(f.name).unlink()
def test_invalid_schema(self):
"""Test loading file with invalid schema."""
data: dict[str, list[str]] = {"not_tags": []}
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
json.dump(data, f)
f.flush()
with pytest.raises(ValidationError):
load_tags_file(f.name)
Path(f.name).unlink()
class TestImportTags:
"""Tests for import_tags function."""
@pytest.fixture
def db_manager(self):
"""Create an in-memory database manager for testing."""
manager = DatabaseManager("sqlite:///:memory:")
manager.create_tables()
yield manager
manager.dispose()
@pytest.fixture
def sample_tags_file(self):
"""Create a sample tags file."""
data = {
"tags": [
{
"public_key": "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
"key": "location",
"value": "San Francisco",
"value_type": "string",
},
{
"public_key": "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
"key": "role",
"value": "gateway",
"value_type": "string",
},
{
"public_key": "fedcba9876543210fedcba9876543210fedcba9876543210fedcba9876543210",
"key": "altitude",
"value": "100",
"value_type": "number",
},
]
}
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
json.dump(data, f)
f.flush()
yield f.name
Path(f.name).unlink()
def test_import_creates_nodes_and_tags(self, db_manager, sample_tags_file):
"""Test that import creates nodes and tags."""
stats = import_tags(sample_tags_file, db_manager, create_nodes=True)
assert stats["total"] == 3
assert stats["created"] == 3
assert stats["updated"] == 0
assert stats["skipped"] == 0
assert stats["nodes_created"] == 2
assert len(stats["errors"]) == 0
# Verify in database
with db_manager.session_scope() as session:
nodes = session.execute(select(Node)).scalars().all()
assert len(nodes) == 2
tags = session.execute(select(NodeTag)).scalars().all()
assert len(tags) == 3
def test_import_updates_existing_tags(self, db_manager, sample_tags_file):
"""Test that import updates existing tags."""
# First import
stats1 = import_tags(sample_tags_file, db_manager, create_nodes=True)
assert stats1["created"] == 3
assert stats1["updated"] == 0
# Second import
stats2 = import_tags(sample_tags_file, db_manager, create_nodes=True)
assert stats2["created"] == 0
assert stats2["updated"] == 3
assert stats2["nodes_created"] == 0
def test_import_skips_unknown_nodes(self, db_manager, sample_tags_file):
"""Test that import skips tags for unknown nodes when create_nodes=False."""
stats = import_tags(sample_tags_file, db_manager, create_nodes=False)
assert stats["total"] == 3
assert stats["created"] == 0
assert stats["skipped"] == 3
assert stats["nodes_created"] == 0
# Verify no nodes or tags in database
with db_manager.session_scope() as session:
nodes = session.execute(select(Node)).scalars().all()
assert len(nodes) == 0
tags = session.execute(select(NodeTag)).scalars().all()
assert len(tags) == 0
def test_import_with_existing_nodes(self, db_manager, sample_tags_file):
"""Test import when nodes already exist."""
# Create a node first
with db_manager.session_scope() as session:
node = Node(
public_key="0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
)
session.add(node)
# Import with create_nodes=False
stats = import_tags(sample_tags_file, db_manager, create_nodes=False)
# Only 2 tags for the existing node should be created
assert stats["created"] == 2
assert stats["skipped"] == 1 # One tag for the non-existent node
assert stats["nodes_created"] == 0
def test_import_nonexistent_file(self, db_manager):
"""Test import with non-existent file."""
stats = import_tags("/nonexistent/tags.json", db_manager)
assert stats["total"] == 0
assert len(stats["errors"]) == 1
assert "Failed to load" in stats["errors"][0]
def test_import_empty_file(self, db_manager):
"""Test import with empty tags list."""
data: dict[str, list[str]] = {"tags": []}
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
json.dump(data, f)
f.flush()
stats = import_tags(f.name, db_manager)
assert stats["total"] == 0
assert stats["created"] == 0
Path(f.name).unlink()
def test_import_preserves_value_type(self, db_manager):
"""Test that import preserves value_type correctly."""
data = {
"tags": [
{
"public_key": "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
"key": "count",
"value": "42",
"value_type": "number",
},
{
"public_key": "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
"key": "active",
"value": "true",
"value_type": "boolean",
},
]
}
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
json.dump(data, f)
f.flush()
import_tags(f.name, db_manager, create_nodes=True)
with db_manager.session_scope() as session:
tags = session.execute(select(NodeTag)).scalars().all()
tag_dict = {t.key: t for t in tags}
assert tag_dict["count"].value_type == "number"
assert tag_dict["active"].value_type == "boolean"
Path(f.name).unlink()