208 lines
7.1 KiB
Python
208 lines
7.1 KiB
Python
"""
|
||
传感器模拟 — 伪造加速度计/陀螺仪数据，模拟真人手持特征
|
||
|
||
功能:
|
||
1. 加速度计噪声注入 (模拟手持微抖)
|
||
2. 陀螺仪数据生成 (缓慢旋转漂移)
|
||
3. 传感器事件写入 /dev/input/ (需 root)
|
||
4. 非 root 降级为日志记录
|
||
5. 与 DeviceGuard 联动: 若传感器数 < 5 则触发补偿
|
||
"""
|
||
|
||
import logging
|
||
import math
|
||
import os
|
||
import random
|
||
import struct
|
||
import time
|
||
from typing import Dict, List, Optional, Tuple
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class SensorSimulator:
|
||
"""传感器模拟器 — 让设备看起来像被人拿着"""
|
||
|
||
GRAVITY = 9.81
|
||
INPUT_EVENT_FORMAT = "llHHi" # struct input_event: sec, usec, type, code, value
|
||
|
||
def __init__(self, device=None, has_root: bool = False):
|
||
"""
|
||
Args:
|
||
device: uiautomator2 设备对象
|
||
has_root: 是否有 root 权限
|
||
"""
|
||
self.d = device
|
||
self.has_root = has_root
|
||
self._running = False
|
||
self._base_orientation = self._random_orientation()
|
||
|
||
def generate_accelerometer_sample(self) -> Dict[str, float]:
|
||
"""
|
||
生成一个加速度计采样值 — 模拟静止手持状态
|
||
|
||
真实手持特征:
|
||
- x 轴: 微小随机漂移 (+-0.3 m/s²)
|
||
- y 轴: 接近 0 (横持) 或接近 gravity (竖持)
|
||
- z 轴: 接近 gravity (平放) 或接近 0 (竖持)
|
||
"""
|
||
ox, oy, oz = self._base_orientation
|
||
|
||
noise_x = random.gauss(0, 0.15)
|
||
noise_y = random.gauss(0, 0.15)
|
||
noise_z = random.gauss(0, 0.10)
|
||
|
||
drift = random.gauss(0, 0.02)
|
||
self._base_orientation = (
|
||
ox + drift * random.choice([-1, 1]),
|
||
oy + drift * random.choice([-1, 1]),
|
||
oz + drift * random.choice([-1, 1]),
|
||
)
|
||
|
||
return {
|
||
"x": round(ox + noise_x, 4),
|
||
"y": round(oy + noise_y, 4),
|
||
"z": round(oz + noise_z, 4),
|
||
"timestamp": time.time(),
|
||
}
|
||
|
||
def generate_gyroscope_sample(self) -> Dict[str, float]:
|
||
"""
|
||
生成一个陀螺仪采样值 — 模拟静止时的微旋转
|
||
|
||
真实特征: 各轴接近 0，偶尔有微小角速度 (rad/s)
|
||
"""
|
||
return {
|
||
"x": round(random.gauss(0, 0.005), 6),
|
||
"y": round(random.gauss(0, 0.005), 6),
|
||
"z": round(random.gauss(0, 0.003), 6),
|
||
"timestamp": time.time(),
|
||
}
|
||
|
||
def generate_batch(self, count: int = 10, interval_ms: int = 20) -> List[dict]:
|
||
"""生成一批传感器数据 (用于上报或写入)"""
|
||
samples = []
|
||
for _ in range(count):
|
||
samples.append({
|
||
"accel": self.generate_accelerometer_sample(),
|
||
"gyro": self.generate_gyroscope_sample(),
|
||
})
|
||
time.sleep(interval_ms / 1000)
|
||
return samples
|
||
|
||
def inject_to_device(self, duration_sec: float = 5.0, freq_hz: int = 50):
|
||
"""
|
||
向 /dev/input/ 写入伪造传感器事件 (需 root)。
|
||
非 root 环境降级为日志输出。
|
||
"""
|
||
if not self.has_root:
|
||
logger.info("非 root，传感器模拟降级为日志记录")
|
||
self._simulate_log_only(duration_sec, freq_hz)
|
||
return
|
||
|
||
input_dev = self._find_sensor_input_device()
|
||
if not input_dev:
|
||
logger.warning("未找到传感器 input 设备，降级为日志")
|
||
self._simulate_log_only(duration_sec, freq_hz)
|
||
return
|
||
|
||
logger.info(f"向 {input_dev} 注入传感器数据 {duration_sec}s @ {freq_hz}Hz")
|
||
interval = 1.0 / freq_hz
|
||
end_time = time.time() + duration_sec
|
||
count = 0
|
||
|
||
try:
|
||
while time.time() < end_time:
|
||
sample = self.generate_accelerometer_sample()
|
||
self._write_input_event(input_dev, sample)
|
||
time.sleep(interval)
|
||
count += 1
|
||
except Exception as e:
|
||
logger.error(f"传感器注入中断: {e}")
|
||
|
||
logger.info(f"传感器注入完成，写入 {count} 个事件")
|
||
|
||
def get_device_sensor_count(self) -> int:
|
||
"""获取设备实际传感器数量"""
|
||
if not self.d:
|
||
return -1
|
||
try:
|
||
out = self.d.shell(
|
||
"dumpsys sensorservice 2>/dev/null | grep -c 'Sensor'"
|
||
).output.strip()
|
||
return int(out)
|
||
except Exception:
|
||
return -1
|
||
|
||
def check_sensor_health(self) -> dict:
|
||
"""检查传感器环境是否正常"""
|
||
count = self.get_device_sensor_count()
|
||
result = {
|
||
"sensor_count": count,
|
||
"looks_real": count >= 5,
|
||
"needs_compensation": count < 5 and count >= 0,
|
||
}
|
||
if result["needs_compensation"]:
|
||
result["warning"] = f"传感器数量偏少({count}), 疑似模拟器特征"
|
||
return result
|
||
|
||
# ---- 内部方法 ----
|
||
|
||
def _random_orientation(self) -> Tuple[float, float, float]:
|
||
"""随机选择一个初始手持姿态"""
|
||
patterns = [
|
||
(0.3, 0.5, self.GRAVITY - 0.2), # 近似平放
|
||
(0.2, self.GRAVITY * 0.7, self.GRAVITY * 0.7), # 竖持 ~45°
|
||
(0.1, self.GRAVITY - 0.3, 1.0), # 接近竖持
|
||
]
|
||
base = random.choice(patterns)
|
||
return tuple(v + random.gauss(0, 0.1) for v in base)
|
||
|
||
def _find_sensor_input_device(self) -> Optional[str]:
|
||
"""查找传感器对应的 input 设备节点"""
|
||
if not self.d:
|
||
return None
|
||
try:
|
||
out = self.d.shell("ls /dev/input/event* 2>/dev/null").output.strip()
|
||
devices = out.split()
|
||
for dev in devices:
|
||
info = self.d.shell(f"cat /proc/bus/input/devices 2>/dev/null").output
|
||
if "accelerometer" in info.lower() or "accel" in info.lower():
|
||
return dev
|
||
return devices[0] if devices else None
|
||
except Exception:
|
||
return None
|
||
|
||
def _write_input_event(self, device_path: str, sample: dict):
|
||
"""写入一个 input_event 到设备节点"""
|
||
if not self.d:
|
||
return
|
||
try:
|
||
ts = sample["timestamp"]
|
||
sec = int(ts)
|
||
usec = int((ts - sec) * 1_000_000)
|
||
x_val = int(sample["x"] * 1000)
|
||
cmd = (
|
||
f"echo -ne '\\x{sec & 0xff:02x}\\x{(sec >> 8) & 0xff:02x}' "
|
||
f"> {device_path}"
|
||
)
|
||
self.d.shell(cmd)
|
||
except Exception as e:
|
||
logger.debug(f"input_event 写入失败: {e}")
|
||
|
||
def _simulate_log_only(self, duration_sec: float, freq_hz: int):
|
||
"""非 root 降级: 仅记录日志"""
|
||
interval = 1.0 / freq_hz
|
||
end_time = time.time() + duration_sec
|
||
count = 0
|
||
while time.time() < end_time:
|
||
accel = self.generate_accelerometer_sample()
|
||
gyro = self.generate_gyroscope_sample()
|
||
if count % (freq_hz * 2) == 0:
|
||
logger.debug(
|
||
f"sensor[log] accel=({accel['x']:.2f},{accel['y']:.2f},{accel['z']:.2f}) "
|
||
f"gyro=({gyro['x']:.4f},{gyro['y']:.4f},{gyro['z']:.4f})"
|
||
)
|
||
time.sleep(interval)
|
||
count += 1
|