Skip to content

cnak-us/cnak-plugin-sdk-python

Folders and files

NameName
Last commit message
Last commit date

Latest commit

 

History

1 Commit
 
 
 
 
 
 

Repository files navigation

CNAK Plugin SDK for Python

Build CNAK plugins in Python with minimal boilerplate.

Install

pip install cnak-plugin-sdk

Or from source:

pip install -e .

Quick Start

from aiohttp import web
from cnak_plugin import Plugin, TrackPoint, CoTEvent, CoTPoint

plugin = Plugin("hello-world", "0.1.0", description="A minimal CNAK plugin")

@plugin.sidebar("Hello", icon="MdWavingHand")
def hello_page():
    pass

@plugin.route("/api/greet")
async def greet(request):
    return web.json_response({"message": "Hello from CNAK plugin!"})

@plugin.on_track_update
async def handle_track(track_point: TrackPoint):
    print(f"{track_point.callsign} at ({track_point.latitude}, {track_point.longitude})")

plugin.run()

CoT Types (via PyCot)

The SDK uses PyCot for its Cursor on Target type system. All CoT models are Pydantic v2 models with full validation.

from cnak_plugin import CoTEvent, CoTPoint, Detail, Track

# Build a CoT event
from CoT.models.core import Event, Point, Detail, Remarks
from CoT.types import CoTTypes, CoTHow

event = Event(
    type=CoTTypes["a-f-G-U-C"],
    how=CoTHow["h-g-i-g-o"],
    point=Point(lat=38.8977, lon=-77.0365, hae=10.0, ce=9999999, le=9999999),
    detail=Detail(remarks=Remarks(text="Hello from plugin")),
)

Available PyCot types

Import Description
CoTEvent (alias for Event) Full CoT event with uid, type, time, point, detail
CoTPoint (alias for Point) Validated lat/lon/hae/ce/le
Detail CoT detail element (remarks, image, track; extras allowed)
Track Course, speed, slope with error margins
Remarks Text remarks
Image Full NITF-spec image metadata
CoTTypes Dictionary of CoT type codes to descriptions
CoTHow Dictionary of how codes to descriptions

CNAK-specific types

Import Description
TrackPoint Backend store.Point format received on NATS tracks.>
Geofence CNAK geofence region definition

How It Works

  1. The SDK starts an aiohttp HTTP server (default port 8200) with a /health endpoint.
  2. It connects to NATS and publishes a PluginRegistration message to cnak.plugin.register.
  3. A heartbeat re-registers every 30 seconds so the CNAK backend knows the plugin is alive.
  4. It responds to cnak.plugin.discover requests by re-announcing itself.
  5. On shutdown it publishes to cnak.plugin.deregister.

API

Plugin(name, version, port=8200, nats_url=None, author="", description="")

Create a plugin instance. nats_url defaults to the NATS_URL environment variable or nats://nats:4222.

Decorators

  • @plugin.sidebar(label, icon="MdExtension", route=None, asset=None) -- Register a sidebar navigation entry.
  • @plugin.on_track_update -- Subscribe to tracks.>. Handler receives a TrackPoint.
  • @plugin.on_geofence_alert -- Subscribe to geofence.alerts.> NATS subject.
  • @plugin.on_nats_message(subject) -- Subscribe to any NATS subject.
  • @plugin.route(path, method="GET") -- Add a custom HTTP endpoint.

Methods

  • plugin.frontend_assets(*files) -- Declare JS asset files for the manifest.
  • plugin.run() -- Blocking entry point (runs asyncio event loop).
  • await plugin.start() -- Non-blocking async start.
  • await plugin.shutdown() -- Graceful shutdown.

Environment Variables

Variable Default Description
NATS_URL nats://nats:4222 NATS server URL

License

Apache-2.0

About

No description, website, or topics provided.

Resources

Stars

Watchers

Forks

Releases

No releases published

Packages

 
 
 

Contributors

Languages