TAK Ecosystem Integration Guide

<?xml version="1.0" encoding="UTF-8"?>
<event version="2.0"
       uid="org.example.vehicle-unit-42"
       type="a-f-G-E-V"
       how="m-g"
       time="2026-01-01T14:30:00.000Z"
       start="2026-01-01T14:30:00.000Z"
       stale="2026-01-01T14:35:00.000Z">
  <!-- WGS-84: hae=height above ellipsoid(m), ce=circular error(m) -->
  <point lat="38.8977" lon="-77.0365" hae="25.4" ce="5" le="10"/>
  <detail>
    <contact callsign="BRAVO-7" endpoint="*:-1:stcp"/>
    <__group name="Blue" role="Team Member"/>
    <status battery="87"/>
    <track speed="12.5" course="270.0"/>    <!-- m/s, degrees true -->
    <remarks>Proceeding to checkpoint DELTA. ETA 14:45Z.</remarks>
    <sensor type="optical" status="active" zoom="4x"/>
    <color argb="-65536"/>   <!-- red -->
    <link uid="org.example.company-alpha" type="a-f-G-U-C"
          relation="p-p" remarks="Parent unit"/>
  </detail>
</event>

import socket, ssl, time

def connect_tls(host, port, certfile, keyfile, cafile):
    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
    ctx.load_cert_chain(certfile, keyfile)
    ctx.load_verify_locations(cafile)
    raw = socket.create_connection((host, port))
    return ctx.wrap_socket(raw, server_hostname=host)

def resilient_stream(host, port, cot_generator):
    delay = 1
    while True:
        try:
            sock = connect_tls(host, port,
                'client.pem', 'client.key', 'ca.pem')
            delay = 1
            for cot in cot_generator():
                sock.sendall(cot.encode('utf-8'))
        except (socket.error, ssl.SSLError) as e:
            print(f'Connection lost: {e}. Retry in {delay}s')
            time.sleep(delay)
            delay = min(delay * 2, 60)

import socket

MCAST_GRP  = '239.2.3.1'
MCAST_PORT = 6969

def send_multicast(cot_xml):
    sock = socket.socket(socket.AF_INET,
                         socket.SOCK_DGRAM, socket.IPPROTO_UDP)
    sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 2)
    sock.sendto(cot_xml.encode(), (MCAST_GRP, MCAST_PORT))
    sock.close()

const ws = new WebSocket('wss://takserver:8443/takproto/1');

ws.onmessage = (ev) => {
    const xml = new DOMParser()
        .parseFromString(ev.data, 'application/xml');
    const uid = xml.querySelector('event').getAttribute('uid');
    const lat = xml.querySelector('point').getAttribute('lat');
    console.log(`${uid} is at ${lat}`);
};

ws.send(buildCotXml('my-asset', 38.8977, -77.0365));

import requests, socket, ssl, datetime, time, logging

log = logging.getLogger('json2cot')
API_URL   = 'https://fleet.example.com/api/v2/vehicles'
API_KEY   = 'YOUR_API_KEY'
TAK_HOST  = 'takserver.example.com'
TAK_PORT  = 8089
POLL_SEC  = 15
STALE_MIN = 2

def iso(dt): return dt.strftime('%Y-%m-%dT%H:%M:%S.000Z')

def vehicle_to_cot(v):
    now   = datetime.datetime.utcnow()
    stale = now + datetime.timedelta(minutes=STALE_MIN)
    uid   = f'fleet-{v["id"]}'
    return f'''<?xml version="1.0"?>
    <event version="2.0" uid="{uid}" type="a-f-G-E-V"
           how="m-g" time="{iso(now)}"
           start="{iso(now)}" stale="{iso(stale)}">
      <point lat="{v['location']['lat']}"
             lon="{v['location']['lng']}" hae="0" ce="10" le="10"/>
      <detail>
        <contact callsign="{v.get('name', uid)}"/>
        <track speed="{v.get('speed_ms',0)}"
               course="{v.get('heading',0)}"/>
        <remarks>Dept: {v.get('department','Unknown')}</remarks>
      </detail>
    </event>'''

def get_conn():
    ctx = ssl.create_default_context()
    ctx.check_hostname = False
    ctx.verify_mode    = ssl.CERT_NONE   # use CERT_REQUIRED + CA in prod
    return ctx.wrap_socket(socket.create_connection((TAK_HOST, TAK_PORT)))

conn = get_conn()
while True:
    try:
        resp = requests.get(API_URL,
                            headers={'X-Api-Key': API_KEY}, timeout=8)
        for v in resp.json().get('vehicles', []):
            conn.sendall(vehicle_to_cot(v).encode())
        time.sleep(POLL_SEC)
    except Exception as e:
        log.error(f'Error: {e}. Reconnecting...')
        try: conn.close()
        except: pass
        time.sleep(5)
        conn = get_conn()

from flask import Flask, Response
from your_db import get_active_assets

app = Flask(__name__)

@app.route('/tak/feed.xml')
def feed():
    events = ''.join(asset_to_cot(a) for a in get_active_assets())
    xml = f'<?xml version="1.0"?><events>{events}</events>'
    return Response(xml, mimetype='application/xml')

# In TAK Server admin: Data Feeds > Add Feed
# Protocol: HTTPS, URL: https://yourserver/tak/feed.xml
# Polling interval: 15-60 seconds

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/tak/callback', methods=['POST'])
def tak_callback():
    payload    = request.json
    event_type = payload.get('event_type')
    unit_uid   = payload.get('uid')
    lat        = payload.get('lat')
    lon        = payload.get('lon')
    if event_type == 'unit_dispatched':
        update_dispatch_record(unit_uid, lat, lon)
        trigger_camera_ptz(lat, lon)   # swing PTZ camera to target
        notify_supervisor(unit_uid)
    return jsonify({'status': 'received'}), 200

import socket, kafka, xml.etree.ElementTree as ET

producer = kafka.KafkaProducer(bootstrap_servers='kafka:9092')

def cot_to_dict(xml_str):
    root = ET.fromstring(xml_str)
    pt   = root.find('point')
    return {'uid': root.get('uid'), 'type': root.get('type'),
            'lat': pt.get('lat'), 'lon': pt.get('lon')}

with socket.create_connection(('takserver', 8087)) as s:
    buf = ''
    while True:
        data = s.recv(4096).decode()
        buf += data
        while '</event>' in buf:
            end = buf.index('</event>') + 8
            try:
                event = cot_to_dict(buf[:end])
                producer.send('tak-events', str(event).encode())
            except ET.ParseError: pass
            buf = buf[end:]

# Return path: Kafka consumer → TAK
consumer = kafka.KafkaConsumer('tak-commands',
                               bootstrap_servers='kafka:9092')
tak_out  = socket.create_connection(('takserver', 8087))
for msg in consumer:
    tak_out.sendall(build_cot_from_command(msg.value).encode())

public class BidirectionalBridge
    implements MapEventDispatcher.MapEventDispatchListener {
    private WebSocket ws;

    public void start() {
        mapView.getMapEventDispatcher()
               .addMapItemEventListener(MapEvent.ITEM_ADDED, this);

        ws = http.newWebSocket(
            new Request.Builder()
                .url("wss://ai-server.example.com/tak-bridge").build(),
            new WebSocketListener() {
                @Override
                public void onMessage(WebSocket ws, String text) {
                    // External server sent command → dispatch as CoT
                    CotDispatcher.getExternalDispatcher()
                                 .dispatchToBroadcast(buildCotFromCommand(text));
                }
            });
    }

    @Override
    public void onMapEvent(MapEvent event) {
        if (event.getType().equals(MapEvent.ITEM_ADDED)) {
            MapItem item = event.getItem();
            JSONObject p = new JSONObject();
            p.put("uid", item.getUID());
            p.put("lat", item.getGeoPointMetaData().get().getLatitude());
            p.put("lon", item.getGeoPointMetaData().get().getLongitude());
            ws.send(p.toString());  // forward map item to external server
        }
    }
}

import socket, threading, xml.etree.ElementTree as ET

def listen_to_tak(host='takserver', port=8087):
    with socket.create_connection((host, port)) as s:
        buf = b''
        while True:
            chunk = s.recv(4096)
            if not chunk: break
            buf += chunk
            while b'</event>' in buf:
                end = buf.index(b'</event>') + 8
                xml = buf[:end].decode('utf-8', errors='replace')
                buf = buf[end:]
                process_incoming_cot(xml)   # AI model, DB, alerting

threading.Thread(target=listen_to_tak, daemon=True).start()

import requests

SERVER = 'https://takserver:8443'
CERT   = ('client.pem', 'client.key')
CA     = 'ta-ca.pem'

# All currently connected clients
resp = requests.get(f'{SERVER}/Marti/api/clients', cert=CERT, verify=CA)
for c in resp.json()['data']:
    print(c['uid'], c['callsign'], c['lastEventPt'])

# Latest CoT for a specific UID
uid  = 'vehicle-unit-42'
resp = requests.get(f'{SERVER}/Marti/api/cot/xml/{uid}',
                    cert=CERT, verify=CA)

# POST a CoT event without a persistent socket
resp = requests.post(f'{SERVER}/Marti/api/cot/xml',
                     data=cot_xml, cert=CERT, verify=CA,
                     headers={'Content-Type': 'application/xml'})

from ultralytics import YOLO
import cv2, socket, datetime, threading, requests

model = YOLO('yolov8n.pt')
tak   = socket.create_connection(('takserver', 8087))

CAMERA_ID    = 'gate-cam-01'
CAMERA_LAT   = 38.8977
CAMERA_LON   = -77.0365
CALLBACK_URL = 'https://cameraserver.example.com/tak/callback'

def detection_to_cot(n, cls, conf, cam_lat, cam_lon):
    now   = datetime.datetime.utcnow()
    stale = now + datetime.timedelta(minutes=3)
    fmt   = '%Y-%m-%dT%H:%M:%S.000Z'
    ctype = 'a-u-G-E-V' if cls == 'car' else 'a-u-G'
    return f'''<?xml version="1.0"?>
    <event version="2.0" uid="cv-{n}" type="{ctype}"
           how="m-g" time="{now.strftime(fmt)}"
           start="{now.strftime(fmt)}" stale="{stale.strftime(fmt)}">
      <point lat="{cam_lat}" lon="{cam_lon}" hae="0" ce="50" le="50"/>
      <detail>
        <contact callsign="AI-DET:{cls.upper()}"/>
        <remarks>Conf:{conf:.2f} | Cam:{CAMERA_ID}</remarks>
        <color argb="-65536"/>
      </detail>
    </event>'''

cap = cv2.VideoCapture('rtsp://camera.local:554/stream1')
n   = 0
while cap.isOpened():
    ret, frame = cap.read()
    if not ret: break
    for r in model(frame, conf=0.6):
        for box in r.boxes:
            cls, conf = model.names[int(box.cls)], float(box.conf)
            n += 1
            tak.sendall(detection_to_cot(n, cls, conf,
                                         CAMERA_LAT, CAMERA_LON).encode())
            threading.Thread(target=requests.post, kwargs={
                'url': CALLBACK_URL,
                'json': {'event':'detection','class':cls,'conf':conf}
            }).start()

import openai, json, socket, datetime

client   = openai.OpenAI(api_key='YOUR_KEY')
tak_sock = socket.create_connection(('takserver', 8087))

SYSTEM_PROMPT = '''You are a military/emergency communications parser.
Extract: unit_id, action, lat, lon, remarks from the transcript.
Respond ONLY as JSON. Use null for unknown fields.'''

def parse_radio(transcript):
    resp = client.chat.completions.create(
        model='gpt-4o',
        messages=[{'role':'system','content':SYSTEM_PROMPT},
                  {'role':'user','content':transcript}],
        response_format={'type':'json_object'})
    return json.loads(resp.choices[0].message.content)

def parsed_to_cot(p):
    if not p.get('lat') or not p.get('lon'): return None
    now   = datetime.datetime.utcnow()
    stale = now + datetime.timedelta(minutes=10)
    fmt   = '%Y-%m-%dT%H:%M:%S.000Z'
    return f'''<?xml version="1.0"?>
    <event version="2.0" uid="llm-{p['unit_id'].replace(' ','-')}"
           type="a-f-G-U-C" how="h-e"
           time="{now.strftime(fmt)}" start="{now.strftime(fmt)}"
           stale="{stale.strftime(fmt)}">
      <point lat="{p['lat']}" lon="{p['lon']}"
             hae="0" ce="500" le="500"/>
      <detail>
        <contact callsign="{p['unit_id']}"/>
        <remarks>AI-parsed: {p.get('remarks','')}</remarks>
      </detail>
    </event>'''

import openai, collections, time, threading

client       = openai.OpenAI(api_key='YOUR_KEY')
event_buffer = collections.deque(maxlen=200)
# Background thread fills buffer from TAK TCP stream (see Section 3.2.1)

def generate_sitrep():
    events  = list(event_buffer)
    if not events: return 'No events recorded.'
    summary = '\n'.join(
        f"{e['uid']} ({e['type']}) at {e['lat']:.4f},{e['lon']:.4f}"
        for e in events[-50:])
    resp = client.chat.completions.create(
        model='gpt-4o',
        messages=[
            {'role':'system','content':
             'You are a military operations staff officer. '
             'Generate a concise SITREP from TAK event data. '
             'Include friendly forces disposition and notable activities.'},
            {'role':'user','content':summary}])
    return resp.choices[0].message.content

def sitrep_loop():
    while True:
        time.sleep(3600)
        dispatch_geochat('AI-SITREP', generate_sitrep())

threading.Thread(target=sitrep_loop, daemon=True).start()

def route_alert_with_llm(alert):
    prompt = f'''
    TAK alert: {alert['type']} at {alert['lat']},{alert['lon']}
    Description: {alert['remarks']}
    Nearby units: Alpha-1 (0.3km), Bravo-2 (1.2km)
    External: camera-server, dispatch-cad, supervisors-sms
    Respond as JSON:
    {{"notify_units":[],"external_webhooks":[],"escalate":false}}
    '''
    resp = client.chat.completions.create(
        model='gpt-4o',
        messages=[{'role':'user','content':prompt}],
        response_format={'type':'json_object'})
    execute_routing(json.loads(resp.choices[0].message.content))

# Install: curl -fsSL https://ollama.ai/install.sh | sh
# Pull model: ollama pull llama3.2

import requests

def local_llm(prompt, model='llama3.2'):
    resp = requests.post('http://localhost:11434/api/generate',
        json={'model':model,'prompt':prompt,'stream':False})
    return resp.json()['response']

sitrep = local_llm(f'Generate SITREP from: {event_summary}')

from shapely.geometry import Point, Polygon
import requests

ZONES = {'restricted-zone-a': Polygon([
    (-77.05,38.90),(-77.03,38.90),(-77.03,38.89),(-77.05,38.89)])}
known = {}

def check_geofences(uid, lat, lon, cot_type):
    pt = Point(lon, lat)
    for zone, poly in ZONES.items():
        inside = poly.contains(pt)
        if inside and not known.get((uid,zone), False):
            if 'hostile' in cot_type or 'unknown' in cot_type:
                tak.sendall(build_alert_cot(uid,lat,lon,zone).encode())
                requests.post('https://cameras/ptz',json={'zone':zone})
        known[(uid, zone)] = inside

from fastapi import FastAPI, Depends, HTTPException, Header
from fastapi.responses import Response
import httpx, datetime

app = FastAPI()
VALID_KEYS = {'tak-client-key-abc123'}

async def verify_key(x_api_key: str = Header(...)):
    if x_api_key not in VALID_KEYS:
        raise HTTPException(403, 'Invalid API key')

def make_placemark(site):
    lat, lon  = site['coordinates']['lat'], site['coordinates']['lon']
    color     = 'ff00ff00' if site['status'] == 'online' else 'ff0000ff'
    return f'''
    <Placemark>
      <name>{site['name']}</name>
      <Style><IconStyle><color>{color}</color></IconStyle></Style>
      <Point><coordinates>{lon},{lat},0</coordinates></Point>
    </Placemark>'''

@app.get('/live.kml', dependencies=[Depends(verify_key)])
async def live_kml():
    async with httpx.AsyncClient() as client:
        resp  = await client.get('https://data.example.com/api/sites',
                                headers={'X-Api-Key':'EXT_KEY'})
    sites = resp.json()['sites']
    kml = f'''<?xml version="1.0" encoding="UTF-8"?>
    <kml xmlns="http://www.opengis.net/kml/2.2">
      <Document>
        <name>Live Sites - {datetime.datetime.utcnow():%H:%MZ}</name>
        {''.join(make_placemark(s) for s in sites)}
      </Document>
    </kml>'''
    return Response(content=kml,
        media_type='application/vnd.google-earth.kml+xml')

# Distribute this Network Link KML to ATAK users:
# <NetworkLink><Link>
#   <href>https://yourserver/live.kml</href>
#   <refreshMode>onInterval</refreshMode>
#   <refreshInterval>30</refreshInterval>
# </Link></NetworkLink>

android {
    compileSdk 34
    defaultConfig { minSdk 21; targetSdk 34 }
}
dependencies {
    compileOnly files('libs/atak-sdk-4.8.1.jar')
    implementation 'com.squareup.okhttp3:okhttp:4.12.0'
    implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-android:1.7.3'
}

MapView mapView = MapView.getMapView();

// Pan and zoom
mapView.getMapController().panTo(new GeoPoint(38.8977, -77.0365), true);
mapView.getMapController().zoomTo(5000.0, true);

// Add a marker
PlacemarkMapItem marker = new PlacemarkMapItem(
    mapView.getContext(),
    new GeoPoint(38.8977, -77.0365),
    UUID.randomUUID().toString());
marker.setTitle("My API Marker");
marker.setType("a-f-G-U-C");
mapView.getRootGroup().addItem(marker);

// Listen for map taps
mapView.getMapEventDispatcher()
       .addMapEventListener(MapEvent.MAP_CLICK, event -> {
    GeoPoint tap = event.getPoint();
    // tap.getLatitude(), tap.getLongitude()
});

// Iterate all map items (full state access)
mapView.getRootGroup().deepForEachItem(item -> {
    String uid      = item.getUID();
    String type     = item.getType();
    GeoPoint pt     = item.getGeoPointMetaData().get();
    String callsign = item.getMetaString("callsign", "Unknown");
    externalServer.reportPosition(uid, callsign, type,
                                  pt.getLatitude(), pt.getLongitude());
    return true;
});

GeoFenceComponent gfc = GeoFenceComponent.getInstance(mapView);

GeoFence fence = new GeoFence(
    "zone-alpha",
    new GeoPoint(38.8977, -77.0365),
    500.0,                       // radius in meters
    GeoFence.Trigger.Entry,      // Entry, Exit, or Both
    GeoFence.MonitoredTypes.All);
gfc.addGeoFence(fence);

gfc.addGeoFenceListener((fenceName, itemUID, trigger) -> {
    if (trigger == GeoFence.Trigger.Entry) {
        notifyExternalServer(itemUID, fenceName, "ENTRY");
        dispatchAlertCot(itemUID, fenceName);
    }
});

CotEventCache.getInstance().addCotEventListener(cot -> {
    if (cot.getType().startsWith("a-h")) {  // hostile contacts
        new Thread(() -> {
            try {
                JSONObject payload = new JSONObject();
                payload.put("uid",  cot.getUID());
                payload.put("type", cot.getType());
                payload.put("lat",  cot.getPoint().getLat());
                payload.put("lon",  cot.getPoint().getLon());
                new OkHttpClient().newCall(
                    new Request.Builder()
                        .url("https://c2.example.com/contacts")
                        .post(RequestBody.create(payload.toString(),
                              MediaType.get("application/json")))
                        .build()).execute();
            } catch (Exception e) { Log.e("Plugin", "Callback failed", e); }
        }).start();
    }
});

import socket, psycopg2, xml.etree.ElementTree as ET

conn = psycopg2.connect(host='db',dbname='taklog',user='tak',password='secret')
cur  = conn.cursor()
cur.execute('''CREATE TABLE IF NOT EXISTS cot_events (
    id SERIAL PRIMARY KEY, uid TEXT, cot_type TEXT,
    lat DOUBLE PRECISION, lon DOUBLE PRECISION,
    callsign TEXT, received TIMESTAMPTZ DEFAULT NOW())''')
conn.commit()

def parse_and_log(xml_str):
    root = ET.fromstring(xml_str)
    pt   = root.find('point')
    if pt is None: return
    detail   = root.find('detail') or ET.Element('d')
    contact  = detail.find('contact')
    callsign = contact.get('callsign','') if contact else ''
    cur.execute('INSERT INTO cot_events '
                '(uid,cot_type,lat,lon,callsign) VALUES (%s,%s,%s,%s,%s)',
        (root.get('uid'), root.get('type'),
         float(pt.get('lat',0)), float(pt.get('lon',0)), callsign))
    conn.commit()

import asyncio, websockets, socket, json, xml.etree.ElementTree as ET

TAK_HOST = 'takserver.local'
TAK_PORT = 8087
WS_PORT  = 8765

tak_out  = socket.create_connection((TAK_HOST, TAK_PORT))
CLIENTS  = {}

async def handle_client(websocket, path):
    hello     = json.loads(await websocket.recv())
    server_id = hello['server_id']
    CLIENTS[server_id] = {'ws': websocket,
                          'caps': hello.get('capabilities', ['send'])}
    async for message in websocket:
        data = json.loads(message)
        if data.get('type') == 'cot':
            tak_out.sendall(data['xml'].encode())

async def tak_listener():
    buf = b''
    with socket.create_connection((TAK_HOST, TAK_PORT)) as listener:
        while True:
            data = listener.recv(4096)
            if not data: break
            buf += data
            while b'</event>' in buf:
                end = buf.index(b'</event>') + 8
                xml_str = buf[:end].decode('utf-8','replace')
                buf     = buf[end:]
                payload = json.dumps({'type':'cot_event','xml':xml_str})
                for srv in list(CLIENTS.values()):
                    if 'receive' in srv['caps']:
                        try: await srv['ws'].send(payload)
                        except: pass

async def main():
    server = await websockets.serve(handle_client, '0.0.0.0', WS_PORT)
    await asyncio.gather(server.wait_closed(), tak_listener())

asyncio.run(main())

<?xml version="1.0" encoding="UTF-8"?>
<videoConnections>
  <VideoConnection>
    <uid>cam-gate-north</uid>
    <alias>Gate North — PTZ</alias>
    <address>rtsp://192.168.10.50:554/axis-media/media.amp</address>
    <port>554</port>
    <protocol>rtsp</protocol>
    <username>admin</username>
    <password>your_password</password>
    <latitude>38.8977</latitude>
    <longitude>-77.0365</longitude>
  </VideoConnection>
</videoConnections>

import gps as gpslib, socket, datetime

tak     = socket.create_connection(('takserver', 8087))
session = gpslib.gps(mode=gpslib.WATCH_ENABLE | gpslib.WATCH_NEWSTYLE)

for report in session:
    if report.get('class') == 'TPV':
        lat = getattr(report, 'lat', None)
        lon = getattr(report, 'lon', None)
        if lat and lon:
            now   = datetime.datetime.utcnow()
            stale = now + datetime.timedelta(minutes=2)
            fmt   = '%Y-%m-%dT%H:%M:%S.000Z'
            cot = f'''<?xml version="1.0"?>
            <event version="2.0" uid="gps-001"
                   type="a-f-G-U-C" how="m-g"
                   time="{now.strftime(fmt)}"
                   start="{now.strftime(fmt)}"
                   stale="{stale.strftime(fmt)}">
              <point lat="{lat}" lon="{lon}"
                     hae="{getattr(report,'alt',0)}"
                     ce="3" le="5"/>
              <detail>
                <contact callsign="GPS-UNIT-1"/>
                <track speed="{getattr(report,'speed',0)}"
                       course="{getattr(report,'track',0)}"/>
              </detail>
            </event>'''
            tak.sendall(cot.encode())

from pyais.stream import SerialReaderNAIS
import socket, datetime

tak = socket.create_connection(('takserver', 8087))
AIS_TYPE_MAP = {'Tanker':'a-n-S-S-G','Cargo':'a-n-S-S-C','HSC':'a-f-S-S'}

for msg in SerialReaderNAIS('/dev/ttyUSB1'):
    try:
        d = msg.decode()
        if not hasattr(d,'lat') or not d.lat: continue
        cot_type = AIS_TYPE_MAP.get(str(d.shiptype),'a-u-S-S')
        name     = getattr(d,'shipname',f'MMSI-{d.mmsi}').strip()
        now   = datetime.datetime.utcnow()
        stale = now + datetime.timedelta(minutes=5)
        fmt   = '%Y-%m-%dT%H:%M:%S.000Z'
        cot   = f'''<?xml version="1.0"?>
        <event version="2.0" uid="ais-{d.mmsi}" type="{cot_type}"
               how="m-g" time="{now.strftime(fmt)}"
               start="{now.strftime(fmt)}" stale="{stale.strftime(fmt)}">
          <point lat="{d.lat}" lon="{d.lon}" hae="0" ce="50" le="50"/>
          <detail>
            <contact callsign="{name}"/>
            <remarks>MMSI:{d.mmsi}</remarks>
          </detail>
        </event>'''
        tak.sendall(cot.encode())
    except Exception as e: print(f'AIS error: {e}')

import socket, aprslib, datetime

tak = socket.create_connection(('takserver', 8087))

def aprs_to_cot(packet):
    if 'latitude' not in packet or 'longitude' not in packet: return
    now   = datetime.datetime.utcnow()
    stale = now + datetime.timedelta(minutes=10)
    fmt   = '%Y-%m-%dT%H:%M:%S.000Z'
    cot = f'''<?xml version="1.0"?>
    <event version="2.0" uid="aprs-{packet['from']}" type="a-f-G"
           how="m-g" time="{now.strftime(fmt)}"
           start="{now.strftime(fmt)}" stale="{stale.strftime(fmt)}">
      <point lat="{packet['latitude']}" lon="{packet['longitude']}"
             hae="0" ce="100" le="100"/>
      <detail>
        <contact callsign="{packet['from']}"/>
        <remarks>{packet.get('comment','APRS')}</remarks>
      </detail>
    </event>'''
    tak.sendall(cot.encode())

AIS = aprslib.IS('N0CALL', passwd='-1',
                  host='rotate.aprs2.net', port=14580)
AIS.set_filter('r/38.8977/-77.0365/50')
AIS.connect()
AIS.consumer(aprs_to_cot, raw=False, immortal=True)

import zipfile, io, hashlib, requests

SERVER = 'https://takserver:8443'
CERT   = ('client.pem', 'client.key')
CA     = 'ta-ca.pem'

def build_data_package(name, files):
    """files: {zip_entry_path: bytes_or_string}"""
    buf      = io.BytesIO()
    contents = []
    with zipfile.ZipFile(buf, 'w', zipfile.ZIP_DEFLATED) as zf:
        for zip_path, content in files.items():
            if isinstance(content, str): content = content.encode('utf-8')
            zf.writestr(zip_path, content)
            contents.append(
                f'<Content ignore="false" zipEntry="{zip_path}"/>')
        uid      = f'pkg-{hashlib.md5(name.encode()).hexdigest()[:8]}'
        manifest = f'''<?xml version="1.0" encoding="UTF-8"?>
        <MissionPackageManifest version="2">
          <Configuration>
            <Parameter name="uid" value="{uid}"/>
            <Parameter name="name" value="{name}"/>
          </Configuration>
          <Contents>{"".join(contents)}</Contents>
        </MissionPackageManifest>'''
        zf.writestr('MANIFEST/manifest.xml', manifest)
    return buf.getvalue()

def push_data_package(name, pkg_bytes):
    resp = requests.post(f'{SERVER}/Marti/sync/upload',
        files={'assetfile': (f'{name}.zip', pkg_bytes,
                             'application/x-zip-compressed')},
        params={'name': name, 'keywords': name},
        cert=CERT, verify=CA)
    return resp.json().get('Hash')

# Convert GeoTIFF to GeoPackage
gdal_translate -of GPKG \
    -co TILING_SCHEME=GoogleMapsCompatible \
    -co ZOOM_LEVEL_STRATEGY=UPPER \
    input_imagery.tif output_map.gpkg

# Add vector features to the same GeoPackage
ogr2ogr -f GPKG -append output_map.gpkg input_features.shp

# Push to TAK Server for device download
import requests
requests.post('https://takserver:8443/Marti/sync/upload',
    files={'assetfile':open('output_map.gpkg','rb')},
    cert=('client.pem','client.key'), verify='ta-ca.pem')

pip install gdal2mbtiles
gdal2mbtiles input.tif output.mbtiles \
    --min-resolution=8 --max-resolution=17 \
    --resampling=lanczos

import requests, socket, datetime, time

AGOL_URL = ('https://services.arcgis.com/YOUR_ORG/arcgis/rest'
            '/services/ActiveIncidents/FeatureServer/0/query')
TOKEN = 'YOUR_ARCGIS_TOKEN'
tak   = socket.create_connection(('takserver', 8087))

def fetch_features():
    return requests.get(AGOL_URL, params={
        'f': 'geojson', 'where': "Status='Active'",
        'outFields': 'IncidentName,Type,Severity,OBJECTID',
        'returnGeometry': 'true', 'token': TOKEN
    }).json()['features']

def feature_to_cot(feat):
    props = feat['properties']
    geom  = feat['geometry']
    lon, lat = geom['coordinates']
    cot_type = 'a-h-G' if props.get('Severity') == 'High' else 'a-u-G'
    now   = datetime.datetime.utcnow()
    stale = now + datetime.timedelta(hours=1)
    fmt   = '%Y-%m-%dT%H:%M:%S.000Z'
    return f'''<?xml version="1.0"?>
    <event version="2.0" uid="esri-{props['OBJECTID']}"
           type="{cot_type}" how="m-g"
           time="{now.strftime(fmt)}" start="{now.strftime(fmt)}"
           stale="{stale.strftime(fmt)}">
      <point lat="{lat:.6f}" lon="{lon:.6f}" hae="0" ce="100" le="100"/>
      <detail><contact callsign="{props['IncidentName']}"/></detail>
    </event>'''

while True:
    for feat in fetch_features():
        tak.sendall(feature_to_cot(feat).encode())
    time.sleep(60)

import psycopg2, select, json, socket, datetime

db = psycopg2.connect('postgresql://gisuser:pass@pgserver/gisdb')
db.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)
cur = db.cursor()
cur.execute('LISTEN asset_changed;')
tak = socket.create_connection(('takserver', 8087))

while True:
    if select.select([db], [], [], 5) == ([], [], []): continue
    db.poll()
    while db.notifies:
        row   = json.loads(db.notifies.pop(0).payload)
        now   = datetime.datetime.utcnow()
        stale = now + datetime.timedelta(minutes=5)
        fmt   = '%Y-%m-%dT%H:%M:%S.000Z'
        cot = f'''<?xml version="1.0"?>
        <event version="2.0" uid="asset-{row['id']}"
               type="a-f-G-E-V" how="m-g"
               time="{now.strftime(fmt)}" start="{now.strftime(fmt)}"
               stale="{stale.strftime(fmt)}">
          <point lat="{row['lat']:.6f}" lon="{row['lon']:.6f}"
                 hae="0" ce="5" le="5"/>
          <detail><contact callsign="{row['name']}"/></detail>
        </event>'''
        tak.sendall(cot.encode())

# Requires PostgreSQL trigger:
# AFTER INSERT OR UPDATE ON assets
# EXECUTE FUNCTION notify_asset_change();

import paho.mqtt.client as mqtt
import socket, json, datetime

tak = socket.create_connection(('takserver', 8087))

def on_connect(client, ud, flags, rc):
    client.subscribe('sensors/+/position')
    client.subscribe('alerts/+/trigger')

def on_message(client, ud, msg):
    payload = json.loads(msg.payload.decode())
    if '/position' in msg.topic:
        sensor_id = msg.topic.split('/')[1]
        cot = sensor_to_cot(sensor_id, payload)
        if cot: tak.sendall(cot.encode())
    elif '/trigger' in msg.topic:
        tak.sendall(build_alert_cot(payload).encode())

c = mqtt.Client()
c.on_connect = on_connect
c.on_message = on_message
c.connect('mqtt-broker.example.com', 1883)
c.loop_forever()

[PluginAttribute(Name="My Integration Plugin", Version="1.0.0")]
public class MyPlugin : AbstractPlugin {
    private Timer _pollTimer;

    public override void Initialize() {
        EventBus.Subscribe<CoTEvent>(OnCoTReceived);
        _pollTimer = new Timer(FetchAndDispatch, null,
                               TimeSpan.Zero,
                               TimeSpan.FromSeconds(30));
    }

    private async void FetchAndDispatch(object state) {
        var resp   = await new HttpClient()
                         .GetStringAsync("https://api.example.com/assets");
        var assets = JsonConvert.DeserializeObject<List<Asset>>(resp);
        foreach (var a in assets)
            EventBus.Publish(new SendCoTEvent { CoT = BuildCoT(a) });
    }

    private void OnCoTReceived(CoTEvent evt) {
        ExternalServer.NotifyAsync(evt.UID, evt.Lat, evt.Lon);
    }
}

from flask import Flask, request
import socket, datetime

app = Flask(__name__)
tak = socket.create_connection(('takserver', 8087))

@app.route('/cad/webhook', methods=['POST'])
def cad_incident():
    inc  = request.json
    pri  = inc.get('Priority', 3)
    now  = datetime.datetime.utcnow()
    stale = now + datetime.timedelta(hours=4)
    fmt  = '%Y-%m-%dT%H:%M:%S.000Z'
    cot  = f'''<?xml version="1.0"?>
    <event version="2.0" uid="cad-{inc['IncidentNumber']}"
           type="{'a-h-G' if pri == 1 else 'a-u-G'}" how="h-e"
           time="{now.strftime(fmt)}" start="{now.strftime(fmt)}"
           stale="{stale.strftime(fmt)}">
      <point lat="{inc['Latitude']}" lon="{inc['Longitude']}"
             hae="0" ce="30" le="30"/>
      <detail>
        <contact callsign="CAD-{inc['IncidentNumber']}"/>
        <remarks>P{pri} {inc['IncidentType']}</remarks>
      </detail>
    </event>'''
    tak.sendall(cot.encode())
    return {'status': 'ok'}, 200

import requests, xml.etree.ElementTree as ET

CAD_API = 'https://cad.example.com/api/v2'
CAD_KEY = 'YOUR_CAD_KEY'

def handle_unit_cot(xml_str):
    root   = ET.fromstring(xml_str)
    uid    = root.get('uid')
    detail = root.find('detail') or ET.Element('detail')
    status = detail.find('status')
    if status is not None:
        val = status.get('value', '')
        if val in ('arrived', 'available', 'en_route', 'busy'):
            requests.post(f'{CAD_API}/units/{uid}/status',
                json={'status': val, 'timestamp': root.get('time')},
                headers={'X-Api-Key': CAD_KEY})

import socket, datetime

tak = socket.create_connection(('takserver', 8087))

def send_geochat(sender_uid, callsign, message, group='__ANON__'):
    now    = datetime.datetime.utcnow()
    stale  = now + datetime.timedelta(hours=1)
    fmt    = '%Y-%m-%dT%H:%M:%S.000Z'
    msg_id = f'GeoChat.{sender_uid}.{int(now.timestamp())}'
    cot = f'''<?xml version="1.0"?>
    <event version="2.0" uid="{msg_id}" type="b-t-f" how="h-g-i"
           time="{now.strftime(fmt)}" start="{now.strftime(fmt)}"
           stale="{stale.strftime(fmt)}">
      <point lat="0" lon="0" hae="0" ce="9999999" le="9999999"/>
      <detail>
        <__chat parent="RootContactGroup"
                chatroom="{group}" id="{group}">
          <chatgrp uid0="{sender_uid}" id="{group}"/>
        </__chat>
        <remarks>{message}</remarks>
      </detail>
    </event>'''
    tak.sendall(cot.encode())

send_geochat('system-bot', 'AI-ALERT',
             'Unknown vehicle detected at Gate North.')

import requests

WEBHOOKS = {
    'hostile_detected': [
        'https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK',
        'https://events.pagerduty.com/v2/enqueue',
    ],
}

def fire_webhooks(event_type, payload):
    for url in WEBHOOKS.get(event_type, []):
        try:
            if 'slack.com' in url:
                requests.post(url, json={
                    'text': f'*TAK ALERT*: {event_type}\n{payload}'})
            elif 'pagerduty' in url:
                requests.post(url, json={
                    'routing_key': 'YOUR_PD_KEY',
                    'event_action': 'trigger',
                    'payload': {'summary': payload,
                                'severity': 'critical'}})
        except Exception as e: print(f'Webhook failed: {e}')

pip install pytak

import asyncio, pytak
from configparser import ConfigParser

class MyWorker(pytak.QueueWorker):
    async def handle_data(self, data):
        event = pytak.cot_event(
            uid=data['uid'], lat=data['lat'],
            lon=data['lon'], cot_type='a-f-G-E-V',
            callsign=data['name'])
        await self.put_queue(event)

    async def run(self):
        while True:
            for asset in await fetch_assets():
                await self.handle_data(asset)
            await asyncio.sleep(15)

async def main():
    config = ConfigParser()
    config['pytak'] = {'COT_URL': 'tcp://takserver:8087'}
    clitool = pytak.CLITool(config)
    await clitool.setup()
    clitool.add_tasks({MyWorker(clitool.tx_queue, config)})
    await clitool.run()

asyncio.run(main())

import re, html

def validate_cot_fields(uid, lat, lon, callsign, remarks):
    if not re.match(r'^[a-zA-Z0-9._-]{1,255}$', uid):
        raise ValueError(f'Invalid UID: {uid}')
    if not (-90 <= float(lat) <= 90):
        raise ValueError(f'Latitude out of range: {lat}')
    if not (-180 <= float(lon) <= 180):
        raise ValueError(f'Longitude out of range: {lon}')
    callsign = html.escape(str(callsign))[:64]
    remarks  = html.escape(str(remarks))[:500]
    return uid, float(lat), float(lon), callsign, remarks

import socket, threading, time, logging

class TakIntegrationHub:
    def __init__(self, tak_host, tak_port):
        self.tak_addr     = (tak_host, tak_port)
        self.tak_conn     = socket.create_connection(self.tak_addr)
        self.integrations = []
        self._start_listener()

    def register(self, integration):
        """Integration must implement get_cot_events() and on_tak_event()"""
        self.integrations.append(integration)
        threading.Thread(target=self._run_integration,
                         args=(integration,), daemon=True).start()

    def _run_integration(self, integration):
        while True:
            try:
                for cot_xml in integration.get_cot_events():
                    self.tak_conn.sendall(cot_xml.encode())
            except Exception as e:
                logging.error(f'{integration}: {e}')
                time.sleep(5)

    def _start_listener(self):
        def listen():
            with socket.create_connection(self.tak_addr) as s:
                buf = b''
                while True:
                    data = s.recv(4096)
                    if not data: break
                    buf += data
                    while b'</event>' in buf:
                        end = buf.index(b'</event>') + 8
                        xml = buf[:end].decode('utf-8','replace')
                        buf = buf[end:]
                        for integ in self.integrations:
                            try: integ.on_tak_event(xml)
                            except: pass
        threading.Thread(target=listen, daemon=True).start()

hub = TakIntegrationHub('takserver', 8087)
hub.register(FleetApiIntegration())
hub.register(CameraSystemIntegration())
hub.register(WeatherFeedIntegration())

version: '3.8'
services:
  freetek:
    image: freetakteam/freetakserver:latest
    ports:
      - "8087:8087"     # CoT TCP
      - "19023:19023"   # REST API
    volumes: [freetek-data:/opt/FreeTAKServer]

  mqtt:
    image: eclipse-mosquitto:latest
    ports: ["1883:1883"]
    volumes: ["./mosquitto.conf:/mosquitto/config/mosquitto.conf"]

  ollama:
    image: ollama/ollama:latest
    ports: ["11434:11434"]
    volumes: [ollama-models:/root/.ollama]

  tile-server:
    image: ghcr.io/maplibre/martin:latest
    ports: ["3000:3000"]
    environment:
      - DATABASE_URL=postgresql://gis:pass@postgis/gisdb

  postgis:
    image: postgis/postgis:15-3.4
    environment:
      POSTGRES_DB: gisdb
      POSTGRES_USER: gis
      POSTGRES_PASSWORD: pass
    volumes: [postgis-data:/var/lib/postgresql/data]

  middleware:
    build: ./middleware
    environment:
      - TAK_HOST=freetek
      - TAK_PORT=8087
      - MQTT_HOST=mqtt
      - OLLAMA_URL=http://ollama:11434
    depends_on: [freetek, mqtt, ollama]

volumes:
  freetek-data:
  ollama-models:
  postgis-data: