import html
import json
import math
from collections import defaultdict
from reports.database import query_all, query_one
from reports.utils import (
format_datetime as format_report_datetime,
)
from reports.template import (
ReportContext,
card,
page_begin,
page_end,
page_header,
)
def esc(value):
return html.escape("" if value is None else str(value), quote=True)
def format_datetime(value):
return format_report_datetime(
value,
"%d.%m.%y-%H.%M.%S"
)
def format_number(value, digits=1):
if value is None:
return "—"
try:
return f"{float(value):.{digits}f}"
except (TypeError, ValueError):
return str(value)
def format_speed(value):
if value is None:
return "—"
try:
speed = float(value)
except (TypeError, ValueError):
return str(value)
if abs(speed) < 0.05:
return "0"
return f"{speed:.1f}"
def haversine_m(lat1, lon1, lat2, lon2):
radius = 6371000.0
phi1 = math.radians(lat1)
phi2 = math.radians(lat2)
dphi = math.radians(lat2 - lat1)
dlambda = math.radians(lon2 - lon1)
a = (
math.sin(dphi / 2.0) ** 2
+ math.cos(phi1)
* math.cos(phi2)
* math.sin(dlambda / 2.0) ** 2
)
return 2.0 * radius * math.asin(min(1.0, math.sqrt(a)))
def weighted_mean(values):
if not values:
return None
total_weight = sum(weight for _, weight in values)
if total_weight <= 0:
return sum(value for value, _ in values) / len(values)
return sum(value * weight for value, weight in values) / total_weight
def percentile(values, fraction):
if not values:
return None
ordered = sorted(values)
if len(ordered) == 1:
return ordered[0]
position = (len(ordered) - 1) * fraction
lower = int(math.floor(position))
upper = int(math.ceil(position))
if lower == upper:
return ordered[lower]
ratio = position - lower
return ordered[lower] + (ordered[upper] - ordered[lower]) * ratio
def observation_weight(rssi):
"""
RSSI is used only as a relative confidence weight.
We deliberately do not convert RSSI to a physical distance because
environmental attenuation, antenna characteristics, reflections and
transmitter power make such a conversion unreliable for this task.
"""
if rssi is None:
return 1.0
try:
value = float(rssi)
except (TypeError, ValueError):
return 1.0
# Keep the influence deliberately moderate.
#
# - -90 dBm -> 1
# - -70 dBm -> 2
# - -50 dBm -> 4
#
# This makes stronger observations useful without allowing a handful
# of unusually strong readings to dominate the whole history.
normalized = max(0.0, min(40.0, value + 90.0))
return 1.0 + normalized / 10.0
def valid_observations(rows):
result = []
for row in rows:
try:
latitude = float(row["latitude"])
longitude = float(row["longitude"])
except (TypeError, ValueError):
continue
if not (-90.0 <= latitude <= 90.0):
continue
if not (-180.0 <= longitude <= 180.0):
continue
if abs(latitude) < 1e-12 and abs(longitude) < 1e-12:
continue
result.append(
{
"id": row["id"],
"session_id": row["session_id"],
"observed_at": row["observed_at"],
"latitude": latitude,
"longitude": longitude,
"speed": row["speed"],
"rssi": row["rssi"],
"channel": row["channel"],
"frequency": row["frequency"],
"essid": row["essid"],
"encryption": row["encryption"],
"cipher": row["cipher"],
"akm": row["akm"],
"country": row["country"],
}
)
return result
def normalize_bursts(observations):
"""
Treat repeated observations at the same time and position as one
independent spatial observation.
RSSI values inside a burst are averaged so that short-lived RSSI
fluctuations do not give duplicate rows additional influence.
"""
if not observations:
return observations
groups = {}
for item in observations:
key = (
item["session_id"],
item["observed_at"],
item["latitude"],
item["longitude"],
)
group = groups.setdefault(
key,
{
"item": item,
"rssi_values": [],
},
)
if item["rssi"] is not None:
try:
group["rssi_values"].append(float(item["rssi"]))
except (TypeError, ValueError):
pass
result = []
for group in groups.values():
item = dict(group["item"])
rssi_values = group["rssi_values"]
if rssi_values:
item["rssi"] = sum(rssi_values) / len(rssi_values)
result.append(item)
return result
def local_projection(observations):
"""
Convert geographic coordinates to a local metre-based coordinate system.
x = east/west
y = north/south
"""
if not observations:
return None, []
center_lat = sum(item["latitude"] for item in observations) / len(observations)
center_lon = sum(item["longitude"] for item in observations) / len(observations)
cos_lat = math.cos(math.radians(center_lat))
if abs(cos_lat) < 1e-8:
cos_lat = 1e-8
projected = []
meters_per_degree = 111320.0
for item in observations:
x = (
(item["longitude"] - center_lon)
* meters_per_degree
* cos_lat
)
y = (item["latitude"] - center_lat) * meters_per_degree
projected.append((item, x, y))
return (center_lat, center_lon), projected
def weighted_center(observations):
if not observations:
return None
center, projected = local_projection(observations)
weighted_x = []
weighted_y = []
for item, x, y in projected:
weight = observation_weight(item["rssi"])
weighted_x.append((x, weight))
weighted_y.append((y, weight))
x = weighted_mean(weighted_x)
y = weighted_mean(weighted_y)
center_lat, center_lon = center
meters_per_degree = 111320.0
cos_lat = math.cos(math.radians(center_lat))
if abs(cos_lat) < 1e-8:
cos_lat = 1e-8
latitude = center_lat + y / meters_per_degree
longitude = center_lon + x / (meters_per_degree * cos_lat)
return {
"latitude": latitude,
"longitude": longitude,
"x": x,
"y": y,
}
def robust_observations(observations):
"""
Remove only strong spatial outliers.
The method intentionally has a conservative threshold. We do not want
to destroy normal driving tracks merely because the AP was observed
from several nearby positions.
"""
if len(observations) < 8:
return observations
center = weighted_center(observations)
if not center:
return observations
distances = []
for item in observations:
distance = haversine_m(
center["latitude"],
center["longitude"],
item["latitude"],
item["longitude"],
)
distances.append((item, distance))
numeric_distances = [distance for _, distance in distances]
median = percentile(numeric_distances, 0.50)
p90 = percentile(numeric_distances, 0.90)
if median is None or p90 is None:
return observations
# Robust scale estimate.
deviations = [abs(distance - median) for distance in numeric_distances]
mad = percentile(deviations, 0.50) or 0.0
robust_limit = median + max(3.0 * mad, 0.0)
# Never trim inside the central 90% envelope.
limit = max(robust_limit, p90)
# Avoid creating an artificially tiny area.
minimum_limit = max(75.0, median * 1.5)
limit = max(limit, minimum_limit)
filtered = [
item
for item, distance in distances
if distance <= limit
]
# If the filter became too aggressive, retain the original data.
if len(filtered) < max(5, int(len(observations) * 0.60)):
return observations
return filtered
def covariance_ellipse(observations):
"""
Calculate an oriented uncertainty ellipse in metres.
The ellipse describes the spatial dispersion of the useful observations,
not a guaranteed physical reception radius of the AP.
"""
if len(observations) < 2:
return None
center, projected = local_projection(observations)
weights = [
observation_weight(item["rssi"])
for item, _, _ in projected
]
total_weight = sum(weights)
if total_weight <= 0:
return None
mean_x = sum(x * weight for (_, x, _), weight in zip(projected, weights))
mean_y = sum(y * weight for (_, _, y), weight in zip(projected, weights))
mean_x /= total_weight
mean_y /= total_weight
covariance_xx = 0.0
covariance_xy = 0.0
covariance_yy = 0.0
for (_, x, y), weight in zip(projected, weights):
dx = x - mean_x
dy = y - mean_y
covariance_xx += weight * dx * dx
covariance_xy += weight * dx * dy
covariance_yy += weight * dy * dy
covariance_xx /= total_weight
covariance_xy /= total_weight
covariance_yy /= total_weight
trace = covariance_xx + covariance_yy
determinant = (
covariance_xx * covariance_yy
- covariance_xy * covariance_xy
)
determinant = max(0.0, determinant)
discriminant = max(
0.0,
trace * trace - 4.0 * determinant,
)
lambda1 = max(
0.0,
(trace + math.sqrt(discriminant)) / 2.0,
)
lambda2 = max(
0.0,
(trace - math.sqrt(discriminant)) / 2.0,
)
if lambda1 <= 0.0:
return None
if abs(covariance_xy) > 1e-9:
axis_x = lambda1 - covariance_yy
axis_y = covariance_xy
elif covariance_xx >= covariance_yy:
axis_x = 1.0
axis_y = 0.0
else:
axis_x = 0.0
axis_y = 1.0
axis_length = math.hypot(axis_x, axis_y)
if axis_length <= 1e-9:
return None
axis_x /= axis_length
axis_y /= axis_length
angle = math.degrees(math.atan2(axis_x, axis_y))
# Approximately 90% confidence ellipse for a 2D Gaussian.
scale = math.sqrt(4.605170186)
semi_major = math.sqrt(lambda1) * scale
semi_minor = math.sqrt(lambda2) * scale
# A tiny covariance can occur when the same point was observed many
# times. Keep the visualization visible but do not claim centimetre
# accuracy.
semi_major = max(25.0, semi_major)
semi_minor = max(20.0, semi_minor)
# Prevent visually misleading enormous ellipses.
semi_major = min(5000.0, semi_major)
semi_minor = min(5000.0, semi_minor)
center_lat, center_lon = center
meters_per_degree = 111320.0
cos_lat = math.cos(math.radians(center_lat))
if abs(cos_lat) < 1e-8:
cos_lat = 1e-8
latitude = center_lat + mean_y / meters_per_degree
longitude = center_lon + mean_x / (meters_per_degree * cos_lat)
return {
"latitude": latitude,
"longitude": longitude,
"semi_major": semi_major,
"semi_minor": semi_minor,
"angle": angle,
}
def ellipse_polygon(ellipse, points=64):
if not ellipse:
return []
lat0 = ellipse["latitude"]
lon0 = ellipse["longitude"]
angle = math.radians(ellipse["angle"])
cos_lat = math.cos(math.radians(lat0))
if abs(cos_lat) < 1e-8:
cos_lat = 1e-8
result = []
for index in range(points + 1):
theta = 2.0 * math.pi * index / points
local_x = ellipse["semi_major"] * math.cos(theta)
local_y = ellipse["semi_minor"] * math.sin(theta)
x = (
local_x * math.cos(angle)
+ local_y * math.sin(angle)
)
y = (
-local_x * math.sin(angle)
+ local_y * math.cos(angle)
)
latitude = lat0 + y / 111320.0
longitude = lon0 + x / (111320.0 * cos_lat)
result.append([longitude, latitude])
return result
def spatial_zones(observations):
"""
Find clearly separated observation groups.
This is intentionally conservative. A zone is reported only when there
are at least three observations in a group and the groups are separated
by roughly 600 metres.
"""
if len(observations) < 6:
return []
remaining = list(observations)
groups = []
while remaining:
seed = remaining.pop(0)
group = [seed]
changed = True
while changed:
changed = False
keep = []
for item in remaining:
close = False
for member in group:
if (
haversine_m(
item["latitude"],
item["longitude"],
member["latitude"],
member["longitude"],
)
<= 300.0
):
close = True
break
if close:
group.append(item)
changed = True
else:
keep.append(item)
remaining = keep
groups.append(group)
significant = [
group
for group in groups
if len(group) >= 3
]
if len(significant) <= 1:
return []
centers = []
for group in significant:
center = weighted_center(group)
if center:
centers.append((group, center))
separated = False
for index, (_, center_a) in enumerate(centers):
for _, center_b in centers[index + 1:]:
if (
haversine_m(
center_a["latitude"],
center_a["longitude"],
center_b["latitude"],
center_b["longitude"],
)
> 600.0
):
separated = True
break
if separated:
break
if not separated:
return []
return [
{
"observations": group,
"center": center,
}
for group, center in centers
]
def load_access_point(conn, bssid):
return query_one(
conn,
"""
SELECT
id,
bssid,
essid,
encryption,
cipher,
akm,
country,
channel,
frequency,
vendor,
first_seen,
last_seen,
times_seen,
last_rssi,
last_latitude,
last_longitude,
last_speed,
has_handshake,
has_pmkid,
is_cracked
FROM access_points
WHERE lower(bssid) = lower(?)
LIMIT 1
""",
(bssid,),
)
def load_observations(conn, access_point_id):
return query_all(
conn,
"""
SELECT
id,
session_id,
observed_at,
latitude,
longitude,
speed,
rssi,
channel,
frequency,
essid,
encryption,
cipher,
akm,
country
FROM access_point_observations
WHERE access_point_id = ?
ORDER BY observed_at, id
""",
(access_point_id,),
)
def load_credentials(conn, access_point_id):
return query_all(
conn,
"""
SELECT
c.password,
c.source,
c.created_at,
c.verified,
c.handshake_id
FROM credentials AS c
WHERE c.access_point_id = ?
ORDER BY c.created_at DESC, c.id DESC
""",
(access_point_id,),
)
def security_text(ap):
values = []
for key in ("encryption", "cipher", "akm"):
value = ap[key]
if value and value not in values:
values.append(str(value))
return " / ".join(values) if values else "—"
def calculate_statistics(observations):
rssis = []
speeds = []
for item in observations:
if item["rssi"] is not None:
try:
rssis.append(float(item["rssi"]))
except (TypeError, ValueError):
pass
if item["speed"] is not None:
try:
speeds.append(float(item["speed"]))
except (TypeError, ValueError):
pass
sessions = {
item["session_id"]
for item in observations
if item["session_id"] is not None
}
return {
"observations": len(observations),
"sessions": len(sessions),
"coordinates": len(valid_observations(observations)),
"rssi_min": min(rssis) if rssis else None,
"rssi_avg": sum(rssis) / len(rssis) if rssis else None,
"rssi_max": max(rssis) if rssis else None,
"speed_avg": sum(speeds) / len(speeds) if speeds else None,
"speed_max": max(speeds) if speeds else None,
}
def build_geojson(observations, ellipse=None, zones=None):
features = []
for item in observations:
try:
latitude = float(item["latitude"])
longitude = float(item["longitude"])
except (TypeError, ValueError):
continue
properties = {
"type": "observation",
"id": item["id"],
"session_id": item["session_id"],
"observed_at": format_datetime(item["observed_at"]),
"rssi": item["rssi"],
"speed": item["speed"],
"map_status": item.get("map_status", "normal"),
}
features.append(
{
"type": "Feature",
"geometry": {
"type": "Point",
"coordinates": [longitude, latitude],
},
"properties": properties,
}
)
if ellipse:
polygon = ellipse_polygon(ellipse)
if polygon:
features.append(
{
"type": "Feature",
"geometry": {
"type": "Polygon",
"coordinates": [polygon],
},
"properties": {
"type": "uncertainty",
},
}
)
if zones:
for index, zone in enumerate(zones, start=1):
center = zone["center"]
features.append(
{
"type": "Feature",
"geometry": {
"type": "Point",
"coordinates": [
center["longitude"],
center["latitude"],
],
},
"properties": {
"type": "zone",
"zone": index,
"observations": len(zone["observations"]),
},
}
)
return {
"type": "FeatureCollection",
"features": features,
}
def render_map(geojson, context, title):
maplibre_css = context.asset_url("maplibre/maplibre-gl.css")
maplibre_js = context.asset_url("maplibre/maplibre-gl.js")
geojson_json = json.dumps(
geojson,
ensure_ascii=False,
separators=(",", ":"),
)
return f"""
Обычное наблюдение — координата, записанная во время сканирования.
Handshake — для этого AP в базе есть перехваченный handshake.
Пароль найден — для этого AP в базе есть credential.
Расчётная область — статистическая оценка положения AP по истории наблюдений. Это не гарантированная зона покрытия и не точная координата устройства.
"""
def render_observation_table(observations):
rows = []
for item in reversed(observations):
coordinates = "—"
if (
item["latitude"] is not None
and item["longitude"] is not None
):
coordinates = (
f"{float(item['latitude']):.6f}, "
f"{float(item['longitude']):.6f}"
)
rows.append(
f"""
Channel
{esc(ap["channel"] if ap["channel"] is not None else "—")}
Frequency
{esc(ap["frequency"] if ap["frequency"] is not None else "—")}
First seen
{esc(format_datetime(ap["first_seen"]))}
Last seen
{esc(format_datetime(ap["last_seen"]))}
"""
)
page += card(
"История наблюдений",
f"""
Наблюдений
{statistics["observations"]}
Сессий
{statistics["sessions"]}
С координатами
{statistics["coordinates"]}
RSSI min
{format_number(statistics["rssi_min"])}
RSSI avg
{format_number(statistics["rssi_avg"])}
RSSI max
{format_number(statistics["rssi_max"])}
Средняя скорость
{format_speed(statistics["speed_avg"])}
Максимальная скорость
{format_speed(statistics["speed_max"])}
"""
)
status_parts = []
if ap["has_handshake"]:
status_parts.append("Handshake")
if ap["has_pmkid"]:
status_parts.append("PMKID")
if ap["is_cracked"]:
status_parts.append("Cracked")
if status_parts:
status_text = ", ".join(status_parts)
else:
status_text = "Нет связанных данных"
credential_rows = []
for credential in credentials:
verified = "Да" if credential["verified"] else "Нет"
credential_rows.append(
f"""
{esc(credential["password"])}
{esc(credential["source"] or "—")}
{esc(verified)}
{esc(format_datetime(credential["created_at"]))}
#{esc(credential["handshake_id"])}
"""
)
credentials_html = ""
if credential_rows:
credentials_html = f"""
Password
Source
Verified
Created
Handshake
{''.join(credential_rows)}
"""
else:
credentials_html = """
Связанных credentials нет.
"""
page += card(
"Связанные данные",
f"""
Статус: {esc(status_text)}
{credentials_html}
"""
)
if estimated_center:
estimated_coordinates = (
f"{estimated_center['latitude']:.6f}, "
f"{estimated_center['longitude']:.6f}"
)
else:
estimated_coordinates = "Недостаточно координат"
if ellipse:
ellipse_size = (
f"{ellipse['semi_major']:.0f} × "
f"{ellipse['semi_minor']:.0f} м"
)
ellipse_direction = f"{ellipse['angle']:.0f}°"
else:
ellipse_size = "Недостаточно данных"
ellipse_direction = "—"
if zones:
zone_message = (
f"История содержит {len(zones)} пространственно "
f"разделённых зоны наблюдений. Поэтому одна общая "
f"область неопределённости может быть misleading."
)
else:
zone_message = (
"Наблюдения не показывают достаточно выраженного "
"разделения на несколько независимых пространственных зон."
)
page += card(
"Расчёт положения",
f"""
Оценочный центр
{esc(estimated_coordinates)}
Размер эллипса
{esc(ellipse_size)}
Ориентация
{esc(ellipse_direction)}
Использовано координат
{len(useful_observations)}
Важно:
расчётная точка является статистической оценкой по истории
наблюдений. Она не означает, что AP физически находится
непосредственно в этой координате.
RSSI используется только как дополнительный относительный вес.
Расчётная область показывает неопределённость оценки, а не
гарантированную дальность действия Wi-Fi.