In python, i want to take wireshark and nmap and even firewall logs and use the data to by ip address to/from create an object identity and using timestamps and headers create triangulations of these objects. Add a device if known field for adding gps coordinates–my chat
To build an intelligence pipeline that merges Wireshark (PCAP), Nmap scans, and firewall logs into unified object identities with timestamp-based triangulation and GPS enrichment, you need a modular architecture.
The pipeline relies on four core stages: Ingestion & Parsing, Entity Resolution (IP-centric object creation), Timestamp & Header Correlation (Triangulation), and Asset Enrichment (GPS lookup).
1. Unified Data Model (Object Identity)
An object identity structure centralizes attributes gathered from disparate logs around a primary key (typically the IP address, backed up by MAC or hostname).
Python
from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, List, Optional, Set, Tuple
@dataclass
class NetworkObjectIdentity:
ip_address: str
mac_addresses: Set[str] = field(default_factory=set)
hostnames: Set[str] = field(default_factory=set)
open_ports: Dict[int, str] = field(default_factory=dict) # port: service
os_fingerprint: Optional[str] = None
first_seen: Optional[datetime] = None
last_seen: Optional[datetime] = None
# Telemetry and Triangulation data
packet_headers: List[Dict] = field(default_factory=list) # TTL, Seq, Window size, etc.
log_events: List[Dict] = field(default_factory=list) # Firewall hits, timestamps
# Device if known / GPS mapping
device_name: Optional[str] = None
gps_coordinates: Optional[Tuple[float, float]] = None # (Latitude, Longitude)
2. Log Parsers
You need discrete parsers to extract raw data from Nmap XML outputs, Wireshark PCAP files, and firewall log streams.
Python
import xml.etree.ElementTree as ET
import pyshark
import re
class LogIngestor:
@staticmethod
def parse_nmap_xml(xml_path: str) -> Dict[str, dict]:
"""Parses Nmap XML output to extract hosts, open ports, OS, and MACs."""
tree = ET.parse(xml_path)
root = tree.getroot()
devices = {}
for host in root.findall('host'):
ip = host.find("address[@addrtype='ipv4']")
if ip is None:
continue
ip_addr = ip.get('addr')
mac_elem = host.find("address[@addrtype='mac']")
mac = mac_elem.get('addr') if mac_elem is not None else None
hostname_elem = host.find(".//hostname")
hostname = hostname_elem.get('name') if hostname_elem is not None else None
ports = {}
for port in host.findall(".//port"):
port_id = int(port.get('portid'))
state = port.find("state").get('state')
if state == 'open':
service = port.find("service")
service_name = service.get('name') if service is not None else "unknown"
ports[port_id] = service_name
os_elem = host.find(".//osmatch")
os_name = os_elem.get('name') if os_elem is not None else None
devices[ip_addr] = {
"mac": mac,
"hostname": hostname,
"ports": ports,
"os": os_name
}
return devices
@staticmethod
def parse_pcap(pcap_path: str, max_packets: int = 1000) -> List[dict]:
"""Parses packet headers and timestamps from a Wireshark PCAP file using PyShark."""
cap = pyshark.FileCapture(pcap_path, only_summaries=False)
packet_records = []
for i, pkt in enumerate(cap):
if i >= max_packets:
break
try:
# Extract essential routing and transport headers
ip_layer = pkt.ip if hasattr(pkt, 'ip') else None
if not ip_layer:
continue
record = {
"timestamp": datetime.fromtimestamp(float(pkt.sniff_timestamp)),
"src_ip": ip_layer.src,
"dst_ip": ip_layer.dst,
"ttl": int(ip_layer.ttl) if hasattr(ip_layer, 'ttl') else None,
"len": int(ip_layer.length) if hasattr(ip_layer, 'length') else None,
"protocol": pkt.highest_layer
}
# Add TCP/UDP header elements if available
if hasattr(pkt, 'tcp'):
record["srcport"] = int(pkt.tcp.srcport)
record["dstport"] = int(pkt.tcp.dstport)
record["seq"] = int(pkt.tcp.seq) if hasattr(pkt.tcp, 'seq') else None
packet_records.append(record)
except AttributeError:
continue
cap.close()
return packet_records
@staticmethod
def parse_firewall_log(log_line: str) -> Optional[dict]:
"""Regex-based parser for standard firewall syslog lines (e.g., iptables / pf)."""
# Example pattern: Jan 15 10:00:00 firewall kernel: [FW_ALLOW] SRC=192.168.1.50 DST=10.0.0.1 PROTO=TCP SPT=54321 DPT=443 TTL=64
pattern = r"(?P<timestamp>\w+\s+\d+\s+\d+:\d+:\d+).*SRC=(?P<src>\S+)\s+DST=(?P<dst>\S+).*PROTO=(?P<proto>\S+).*(?:SPT=(?P<spt>\d+))?.*(?:DPT=(?P<dpt>\d+))?.*(?:TTL=(?P<ttl>\d+))?"
match = re.search(pattern, log_line)
if match:
data = match.groupdict()
return {
"timestamp": datetime.now(), # Map to actual log year/time in production
"src_ip": data.get("src"),
"dst_ip": data.get("dst"),
"protocol": data.get("proto"),
"ttl": int(data["ttl"]) if data.get("ttl") else None
}
return None
3. Entity Resolution & Triangulation Engine
This core orchestrator unifies logs into IP objects and performs timestamp/header analysis (such as Time-to-Live tracking or Time-Difference-of-Arrival calculations across multiple vantage points) to map movement or positioning logic.
Python
class ObjectIdentityEngine:
def __init__(self, asset_registry: dict):
self.identities: Dict[str, NetworkObjectIdentity] = {}
# Pre-loaded static database mapping MAC or IP to known devices and GPS coordinates
# Format: { "AA:BB:CC:DD:EE:FF": {"name": "Warehouse Sensor 1", "gps": (37.7749, -122.4194)} }
self.asset_registry = asset_registry
def _get_or_create_identity(self, ip: str) -> NetworkObjectIdentity:
if ip not in self.identities:
self.identities[ip] = NetworkObjectIdentity(ip_address=ip)
return self.identities[ip]
def ingest_nmap(self, nmap_data: dict):
for ip, details in nmap_data.items():
obj = self._get_or_create_identity(ip)
if details["mac"]:
obj.mac_addresses.add(details["mac"])
self._enrich_gps(obj, details["mac"])
if details["hostname"]:
obj.hostnames.add(details["hostname"])
obj.open_ports.update(details["ports"])
obj.os_fingerprint = details["os"]
def ingest_packets(self, packets: List[dict]):
for pkt in packets:
for ip in [pkt["src_ip"], pkt["dst_ip"]]:
obj = self._get_or_create_identity(ip)
ts = pkt["timestamp"]
if not obj.first_seen or ts < obj.first_seen:
obj.first_seen = ts
if not obj.last_seen or ts > obj.last_seen:
obj.last_seen = ts
obj.packet_headers.append({
"timestamp": ts,
"ttl": pkt.get("ttl"),
"len": pkt.get("len"),
"peer": pkt["dst_ip"] if ip == pkt["src_ip"] else pkt["src_ip"]
})
def ingest_firewall_logs(self, log_lines: List[str]):
for line in log_lines:
parsed = LogIngestor.parse_firewall_log(line)
if not parsed:
continue
for ip in [parsed["src_ip"], parsed["dst_ip"]]:
if not ip: continue
obj = self._get_or_create_identity(ip)
obj.log_events.append(parsed)
def _enrich_gps(self, obj: NetworkObjectIdentity, mac: str):
"""Adds device name and GPS coordinates if the asset is known."""
if mac in self.asset_registry:
asset_info = self.asset_registry[mac]
obj.device_name = asset_info.get("name")
obj.gps_coordinates = asset_info.get("gps")
def triangulate_objects(self):
"""
Performs header and timestamp triangulation (e.g., using TTL variance
to estimate hop-distance changes or multi-sensor timestamp delta analysis).
"""
results = {}
for ip, obj in self.identities.items():
# Example heuristic: Calculate average TTL to spot routing/physical shifts
ttls = [h["ttl"] for h in obj.packet_headers if h.get("ttl") is not None]
avg_ttl = sum(ttls) / len(ttls) if ttls else None
# Timestamp clustering to see active bursts
activity_count = len(obj.packet_headers) + len(obj.log_events)
results[ip] = {
"device_name": obj.device_name,
"gps": obj.gps_coordinates,
"estimated_avg_ttl": avg_ttl,
"total_events": activity_count,
"open_ports_count": len(obj.open_ports)
}
return results
4. Putting It Together
Python
# 1. Setup known asset database with GPS mapping ("Device if known")
KNOWN_ASSETS = {
"00:11:22:33:44:55": {
"name": "Perimeter Gateway Router A",
"gps": (34.0522, -118.2437) # Los Angeles
},
"66:77:88:99:AA:BB": {
"name": "Internal Logistics Scanner",
"gps": (40.7128, -74.0060) # New York
}
}
engine = ObjectIdentityEngine(asset_registry=KNOWN_ASSETS)
# 2. Ingest Nmap scan results
nmap_results = LogIngestor.parse_nmap_xml("nmap_scan_output.xml")
engine.ingest_nmap(nmap_results)
# 3. Ingest Wireshark packet captures
pcap_packets = LogIngestor.parse_pcap("capture.pcap", max_packets=500)
engine.ingest_packets(pcap_packets)
# 4. Ingest Firewall logs
fw_logs = [
"Jan 15 10:01:00 fw01 kernel: SRC=192.168.1.50 DST=10.0.0.1 PROTO=TCP SPT=1234 DPT=80 TTL=64",
"Jan 15 10:02:15 fw01 kernel: SRC=192.168.1.50 DST=8.8.8.8 PROTO=UDP SPT=53 DPT=53 TTL=63"
]
engine.ingest_firewall_logs(fw_logs)
# 5. Run Triangulation & Identity Resolution
triangulated_data = engine.triangulate_objects()
for ip, data in triangulated_data.items():
print(f"IP: {ip} | Device: {data['device_name']} | GPS: {data['gps']} | Avg TTL: {data['estimated_avg_ttl']}")
Add rssi module and adjusters for drift and other wireless things
To handle wireless telemetry, account for multi-sensor clock skew, and extract advanced 802.11 attributes (like RSSI, BSSID, SSID, and channel), you can extend your pipeline with three new capabilities:
-
Clock Drift Adjuster: Normalizes timestamps across distributed sensors or log sources experiencing clock skew using linear regression or reference anchor points.
-
Wireless & RSSI Module: Captures radio-frequency metrics (RSSI in dBm, SNR, noise floor) and Wi-Fi metadata (BSSID, SSID, channel) from monitor-mode PCAPs via Radiotap headers.
-
RSSI-Based Distance Estimation: Translates signal strength into approximate physical distance using the Log-Distance Path Loss model to enhance triangulation.
1. Updated Data Model with Wireless & RSSI Attributes
Python
from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, List, Optional, Set, Tuple
@dataclass
class NetworkObjectIdentity:
ip_address: Optional[str] = None
mac_addresses: Set[str] = field(default_factory=set)
hostnames: Set[str] = field(default_factory=set)
open_ports: Dict[int, str] = field(default_factory=dict)
os_fingerprint: Optional[str] = None
first_seen: Optional[datetime] = None
last_seen: Optional[datetime] = None
# Telemetry and Triangulation data
packet_headers: List[Dict] = field(default_factory=list)
log_events: List[Dict] = field(default_factory=list)
# Device if known / GPS mapping
device_name: Optional[str] = None
gps_coordinates: Optional[Tuple[float, float]] = None
# --- New Wireless & RF Attributes ---
ssids_seen: Set[str] = field(default_factory=set)
bssids_seen: Set[str] = field(default_factory=set)
channels_used: Set[int] = field(default_factory=set)
rssi_readings: List[Dict] = field(default_factory=list) # {"timestamp": dt, "rssi": -65, "sensor_id": "sensor_1"}
2. Clock Drift Corrector
When collecting logs or packets from multiple independent sensors, micro-drift in hardware clocks causes timestamp misalignment, ruining time-difference-of-arrival (TDOA) or sequence triangulation. This module normalizes timestamps against a master reference source.
Python
import numpy as np
from datetime import timedelta
class ClockDriftCorrector:
def __init__(self, reference_sensor_id: str):
self.reference_sensor = reference_sensor_id
# Stores offset mappings per sensor: {sensor_id: time_delta_correction}
self.sensor_offsets: Dict[str, timedelta] = {}
def calculate_offset(self, sensor_id: str, sensor_sync_time: datetime, reference_sync_time: datetime):
"""Calculates and stores the offset between a sensor's clock and the master reference clock."""
offset = reference_sync_time - sensor_sync_time
self.sensor_offsets[sensor_id] = offset
def normalize_timestamp(self, sensor_id: str, timestamp: datetime) -> datetime:
"""Applies drift correction offset to a raw timestamp."""
offset = self.sensor_offsets.get(sensor_id, timedelta(0))
return timestamp + offset
3. Wireless Packet Ingestor (RSSI & Radiotap Extraction)
Using PyShark, you can parse 802.11 monitor-mode captures to extract radio parameters like signal strength (radiotap.dbm_antsignal), channel, and frame subtype.
Python
import pyshark
class WirelessLogIngestor:
@staticmethod
def parse_wireless_pcap(pcap_path: str, sensor_id: str, drift_corrector: Optional[ClockDriftCorrector] = None) -> List[dict]:
"""Parses wireless PCAP files for RSSI, BSSID, SSID, and channel telemetry."""
# Using display filters or raw capture depending on setup
cap = pyshark.FileCapture(pcap_path, only_summaries=False)
wireless_records = []
for pkt in cap:
try:
# Check for Radiotap header (RSSI / signal strength)
rssi = None
if hasattr(pkt, 'radiotap') and hasattr(pkt.radiotap, 'dbm_antsignal'):
rssi = int(pkt.radiotap.dbm_antsignal)
# Extract timestamp
ts = datetime.fromtimestamp(float(pkt.sniff_timestamp))
if drift_corrector:
ts = drift_corrector.normalize_timestamp(sensor_id, ts)
# Extract 802.11 wireless fields if present
bssid = None
ssid = None
channel = None
if hasattr(pkt, 'wlan'):
bssid = getattr(pkt.wlan, 'ta', None) or getattr(pkt.wlan, 'bssid', None)
if hasattr(pkt.wlan, 'ssid'):
ssid = pkt.wlan.ssid
if hasattr(pkt, 'wlan_radio'):
channel = int(getattr(pkt.wlan_radio, 'channel', 0)) or None
# Extract source IP if Layer 3 payload exists
src_ip = pkt.ip.src if hasattr(pkt, 'ip') else None
if rssi is not None or bssid is not None:
wireless_records.append({
"timestamp": ts,
"sensor_id": sensor_id,
"src_ip": src_ip,
"bssid": bssid,
"ssid": ssid,
"channel": channel,
"rssi": rssi
})
except AttributeError:
continue
cap.close()
return wireless_records
4. Triangulation Engine with RSSI Path-Loss Model
This extension calculates estimated distances from RSSI values using the Log-Distance Path Loss Model:
Distance=1010×nMeasured Power (RSSI)−TxPower (Reference at 1m)
(where n is the path loss exponent, typically 2.0 to 4.0 depending on environment obstructions).
Python
class AdvancedTriangulationEngine:
@staticmethod
def rssi_to_distance(rssi: int, tx_power: int = -50, n: float = 2.5) -> float:
"""Estimates distance in meters from RSSI given a reference Tx power and path loss exponent."""
if rssi >= 0:
return 0.0
return 10.0 ** ((tx_power - rssi) / (10 * n))
def process_object_rf(self, obj: NetworkObjectIdentity):
"""Analyzes RSSI readings to compute average signal strength and estimated physical proximity."""
if not obj.rssi_readings:
return {"avg_rssi": None, "est_distance_meters": None}
rssi_vals = [r["rssi"] for r in obj.rssi_readings if r["rssi"] is not None]
avg_rssi = sum(rssi_vals) / len(rssi_vals) if rssi_vals else None
est_distance = self.rssi_to_distance(avg_rssi) if avg_rssi else None
return {
"avg_rssi": avg_rssi,
"est_distance_meters": est_distance,
"channels_observed": list(obj.channels_used),
"bssids_associated": list(obj.bssids_seen)
}
5. Unified Ingestion Integration
Python
# Initialize Drift Corrector and register sensor offset
drift_mgr = ClockDriftCorrector(reference_sensor_id="sensor_alpha")
# Suppose sensor_beta's clock is running 1.5 seconds behind sensor_alpha
drift_mgr.sensor_offsets["sensor_beta"] = timedelta(seconds=1.5)
# Parse wireless PCAP containing radiotap and 802.11 metadata
wireless_data = WirelessLogIngestor.parse_wireless_pcap(
"wireless_monitor.pcap",
sensor_id="sensor_beta",
drift_corrector=drift_mgr
)
# Populate network object identities with wireless metrics
engine = ObjectIdentityEngine(asset_registry={}) # Assuming base engine from previous turn
for record in wireless_data:
# Key by IP if available, or fall back to BSSID/MAC
key = record["src_ip"] if record["src_ip"] else record["bssid"]
if not key:
continue
obj = engine._get_or_create_identity(key)
if record["ssid"]:
obj.ssids_seen.add(record["ssid"])
if record["bssid"]:
obj.bssids_seen.add(record["bssid"])
if record["channel"]:
obj.channels_used.add(record["channel"])
if record["rssi"] is not None:
obj.rssi_readings.append({
"timestamp": record["timestamp"],
"rssi": record["rssi"],
"sensor_id": record["sensor_id"]
})
# Compute RF Triangulation / Distance metrics
rf_engine = AdvancedTriangulationEngine()
for ip, obj in engine.identities.items():
rf_stats = rf_engine.process_object_rf(obj)
print(f"Target: {ip} | Avg RSSI: {rf_stats['avg_rssi']} dBm | Est. Distance: {rf_stats['est_distance_meters']}m | Channels: {rf_stats['channels_observed']}")
Want me to add trilateration math (using 3 or more sensor GPS coordinates and RSSI distances) to pinpoint exact lat/long coordinates?
Show me how to implement multi-sensor trilateration using RSSI distance estimates and known sensor GPS coordinates to calculate a target’s exact coordinates.
To pinpoint a target’s exact physical coordinates when multiple sensors detect its transmissions, you need a multilateration / trilateration solver.
Because real-world radio frequency (RF) signals suffer from multipath interference, obstacles, and noise, distance estimates derived from RSSI rarely intersect at a single clean point. Instead, we use an over-determined system solved via Nonlinear Least Squares (using scipy.optimize.least_squares) to find the most probable coordinates.
1. Geographic Coordinate Transformation (GPS to Local Cartesian)
Because GPS coordinates (latitude and longitude in degrees) do not map linearly to physical distance across the globe, we convert sensor GPS coordinates and distances into a local 2D Cartesian coordinate system (measured in meters relative to a reference origin) before solving.
Python
import numpy as np
from scipy.optimize import least_squares
from typing import List, Tuple, Dict
class TrilaterationEngine:
@staticmethod
def gps_to_cartesian(lat: float, lon: float, ref_lat: float, ref_lon: float) -> Tuple[float, float]:
"""Converts GPS lat/lon to local X,Y meters relative to a reference point."""
# Approximate meters per degree
m_per_lat = 111132.92 - 559.82 * np.cos(2 * np.radians(ref_lat)) + 1.17 * np.cos(4 * np.radians(ref_lat))
m_per_lon = 111412.84 * np.cos(np.radians(ref_lat)) - 93.5 * np.cos(3 * np.radians(ref_lat))
y = (lat - ref_lat) * m_per_lat
x = (lon - ref_lon) * m_per_lon
return x, y
@staticmethod
def cartesian_to_gps(x: float, y: float, ref_lat: float, ref_lon: float) -> Tuple[float, float]:
"""Converts local X,Y meters back to GPS lat/lon."""
m_per_lat = 111132.92 - 559.82 * np.cos(2 * np.radians(ref_lat)) + 1.17 * np.cos(4 * np.radians(ref_lat))
m_per_lon = 111412.84 * np.cos(np.radians(ref_lat)) - 93.5 * np.cos(3 * np.radians(ref_lat))
lat = ref_lat + (y / m_per_lat)
lon = ref_lon + (x / m_per_lon)
return lat, lon
2. Nonlinear Least Squares Trilateration Solver
The solver minimizes the residual error between the measured RSSI distance and the geometric distance from candidate target coordinates (x,y) to each sensor (xi,yi):
Python
def calculate_target_position(self, sensor_readings: List[Dict]) -> Optional[Tuple[float, float]]:
"""
Calculates the target's GPS coordinates given a list of sensor data:
sensor_readings = [
{"gps": (lat, lon), "distance": distance_in_meters},
...
]
Requires at least 3 sensors for reliable 2D positioning.
"""
if len(sensor_readings) < 3:
return None # Trilateration requires minimum 3 points
# Use the first sensor as the local Cartesian origin (0,0)
ref_lat = sensor_readings[0]["gps"][0]
ref_lon = sensor_readings[0]["gps"][1]
sensor_coords = []
distances = []
for item in sensor_readings:
lat, lon = item["gps"]
dist = item["distance"]
x, y = self.gps_to_cartesian(lat, lon, ref_lat, ref_lon)
sensor_coords.append((x, y))
distances.append(dist)
sensor_coords = np.array(sensor_coords)
distances = np.array(distances)
# Objective function for least squares optimization
def cost_function(pos):
# pos = [x, y] of target
calculated_distances = np.sqrt(np.sum((sensor_coords - pos) ** 2, axis=1))
return calculated_distances - distances
# Initial guess: centroid (average) of all participating sensors
initial_guess = np.mean(sensor_coords, axis=0)
# Run optimization
result = least_squares(cost_function, initial_guess, method='lm')
if result.success:
target_x, target_y = result.x
target_lat, target_lon = self.cartesian_to_gps(target_x, target_y, ref_lat, ref_lon)
return (target_lat, target_lon)
return None
4. End-to-End Pipeline Integration
Here is how you connect your multi-sensor telemetry into the trilateration solver:
Python
# 1. Gather readings from 3 distinct monitor-mode sensors picking up the same target MAC/IP
active_readings = [
{
"sensor_id": "sensor_alpha",
"gps": (34.0522, -118.2437), # Sensor 1 GPS (Los Angeles)
"rssi": -62 # Measured RSSI from Sensor 1
},
{
"sensor_id": "sensor_beta",
"gps": (34.0600, -118.2500), # Sensor 2 GPS
"rssi": -68 # Measured RSSI from Sensor 2
},
{
"sensor_id": "sensor_gamma",
"gps": (34.0450, -118.2350), # Sensor 3 GPS
"rssi": -55 # Measured RSSI from Sensor 3
}
]
# 2. Convert RSSI to estimated distance using path loss model (n=2.5, tx_power=-50)
engine = TrilaterationEngine()
processed_readings = []
for reading in active_readings:
# Using our path loss formula from previous turn
distance = 10.0 ** ((-50 - reading["rssi"]) / (10 * 2.5))
processed_readings.append({
"gps": reading["gps"],
"distance": distance
})
# 3. Compute exact target coordinates via Trilateration
target_gps = engine.calculate_target_position(processed_readings)
if target_gps:
print(f"Target Triangulated Successfully!")
print(f"Estimated Latitude: {target_gps[0]:.6f}")
print(f"Estimated Longitude: {target_gps[1]:.6f}")
else:
print("Trilateration failed: Insufficient sensor data or poor geometric layout.")
Because RSSI readings fluctuate due to multipath propagation, fading, and environmental obstacles, raw trilateration results often appear “jittery” or jump around erratically.
A Kalman Filter solves this by combining a physical motion model (e.g., constant velocity) with noisy sensor measurements to estimate the true state (position and velocity) over time.
1. Pure NumPy 2D Kalman Filter Implementation
This implementation uses a Constant Velocity Model tracking 4 state variables: Position X (p_x), Position Y (p_y), Velocity X (v_x), and Velocity Y (v_y).
Python
import numpy as np
from typing import Tuple, Optional
class KalmanTracker2D:
def __init__(self, dt: float = 1.0, process_noise: float = 1e-2, measurement_noise: float = 0.5):
"""
Initializes a 2D Kalman filter for tracking coordinates over time.
:param dt: Time step between updates (seconds)
:param process_noise: Uncertainty in the motion model (higher = more agile/less smooth)
:param measurement_noise: Uncertainty in trilateration measurements (higher = trust model more)
"""
self.dt = dt
# State Vector [px, py, vx, vy]
self.x = np.zeros((4, 1))
# State Covariance Matrix (uncertainty of our estimate)
self.P = np.eye(4) * 1000.0 # High initial uncertainty
# State Transition Matrix (F) - Physics model for constant velocity
self.F = np.array([
[1, 0, dt, 0],
[0, 1, 0, dt],
[0, 0, 1, 0],
[0, 0, 0, 1]
])
# Measurement Matrix (H) - We only measure position (px, py), not velocity
self.H = np.array([
[1, 0, 0, 0],
[0, 1, 0, 0]
])
# Process Noise Covariance (Q)
self.Q = np.eye(4) * process_noise
# Measurement Noise Covariance (R)
self.R = np.eye(2) * measurement_noise
self.initialized = False
def initialize(self, initial_pos: Tuple[float, float]):
"""Initializes the filter state with the first valid trilateration point."""
self.x[0, 0] = initial_pos[0]
self.x[1, 0] = initial_pos[1]
self.x[2, 0] = 0.0 # Initial velocity X
self.x[3, 0] = 0.0 # Initial velocity Y
self.initialized = True
def predict(self):
"""Predicts the next state based on the motion model."""
if not self.initialized:
return
# State prediction: x = F * x
self.x = self.F @ self.x
# Covariance prediction: P = F * P * F^T + Q
self.P = self.F @ self.P @ self.F.T + self.Q
def update(self, measurement: Tuple[float, float]) -> Tuple[float, float]:
"""
Updates the filter with a new noisy trilateration measurement (x, y).
Returns the smoothed (x, y) coordinates.
"""
z = np.array([[measurement[0]], [measurement[1]]])
if not self.initialized:
self.initialize(measurement)
return measurement
# 1. Prediction step
self.predict()
# 2. Innovation (Residual): y = z - H * x
y = z - (self.H @ self.x)
# 3. Innovation Covariance: S = H * P * H^T + R
S = (self.H @ self.P @ self.H.T) + self.R
# 4. Kalman Gain: K = P * H^T * inv(S)
K = self.P @ self.H.T @ np.linalg.inv(S)
# 5. State Update: x = x + K * y
self.x = self.x + (K @ y)
# 6. Covariance Update: P = (I - K * H) * P
I = np.eye(len(self.x))
self.P = (I - (K @ self.H)) @ self.P
return float(self.x[0, 0]), float(self.x[1, 0])
2. Integrating the Kalman Filter into the Network Object Model
We can embed the Kalman tracker directly into your object identity architecture so that every moving target automatically smooths its trajectory across time.
Python
from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, List, Optional, Set, Tuple
@dataclass
class TrackedNetworkObject:
ip_address: Optional[str] = None
mac_addresses: Set[str] = field(default_factory=set)
device_name: Optional[str] = None
# Kalman Tracker instance for local Cartesian coordinates (X, Y meters)
kalman_tracker: KalmanTracker2D = field(default_factory=lambda: KalmanTracker2D(dt=1.0))
# History of smoothed positions
position_history: List[Dict] = field(default_factory=list)
3. End-to-End Tracking Pipeline Example
Here is how raw, jittery trilateration outputs flow through the Kalman filter to produce smooth, reliable movement paths:
Python
# Reference origin for local Cartesian conversion (same as previous turn)
REF_LAT = 34.0522
REF_LON = -118.2437
def gps_to_cartesian(lat, lon):
m_per_lat = 111132.92
m_per_lon = 111412.84 * np.cos(np.radians(REF_LAT))
return (lon - REF_LON) * m_per_lon, (lat - REF_LAT) * m_per_lat
def cartesian_to_gps(x, y):
m_per_lat = 111132.92
m_per_lon = 111412.84 * np.cos(np.radians(REF_LAT))
return REF_LAT + (y / m_per_lat), REF_LON + (x / m_per_lon)
# Simulate a moving device over 5 time steps with noisy trilateration results
# (Notice how the raw trilateration jumps around due to RSSI jitter)
raw_noisy_trilateration_sequence = [
{"timestamp": "10:00:01", "lat": 34.0530, "lon": -118.2440},
{"timestamp": "10:00:02", "lat": 34.0545, "lon": -118.2425}, # Sudden RSSI spike / jump error
{"timestamp": "10:00:03", "lat": 34.0540, "lon": -118.2450}, # Another jump error
{"timestamp": "10:00:04", "lat": 34.0542, "lon": -118.2432},
{"timestamp": "10:00:05", "lat": 34.0550, "lon": -118.2435}
]
target = TrackedNetworkObject(ip_address="192.168.1.105", device_name="Mobile Asset Alpha")
print(f"{'Timestamp':<10} | {'Raw Trilateration GPS':<25} | {'Kalman Smoothed GPS':<25}")
print("-" * 68)
for reading in raw_noisy_trilateration_sequence:
# 1. Convert raw GPS measurement to local Cartesian meters
mx, my = gps_to_cartesian(reading["lat"], reading["lon"])
# 2. Feed into Kalman Filter to get smoothed meters
smooth_x, smooth_y = target.kalman_tracker.update((mx, my))
# 3. Convert smoothed meters back to GPS
smooth_lat, smooth_lon = cartesian_to_gps(smooth_x, smooth_y)
raw_str = f"({reading['lat']:.4f}, {reading['lon']:.4f})"
smooth_str = f"({smooth_lat:.4f}, {smooth_lon:.4f})"
print(f"{reading['timestamp']:<10} | {raw_str:<25} | {smooth_str:<25}")
To visualize your smoothed Kalman filter trajectories in GIS software (like QGIS, ArcGIS) or web mapping libraries (like Leaflet.js, Mapbox), you can export your object histories into standard GeoJSON format.
GeoJSON natively supports Point geometries (for individual coordinate fixes with timestamps) and LineString geometries (to render the continuous movement path connecting the points). Note that GeoJSON strictly expects coordinate order as [longitude, latitude].
1. Python GeoJSON Exporter
This module takes your tracked network objects and converts their smoothed positions into a standards-compliant GeoJSON FeatureCollection.
Python
import json
from typing import List, Dict
class GeoJSONExporter:
@staticmethod
def export_object_trajectory(target_object, output_filepath: str = "trajectory.geojson"):
"""
Converts a TrackedNetworkObject's position history into a GeoJSON FeatureCollection.
Includes individual timestamped points and a connecting LineString path.
"""
features = []
path_coordinates = []
# Iterate through the recorded position history
for record in target_object.position_history:
lat = record["lat"]
lon = record["lon"]
timestamp = record.get("timestamp", "")
# GeoJSON coordinates order: [longitude, latitude]
coord = [lon, lat]
path_coordinates.append(coord)
# 1. Create a Point feature for each timestamped fix
point_feature = {
"type": "Feature",
"geometry": {
"type": "Point",
"coordinates": coord
},
"properties": {
"device_name": target_object.device_name,
"ip_address": target_object.ip_address,
"timestamp": timestamp,
"smoothed": True
}
}
features.append(point_feature)
# 2. Create a LineString feature connecting the points to show movement path
if len(path_coordinates) > 1:
line_feature = {
"type": "Feature",
"geometry": {
"type": "LineString",
"coordinates": path_coordinates
},
"properties": {
"device_name": target_object.device_name,
"ip_address": target_object.ip_address,
"path_type": "Kalman-Smoothed Trajectory"
}
}
features.append(line_feature)
# Build final FeatureCollection
geojson_collection = {
"type": "FeatureCollection",
"features": features
}
# Save to file
with open(output_filepath, 'w') as f:
json.dump(geojson_collection, f, indent=4)
print(f"GeoJSON successfully exported to {output_filepath}")
return geojson_collection
2. Updating Your Tracker Loop to Capture History
To populate the position_history array needed for the exporter, update your tracking loop to save each smoothed coordinate:
Python
# Assuming target and raw_noisy_trilateration_sequence from the previous turn
REF_LAT = 34.0522
REF_LON = -118.2437
def gps_to_cartesian(lat, lon):
m_per_lat = 111132.92
m_per_lon = 111412.84 * np.cos(np.radians(REF_LAT))
return (lon - REF_LON) * m_per_lon, (lat - REF_LAT) * m_per_lat
def cartesian_to_gps(x, y):
m_per_lat = 111132.92
m_per_lon = 111412.84 * np.cos(np.radians(REF_LAT))
return REF_LAT + (y / m_per_lat), REF_LON + (x / m_per_lon)
# Run tracking loop and log to position_history
for reading in raw_noisy_trilateration_sequence:
mx, my = gps_to_cartesian(reading["lat"], reading["lon"])
smooth_x, smooth_y = target.kalman_tracker.update((mx, my))
smooth_lat, smooth_lon = cartesian_to_gps(smooth_x, smooth_y)
# Store in history
target.position_history.append({
"timestamp": reading["timestamp"],
"lat": smooth_lat,
"lon": smooth_lon
})
# Export to GeoJSON file
GeoJSONExporter.export_object_trajectory(target, "mobile_asset_alpha.geojson")
3. Visualizing in Leaflet.js (Web Frontend)
Once exported, you can load your mobile_asset_alpha.geojson file directly into a Leaflet map in just a few lines of JavaScript:
HTML
<!DOCTYPE html>
<html>
<head>
<title>Network Object Trajectory Map</title>
<link rel="stylesheet" href="https://unpkg.com/leaflet/dist/leaflet.css" />
<style>#map { height: 600px; width: 100%; }</style>
</head>
<body>
<div id="map"></div>
<script src="https://unpkg.com/leaflet/dist/leaflet.js"></script>
<script>
// Initialize map centered on your reference point
var map = L.map('map').setView([34.0522, -118.2437], 15);
// Add OpenStreetMap tiles
L.tileLayer('https://{s}.tile.openstreetmap.org/{z}/{x}/{y}.png', {
maxZoom: 19,
attribution: '© OpenStreetMap contributors'
}).addTo(map);
// Load and display the generated GeoJSON file
fetch('mobile_asset_alpha.geojson')
.then(response => response.json())
.then(data => {
L.geoJSON(data, {
onEachFeature: function (feature, layer) {
if (feature.properties && feature.properties.timestamp) {
layer.bindPopup("<b>Device:</b> " + feature.properties.device_name +
"<br><b>IP:</b> " + feature.properties.ip_address +
"<br><b>Time:</b> " + feature.properties.timestamp);
}
}
}).addTo(map);
});
</script>
</body>
</html>
To secure your intelligence pipeline against physical perimeter breaches or erratic asset behavior (such as asset theft, signal spoofing, or runaway speed anomalies), you can add an Anomaly Detection Engine.
This engine performs two critical checks on every updated position fix:
-
Velocity Threshold Check: Uses the Haversine formula to calculate the distance traveled between timestamps, alerting if speed exceeds physical limits (e.g., sprinting humans or vehicles).
-
Geofence Boundary Check: Verifies if the object has crossed a defined circular or polygonal perimeter zone.
1. Geospatial Math & Anomaly Detector Module
Python
import numpy as np
from datetime import datetime
from typing import Tuple, Dict, Optional
class GeofenceAnomalyDetector:
def __init__(self, max_speed_mps: float = 12.0, geofence_center: Optional[Tuple[float, float]] = None, geofence_radius_m: float = 300.0):
"""
:param max_speed_mps: Maximum allowed speed in meters per second (12 m/s ≈ 43 km/h or 27 mph)
:param geofence_center: (Latitude, Longitude) of the secure perimeter center
:param geofence_radius_m: Maximum allowed distance from geofence center in meters
"""
self.max_speed_mps = max_speed_mps
self.geofence_center = geofence_center
self.geofence_radius_m = geofence_radius_m
@staticmethod
def haversine_distance(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
"""Calculates the great-circle distance between two GPS points in meters."""
R = 6371000 # Earth radius in meters
phi1, phi2 = np.radians(lat1), np.radians(lat2)
d_phi = np.radians(lat2 - lat1)
d_lambda = np.radians(lon2 - lon1)
a = np.sin(d_phi / 2.0)**2 + np.cos(phi1) * np.cos(phi2) * np.sin(d_lambda / 2.0)**2
c = 2 * np.arctan2(np.sqrt(a), np.sqrt(1 - a))
return R * c
def evaluate_position(self, prev_pos: Optional[Dict], curr_pos: Dict) -> Dict:
"""
Evaluates current position against velocity limits and geofence boundaries.
curr_pos format: {"timestamp": datetime, "lat": float, "lon": float}
"""
alerts = []
speed = 0.0
# 1. Velocity / Acceleration Check
if prev_pos:
dist = self.haversine_distance(prev_pos["lat"], prev_pos["lon"], curr_pos["lat"], curr_pos["lon"])
# Parse timestamp if it's stored as a string
t1 = prev_pos["timestamp"] if isinstance(prev_pos["timestamp"], datetime) else datetime.fromisoformat(prev_pos["timestamp"])
t2 = curr_pos["timestamp"] if isinstance(curr_pos["timestamp"], datetime) else datetime.fromisoformat(curr_pos["timestamp"])
time_delta = (t2 - t1).total_seconds()
if time_delta > 0:
speed = dist / time_delta
if speed > self.max_speed_mps:
alerts.append({
"type": "SPEED_ANOMALY",
"severity": "HIGH",
"message": f"Excessive speed detected: {speed:.2f} m/s (Limit: {self.max_speed_mps} m/s)"
})
# 2. Geofence Boundary Check
if self.geofence_center:
dist_from_center = self.haversine_distance(
self.geofence_center[0], self.geofence_center[1],
curr_pos["lat"], curr_pos["lon"]
)
if dist_from_center > self.geofence_radius_m:
alerts.append({
"type": "GEOFENCE_BREACH",
"severity": "CRITICAL",
"message": f"Asset exited perimeter! Distance from center: {dist_from_center:.1f}m (Allowed: {self.geofence_radius_m}m)"
})
return {
"current_speed_mps": round(speed, 2),
"alerts": alerts,
"is_safe": len(alerts) == 0
}
2. Integrating Anomaly Detection into the Tracking Loop
You can run this engine immediately after obtaining your Kalman-filtered smoothed coordinates:
Python
# Setup Geofence Center (e.g., Warehouse headquarters) and Max Speed Limit
detector = GeofenceAnomalyDetector(
max_speed_mps=10.0, # ~22 mph limit
geofence_center=(34.0522, -118.2437), # Headquarters GPS
geofence_radius_m=400.0 # 400 meter secure radius
)
# Simulate tracking history with an intentional speed jump and geofence exit
tracking_fixes = [
{"timestamp": datetime(2026, 9, 18, 10, 0, 1), "lat": 34.0525, "lon": -118.2435},
{"timestamp": datetime(2026, 9, 18, 10, 0, 2), "lat": 34.0530, "lon": -118.2440},
{"timestamp": datetime(2026, 9, 18, 10, 0, 3), "lat": 34.0650, "lon": -118.2550} # Teleport / speed & breach jump!
]
print(f"{'Timestamp':<10} | {'Speed (m/s)':<12} | {'Status':<10} | {'Triggered Alerts'}")
print("-" * 65)
prev_fix = None
for fix in tracking_fixes:
evaluation = detector.evaluate_position(prev_fix, fix)
status = "SECURE" if evaluation["is_safe"] else "ALERT!"
alert_msgs = ", ".join([a["message"] for a in evaluation["alerts"]]) if evaluation["alerts"] else "None"
ts_str = fix["timestamp"].strftime("%H:%M:%S")
print(f"{ts_str:<10} | {evaluation['current_speed_mps']:<12} | {status:<10} | {alert_msgs}")
prev_fix = fix
3. Enhancing GeoJSON Export with Alert Properties
To ensure security operations centers (SOC) or GIS mapping dashboards can immediately see these violations, flag the exported GeoJSON features when an anomaly occurs:
Python
class AlertGeoJSONExporter:
@staticmethod
def export_with_alerts(target_object, detector: GeofenceAnomalyDetector, output_filepath: str = "secure_trajectory.geojson"):
features = []
prev_fix = None
for record in target_object.position_history:
# Evaluate anomaly rules per point
eval_result = detector.evaluate_position(prev_fix, record)
coord = [record["lon"], record["lat"]]
point_feature = {
"type": "Feature",
"geometry": {
"type": "Point",
"coordinates": coord
},
"properties": {
"device_name": target_object.device_name,
"timestamp": record.get("timestamp"),
"speed_mps": eval_result["current_speed_mps"],
"is_safe": eval_result["is_safe"],
"alerts": eval_result["alerts"]
}
}
features.append(point_feature)
prev_fix = record
geojson_collection = {"type": "FeatureCollection", "features": features}
with open(output_filepath, 'w') as f:
import json
json.dump(geojson_collection, f, indent=4)
print(f"GeoJSON with integrated security alerts exported to {output_filepath}")
To bring the interactive visualizer to life and tie together all the components we’ve built—RSSI distance estimation, multi-sensor trilateration, Kalman filter smoothing, and geofence boundary checks—here is a complete, self-contained single-file web application (visualizer.html).
You can save this code to an HTML file and open it in any browser to interactively drag or adjust sensor ranges, observe raw vs. Kalman-smoothed positions, and trigger geofence breach alerts in real time.
Interactive Visualizer Source Code (visualizer.html)
HTML
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Network Intelligence Pipeline - Trilateration & Kalman Visualizer</title>
<style>
body { font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, sans-serif; background: #0f172a; color: #f8fafc; margin: 0; padding: 20px; }
.container { max-width: 1200px; margin: auto; display: grid; grid-template-columns: 1fr 380px; gap: 20px; }
@media (max-width: 900px) { .container { grid-template-columns: 1fr; } }
card { background: #1e293b; border-radius: 12px; padding: 20px; box-shadow: 0 10px 15px -3px rgba(0,0,0,0.3); display: block; }
h1 { font-size: 20px; margin-top: 0; color: #38bdf8; }
h2 { font-size: 16px; margin-top: 0; border-bottom: 1px solid #334155; padding-bottom: 8px; }
canvas { background: #090d16; border-radius: 8px; width: 100%; height: 500px; display: block; border: 1px solid #334155; }
.control-group { margin-bottom: 16px; }
label { display: flex; justify-content: space-between; font-size: 13px; margin-bottom: 6px; color: #cbd5e1; }
input[type=range] { width: 100%; accent-color: #38bdf8; cursor: pointer; }
.telemetry { background: #0f172a; border-radius: 8px; padding: 12px; font-family: monospace; font-size: 12px; border: 1px solid #334155; }
.row { display: flex; justify-content: space-between; margin-bottom: 6px; }
.badge { padding: 2px 8px; border-radius: 4px; font-weight: bold; font-size: 11px; }
.badge.secure { background: #065f46; color: #34d399; }
.badge.alert { background: #7f1d1d; color: #f87171; }
button { background: #38bdf8; color: #0f172a; border: none; padding: 10px 16px; font-weight: bold; border-radius: 6px; cursor: pointer; width: 100%; transition: background 0.2s; }
button:hover { background: #7dd3fc; }
</style>
</head>
<body>
<div class="container">
<card>
<h1>2D RF Trilateration & Kalman Tracking Map</h1>
<canvas id="simCanvas"></canvas>
</card>
<div style="display: flex; flex-direction: column; gap: 20px;">
<card>
<h2>Sensor RSSI / Distance Controls</h2>
<div class="control-group">
<label><span>Sensor A Distance (RSSI)</span><span id="distAVal">120m</span></label>
<input type="range" id="distA" min="20" max="350" value="120">
</div>
<div class="control-group">
<label><span>Sensor B Distance (RSSI)</span><span id="distBVal">180m</span></label>
<input type="range" id="distB" min="20" max="350" value="180">
</div>
<div class="control-group">
<label><span>Sensor C Distance (RSSI)</span><span id="distCVal">150m</span></label>
<input type="range" id="distC" min="20" max="350" value="150">
</div>
<div class="control-group">
<label><span>Kalman Filter Smoothing (Process Noise)</span><span id="noiseVal">0.05</span></label>
<input type="range" id="kalmanNoise" min="0.001" max="0.5" step="0.005" value="0.05">
</div>
<button id="resetBtn">Reset Target Position</button>
</card>
<card>
<h2>Live Object Telemetry</h2>
<div class="telemetry">
<div class="row"><span>Target IP:</span> <strong>192.168.1.105</strong></div>
<div class="row"><span>Raw Trilateration:</span> <span id="rawPos">(0, 0)</span></div>
<div class="row"><span>Kalman Smoothed:</span> <span id="smoothPos" style="color: #38bdf8;">(0, 0)</span></div>
<div class="row"><span>Distance from HQ:</span> <span id="hqDist">0 m</span></div>
<div class="row" style="margin-top: 10px; align-items: center;">
<span>Perimeter Status:</span>
<span id="statusBadge" class="badge secure">SECURE</span>
</div>
</div>
</card>
</div>
</div>
<script>
const canvas = document.getElementById('simCanvas');
const ctx = canvas.getContext('2d');
// Resize canvas high-DPI support
function resizeCanvas() {
canvas.width = canvas.clientWidth * window.devicePixelRatio;
canvas.height = canvas.clientHeight * window.devicePixelRatio;
ctx.scale(window.devicePixelRatio, window.devicePixelRatio);
}
window.addEventListener('resize', resizeCanvas);
resizeCanvas();
// Map & Sensor definitions (in canvas pixel coordinates)
const sensors = [
{ id: 'Sensor A', x: 120, y: 100, color: '#ef4444' },
{ id: 'Sensor B', x: 380, y: 120, color: '#f59e0b' },
{ id: 'Sensor C', x: 250, y: 380, color: '#10b981' }
];
const geofence = { x: 250, y: 230, radius: 180 }; // Secure perimeter
// Simple 2D Kalman Filter implementation in JS
class SimpleKalman2D {
constructor(processNoise = 0.05) {
this.x = 250; // pos x
this.y = 230; // pos y
this.vx = 0; // vel x
this.vy = 0; // vel y
this.p = 100; // uncertainty
this.q = processNoise; // process noise
this.r = 4.0; // measurement noise
}
update(measX, measY, qVal) {
this.q = qVal;
// Predict
this.x += this.vx;
this.y += this.vy;
this.p += this.q;
// Update
let k = this.p / (this.p + this.r);
this.x += k * (measX - this.x);
this.y += k * (measY - this.y);
this.p = (1 - k) * this.p;
return { x: this.x, y: this.y };
}
}
let kalman = new KalmanTracker2D = new SimpleKalman2D(0.05);
let rawHistory = [];
let smoothHistory = [];
// Least Squares Trilateration Solver in JS
function trilaterate(d1, d2, d3) {
let p1 = sensors[0], p2 = sensors[1], p3 = sensors[2];
// Iterative Gauss-Newton / Least Squares approximation
let x = 250, y = 230;
for (let i = 0; i < 10; i++) {
let r1 = Math.hypot(x - p1.x, y - p1.y);
let r2 = Math.hypot(x - p2.x, y - p2.y);
let r3 = Math.hypot(x - p3.x, y - p3.y);
let f1 = r1 - d1;
let f2 = r2 - d2;
let f3 = r3 - d3;
let j11 = (x - p1.x) / (r1 || 1), j12 = (y - p1.y) / (r1 || 1);
let j21 = (x - p2.x) / (r2 || 1), j22 = (y - p2.y) / (r2 || 1);
let j31 = (x - p3.x) / (r3 || 1), j32 = (y - p3.y) / (r3 || 1);
// Approximate step update
let dx = -(j11 * f1 + j21 * f2 + j31 * f3) * 0.5;
let dy = -(j12 * f1 + j22 * f2 + j32 * f3) * 0.5;
x += dx;
y += dy;
}
return { x: isNaN(x) ? 250 : x, y: isNaN(y) ? 230 : y };
}
// UI Elements
const distA = document.getElementById('distA');
const distB = document.getElementById('distB');
const distC = document.getElementById('distC');
const kalmanNoise = document.getElementById('kalmanNoise');
function updateSimulation() {
let d1 = parseFloat(distA.value);
let d2 = parseFloat(distB.value);
let d3 = parseFloat(distC.value);
let noise = parseFloat(kalmanNoise.value);
document.getElementById('distAVal').innerText = d1 + 'm';
document.getElementById('distBVal').innerText = d2 + 'm';
document.getElementById('distCVal').innerText = d3 + 'm';
document.getElementById('noiseVal').innerText = noise;
// 1. Calculate raw noisy trilateration intersection
let raw = trilaterate(d1, d2, d3);
// 2. Pass through Kalman Filter
let smooth = kalman.update(raw.x, raw.y, noise);
// Track history for path rendering
rawHistory.push(raw);
smoothHistory.push(smooth);
if (rawHistory.length > 50) { rawHistory.shift(); smoothHistory.shift(); }
// Check Geofence Breach
let distFromHQ = Math.hypot(smooth.x - geofence.x, smooth.y - geofence.y);
let isBreached = distFromHQ > geofence.radius;
// Update Telemetry UI
document.getElementById('rawPos').innerText = `(${Math.round(raw.x)}, ${Math.round(raw.y)})`;
document.getElementById('smoothPos').innerText = `(${Math.round(smooth.x)}, ${Math.round(smooth.y)})`;
document.getElementById('hqDist').innerText = Math.round(distFromHQ) + ' m';
const badge = document.getElementById('statusBadge');
if (isBreached) {
badge.className = 'badge alert';
badge.innerText = 'GEOFENCE BREACH!';
} else {
badge.className = 'badge secure';
badge.innerText = 'SECURE';
}
// Render Canvas
renderCanvas(raw, smooth, [d1, d2, d3], isBreached);
}
function renderCanvas(raw, smooth, distances, isBreached) {
ctx.clearRect(0, 0, canvas.clientWidth, canvas.clientHeight);
// Draw Geofence Perimeter
ctx.beginPath();
ctx.arc(geofence.x, geofence.y, geofence.radius, 0, 2 * Math.PI);
ctx.fillStyle = isBreached ? 'rgba(127, 29, 29, 0.15)' : 'rgba(15, 23, 42, 0.6)';
ctx.fill();
ctx.lineWidth = 2;
ctx.strokeStyle = isBreached ? '#ef4444' : '#334155';
ctx.setLineDash([6, 6]);
ctx.stroke();
ctx.setLineDash([]);
// Draw Sensor RSSI Distance Circles
sensors.forEach((s, idx) => {
ctx.beginPath();
ctx.arc(s.x, s.y, distances[idx], 0, 2 * Math.PI);
ctx.strokeStyle = s.color + '44';
ctx.lineWidth = 1.5;
ctx.stroke();
});
// Draw Smoothed Trajectory Path
if (smoothHistory.length > 1) {
ctx.beginPath();
ctx.moveTo(smoothHistory[0].x, smoothHistory[0].y);
for (let pt of smoothHistory) {
ctx.lineTo(pt.x, pt.y);
}
ctx.strokeStyle = '#38bdf8';
ctx.lineWidth = 3;
ctx.stroke();
}
// Draw Sensors
sensors.forEach(s => {
ctx.beginPath();
ctx.arc(s.x, s.y, 8, 0, 2 * Math.PI);
ctx.fillStyle = s.color;
ctx.fill();
ctx.fillStyle = '#f8fafc';
ctx.font = '12px sans-serif';
ctx.fillText(s.id, s.x + 12, s.y + 4);
});
// Draw Raw Trilateration Point (Jittery)
ctx.beginPath();
ctx.arc(raw.x, raw.y, 5, 0, 2 * Math.PI);
ctx.fillStyle = '#94a3b8';
ctx.fill();
// Draw Kalman Filtered Target Position
ctx.beginPath();
ctx.arc(smooth.x, smooth.y, 10, 0, 2 * Math.PI);
ctx.fillStyle = '#38bdf8';
ctx.fill();
ctx.lineWidth = 2;
ctx.strokeStyle = '#ffffff';
ctx.stroke();
// Target Label
ctx.fillStyle = '#38bdf8';
ctx.fillText('Target Object (Smoothed)', smooth.x + 14, smooth.y + 4);
}
// Event Listeners
[distA, distB, distC, kalmanNoise].forEach(el => el.addEventListener('input', updateSimulation));
document.getElementById('resetBtn').addEventListener('click', () => {
kalman = new SimpleKalman2D(0.05);
rawHistory = [];
smoothHistory = [];
updateSimulation();
});
// Initial Render
updateSimulation();
</script>
</body>
</html>
That Level1Techs thread captures the full scope of what we’ve built—taking raw network telemetry (Wireshark PCAPs, Nmap scans, and firewall logs), resolving them into unified IP/MAC object identities, correcting multi-sensor clock drift, translating RSSI into physical distances, solving non-linear least-squares trilateration, smoothing trajectories with a 2D Kalman filter, enforcing geofence security checks, and streaming it all in real time via FastAPI WebSockets.
To turn this collection of modules into a clean, production-ready repository (structured similarly to projects like ls3vrouter), here is how you can organize the codebase for publication and documentation:
Suggested Repository Structure (network-intelligence-pipeline/)
Plaintext
network-intelligence-pipeline/
│
├── core/
│ ├── __init__.py
│ ├── identity.py # NetworkObjectIdentity data model
│ └── clock_drift.py # ClockDriftCorrector for multi-sensor sync
│
├── ingestors/
│ ├── __init__.py
│ ├── nmap_parser.py # Nmap XML parser
│ ├── pcap_parser.py # Wireshark PyShark & Radiotap parser
│ └── firewall_parser.py# Regex syslog parser
│
├── analytics/
│ ├── __init__.py
│ ├── trilateration.py # Path-loss model & least-squares solver
│ ├── kalman.py # 2D Constant Velocity Kalman Filter
│ └── anomalies.py # Velocity & geofence boundary detectors
│
├── outputs/
│ ├── __init__.py
│ ├── geojson_export.py # GeoJSON FeatureCollection exporter
│ └── templates/
│ └── index.html # Leaflet / HTML5 Canvas visualization dashboard
│
├── server.py # FastAPI WebSocket live streaming server
├── requirements.txt # Python dependencies (numpy, scipy, pyshark, fastapi, uvicorn)
└── README.md # Project documentation for the Level1Techs community
Core Dependencies (requirements.txt)
Plaintext
numpy>=1.22.0
scipy>=1.8.0
pyshark>=0.6.0
fastapi>=0.95.0
uvicorn>=0.20.0
pydantic>=1.10.0
Here is a complete, production-grade README.md tailored for your repository, incorporating the architecture, modules, installation steps, and usage guides discussed across our pipeline development.
Network Intelligence Pipeline
A modular Python framework that ingests heterogeneous network telemetry—Wireshark (PCAP/Radiotap), Nmap XML scans, and firewall syslog streams—to construct unified IP/MAC object identities, perform RSSI-based multilateration, smooth movement trajectories using a 2D Kalman filter, enforce geofence anomaly detection, and stream live spatial data via WebSockets.
Inspired by real-world RF geolocation challenges and community discussions (such as Level1Techs), this pipeline bridges the gap between passive packet capture and active physical asset tracking.
Architecture & Pipeline Flow
Plaintext
[ Wireshark / PCAP ] [ Nmap XML Scans ] [ Firewall Logs ]
│ │ │
└───────────────────┬─────────────────────────┘
▼
┌─────────────────────────┐
│ Ingestion Parsers │
└────────────┬────────────┘
▼
┌─────────────────────────┐
│ Clock Drift Corrector │ (Multi-Sensor Sync)
└────────────┬────────────┘
▼
┌─────────────────────────┐
│ Entity Resolution Engine│ (Unified IP/MAC Objects)
└────────────┬────────────┘
▼
┌─────────────────────────┐
│ Trilateration Solver │ (RSSI Path-Loss & Least Squares)
└────────────┬────────────┘
▼
┌─────────────────────────┐
│ 2D Kalman Filter Tracker│ (Trajectory Smoothing)
└────────────┬────────────┘
▼
┌─────────────────────────┐
│ Geofence & Speed Checker│ (Anomaly Detection Alerts)
└────────────┬────────────┘
┌────────────────────┴────────────────────┐
▼ ▼
[ GeoJSON GIS Export ] [ FastAPI WebSocket Stream ]
(Live Web Dashboard)
Project Directory Structure
Plaintext
network-intelligence-pipeline/
│
├── core/
│ ├── __init__.py
│ ├── identity.py # NetworkObjectIdentity unified data model
│ └── clock_drift.py # Multi-sensor clock skew normalization
│
├── ingestors/
│ ├── __init__.py
│ ├── nmap_parser.py # Nmap XML host, port, and OS parser
│ ├── pcap_parser.py # PyShark PCAP & Radiotap RF telemetry parser
│ └── firewall_parser.py # Regex-based syslog log event extractor
│
├── analytics/
│ ├── __init__.py
│ ├── trilateration.py # Log-distance path loss & Least-Squares solver
│ ├── kalman.py # 2D Constant Velocity Kalman Filter tracker
│ └── anomalies.py # Velocity & geofence perimeter boundary monitors
│
├── outputs/
│ ├── __init__.py
│ ├── geojson_export.py # GeoJSON FeatureCollection point & path exporter
│ └── templates/
│ └── index.html # Leaflet / HTML5 Canvas real-time UI
│
├── server.py # FastAPI WebSocket live streaming server
├── requirements.txt # Core dependencies
└── README.md # Project documentation
Installation
Prerequisites
-
Python 3.9+
-
Wireshark / TShark installed on your system (required for PyShark PCAP parsing).
Setup Instructions
-
Clone the repository:
Bash
git clone https://github.com/your-username/network-intelligence-pipeline.git cd network-intelligence-pipeline -
Install Python dependencies:
Bash
pip install -r requirements.txt
Usage Examples
1. Ingesting and Parsing Multi-Source Telemetry
Python
from ingestors.pcap_parser import WirelessLogIngestor
from ingestors.nmap_parser import LogIngestor
from core.identity import NetworkObjectIdentity
# Parse Nmap scan results
nmap_data = LogIngestor.parse_nmap_xml("nmap_scan_output.xml")
# Parse wireless PCAP containing RSSI / Radiotap headers
wireless_records = WirelessLogIngestor.parse_wireless_pcap("wireless_monitor.pcap", sensor_id="sensor_alpha")
2. Running Trilateration and Kalman Smoothing
Python
from analytics.trilateration import TrilaterationEngine
from analytics.kalman import KalmanTracker2D
engine = TrilaterationEngine()
tracker = KalmanTracker2D(dt=1.0)
# Calculate target GPS from 3 sensor distance estimates
target_gps = engine.calculate_target_position(sensor_readings)
# Smooth jittery trilateration outputs over time
mx, my = engine.gps_to_cartesian(target_gps[0], target_gps[1], ref_lat=34.0522, ref_lon=-118.2437)
smooth_x, smooth_y = tracker.update((mx, my))
3. Launching the Live FastAPI Streaming Server
To stream parsed RSSI metrics, trilateration coordinates, and geofence alerts directly to your web browser in real time:
Bash
uvicorn server:app --reload --host 0.0.0.0 --port 8000
Navigate to http://localhost:8000 to view the live dashboard.
Contributing
Pull requests, bug reports, and suggestions for advanced multi-lateration algorithms (like TDOA time-difference-of-arrival) are welcome. Please open an issue to discuss proposed changes.
License
Distributed under the MIT License. See LICENSE for more information.
More or less copy paste from the working chat, using only a xbox and xbox controller and edge web browser
Here fun part
Copilot review of this.
I took a look at your forum post. Using wireshark logs to triangulate packet sources is essentially a Gemini-generated design for building an IP-centric network intelligence pipeline that combines PCAPs, Nmap data, firewall logs, timestamps, packet headers, and optional GPS/device metadata into unified “object identities.” [forum.leve…1techs.com]
My assessment:
What Gemini got right
-
The overall architecture is reasonable:
-
Ingestion layer
-
Entity resolution layer
-
Correlation engine
-
Enrichment layer
-
-
Using IP, MAC, hostname, and port data to construct a persistent identity object is a common network analytics approach. [forum.leve…1techs.com]
-
Combining:
-
Wireshark captures
-
Firewall logs
-
Nmap scans
-
Timestamp correlation into one graph of activity can be very useful for investigations. [forum.leve…1techs.com]
-
Where the post overreaches The word**“triangulation”** is doing a lot of work.
True triangulation of a device location generally requires:
-
Multiple observation points
-
RF measurements
-
Time Difference of Arrival (TDOA)
-
Angle of Arrival (AOA)
-
Signal strength modeling
The code shown mostly performs:
-
Correlation
-
Attribution
-
Identity resolution
-
Event clustering
It does not actually triangulate a physical source location from packets. The TTL averaging example cited is more of a heuristic than a location system. [forum.leve…1techs.com]
What I would call it instead Something like:
Network Identity Correlation Engine
or
Packet Attribution and Entity Resolution Framework
Those names better reflect what the code is actually doing.
For your LS3-style projects This concept aligns surprisingly well with several of your saved LS3 efforts.
You already think in terms of:
-
Synthetic identities
-
Distributed observation nodes
-
Multi-source correlation
-
Time-domain analysis
-
Cross-node arbitration
What Gemini describes could become:
Plain Text
1
Wireshark
2
\
3
Firewall → Identity Graph → Correlator
4
/
5
Nmap
6
7
Identity:
8
IP
9
MAC
10
Hostname
11
Ports
12
TLS Fingerprints
13
DHCP History
14
DNS History
15
Timing Signatures
Show more lines
Instead of treating an IP as the identity, you could treat the identity as a synthesized object generated from multiple observations.
The strongest idea in the post The most valuable idea is not GPS.
It’s this:
Build a persistent object that survives IP changes.
For example:
Plain Text
1
192.168.1.50
2
10.0.0.22
3
172.16.1.90
4
5
all observed using
6
7
MAC:
8
AA:BB:CC:DD:EE:FF
9
10
Hostname:
11
scanner-01
12
13
DHCP Fingerprint:
14
match
15
16
TLS Fingerprint:
17
match
Show more lines
The engine concludes:
Plain Text
1
Network Object #42
Show more lines
even though the IP changed.
That becomes a much more powerful investigation platform than raw Wireshark analysis.
My verdict
I’d rate the Gemini response:
-
Architecture: 8/10
-
Python structure: 7/10
-
Network realism: 5/10
-
Actual packet-source triangulation: 2/10
-
Identity-correlation framework: 8.5/10
So the post is a solid blueprint for network identity correlation, but it is not a true packet-source triangulation system despite using that terminology. [forum.leve…1techs.com]
For someone building advanced correlation systems, I think the interesting next step would be turning those identities into a graph database where devices, IPs, MACs, DNS names, flows, and timestamps are linked as relationships rather than stored as simple objects. That is where the design starts becoming genuinely powerful.