voice_input.py - master - nodpc/M_Pro

voice_input.py
421 строка14 KB
nodpc-gh
Initial commit: голосовой ввод текста (VoiceInput)
07 июл 2026, 16:58
07 июл 2026, 16:5838453a2
import ctypes
import ctypes.wintypes
import os
import queue
import re
import threading
import time
import winsound
import keyboard
import numpy as np
import sounddevice as sd
import speech_recognition as sr
from scipy import signal

DEBUG = os.path.expanduser("~/voice_input_debug.txt")

def dbg(msg):
    with open(DEBUG, "a", encoding="utf-8") as f:
        f.write(f"[{time.strftime('%H:%M:%S')}] {msg}\n")

user32 = ctypes.windll.user32
gdi32 = ctypes.windll.gdi32
kernel32 = ctypes.windll.kernel32

user32.FillRect.argtypes = [ctypes.c_void_p, ctypes.c_void_p, ctypes.c_void_p]
user32.FillRect.restype = ctypes.c_int
user32.DefWindowProcW.argtypes = [ctypes.c_void_p, ctypes.c_uint, ctypes.c_uint64, ctypes.c_int64]
user32.DefWindowProcW.restype = ctypes.c_int64
user32.CreateWindowExW.argtypes = [
    ctypes.c_uint, ctypes.c_wchar_p, ctypes.c_wchar_p,
    ctypes.c_uint, ctypes.c_int, ctypes.c_int,
    ctypes.c_int, ctypes.c_int,
    ctypes.c_void_p, ctypes.c_void_p, ctypes.c_void_p, ctypes.c_void_p,
]
user32.CreateWindowExW.restype = ctypes.c_void_p
user32.RegisterHotKey.argtypes = [ctypes.c_void_p, ctypes.c_int, ctypes.c_uint, ctypes.c_uint]
user32.RegisterHotKey.restype = ctypes.c_int
user32.PeekMessageW.argtypes = [ctypes.POINTER(ctypes.wintypes.MSG), ctypes.c_void_p, ctypes.c_uint, ctypes.c_uint, ctypes.c_uint]
user32.PeekMessageW.restype = ctypes.c_bool
user32.TranslateMessage.argtypes = [ctypes.POINTER(ctypes.wintypes.MSG)]
user32.DispatchMessageW.argtypes = [ctypes.POINTER(ctypes.wintypes.MSG)]
user32.GetForegroundWindow.restype = ctypes.c_void_p
user32.GetWindowTextW.argtypes = [ctypes.c_void_p, ctypes.c_wchar_p, ctypes.c_int]
user32.GetWindowTextW.restype = ctypes.c_int
user32.IsWindow.argtypes = [ctypes.c_void_p]
user32.IsWindow.restype = ctypes.c_bool
user32.SetForegroundWindow.argtypes = [ctypes.c_void_p]
user32.SetForegroundWindow.restype = ctypes.c_bool
user32.IsIconic.argtypes = [ctypes.c_void_p]
user32.IsIconic.restype = ctypes.c_bool
user32.ShowWindow.argtypes = [ctypes.c_void_p, ctypes.c_int]
user32.ShowWindow.restype = ctypes.c_bool
user32.EnumWindows.argtypes = [ctypes.c_void_p, ctypes.c_void_p]
user32.EnumWindows.restype = ctypes.c_bool

class WNDCLASSW(ctypes.Structure):
    _fields_ = [
        ("style", ctypes.c_uint),
        ("lpfnWndProc", ctypes.c_void_p),
        ("cbClsExtra", ctypes.c_int),
        ("cbWndExtra", ctypes.c_int),
        ("hInstance", ctypes.c_void_p),
        ("hIcon", ctypes.c_void_p),
        ("hCursor", ctypes.c_void_p),
        ("hbrBackground", ctypes.c_void_p),
        ("lpszMenuName", ctypes.c_wchar_p),
        ("lpszClassName", ctypes.c_wchar_p),
    ]

class PAINTSTRUCT(ctypes.Structure):
    _fields_ = [
        ("hdc", ctypes.c_void_p),
        ("fErase", ctypes.c_int),
        ("rcPaint", ctypes.wintypes.RECT),
        ("fRestore", ctypes.c_int),
        ("fIncUpdate", ctypes.c_int),
        ("rgbReserved", ctypes.c_byte * 32),
    ]

SAMPLE_RATE = 16000
SILENCE_DB = 15
SPEECH_DB = 28
WAKE_WORDS = ("нод", "nod", "not", "нот")
KEYWORD_ENTER = "отправить"
AGENT_BROWSER_KEYWORD = "Агент"

COLORS = {
    "wait": 0xCC6600,
    "listen": 0x33CC33,
    "proc": 0x33CCCC,
    "muted": 0x555555,
}

recognizer = sr.Recognizer()
muted = False
recording = False
buf = []
stop_event = threading.Event()
q = queue.Queue()
status = "wait"
current_sr = 16000
last_type_time = 0
dynamic_target_hwnd = None

WM_HOTKEY = 0x0312
VK_Z = 0x5A
VK_X = 0x58
MOD_CTRL_ALT = 0x0001 | 0x0002

WNDPROC = ctypes.WINFUNCTYPE(ctypes.c_int64, ctypes.c_void_p, ctypes.c_uint, ctypes.c_uint64, ctypes.c_int64)
_wndproc = None
indicator_hwnd = None

def wnd_proc(hwnd, msg, wParam, lParam):
    if msg == 0x000F:
        ps = PAINTSTRUCT()
        hdc = user32.BeginPaint(hwnd, ctypes.byref(ps))
        rect = ctypes.wintypes.RECT()
        user32.GetClientRect(hwnd, ctypes.byref(rect))
        color = COLORS.get(status, 0xCC6600)
        brush = gdi32.CreateSolidBrush(color)
        user32.FillRect(hdc, ctypes.byref(rect), brush)
        gdi32.DeleteObject(brush)
        user32.EndPaint(hwnd, ctypes.byref(ps))
        return 0
    if msg == 0x0002:
        return 0
    return user32.DefWindowProcW(hwnd, msg, wParam, lParam)

def create_indicator():
    global _wndproc, indicator_hwnd
    _wndproc = WNDPROC(wnd_proc)
    wc = WNDCLASSW()
    wc.style = 0
    wc.lpfnWndProc = ctypes.cast(_wndproc, ctypes.c_void_p)
    wc.cbClsExtra = 0
    wc.cbWndExtra = 0
    hmod = kernel32.GetModuleHandleW(None)
    wc.hInstance = ctypes.c_void_p(hmod & 0xFFFFFFFFFFFFFFFF)
    wc.hIcon = None
    wc.hCursor = None
    wc.hbrBackground = None
    wc.lpszMenuName = None
    wc.lpszClassName = "VoiceInputInd"
    if not user32.RegisterClassW(ctypes.byref(wc)):
        return False
    sw = user32.GetSystemMetrics(0)
    cx, cy = 120, 20
    x = sw // 2 - cx // 2
    y = 0
    indicator_hwnd = user32.CreateWindowExW(
        0x00000008 | 0x00000080,
        "VoiceInputInd", None,
        0x80000000,
        x, y, cx, cy,
        None, None, wc.hInstance, None,
    )
    if not indicator_hwnd:
        return False
    user32.ShowWindow(indicator_hwnd, 1)
    user32.UpdateWindow(indicator_hwnd)
    return True

def update_bar():
    if indicator_hwnd:
        user32.InvalidateRect(indicator_hwnd, None, True)
        user32.UpdateWindow(indicator_hwnd)

def set_status(s):
    global status
    status = s
    update_bar()

def rms_db(block_int16):
    rms = np.sqrt(np.mean(block_int16.astype(np.float64) ** 2))
    return 20 * np.log10(max(rms, 1e-10))

def callback(indata, frames, time_info, status_flags):
    if not status_flags:
        q.put(indata.copy())

def type_text(text):
    if not text.strip():
        return
    keyboard.write(text, delay=0.005)

def recognize(audio_int16):
    try:
        data = audio_int16
        if current_sr == 48000:
            data = data[::3]
        elif current_sr != 16000:
            ratio = 16000 / current_sr
            new_len = int(len(data) * ratio)
            data = signal.resample(data, new_len).astype(np.int16)
        raw = data.tobytes()
        audio_data = sr.AudioData(raw, 16000, 2)
        text = recognizer.recognize_google(audio_data, language="ru-RU")
        return text
    except sr.UnknownValueError:
        return ""
    except sr.RequestError as e:
        dbg(f"RequestError: {e}")
        return ""

def force_focus_window(hwnd):
    if not hwnd or not user32.IsWindow(hwnd):
        return False
    if user32.IsIconic(hwnd):
        user32.ShowWindow(hwnd, 9)
    user32.SetForegroundWindow(hwnd)
    time.sleep(0.15)
    return True

def find_browser_agent_hwnd():
    found_hwnd = [None]
    def enum_windows_callback(hwnd, lParam):
        buf = ctypes.create_unicode_buffer(512)
        user32.GetWindowTextW(hwnd, buf, 512)
        if AGENT_BROWSER_KEYWORD.lower() in buf.value.lower():
            found_hwnd[0] = hwnd
            return False
        return True
    WNDENUMPROC = ctypes.WINFUNCTYPE(ctypes.c_bool, ctypes.c_void_p, ctypes.c_void_p)
    user32.EnumWindows(WNDENUMPROC(enum_windows_callback), 0)
    return found_hwnd[0]

def set_custom_target_window():
    global dynamic_target_hwnd
    current_hwnd = user32.GetForegroundWindow()
    if current_hwnd:
        buf = ctypes.create_unicode_buffer(256)
        user32.GetWindowTextW(current_hwnd, buf, 256)
        if AGENT_BROWSER_KEYWORD.lower() in buf.value.lower():
            dynamic_target_hwnd = None
            winsound.Beep(800, 150)
            dbg("Target reset to default: Agent in browser (Priority 1)")
        else:
            dynamic_target_hwnd = current_hwnd
            winsound.Beep(600, 150)
            dbg(f"Custom priority target locked: '{buf.value}' (HWND: {current_hwnd})")

def handle_text(text):
    global last_type_time
    dbg(f"Recognized: '{text}'")
    if not text:
        return
    cmd = text.strip()
    if not cmd:
        return
    cmd_lower = cmd.lower()
    is_enter_needed = KEYWORD_ENTER in cmd_lower

    enter_pattern = rf"\s*,?\s*\b{re.escape(KEYWORD_ENTER)}\b\s*,?\s*"
    cmd = re.sub(enter_pattern, " ", cmd, flags=re.IGNORECASE)
    for wake_word in WAKE_WORDS:
        wake_pattern = rf"\b{re.escape(wake_word)}\b\s*,?\s*"
        cmd = re.sub(wake_pattern, "", cmd, flags=re.IGNORECASE)
    cmd = cmd.strip()
    cmd = re.sub(r"\s+", " ", cmd)
    if not cmd:
        dbg("Text is empty after removing keywords.")
        return

    target_focused = False
    if dynamic_target_hwnd and user32.IsWindow(dynamic_target_hwnd):
        target_focused = force_focus_window(dynamic_target_hwnd)
        if target_focused:
            dbg("Text sent to Custom Window assigned by Hotkey")
    if not target_focused:
        agent_hwnd = find_browser_agent_hwnd()
        if agent_hwnd:
            force_focus_window(agent_hwnd)
            dbg("Text sent to Priority 1: Agent Browser Tab")
        else:
            dbg("Priority 1 Window not found. Typing into current active window.")

    now = time.time()
    if 0 < now - last_type_time < 5:
        cmd = " " + cmd
    last_type_time = now

    dbg(f"Typing clean cmd: '{cmd}', press enter={is_enter_needed}")
    if is_enter_needed:
        keyboard.write(cmd, delay=0.005)
        time.sleep(0.2)
        keyboard.send("enter")
    else:
        type_text(cmd)

def process():
    global recording, buf
    silence = 0
    log_n = 0
    max_level = -100
    dbg("Process thread running")
    while not stop_event.is_set():
        try:
            try:
                chunk = q.get(timeout=0.3)
            except queue.Empty:
                if recording:
                    silence += 1
                    if silence > 20:
                        audio = np.concatenate(buf)
                        recording = False
                        silence = 0
                        dur = len(audio) / current_sr
                        if dur > 0.3:
                            set_status("proc")
                            dbg(f"Sending to Google ({dur:.1f}s)...")
                            text = recognize(audio)
                            dbg(f"Google result: '{text}'")
                            handle_text(text)
                        set_status("wait" if not muted else "muted")
                continue
            if muted:
                continue
            level = rms_db(chunk)
            if level > max_level:
                max_level = level
            log_n += 1
            if log_n >= 50:
                dbg(f"Max audio level last ~3s: {max_level:.1f} dB, recording={recording}")
                log_n = 0
                max_level = -100
            if level > SPEECH_DB and not recording:
                recording = True
                buf = [chunk]
                silence = 0
                dbg(f"Start recording (level={level:.1f} dB)")
                set_status("listen")
            elif recording:
                buf.append(chunk)
                if level < SILENCE_DB:
                    silence += 1
                else:
                    silence = 0
                total = len(np.concatenate(buf)) / current_sr
                if silence > 20 or total > 12:
                    audio = np.concatenate(buf)
                    recording = False
                    silence = 0
                    dur = len(audio) / current_sr
                    if dur > 0.3:
                        set_status("proc")
                        dbg(f"Sending to Google ({dur:.1f}s)...")
                        text = recognize(audio)
                        dbg(f"Google result: '{text}'")
                        handle_text(text)
                    set_status("wait" if not muted else "muted")
        except Exception as e:
            dbg(f"Process error: {e}")

def toggle_mute():
    global muted, recording, buf
    muted = not muted
    recording = False
    buf = []
    set_status("muted" if muted else "wait")
    winsound.Beep(1000 if not muted else 600, 100)
    dbg(f"Mute toggled. Muted={muted}")

def msg_loop():
    msg = ctypes.wintypes.MSG()
    while not stop_event.is_set():
        ret = user32.PeekMessageW(ctypes.byref(msg), None, 0, 0, 1)
        if ret:
            if msg.message == 0x0012:
                break
            if msg.message == WM_HOTKEY:
                dbg(f"WM_HOTKEY id={msg.wParam}")
                if msg.wParam == 1:
                    toggle_mute()
                elif msg.wParam == 2:
                    stop_event.set()
            else:
                user32.TranslateMessage(ctypes.byref(msg))
                user32.DispatchMessageW(ctypes.byref(msg))
        else:
            time.sleep(0.05)

def main():
    with open(DEBUG, "w", encoding="utf-8") as f:
        f.write("=== Voice Input (continuous) ===\n")
    if not create_indicator():
        dbg("Indicator FAILED!")
        return
    set_status("wait")
    stream = None
    devices_to_try = [(12, 48000), (None, 16000), (0, 44100), (6, 44100), (14, 44100), (1, 44100)]
    for dev, sr in devices_to_try:
        try:
            stream = sd.InputStream(
                samplerate=sr, channels=1, dtype="int16",
                callback=callback, device=dev,
            )
            stream.start()
            dbg(f"Audio stream OK on device={dev} sr={sr}")
            global current_sr
            current_sr = sr
            break
        except Exception as e:
            dbg(f"Audio device={dev} sr={sr} failed: {e}")
            continue
    if not stream:
        dbg("All audio devices failed!")
        return
    t = threading.Thread(target=process, daemon=True)
    t.start()
    r1 = user32.RegisterHotKey(indicator_hwnd, 1, MOD_CTRL_ALT | 0x4000, VK_Z)
    r2 = user32.RegisterHotKey(indicator_hwnd, 2, MOD_CTRL_ALT | 0x4000, VK_X)
    dbg(f"RegisterHotKey Z={r1} X={r2}")
    keyboard.add_hotkey("ctrl+alt+s", set_custom_target_window)
    dbg("Hotkeys: Ctrl+Alt+Z=mute, Ctrl+Alt+X=quit, Ctrl+Alt+S=set target window")
    dbg("Entering msg_loop")
    msg_loop()
    stream.stop()

if __name__ == "__main__":
    main()