yaro/wifi_test
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Some adjustments to make this work a little better with existing structure.

Browse source
This commit is contained in:
yaro 2025-04-09 14:05:03 -05:00
parent d327711b5c
commit 38872f0c37
1 changed files with 45 additions and 31 deletions
Download patch file Download diff file Expand all files Collapse all files

76
kismet_enrich_csv.py
View file

@ -2,45 +2,47 @@
import sqlite3
import csv
import argparse
from datetime import datetime, timedelta
from datetime import datetime
def parse_args():
parser = argparse.ArgumentParser()
parser.add_argument("--csv", required=True, help="Original speed test CSV file")
parser.add_argument("--kismet", required=True, help=".kismet SQLite log file")
parser.add_argument("--output", required=True, help="Output enriched CSV file")
parser.add_argument("--csv", required=True, help="Input speedtest CSV")
parser.add_argument("--kismet", required=True, help="Path to .kismet log file")
parser.add_argument("--output", required=True, help="Output enriched CSV")
return parser.parse_args()
def convert_timestamp_to_epoch(ts_string):
try:
return int(datetime.strptime(ts_string.split(".")[0], "%Y-%m-%dT%H:%M:%S").timestamp())
except Exception as e:
print(f"[!] Invalid timestamp format: {ts_string}")
return None
def get_rf_metrics(cursor, bssid, channel, timestamp):
# Match devices associated with the BSSID
cursor.execute("""
SELECT COUNT(*) FROM devicelink
WHERE type = 'Wi-Fi Client' AND mac = ? AND last_time >= ? - 10 AND last_time <= ? + 10
""", (bssid, timestamp, timestamp))
WHERE type = 'Wi-Fi Client' AND mac = ? AND last_time BETWEEN ? AND ?
""", (bssid.lower(), timestamp - 10, timestamp + 10))
clients_on_ap = cursor.fetchone()[0]
# Match clients on the same channel
cursor.execute("""
SELECT COUNT(DISTINCT devicelink.remote_mac)
FROM devicelink
JOIN devices ON devicelink.mac = devices.base_mac
WHERE devicelink.type = 'Wi-Fi Client'
AND devices.channel = ?
AND devicelink.last_time >= ? - 10 AND devicelink.last_time <= ? + 10
""", (channel, timestamp, timestamp))
AND devicelink.last_time BETWEEN ? AND ?
""", (channel, timestamp - 10, timestamp + 10))
clients_on_channel = cursor.fetchone()[0]
# Count APs on the same channel
cursor.execute("""
SELECT COUNT(*) FROM devices
WHERE type = 'Wi-Fi AP' AND channel = ?
AND last_time >= ? - 10 AND last_time <= ? + 10
""", (channel, timestamp, timestamp))
AND last_time BETWEEN ? AND ?
""", (channel, timestamp - 10, timestamp + 10))
aps_on_channel = cursor.fetchone()[0]
congestion_score = (
round(clients_on_channel / aps_on_channel, 2) if aps_on_channel > 0 else 0.0
)
congestion_score = round(clients_on_channel / aps_on_channel, 2) if aps_on_channel else 0.0
return clients_on_ap, clients_on_channel, aps_on_channel, congestion_score
@ -49,27 +51,39 @@ def main():
conn = sqlite3.connect(args.kismet)
cursor = conn.cursor()
with open(args.csv, newline="") as infile, open(args.output, "w", newline="") as outfile:
reader = csv.reader(infile)
writer = csv.writer(outfile)
headers = next(reader)
headers += ["ClientsOnAP", "ClientsOnChannel", "APsOnChannel", "CongestionScore"]
writer.writerow(headers)
with open(args.csv, newline='') as infile, open(args.output, 'w', newline='') as outfile:
reader = csv.DictReader(infile)
fieldnames = reader.fieldnames + [
"ClientsOnAP", "ClientsOnChannel", "APsOnChannel", "CongestionScore"
]
writer = csv.DictWriter(outfile, fieldnames=fieldnames)
writer.writeheader()
for row in reader:
ts = convert_timestamp_to_epoch(row["Timestamp"])
try:
timestamp_str = row[0] # assuming first column is timestamp
bssid = row[3].strip()
channel = int(row[-4]) # assuming you store channel near the end
timestamp = int(datetime.strptime(timestamp_str, "%Y-%m-%d_%H:%M:%S").timestamp())
rf = get_rf_metrics(cursor, bssid, channel, timestamp)
writer.writerow(row + list(rf))
bssid = row["BSSID"].strip().lower()
channel = int(row["Channel"])
except Exception as e:
print(f"[!] Failed to process row: {row}\n Error: {e}")
writer.writerow(row + ["ERR", "ERR", "ERR", "ERR"])
print(f"[!] Failed to extract BSSID/Channel: {e}")
writer.writerow(row)
continue
if ts is None:
writer.writerow(row)
continue
clients_ap, clients_chan, aps_chan, congestion = get_rf_metrics(cursor, bssid, channel, ts)
row["ClientsOnAP"] = clients_ap
row["ClientsOnChannel"] = clients_chan
row["APsOnChannel"] = aps_chan
row["CongestionScore"] = congestion
writer.writerow(row)
conn.close()
print(f"[+] Enrichment complete: {args.output}")
if __name__ == "__main__":
main()