Update research_manager.py for windows

Replaced termios-based input handling with msvcrt
Modified TerminalUI class
Added Windows-specific NonBlockingInput class
Updated get_multiline_conversation_input method
Removed Unix-specific imports
This commit is contained in:
Hafeez 2024-11-20 21:46:46 +05:30 committed by GitHub
parent aa73019d70
commit c53dddc457
No known key found for this signature in database
GPG key ID: B5690EEEBB952194

View file

@ -1,3 +1,4 @@
import msvcrt
import os import os
import sys import sys
import threading import threading
@ -14,17 +15,15 @@ from datetime import datetime
from io import StringIO from io import StringIO
from colorama import init, Fore, Style from colorama import init, Fore, Style
import select import select
import termios
import tty
from threading import Event from threading import Event
from urllib.parse import urlparse from urllib.parse import urlparse
from pathlib import Path from pathlib import Path
# Initialize colorama for cross-platform color support # Initialize colorama for cross-platform color support
if os.name == 'nt': # Windows-specific initialization if os.name == 'nt':
init(convert=True, strip=False, wrap=True) import msvcrt
else: else:
init() raise ImportError("This version is Windows-specific. Please use the Unix version for other operating systems.")
# Set up logging # Set up logging
log_directory = 'logs' log_directory = 'logs'
@ -479,7 +478,7 @@ class TerminalUI:
pass pass
def get_input(self, prompt: Optional[str] = None) -> Optional[str]: def get_input(self, prompt: Optional[str] = None) -> Optional[str]:
"""Enhanced input handling with mouse scroll support""" """Windows-compatible input handling"""
try: try:
if prompt: if prompt:
self.update_status(prompt) self.update_status(prompt)
@ -492,18 +491,10 @@ class TerminalUI:
if self.should_terminate.is_set(): if self.should_terminate.is_set():
return None return None
try: if msvcrt.kbhit():
ch = self.input_win.getch() ch = msvcrt.getch()
if ch == curses.KEY_MOUSE: if ch == b'\x04': # Ctrl+D
try:
mouse_event = curses.getmouse()
# Ignore mouse events entirely for now
continue
except curses.error:
continue
if ch == 4: # Ctrl+D
result = self.input_buffer.strip() result = self.input_buffer.strip()
self.input_buffer = "" self.input_buffer = ""
if not result: if not result:
@ -511,32 +502,25 @@ class TerminalUI:
return "@quit" return "@quit"
return result return result
elif ch == 3: # Ctrl+C elif ch == b'\x03': # Ctrl+C
self.should_terminate.set() self.should_terminate.set()
self.cleanup() self.cleanup()
return "@quit" return "@quit"
elif ch == ord('\n'): # Enter elif ch == b'\r': # Enter
result = self.input_buffer.strip() result = self.input_buffer.strip()
if result: if result:
self.input_buffer = "" self.input_buffer = ""
return result return result
continue continue
elif ch == curses.KEY_BACKSPACE or ch == 127: # Backspace elif ch == b'\x08': # Backspace
if self.input_buffer: if self.input_buffer:
self.input_buffer = self.input_buffer[:-1] self.input_buffer = self.input_buffer[:-1]
self._refresh_input_prompt() self._refresh_input_prompt()
elif 32 <= ch <= 126: # Printable characters elif 32 <= ord(ch[0]) <= 126: # Printable characters
self.input_buffer += chr(ch) self.input_buffer += ch.decode('utf-8')
self._refresh_input_prompt()
except KeyboardInterrupt:
self.should_terminate.set()
self.cleanup()
return "@quit"
except curses.error:
self._refresh_input_prompt() self._refresh_input_prompt()
except Exception as e: except Exception as e:
@ -559,33 +543,21 @@ class TerminalUI:
os._exit(0) # Ensure exit os._exit(0) # Ensure exit
class NonBlockingInput: class NonBlockingInput:
"""Handles non-blocking keyboard input for Unix-like systems""" """Windows-compatible non-blocking input"""
def __init__(self): def __init__(self):
self.old_settings = None pass
def __enter__(self): def __enter__(self):
if os.name == 'nt': # Windows
return self
self.old_settings = termios.tcgetattr(sys.stdin)
tty.setcbreak(sys.stdin.fileno())
return self return self
def __exit__(self, type, value, traceback): def __exit__(self, type, value, traceback):
if os.name != 'nt': # Unix-like pass
termios.tcsetattr(sys.stdin, termios.TCSADRAIN, self.old_settings)
def check_input(self, timeout=0.1): def check_input(self, timeout=0.1):
"""Check for input without blocking, cross-platform""" """Check for input without blocking"""
if os.name == 'nt': # Windows
import msvcrt
if msvcrt.kbhit(): if msvcrt.kbhit():
return msvcrt.getch().decode('utf-8') return msvcrt.getch().decode('utf-8')
return None return None
else: # Unix-like
ready_to_read, _, _ = select.select([sys.stdin], [], [], timeout)
if ready_to_read:
return sys.stdin.read(1)
return None
class ResearchManager: class ResearchManager:
"""Manages the research process including analysis, search, and documentation""" """Manages the research process including analysis, search, and documentation"""
@ -1380,55 +1352,48 @@ Answer:
logger.error(f"Error generating response: {str(e)}") logger.error(f"Error generating response: {str(e)}")
return f"I apologize, but I encountered an error processing your question: {str(e)}" return f"I apologize, but I encountered an error processing your question: {str(e)}"
def get_multiline_conversation_input(self) -> str: def get_multiline_conversation_input(self) -> str:
"""Get multiline input with CTRL+D handling for conversation mode""" """Windows-compatible multiline input"""
buffer = [] buffer = []
current_line = []
# Save original terminal settings
fd = sys.stdin.fileno()
old_settings = termios.tcgetattr(fd)
try: try:
# Set terminal to raw mode
tty.setraw(fd)
current_line = []
while True: while True:
char = sys.stdin.read(1) if msvcrt.kbhit():
char = msvcrt.getch()
# CTRL+D detection # CTRL+D or CTRL+Z detection
if not char or ord(char) == 4: # EOF or CTRL+D if char in [b'\x04', b'\x1a']:
sys.stdout.write('\n') sys.stdout.write('\n')
if current_line: if current_line:
buffer.append(''.join(current_line)) buffer.append(''.join(current_line))
return ' '.join(buffer).strip() return ' '.join(buffer).strip()
# Handle special characters # Handle special characters
elif ord(char) == 13: # Enter elif char == b'\r': # Enter
sys.stdout.write('\n') sys.stdout.write('\n')
buffer.append(''.join(current_line)) buffer.append(''.join(current_line))
current_line = [] current_line = []
elif ord(char) == 127: # Backspace elif char == b'\x08': # Backspace
if current_line: if current_line:
current_line.pop() current_line.pop()
sys.stdout.write('\b \b') sys.stdout.write('\b \b')
elif ord(char) == 3: # CTRL+C elif char == b'\x03': # CTRL+C
sys.stdout.write('\n') sys.stdout.write('\n')
return 'quit' return 'quit'
# Normal character # Normal character
elif 32 <= ord(char) <= 126: # Printable characters elif 32 <= ord(char[0]) <= 126:
current_line.append(char) current_line.append(char.decode('utf-8'))
sys.stdout.write(char) sys.stdout.write(char.decode('utf-8'))
sys.stdout.flush() sys.stdout.flush()
finally: except Exception as e:
# Restore terminal settings logger.error(f"Error in multiline input: {str(e)}")
termios.tcsetattr(fd, termios.TCSADRAIN, old_settings) return 'quit'
print() # New line for clean display
if __name__ == "__main__": if __name__ == "__main__":
from llm_wrapper import LLMWrapper from llm_wrapper import LLMWrapper