183 lines
7.7 KiB
Python
183 lines
7.7 KiB
Python
from multiprocessing import Process, shared_memory, Value
|
|
from pynput.mouse import Controller
|
|
import numpy as np
|
|
import time
|
|
import sounddevice as sd
|
|
from .args import args
|
|
from .utils.shared_mem_guard import SharedMemoryGuard
|
|
from .utils.timer_wait import wait_until
|
|
from OneEuroFilter import OneEuroFilter
|
|
from .utils.fps import FPS
|
|
from .utils.timer_wait import wait_until
|
|
|
|
class MouseClientProcess(Process):
|
|
def __init__(self, pose_position_shm: shared_memory.SharedMemory):
|
|
super().__init__()
|
|
self.pose_position_shm = pose_position_shm
|
|
self.fps = Value('f', 60.0)
|
|
self.audio_volume = Value('f', 0.0)
|
|
self.audio_running = False
|
|
self.audio_callback_fps = FPS(60)
|
|
|
|
def audio_callback(self, indata, frames, time_info, status):
|
|
"""音频回调函数，计算音量"""
|
|
if status:
|
|
print(f"Audio status: {status}")
|
|
|
|
self.audio_callback_fps()
|
|
# 计算 RMS (均方根) 音量
|
|
volume_norm = np.linalg.norm(indata) * 10
|
|
|
|
# 应用阈值和灵敏度
|
|
if volume_norm < args.audio_threshold:
|
|
volume_norm = 0
|
|
else:
|
|
volume_norm = (volume_norm - args.audio_threshold) * args.audio_sensitivity
|
|
# 限制在 0-1 范围
|
|
volume_norm = np.clip(volume_norm, 0, 1)
|
|
self.audio_volume.value = volume_norm
|
|
|
|
def run(self):
|
|
mouse = Controller()
|
|
pose_position_shm_guard = SharedMemoryGuard(self.pose_position_shm, ctrl_name="pose_position_shm_ctrl")
|
|
np_pose_shm = np.ndarray((45,), dtype=np.float32, buffer=self.pose_position_shm.buf[:45 * 4])
|
|
np_position_shm = np.ndarray((4,), dtype=np.float32, buffer=self.pose_position_shm.buf[45 * 4:45 * 4 + 4 * 4])
|
|
|
|
# posLimit = [0, 0, 1920, 1080]
|
|
posLimit = [int(x) for x in args.mouse_input.split(',')]
|
|
|
|
if args.mouse_audio_input:
|
|
# 启动音频流 (使用 WASAPI)
|
|
try:
|
|
# 获取默认输入设备（麦克风或系统音频）
|
|
# 如果要捕获系统音频输出，需要使用 loopback 设备
|
|
audio_stream = sd.InputStream(
|
|
callback=self.audio_callback,
|
|
channels=1,
|
|
samplerate=16000,
|
|
blocksize=1024,
|
|
)
|
|
audio_stream.start()
|
|
self.audio_running = True
|
|
wait_until(time.perf_counter() + 1.0) # 等待一秒以稳定音频输入
|
|
print("Audio capture started (WASAPI)")
|
|
except Exception as e:
|
|
print(f"Failed to start audio capture: {e}")
|
|
print("Continuing without audio input...")
|
|
self.audio_running = False
|
|
audio_filter = OneEuroFilter(freq=self.audio_callback_fps.view(), mincutoff=10.0, beta=0.0)
|
|
|
|
prev = {
|
|
'eye_l_h_temp': 0,
|
|
'eye_r_h_temp': 0,
|
|
'mouth_ratio': 0,
|
|
'eye_y_ratio': 0,
|
|
'eye_x_ratio': 0,
|
|
'x_angle': 0,
|
|
'y_angle': 0,
|
|
'z_angle': 0,
|
|
}
|
|
last_time : float = time.perf_counter()
|
|
interval : float = 1.0 / 60 # 60 FPS
|
|
|
|
# 眨眼相关参数
|
|
blink_interval = args.blink_interval # 眨眼间隔时间（秒）
|
|
blink_duration = 0.25 # 眨眼持续时间（秒）
|
|
last_blink_time = time.perf_counter()
|
|
|
|
# 呼吸循环参数
|
|
breath_start_time = time.perf_counter()
|
|
|
|
print("Mouse Input Running at 60 FPS")
|
|
position_vector = np.array([0, 0, 0, 1], dtype=np.float32)
|
|
while True:
|
|
pos = mouse.position
|
|
# print(pos)
|
|
eye_limit = [0.8, 0.5]
|
|
head_eye_reduce = 0.6
|
|
head_slowness = 0.2
|
|
|
|
# 计算眨眼效果
|
|
current_time = time.perf_counter()
|
|
time_since_last_blink = current_time - last_blink_time
|
|
|
|
# 每隔 blink_interval 秒眨一次眼
|
|
if time_since_last_blink >= blink_interval:
|
|
last_blink_time = current_time
|
|
time_since_last_blink = 0
|
|
|
|
# 计算眨眼动画（使用尖锐的 sin 函数）
|
|
if time_since_last_blink < blink_duration:
|
|
# 在眨眼持续时间内，使用 sin 函数从 0 升到接近 1 再返回 0
|
|
# 使用 sin^2 使曲线更尖锐
|
|
blink_progress = time_since_last_blink / blink_duration * np.pi
|
|
blink_value = np.sin(blink_progress) ** 2
|
|
eye_close = blink_value * 1.0 # 升到1，返回到 0
|
|
else:
|
|
eye_close = 0.0
|
|
|
|
# 计算呼吸效果（使用 sin 函数，在 breath_cycle 时间内从 0 到 1 再到 0）
|
|
breath_elapsed = (current_time - breath_start_time) % args.breath_cycle
|
|
# 使用 sin 函数，让值在一个周期内从 0 -> 1 -> 0
|
|
# sin 在 0 到 π 之间从 0 到 1 到 0
|
|
breath_value = np.sin(breath_elapsed / args.breath_cycle * np.pi)
|
|
|
|
# 获取音频音量并映射到 mouth_ratio
|
|
current_mouth_ratio = audio_filter(self.audio_volume.value, time.perf_counter()) if self.audio_running else 0
|
|
mouse_data = {
|
|
'eye_l_h_temp': eye_close,
|
|
'eye_r_h_temp': eye_close,
|
|
'mouth_ratio': current_mouth_ratio,
|
|
'eye_y_ratio': np.interp(pos[1], [posLimit[1], posLimit[3]], [1, -1]) * eye_limit[1],
|
|
'eye_x_ratio': np.interp(pos[0], [posLimit[0], posLimit[2]], [1, -1]) * eye_limit[0],
|
|
'x_angle': np.interp(pos[1], [posLimit[1], posLimit[3]], [1, -1]),
|
|
'y_angle': np.interp(pos[0], [posLimit[0], posLimit[2]], [1, -1]),
|
|
'z_angle': 0,
|
|
}
|
|
mouse_data['x_angle'] = np.interp(head_slowness, [0, 1], [prev['x_angle'], mouse_data['x_angle']])
|
|
mouse_data['y_angle'] = np.interp(head_slowness, [0, 1], [prev['y_angle'], mouse_data['y_angle']])
|
|
mouse_data['eye_y_ratio'] -= mouse_data['x_angle'] * eye_limit[1] * head_eye_reduce
|
|
mouse_data['eye_x_ratio'] -= mouse_data['y_angle'] * eye_limit[0] * head_eye_reduce
|
|
if args.bongo:
|
|
mouse_data['y_angle'] += 0.05
|
|
mouse_data['x_angle'] += 0.05
|
|
prev = mouse_data
|
|
|
|
eye_l_h_temp = mouse_data['eye_l_h_temp']
|
|
eye_r_h_temp = mouse_data['eye_r_h_temp']
|
|
mouth_ratio = mouse_data['mouth_ratio']
|
|
eye_y_ratio = mouse_data['eye_y_ratio']
|
|
eye_x_ratio = mouse_data['eye_x_ratio']
|
|
x_angle = mouse_data['x_angle']
|
|
y_angle = mouse_data['y_angle']
|
|
z_angle = mouse_data['z_angle']
|
|
|
|
eyebrow_vector = [0.0] * 12
|
|
mouth_eye_vector = [0.0] * 27
|
|
pose_vector = [0.0] * 6
|
|
|
|
mouth_eye_vector[2] = eye_l_h_temp
|
|
mouth_eye_vector[3] = eye_r_h_temp
|
|
|
|
mouth_eye_vector[14] = mouth_ratio * 1.5
|
|
|
|
mouth_eye_vector[25] = eye_y_ratio
|
|
mouth_eye_vector[26] = eye_x_ratio
|
|
|
|
pose_vector[0] = x_angle
|
|
pose_vector[1] = y_angle
|
|
pose_vector[2] = z_angle
|
|
pose_vector[3] = pose_vector[1]
|
|
pose_vector[4] = pose_vector[2]
|
|
pose_vector[5] = breath_value
|
|
|
|
model_input_arr = eyebrow_vector
|
|
model_input_arr.extend(mouth_eye_vector)
|
|
model_input_arr.extend(pose_vector)
|
|
|
|
with pose_position_shm_guard.lock():
|
|
np_pose_shm[:] = np.array(model_input_arr, dtype=np.float32)
|
|
np_position_shm[:] = position_vector
|
|
|
|
wait_until(last_time + interval)
|
|
last_time += interval |