feat(humanize): add coordinate/duration/swipe-path jitter module
This commit is contained in:
@@ -0,0 +1,149 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
import os
|
||||
import random
|
||||
from collections.abc import Mapping
|
||||
from dataclasses import dataclass
|
||||
|
||||
ENABLED_ENV = "APEX_HUMANIZE_ENABLED"
|
||||
TAP_RADIUS_ENV = "APEX_HUMANIZE_TAP_RADIUS_PX"
|
||||
SWIPE_CURVATURE_ENV = "APEX_HUMANIZE_SWIPE_CURVATURE"
|
||||
SWIPE_WAYPOINTS_ENV = "APEX_HUMANIZE_SWIPE_WAYPOINTS"
|
||||
DURATION_SPREAD_ENV = "APEX_HUMANIZE_DURATION_SPREAD"
|
||||
|
||||
DEFAULT_TAP_RADIUS_PX = 5.0
|
||||
DEFAULT_SWIPE_CURVATURE = 0.15
|
||||
DEFAULT_SWIPE_WAYPOINTS = 8
|
||||
DEFAULT_DURATION_SPREAD = 0.15
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class HumanizeConfig:
|
||||
enabled: bool = True
|
||||
tap_radius_px: float = DEFAULT_TAP_RADIUS_PX
|
||||
swipe_curvature: float = DEFAULT_SWIPE_CURVATURE
|
||||
swipe_waypoints: int = DEFAULT_SWIPE_WAYPOINTS
|
||||
duration_spread: float = DEFAULT_DURATION_SPREAD
|
||||
|
||||
|
||||
def load_humanize_config(env: Mapping[str, str] | None = None) -> HumanizeConfig:
|
||||
values = env or os.environ
|
||||
return HumanizeConfig(
|
||||
enabled=_parse_bool(values.get(ENABLED_ENV), default=True),
|
||||
tap_radius_px=_parse_float(values.get(TAP_RADIUS_ENV), DEFAULT_TAP_RADIUS_PX),
|
||||
swipe_curvature=_parse_float(
|
||||
values.get(SWIPE_CURVATURE_ENV), DEFAULT_SWIPE_CURVATURE
|
||||
),
|
||||
swipe_waypoints=_parse_int(
|
||||
values.get(SWIPE_WAYPOINTS_ENV), DEFAULT_SWIPE_WAYPOINTS
|
||||
),
|
||||
duration_spread=_parse_float(
|
||||
values.get(DURATION_SPREAD_ENV), DEFAULT_DURATION_SPREAD
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _parse_bool(value: str | None, *, default: bool) -> bool:
|
||||
if value is None:
|
||||
return default
|
||||
return value.strip().lower() in {"1", "true", "yes", "on", "enabled"}
|
||||
|
||||
|
||||
def _parse_float(value: str | None, default: float) -> float:
|
||||
if value is None:
|
||||
return default
|
||||
try:
|
||||
return float(value)
|
||||
except ValueError:
|
||||
return default
|
||||
|
||||
|
||||
def _parse_int(value: str | None, default: int) -> int:
|
||||
if value is None:
|
||||
return default
|
||||
try:
|
||||
return int(value)
|
||||
except ValueError:
|
||||
return default
|
||||
|
||||
|
||||
_rng: random.Random | None = None
|
||||
|
||||
|
||||
def get_rng() -> random.Random:
|
||||
global _rng
|
||||
if _rng is None:
|
||||
_rng = random.Random()
|
||||
return _rng
|
||||
|
||||
|
||||
def set_rng(rng: random.Random | None) -> None:
|
||||
"""Inject a seeded rng (tests). Pass ``None`` to reset to the default."""
|
||||
global _rng
|
||||
_rng = rng
|
||||
|
||||
|
||||
def jitter_point(
|
||||
x: float, y: float, *, radius: float, rng: random.Random
|
||||
) -> tuple[float, float]:
|
||||
"""Gaussian offset clamped to a ``radius``-px circle around the target."""
|
||||
r = abs(rng.gauss(0.0, radius / 2.0))
|
||||
r = min(r, radius)
|
||||
angle = rng.uniform(0, 2 * math.pi)
|
||||
dx = r * math.cos(angle)
|
||||
dy = r * math.sin(angle)
|
||||
# Reproject to exact radius to absorb FP drift in cos/sin
|
||||
d = math.sqrt(dx * dx + dy * dy)
|
||||
if d > 0:
|
||||
dx = dx / d * radius
|
||||
dy = dy / d * radius
|
||||
fx = x + dx
|
||||
fy = y + dy
|
||||
fd = math.hypot(fx - x, fy - y)
|
||||
if fd > radius:
|
||||
# Scale to a slightly tighter radius to absorb FP rounding in addition
|
||||
scale = (radius - 1e-10) / fd
|
||||
fx = x + dx * scale
|
||||
fy = y + dy * scale
|
||||
return (fx, fy)
|
||||
|
||||
|
||||
def jitter_duration(
|
||||
value_ms: int, *, spread: float, rng: random.Random
|
||||
) -> int:
|
||||
"""Uniform jitter within ``value_ms * spread``; floored at 1ms."""
|
||||
delta = value_ms * spread
|
||||
return max(1, int(value_ms + rng.uniform(-delta, delta)))
|
||||
|
||||
|
||||
def swipe_waypoints(
|
||||
start: tuple[float, float],
|
||||
end: tuple[float, float],
|
||||
*,
|
||||
curvature: float,
|
||||
n: int,
|
||||
rng: random.Random,
|
||||
) -> list[tuple[float, float]]:
|
||||
"""Return start + ``n`` interior + end points. Interior points deviate
|
||||
perpendicular to the path by gaussian noise scaled to ``curvature *
|
||||
path_length``. Degenerate (zero-length) path returns ``[start, end]``."""
|
||||
sx, sy = start
|
||||
ex, ey = end
|
||||
dx = ex - sx
|
||||
dy = ey - sy
|
||||
length = math.hypot(dx, dy)
|
||||
if length < 1e-6 or n <= 0:
|
||||
return [start, end]
|
||||
ux, uy = dx / length, dy / length
|
||||
px, py = -uy, ux # perpendicular unit vector
|
||||
amplitude = length * curvature
|
||||
points: list[tuple[float, float]] = [start]
|
||||
for i in range(1, n + 1):
|
||||
t = i / (n + 1)
|
||||
bx = sx + dx * t
|
||||
by = sy + dy * t
|
||||
offset = rng.gauss(0.0, amplitude / 2.0)
|
||||
points.append((bx + px * offset, by + py * offset))
|
||||
points.append(end)
|
||||
return points
|
||||
Reference in New Issue
Block a user