Spaces:
Sleeping
Sleeping
| #!/usr/bin/env python3 | |
| """ | |
| ResearchMate Development Server | |
| A complete Python-based development environment for ResearchMate | |
| """ | |
| import os | |
| import sys | |
| import subprocess | |
| import threading | |
| import time | |
| import logging | |
| from pathlib import Path | |
| from typing import Dict, List, Optional | |
| import signal | |
| import webbrowser | |
| from watchdog.observers import Observer | |
| from watchdog.events import FileSystemEventHandler | |
| import platform | |
| import uvicorn | |
| import socket | |
| # Add the project root to Python path | |
| sys.path.append(str(Path(__file__).parent.parent.parent)) | |
| # Import the main app from main.py | |
| from ResearchMate.app import app | |
| # Setup logging | |
| logging.basicConfig( | |
| level=logging.INFO, | |
| format='%(asctime)s - %(levelname)s - %(message)s', | |
| handlers=[ | |
| logging.FileHandler(Path(__file__).parent.parent.parent / 'logs' / 'development.log'), | |
| logging.StreamHandler() | |
| ] | |
| ) | |
| logger = logging.getLogger(__name__) | |
| class FileChangeHandler(FileSystemEventHandler): | |
| """Handle file changes for auto-reload""" | |
| def __init__(self, callback): | |
| self.callback = callback | |
| self.last_modified = {} | |
| def on_modified(self, event): | |
| if event.is_directory: | |
| return | |
| # Only watch Python files | |
| if not event.src_path.endswith('.py'): | |
| return | |
| # Debounce rapid changes | |
| current_time = time.time() | |
| if event.src_path in self.last_modified: | |
| if current_time - self.last_modified[event.src_path] < 1: | |
| return | |
| self.last_modified[event.src_path] = current_time | |
| logger.info(f"File changed: {event.src_path}") | |
| self.callback() | |
| class ResearchMateDevServer: | |
| """Development server for ResearchMate""" | |
| def __init__(self, project_root: Optional[Path] = None): | |
| self.project_root = project_root or Path(__file__).parent.parent.parent | |
| self.venv_path = self.project_root / "venv" | |
| self.server_thread = None | |
| self.observer = None | |
| self.is_running = False | |
| self.is_windows = platform.system() == "Windows" | |
| # Store server config for restarts | |
| self.server_host = "127.0.0.1" | |
| self.server_port = 8000 | |
| def print_banner(self): | |
| """Print development server banner""" | |
| banner = """ | |
| ResearchMate Development Server | |
| ================================= | |
| AI Research Assistant - Development Mode | |
| Auto-reload enabled for Python files | |
| """ | |
| print(banner) | |
| logger.info("Starting ResearchMate development server") | |
| def get_venv_python(self) -> Path: | |
| """Get path to Python executable in virtual environment""" | |
| # If we're already in a virtual environment (including Conda), use the current Python executable | |
| if sys.prefix != sys.base_prefix or 'CONDA_DEFAULT_ENV' in os.environ: | |
| return Path(sys.executable) | |
| # Otherwise, construct the path to the venv Python executable | |
| if self.is_windows: | |
| return self.venv_path / "Scripts" / "python.exe" | |
| else: | |
| return self.venv_path / "bin" / "python" | |
| def check_virtual_environment(self) -> bool: | |
| """Check if virtual environment exists""" | |
| # Since we're importing directly, just check if we can import the modules | |
| try: | |
| import ResearchMate.app as app | |
| logger.info("Successfully imported main application") | |
| return True | |
| except ImportError as e: | |
| logger.error(f"Failed to import main application: {e}") | |
| logger.error("Make sure you're in the correct environment with all dependencies installed") | |
| return False | |
| def check_port_available(self, host: str, port: int) -> bool: | |
| """Check if a port is available""" | |
| try: | |
| with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: | |
| sock.bind((host, port)) | |
| return True | |
| except OSError: | |
| return False | |
| def find_available_port(self, host: str, start_port: int = 8000, max_attempts: int = 10) -> Optional[int]: | |
| """Find an available port starting from start_port""" | |
| for port in range(start_port, start_port + max_attempts): | |
| if self.check_port_available(host, port): | |
| return port | |
| return None | |
| def start_server_process(self, host: str = "127.0.0.1", port: int = 8000): | |
| """Start the server using uvicorn directly with the imported app""" | |
| try: | |
| # Check if the requested port is available | |
| if not self.check_port_available(host, port): | |
| logger.warning(f"Port {port} is already in use on {host}") | |
| available_port = self.find_available_port(host, port) | |
| if available_port: | |
| logger.info(f"Using available port {available_port} instead") | |
| port = available_port | |
| self.server_port = port # Update stored port | |
| else: | |
| logger.error(f"No available ports found starting from {port}") | |
| return False | |
| logger.info(f"Starting server on {host}:{port}") | |
| # Run uvicorn with the imported app in a separate thread | |
| def run_server(): | |
| uvicorn.run( | |
| app, | |
| host=host, | |
| port=port, | |
| reload=False, # We handle reload ourselves with file watcher | |
| log_level="info" | |
| ) | |
| # Start server in background thread | |
| self.server_thread = threading.Thread(target=run_server, daemon=True) | |
| self.server_thread.start() | |
| # Wait a moment for server to start | |
| time.sleep(2) | |
| logger.info("Server process started successfully") | |
| return True | |
| except Exception as e: | |
| logger.error(f"Failed to start server: {e}") | |
| import traceback | |
| logger.error(f"Traceback: {traceback.format_exc()}") | |
| return False | |
| def stop_server_process(self): | |
| """Stop the server process""" | |
| if self.server_thread: | |
| logger.info("Stopping server...") | |
| # Note: For development, we'll let the thread finish naturally | |
| # In a real implementation, you might want to implement graceful shutdown | |
| self.server_thread = None | |
| def restart_server(self): | |
| """Restart the server""" | |
| logger.info("File change detected - restarting server...") | |
| logger.info("Note: For full restart, please stop and start the dev server manually") | |
| # Note: Auto-restart is complex with embedded uvicorn | |
| # For now, just log the change. User can manually restart. | |
| def setup_file_watcher(self): | |
| """Setup file watcher for auto-reload""" | |
| try: | |
| self.observer = Observer() | |
| # Watch source files | |
| watch_paths = [ | |
| self.project_root / "src", | |
| self.project_root / "main.py" | |
| ] | |
| handler = FileChangeHandler(self.restart_server) | |
| for path in watch_paths: | |
| if path.exists(): | |
| if path.is_file(): | |
| self.observer.schedule(handler, str(path.parent), recursive=False) | |
| else: | |
| self.observer.schedule(handler, str(path), recursive=True) | |
| self.observer.start() | |
| logger.info("File watcher started") | |
| except Exception as e: | |
| logger.error(f"Failed to setup file watcher: {e}") | |
| def stop_file_watcher(self): | |
| """Stop file watcher""" | |
| if self.observer: | |
| self.observer.stop() | |
| self.observer.join() | |
| self.observer = None | |
| def open_browser(self, url: str): | |
| """Open browser after server starts""" | |
| def open_after_delay(): | |
| time.sleep(3) # Wait for server to start | |
| try: | |
| webbrowser.open(url) | |
| logger.info(f"Opened browser at {url}") | |
| except Exception as e: | |
| logger.warning(f"Could not open browser: {e}") | |
| thread = threading.Thread(target=open_after_delay) | |
| thread.daemon = True | |
| thread.start() | |
| def run_tests(self): | |
| """Run project tests""" | |
| try: | |
| logger.info("Running tests...") | |
| logger.info("No tests configured - skipping test run") | |
| except Exception as e: | |
| logger.error(f"Failed to run tests: {e}") | |
| def check_code_quality(self): | |
| """Check code quality with linting""" | |
| try: | |
| logger.info("Checking code quality...") | |
| python_path = self.get_venv_python() | |
| # Run flake8 if available | |
| try: | |
| result = subprocess.run([ | |
| str(python_path), "-m", "flake8", | |
| "src/", "main.py", "--max-line-length=88" | |
| ], cwd=self.project_root, capture_output=True, text=True) | |
| if result.returncode == 0: | |
| logger.info("Code quality checks passed") | |
| else: | |
| logger.warning("Code quality issues found:") | |
| print(result.stdout) | |
| except FileNotFoundError: | |
| logger.info("flake8 not installed, skipping code quality check") | |
| except Exception as e: | |
| logger.error(f"Failed to check code quality: {e}") | |
| def start(self, host: str = "127.0.0.1", port: int = 8000, open_browser: bool = True): | |
| """Start the development server""" | |
| self.print_banner() | |
| if not self.check_virtual_environment(): | |
| return False | |
| # Setup signal handlers | |
| def signal_handler(signum, frame): | |
| logger.info("Received interrupt signal") | |
| self.stop() | |
| sys.exit(0) | |
| signal.signal(signal.SIGINT, signal_handler) | |
| signal.signal(signal.SIGTERM, signal_handler) | |
| try: | |
| self.is_running = True | |
| # Store server config for restarts | |
| self.server_host = host | |
| self.server_port = port | |
| # Start server | |
| if not self.start_server_process(host, port): | |
| return False | |
| # Use the actual port (might have changed if original was busy) | |
| actual_port = self.server_port | |
| # Setup file watcher | |
| self.setup_file_watcher() | |
| # Open browser | |
| if open_browser: | |
| self.open_browser(f"http://{host}:{actual_port}") | |
| logger.info("Development server started successfully!") | |
| logger.info(f"Web Interface: http://{host}:{actual_port}") | |
| logger.info(f"API Documentation: http://{host}:{actual_port}/docs") | |
| logger.info("File watcher enabled (manual restart required for changes)") | |
| logger.info("Use Ctrl+C to stop") | |
| # Keep the main thread alive | |
| while self.is_running: | |
| time.sleep(1) | |
| except KeyboardInterrupt: | |
| logger.info("Server stopped by user") | |
| except Exception as e: | |
| logger.error(f"Development server error: {e}") | |
| finally: | |
| self.stop() | |
| def stop(self): | |
| """Stop the development server""" | |
| self.is_running = False | |
| self.stop_file_watcher() | |
| self.stop_server_process() | |
| logger.info("Development server stopped") | |
| def main(): | |
| """Main development server function""" | |
| import argparse | |
| parser = argparse.ArgumentParser(description="ResearchMate Development Server") | |
| parser.add_argument("--host", default="127.0.0.1", help="Host to bind to") | |
| parser.add_argument("--port", type=int, default=8000, help="Port to bind to") | |
| parser.add_argument("--no-browser", action="store_true", help="Don't open browser") | |
| parser.add_argument("--test", action="store_true", help="Run tests only") | |
| parser.add_argument("--lint", action="store_true", help="Check code quality only") | |
| args = parser.parse_args() | |
| dev_server = ResearchMateDevServer() | |
| if args.test: | |
| dev_server.run_tests() | |
| elif args.lint: | |
| dev_server.check_code_quality() | |
| else: | |
| dev_server.start( | |
| host=args.host, | |
| port=args.port, | |
| open_browser=not args.no_browser | |
| ) | |
| if __name__ == "__main__": | |
| main() | |