-
Notifications
You must be signed in to change notification settings - Fork 0
Performance Tips
Dragon edited this page Dec 22, 2025
·
1 revision
Maximize speed, reduce memory usage, and optimize your NaviDuck experience with these performance tips.
Last updated: 12/22/2025
- ⚡ Immediate Speed Boosts
- 💾 Memory Optimization
- 🌐 Network Performance
- 🔍 Search Optimization
- 🤖 AI Performance
- 🎯 UI Responsiveness
- 🛠️ Advanced Optimizations
- 📊 Benchmarking
| Action | Expected Improvement | Time Required |
|---|---|---|
| Switch to ddg_api engine | 50-70% faster searches | 30 seconds |
| Clear history/bookmarks | 20% faster startup | 10 seconds |
| Enable search caching | 60% faster repeat searches | 1 minute |
| Disable unused engines | 15% faster searches | 30 seconds |
| Use faster terminal | 40% better UI response | 5 minutes |
# Benchmark results (average response time):
# 1. ddg_api 0.8-1.2 seconds (JSON API, fastest)
# 2. wikipedia 1.2-1.8 seconds (API)
# 3. brave 2.0-3.5 seconds (HTML)
# 4. google 3.0-5.0+ seconds (Often has CAPTCHA delays)
# 5. ddg 4.0-6.0+ seconds (HTML + CAPTCHA issues)
# Change default to fastest:
settings
# Select option 1 → choose "DuckDuckGo API"# In SEARCH_ENGINES dictionary, add performance settings:
"ddg_api": {
"name": "DuckDuckGo API",
"url": "https://api.duckduckgo.com/",
"params": {"q": "{query}", "format": "json", "no_html": "1"},
"icon": "DDG",
"requires_tor": False,
"enabled": True,
"type": "api",
"timeout": 3, # Lower timeout for faster failures
"cache_ttl": 300, # Cache for 5 minutes
"priority": 1 # Highest priority
},# In NaviDuck:
settings → option 4 # Clear history
settings → option 5 # Clear bookmarks
# Or manually clean config files:
rm ~/.naviduck_data.json
rm ~/.naviduck_config.json
# Restart NaviDuck for fresh start# save as cleanup.py and run
import json
import os
def cleanup_history(max_entries=100):
data_file = os.path.expanduser("~/.naviduck_data.json")
if os.path.exists(data_file):
with open(data_file, 'r') as f:
data = json.load(f)
# Keep only recent entries
data['history'] = data['history'][-max_entries:]
with open(data_file, 'w') as f:
json.dump(data, f, indent=2)
print(f"Reduced history to {max_entries} entries")
cleanup_history(100) # Keep only 100 most recent# Windows: Use Windows Terminal (not CMD/PowerShell)
# Features: GPU acceleration, better font rendering
# Linux: Use a performant terminal:
# Fastest options:
# 1. Alacritty (GPU accelerated)
# 2. Kitty (GPU accelerated)
# 3. WezTerm (GPU accelerated)
# Mac: Use iTerm2 with GPU rendering enabled
# Disable animations/effects:
# In terminal settings, disable:
# - Blinking cursor
# - Visual bell
# - Scroll animations
# - Transparency effects# Use monospace fonts with good Unicode support:
# Recommended fonts (fast rendering):
# 1. Cascadia Code (Windows)
# 2. JetBrains Mono (Cross-platform)
# 3. Fira Code (Cross-platform)
# 4. Source Code Pro (Cross-platform)
# Avoid: Consolas (slow Unicode), Comic Sans (just no)# Disable slow/blocked engines:
engines
# Then: disable [number] for google, ddg if not needed
# Keep only what you use:
# Minimum recommended: brave, ddg_api, wikipedia# Create performance profile in config:
performance_profile = {
"minimal": {
"engines": ["brave", "ddg_api"],
"cache_enabled": True,
"history_size": 50,
"ai_timeout": 2,
"tor_enabled": False
},
"balanced": {
"engines": ["brave", "ddg_api", "wikipedia"],
"cache_enabled": True,
"history_size": 100,
"ai_timeout": 5,
"tor_enabled": False
},
"full": {
"engines": ["brave", "ddg_api", "wikipedia", "google", "ddg"],
"cache_enabled": True,
"history_size": 500,
"ai_timeout": 10,
"tor_enabled": False # Tor slows everything
}
}# In BrowserState.load_data():
def load_data(self):
if os.path.exists(self.data_file):
try:
with open(self.data_file, 'r', encoding='utf-8') as f:
data = json.load(f)
# Keep only last N entries
self.history = data.get('history', [])[-500:] # Limit to 500
self.bookmarks = data.get('bookmarks', [])[:100] # Limit to 100
except:
pass# Load data only when needed
class LazyBrowserState(BrowserState):
def __init__(self):
self._history_loaded = False
self._bookmarks_loaded = False
self._history = []
self._bookmarks = []
@property
def history(self):
if not self._history_loaded:
self._load_history()
return self._history
@property
def bookmarks(self):
if not self._bookmarks_loaded:
self._load_bookmarks()
return self._bookmarks# Add cleanup method
def cleanup_memory(self):
"""Clear cached data to free memory"""
import gc
# Clear large variables
self.current_page = ""
self.current_results = []
# Force garbage collection
gc.collect()
# Clear Python's internal caches
import sys
if hasattr(sys, 'getallocatedblocks'):
# Python 3.4+
allocated_before = sys.getallocatedblocks()
gc.collect()
allocated_after = sys.getallocatedblocks()
print(f"Freed {allocated_before - allocated_after} blocks")# Add to UIManager.show_banner():
import psutil
import os
def show_memory_usage(self):
process = psutil.Process(os.getpid())
memory_mb = process.memory_info().rss / 1024 / 1024
if memory_mb > 100: # If using more than 100MB
print(f"{Colors.WARNING}⚠️ High memory: {memory_mb:.1f} MB{Colors.RESET}")
print(f"{Colors.INFO}Tip: Clear history or restart to free memory{Colors.RESET}")
return f"{Colors.GRAY}Memory: {memory_mb:.1f} MB{Colors.RESET}"# Track memory usage over time
class MemoryMonitor:
def __init__(self):
self.samples = []
self.leak_threshold = 10 # MB increase threshold
def sample(self):
import psutil
process = psutil.Process(os.getpid())
self.samples.append(process.memory_info().rss / 1024 / 1024)
# Keep only last 10 samples
if len(self.samples) > 10:
self.samples.pop(0)
# Check for leak
if len(self.samples) == 10:
increase = self.samples[-1] - self.samples[0]
if increase > self.leak_threshold:
print(f"⚠️ Possible memory leak: +{increase:.1f} MB")
return False
return True# Replace lists with more efficient structures
from collections import deque
class OptimizedBrowserState(BrowserState):
def __init__(self):
super().__init__()
# Use deque for history (faster appends/pops)
self.history = deque(maxlen=1000) # Fixed size, auto-truncates
# Use set for fast bookmark lookup
self._bookmark_urls = set()
def add_bookmark(self, title, url):
if url not in self._bookmark_urls:
self.bookmarks.append({
'title': title[:80],
'url': url,
'added': datetime.now().isoformat()
})
self._bookmark_urls.add(url)
self.save_data()
return True
return False# Compress JSON data on disk
import gzip
import json
def save_compressed_data(self):
data = {
'history': list(self.history),
'bookmarks': self.bookmarks,
}
# Compress before saving
with gzip.open(self.data_file + '.gz', 'wt', encoding='utf-8') as f:
json.dump(data, f, separators=(',', ':')) # Minify JSON
# Also keep uncompressed for compatibility
with open(self.data_file, 'w', encoding='utf-8') as f:
json.dump(data, f, indent=2)
def load_compressed_data(self):
# Try compressed first, then uncompressed
compressed_file = self.data_file + '.gz'
if os.path.exists(compressed_file):
with gzip.open(compressed_file, 'rt', encoding='utf-8') as f:
data = json.load(f)
elif os.path.exists(self.data_file):
with open(self.data_file, 'r', encoding='utf-8') as f:
data = json.load(f)
else:
return
self.history = data.get('history', [])[-500:]
self.bookmarks = data.get('bookmarks', [])# Use faster DNS resolution
import socket
def set_fast_dns():
# Use Cloudflare or Google DNS
fast_dns_servers = ['1.1.1.1', '8.8.8.8', '1.0.0.1', '8.8.4.4']
# Create custom resolver
import dns.resolver # pip install dnspython
resolver = dns.resolver.Resolver()
resolver.nameservers = fast_dns_servers
# Override socket.getaddrinfo
original_getaddrinfo = socket.getaddrinfo
def fast_getaddrinfo(host, port, family=0, type=0, proto=0, flags=0):
try:
# Try fast DNS first
answers = resolver.resolve(host, 'A')
for answer in answers:
return [(socket.AF_INET, socket.SOCK_STREAM, 6, '', (str(answer), port))]
except:
pass
# Fallback to original
return original_getaddrinfo(host, port, family, type, proto, flags)
socket.getaddrinfo = fast_getaddrinfo# Optimize requests session
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
class OptimizedNetworkManager(NetworkManager):
def __init__(self, state):
super().__init__(state)
# Configure connection pooling
adapter = HTTPAdapter(
pool_connections=10, # Number of connection pools
pool_maxsize=20, # Max connections per pool
max_retries=Retry( # Retry configuration
total=3,
backoff_factor=0.5,
status_forcelist=[500, 502, 503, 504]
)
)
self.session.mount('http://', adapter)
self.session.mount('https://', adapter)
# Keep-alive settings
self.session.headers.update({
'Connection': 'keep-alive',
'Accept-Encoding': 'gzip, deflate',
})# Search multiple engines in parallel
import concurrent.futures
import asyncio
import aiohttp # pip install aiohttp
async def parallel_search_async(query, engines):
"""Search multiple engines simultaneously"""
async with aiohttp.ClientSession() as session:
tasks = []
for engine in engines:
task = self.search_engine_async(session, query, engine)
tasks.append(task)
# Gather all results
results = await asyncio.gather(*tasks, return_exceptions=True)
# Filter out errors and combine results
all_results = []
for result in results:
if isinstance(result, list):
all_results.extend(result)
return all_results[:10] # Return top 10 combined resultsimport hashlib
import pickle
import os
import time
class CachedSearchManager(SearchManager):
def __init__(self, state, network):
super().__init__(state, network)
self.cache_dir = os.path.expanduser("~/.naviduck_cache")
os.makedirs(self.cache_dir, exist_ok=True)
# Cache statistics
self.cache_hits = 0
self.cache_misses = 0
def search(self, query, engine=None):
engine = engine or self.state.current_engine
# Generate cache key
cache_key = hashlib.md5(f"{query}_{engine}".encode()).hexdigest()
cache_file = os.path.join(self.cache_dir, f"{cache_key}.pkl")
# Check cache
if os.path.exists(cache_file):
cache_age = time.time() - os.path.getmtime(cache_file)
# Different TTLs for different engines
ttl = {
'ddg_api': 300, # 5 minutes (API data changes slowly)
'wikipedia': 3600, # 1 hour (Wikipedia changes slowly)
'brave': 180, # 3 minutes
'google': 60, # 1 minute (Google changes frequently)
'ddg': 60, # 1 minute
}.get(engine, 300)
if cache_age < ttl:
self.cache_hits += 1
with open(cache_file, 'rb') as f:
print(f"{Colors.INFO}⚡ Cache hit ({int(cache_age)}s old){Colors.RESET}")
return pickle.load(f)
# Cache miss - perform actual search
self.cache_misses += 1
results = super().search(query, engine)
# Cache results
with open(cache_file, 'wb') as f:
pickle.dump(results, f)
# Print cache stats occasionally
if (self.cache_hits + self.cache_misses) % 10 == 0:
hit_rate = self.cache_hits / (self.cache_hits + self.cache_misses) * 100
print(f"{Colors.INFO}📊 Cache: {hit_rate:.0f}% hit rate{Colors.RESET}")
return results# Compress cached data to save space
import gzip
def save_compressed_cache(self, key, data):
cache_file = os.path.join(self.cache_dir, f"{key}.pkl.gz")
with gzip.open(cache_file, 'wb') as f:
pickle.dump(data, f)
def load_compressed_cache(self, key, max_age):
cache_file = os.path.join(self.cache_dir, f"{key}.pkl.gz")
if os.path.exists(cache_file):
file_age = time.time() - os.path.getmtime(cache_file)
if file_age < max_age:
with gzip.open(cache_file, 'rb') as f:
return pickle.load(f)
return None# Add in-memory cache on top of disk cache
from functools import lru_cache
class MemoryCachedSearchManager(SearchManager):
def __init__(self, state, network):
super().__init__(state, network)
self.memory_cache = {}
self.max_memory_cache_size = 100 # Store 100 queries in memory
@lru_cache(maxsize=100)
def search_cached(self, query: str, engine: str):
"""LRU cache decorator for memory caching"""
return self._search_uncached(query, engine)
def search(self, query, engine=None):
engine = engine or self.state.current_engine
# Try memory cache first
if (query, engine) in self.memory_cache:
cached_data = self.memory_cache[(query, engine)]
if time.time() - cached_data['timestamp'] < 60: # 1 minute TTL
print(f"{Colors.INFO}⚡ Memory cache hit{Colors.RESET}")
return cached_data['results']
# Fall back to parent method
results = super().search(query, engine)
# Store in memory cache
if len(self.memory_cache) >= self.max_memory_cache_size:
# Remove oldest entry
oldest_key = min(self.memory_cache.keys(),
key=lambda k: self.memory_cache[k]['timestamp'])
del self.memory_cache[oldest_key]
self.memory_cache[(query, engine)] = {
'results': results,
'timestamp': time.time()
}
return results# Different timeouts for different operations
class AdaptiveNetworkManager(NetworkManager):
def get(self, url, use_tor=False, timeout=None):
if timeout is None:
# Auto-detect timeout based on URL pattern
if 'api.duckduckgo.com' in url:
timeout = 2 # Fast API
elif 'wikipedia.org' in url:
timeout = 3 # Wikipedia API
elif 'google.com' in url:
timeout = 5 # Google can be slow
elif use_tor:
timeout = 10 # Tor is slower
else:
timeout = 5 # Default
# Add jitter to avoid thundering herd
jitter = random.uniform(0, 0.5)
time.sleep(jitter)
return super().get(url, use_tor=use_tor, timeout=timeout)# Reuse connections for same domains
from urllib3 import PoolManager
class ConnectionReuseManager:
def __init__(self):
self.pool_manager = PoolManager(
maxsize=10,
block=True,
timeout=5.0,
retries=3
)
def get(self, url):
# Extract domain for connection pooling
domain = urlparse(url).netloc
# Reuse connection for same domain
return self.pool_manager.request('GET', url)def optimize_query(query):
"""Rewrite queries for better search performance"""
query = query.strip().lower()
# Remove common stop words for faster searching
stop_words = {'the', 'a', 'an', 'and', 'or', 'but', 'in', 'on', 'at', 'to', 'for', 'of', 'with', 'by'}
words = query.split()
optimized_words = [w for w in words if w not in stop_words]
if not optimized_words:
optimized_words = words[-1:] # Keep at least one word
# Add site: filter for common domains
if any(tech in query for tech in ['python', 'javascript', 'java', 'c++', 'go', 'rust']):
# Add programming sites for tech queries
optimized_query = f"{' '.join(optimized_words)} site:stackoverflow.com OR site:github.com"
elif any(word in query for word in ['how', 'tutorial', 'guide', 'learn']):
# Add tutorial sites
optimized_query = f"{' '.join(optimized_words)} site:w3schools.com OR site:realpython.com"
else:
optimized_query = ' '.join(optimized_words)
return optimized_query
# Use in SearchManager:
def search(self, query, engine=None):
optimized_query = optimize_query(query)
print(f"Optimized query: {optimized_query}")
return self._actual_search(optimized_query, engine)def optimize_for_engine(query, engine):
"""Apply engine-specific optimizations"""
optimizations = {
'ddg_api': {
'preprocess': lambda q: q[:200], # API has query length limits
'add_params': {'format': 'json', 'no_html': '1', 'skip_disambig': '1'},
},
'wikipedia': {
'preprocess': lambda q: q.title(), # Wikipedia prefers Title Case
'add_params': {'action': 'opensearch', 'limit': '10', 'format': 'json'},
},
'google': {
'preprocess': lambda q: f'"{q}"' if len(q.split()) > 2 else q,
'add_params': {'num': '10', 'hl': 'en', 'lr': 'lang_en'},
},
'brave': {
'preprocess': lambda q: q,
'add_params': {'q': query, 'source': 'web'},
}
}
config = optimizations.get(engine, {})
processed_query = config.get('preprocess', lambda x: x)(query)
additional_params = config.get('add_params', {})
return processed_query, additional_paramsimport multiprocessing
from concurrent.futures import ThreadPoolExecutor
def parallel_parse_results(self, response, engine, query):
"""Parse HTML results in parallel"""
html = response.text
# Split HTML into chunks for parallel processing
num_workers = min(multiprocessing.cpu_count(), 4)
chunk_size = len(html) // num_workers
chunks = []
for i in range(num_workers):
start = i * chunk_size
end = start + chunk_size if i < num_workers - 1 else len(html)
chunks.append(html[start:end])
# Parse chunks in parallel
with ThreadPoolExecutor(max_workers=num_workers) as executor:
futures = []
for chunk in chunks:
future = executor.submit(self._parse_chunk, chunk, engine)
futures.append(future)
# Combine results
all_results = []
for future in futures:
try:
results = future.result(timeout=2)
all_results.extend(results)
except:
pass
# Deduplicate and sort
return self._deduplicate_results(all_results)[:10]def display_results_streaming(self, results):
"""Display results as they come in (not waiting for all)"""
print(f"{Colors.INFO}Fetching results...{Colors.RESET}")
for i, result in enumerate(results, 1):
# Display immediately without waiting for all
engine_icon = self.state.get_icon(result.get('engine', 'SEARCH'))
title = result['title']
print(f"{Colors.CYAN}{i:2d}.{Colors.RESET} {engine_icon} {title}")
# Short pause for readability
if i % 5 == 0:
print(f"{Colors.GRAY} (loading more...){Colors.RESET}")
time.sleep(0.1)
return len(results)class PredictiveSearchManager(SearchManager):
def __init__(self, state, network):
super().__init__(state, network)
self.search_patterns = {}
self.prefetch_cache = {}
def learn_search_patterns(self):
"""Learn user's search patterns for prefetching"""
# Analyze history for common search patterns
search_history = [h for h in self.state.history if h['type'] == 'search']
if len(search_history) < 10:
return
# Find common prefixes
from collections import Counter
prefixes = Counter()
for entry in search_history[-50:]:
query = entry['query'].lower()
words = query.split()
if len(words) > 1:
# Track first word patterns
prefixes[words[0]] += 1
# Store common prefixes
self.common_prefixes = [prefix for prefix, count in prefixes.most_common(5)]
def predictive_prefetch(self, partial_query):
"""Prefetch likely completions"""
if not hasattr(self, 'common_prefixes'):
self.learn_search_patterns()
for prefix in self.common_prefixes:
if partial_query.startswith(prefix):
# Prefetch full query based on pattern
full_query = f"{prefix} {partial_query[len(prefix):].strip()}"
if full_query and full_query not in self.prefetch_cache:
# Start async prefetch
threading.Thread(
target=self._prefetch_search,
args=(full_query, self.state.current_engine)
).start()def rank_results(self, results, query):
"""Intelligent ranking of search results"""
query_words = set(query.lower().split())
ranked_results = []
for result in results:
score = 0
# Title relevance
title_lower = result['title'].lower()
for word in query_words:
if word in title_lower:
score += 10
elif word[:3] in title_lower:
score += 3
# URL relevance
url_lower = result['url'].lower()
if any(word in url_lower for word in query_words):
score += 5
# Snippet relevance
snippet = result.get('snippet', '').lower()
for word in query_words:
if word in snippet:
score += 3
# Domain authority (simple heuristic)
domain = urlparse(result['url']).netloc
authoritative_domains = [
'github.com', 'stackoverflow.com', 'wikipedia.org',
'docs.python.org', 'developer.mozilla.org'
]
if any(auth in domain for auth in authoritative_domains):
score += 15
# Recency (if available)
if 'date' in result:
# Convert date to score
score += 5
ranked_results.append((score, result))
# Sort by score descending
ranked_results.sort(key=lambda x: x[0], reverse=True)
return [result for score, result in ranked_results]class CachedNavAI(NavAI):
def __init__(self, icons):
super().__init__(icons)
self.cache = {}
self.cache_file = os.path.expanduser("~/.navai_cache.json")
self.load_cache()
def ask(self, question: str) -> str:
question_lower = question.strip().lower()
# Check cache first
if question_lower in self.cache:
cached = self.cache[question_lower]
if time.time() - cached['timestamp'] < 3600: # 1 hour TTL
return cached['answer']
# Get fresh answer
answer = super().ask(question)
# Cache it
self.cache[question_lower] = {
'answer': answer,
'timestamp': time.time()
}
# Prune old cache entries
self._prune_cache()
# Save cache periodically
if random.random() < 0.1: # 10% chance to save
self.save_cache()
return answer
def _prune_cache(self, max_size=100):
"""Keep only recent cache entries"""
if len(self.cache) > max_size:
# Sort by timestamp, keep newest
sorted_items = sorted(self.cache.items(),
key=lambda x: x[1]['timestamp'],
reverse=True)
self.cache = dict(sorted_items[:max_size])def ask_parallel(self, question: str) -> str:
"""Query multiple knowledge sources in parallel"""
import concurrent.futures
sources = [
self._get_duckduckgo_answer,
self._get_wikipedia_summary,
self._get_local_knowledge,
]
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
futures = {executor.submit(source, question): source.__name__
for source in sources}
for future in concurrent.futures.as_completed(futures, timeout=3):
try:
answer = future.result(timeout=2)
if answer and answer != "No answer found.":
return answer
except:
continue
return f"{self.icons['INFO']} I couldn't find a direct answer. Try: 'search {question}'"def _get_local_knowledge(self, query: str) -> str:
"""Local knowledge base for instant answers"""
knowledge_base = {
# NaviDuck specific
"naviduck commands": "Type 'help' for complete command list",
"naviduck features": "Search, AI assistant, Tor browsing, bookmarks, history",
"naviduck version": "Enhanced CLI Browser with REAL AI & Working Search",
# Programming
"python install": "Download from python.org or use package manager",
"python hello world": "print('Hello, World!')",
"python list": "my_list = [1, 2, 3]",
# Common tech
"github": "GitHub is a code hosting platform for version control",
"stack overflow": "Q&A site for programmers",
"wikipedia": "Free online encyclopedia",
# Quick facts
"capital of france": "Paris",
"largest ocean": "Pacific Ocean",
"population of earth": "~8 billion",
}
query_lower = query.lower()
for key, answer in knowledge_base.items():
if key in query_lower:
return f"{self.icons['INFO']} {answer}"
return "No answer found."def optimize_response_length(self, response, max_length=500):
"""Trim responses to optimal length"""
if len(response) <= max_length:
return response
# Try to find a natural breakpoint
sentences = response.split('. ')
truncated = []
current_length = 0
for sentence in sentences:
if current_length + len(sentence) < max_length - 50: # Leave room for "..."
truncated.append(sentence)
current_length += len(sentence) + 2 # +2 for ". "
else:
break
if truncated:
result = '. '.join(truncated) + '.'
if len(result) < len(response):
result += " (truncated)"
return result
# Fallback: simple truncation
return response[:max_length-3] + "..."def select_best_answer(self, answers):
"""Choose the best answer from multiple sources"""
if not answers:
return "No answer found."
# Score each answer
scored_answers = []
for answer in answers:
score = 0
# Length scoring (medium length is best)
length = len(answer)
if 50 <= length <= 300:
score += 20
elif length > 500:
score -= 10
# Completeness scoring
if answer.endswith('.'):
score += 5
# Information density
words = answer.split()
unique_words = set(words)
if len(words) > 0:
density = len(unique_words) / len(words)
if density > 0.7:
score += 10
scored_answers.append((score, answer))
# Return highest scoring answer
scored_answers.sort(key=lambda x: x[0], reverse=True)
return scored_answers[0][1]def optimized_clear_screen():
"""Clear screen with minimal overhead"""
if os.name == 'nt':
# Windows - most efficient method
os.system('cls')
else:
# Unix - use ANSI escape codes (fastest)
print('\033[2J\033[H', end='')
sys.stdout.flush()
def fast_print(text):
"""Print without immediate flush for batch operations"""
print(text, end='', flush=False)
def flush_screen():
"""Flush all pending output at once"""
sys.stdout.flush()def progressive_display(self, items, batch_size=5, delay=0.05):
"""Display items progressively for perceived speed"""
print(f"{Colors.INFO}Loading results...{Colors.RESET}")
for i, item in enumerate(items, 1):
# Display item
self._display_item(item, i)
# Batch flush for efficiency
if i % batch_size == 0:
sys.stdout.flush()
time.sleep(delay) # Small delay for perceived responsiveness
sys.stdout.flush()import threading
import queue
class AsyncInputHandler:
def __init__(self):
self.input_queue = queue.Queue()
self.running = False
def start(self):
self.running = True
threading.Thread(target=self._input_thread, daemon=True).start()
def _input_thread(self):
while self.running:
try:
user_input = input()
self.input_queue.put(user_input)
except:
break
def get_input(self, timeout=0.1):
"""Get input without blocking main thread"""
try:
return self.input_queue.get(timeout=timeout)
except queue.Empty:
return None
def stop(self):
self.running = Falsedef buffered_input(prompt="", buffer_size=10):
"""Buffer input for faster response"""
input_buffer = []
while True:
try:
# Try to get from buffer first
if input_buffer:
return input_buffer.pop(0)
# Get fresh input
user_input = input(prompt)
# If multiple commands separated by semicolons
if ';' in user_input:
commands = user_input.split(';')
# Execute first, buffer rest
input_buffer.extend(commands[1:])
return commands[0].strip()
else:
return user_input
except (KeyboardInterrupt, EOFError):
raiseclass VirtualScroller:
def __init__(self, items, page_size=10):
self.items = items
self.page_size = page_size
self.current_page = 0
def display_page(self):
"""Display only visible page of items"""
start = self.current_page * self.page_size
end = start + self.page_size
for i, item in enumerate(self.items[start:end], start + 1):
self._display_item(item, i)
# Show navigation
total_pages = (len(self.items) + self.page_size - 1) // self.page_size
print(f"\nPage {self.current_page + 1}/{total_pages}")
print("n: next, p: previous, q: quit")
def handle_navigation(self, command):
if command == 'n' and (self.current_page + 1) * self.page_size < len(self.items):
self.current_page += 1
return True
elif command == 'p' and self.current_page > 0:
self.current_page -= 1
return True
return False# Pre-compute color strings to avoid string concatenation overhead
class FastColors:
def __init__(self):
# Cache formatted strings
self.cache = {}
def get(self, text, color_code):
key = (text, color_code)
if key not in self.cache:
self.cache[key] = f"{color_code}{text}\033[0m"
# Limit cache size
if len(self.cache) > 1000:
# Remove oldest entries (simple FIFO)
for k in list(self.cache.keys())[:500]:
del self.cache[k]
return self.cache[key]# Optional: Install numba for JIT compilation
# pip install numba
try:
from numba import jit
HAS_NUMBA = True
except ImportError:
HAS_NUMBA = False
if HAS_NUMBA:
@jit(nopython=True, cache=True)
def score_result_fast(title, query_words, domain):
"""JIT-compiled scoring function"""
score = 0
title_lower = title.lower()
for word in query_words:
if word in title_lower:
score += 10
# Fast domain scoring
authoritative = ['github', 'stackoverflow', 'wikipedia']
for auth in authoritative:
if auth in domain:
score += 15
break
return score
else:
# Fallback to Python version
def score_result_fast(title, query_words, domain):
# Python implementation
pass# Optimizations for PyPy JIT
def pypy_optimized_search(self, query, engine):
"""Code optimized for PyPy's JIT compiler"""
# PyPy optimizes Python code, avoid:
# - Excessive object creation
# - Deep recursion
# - Global variable access
# Use local variables
local_engine = engine or self.state.current_engine
local_config = SEARCH_ENGINES.get(local_engine)
# Pre-compute values
encoded_query = quote(query)
params = {}
for key, value in local_config["params"].items():
params[key] = value.format(query=encoded_query)
# Build URL efficiently
url_parts = [local_config["url"], "?"]
url_parts.extend(f"{k}={v}" for k, v in params.items())
url = "&".join(url_parts)
return urlimport mmap
import struct
class MappedCache:
def __init__(self, cache_file, max_size_mb=100):
self.cache_file = cache_file
self.max_size = max_size_mb * 1024 * 1024
# Create or open memory-mapped file
if not os.path.exists(cache_file):
with open(cache_file, 'wb') as f:
f.write(b'\x00' * self.max_size)
self.fd = os.open(cache_file, os.O_RDWR)
self.mmap = mmap.mmap(self.fd, self.max_size, access=mmap.ACCESS_WRITE)
# Simple hash table in memory-mapped file
self.hash_table = {}
def put(self, key, value):
key_hash = hash(key) % 10000
position = key_hash * 1000 # Fixed-size slots
# Serialize value
serialized = pickle.dumps(value)
if len(serialized) > 900: # Leave room for metadata
return False
# Write to mmap
self.mmap.seek(position)
self.mmap.write(struct.pack('I', len(serialized)))
self.mmap.write(serialized)
self.hash_table[key] = position
return True
def get(self, key):
if key not in self.hash_table:
return None
position = self.hash_table[key]
self.mmap.seek(position)
length = struct.unpack('I', self.mmap.read(4))[0]
serialized = self.mmap.read(length)
return pickle.loads(serialized)import sqlite3
import threading
class SQLiteStorage:
def __init__(self, db_file="naviduck.db"):
self.db_file = db_file
self.local = threading.local()
self._init_db()
def _get_conn(self):
if not hasattr(self.local, 'conn'):
self.local.conn = sqlite3.connect(self.db_file, check_same_thread=False)
self.local.conn.row_factory = sqlite3.Row
return self.local.conn
def _init_db(self):
conn = self._get_conn()
conn.execute("""
CREATE TABLE IF NOT EXISTS search_cache (
query TEXT,
engine TEXT,
results BLOB,
timestamp INTEGER,
PRIMARY KEY (query, engine)
)
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_timestamp
ON search_cache(timestamp)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
type TEXT,
query TEXT,
url TEXT,
title TEXT,
timestamp INTEGER,
engine TEXT
)
""")
conn.commit()
def cache_search(self, query, engine, results):
conn = self._get_conn()
conn.execute("""
INSERT OR REPLACE INTO search_cache
VALUES (?, ?, ?, ?)
""", (query, engine, pickle.dumps(results), int(time.time())))
conn.commit()
def get_cached_search(self, query, engine, max_age=300):
conn = self._get_conn()
cursor = conn.execute("""
SELECT results FROM search_cache
WHERE query = ? AND engine = ?
AND timestamp > ?
""", (query, engine, int(time.time()) - max_age))
row = cursor.fetchone()
if row:
return pickle.loads(row['results'])
return None# search_engine.pyx - Cython optimized module
# cython: language_level=3
import re
from libc.string cimport strlen
cdef class FastParser:
cdef dict patterns
def __init__(self):
self.patterns = {
'link': re.compile(b'<a[^>]+href="([^"]+)"[^>]*>([^<]+)</a>'),
'title': re.compile(b'<title[^>]*>(.*?)</title>'),
}
cpdef list parse_links(self, bytes html):
"""C-optimized link parsing"""
cdef list results = []
cdef object match
cdef bytes url, title
for match in self.patterns['link'].finditer(html):
url = match.group(1)
title = match.group(2)
# Fast byte processing
if b'http' in url and strlen(url) < 500:
results.append((
url.decode('utf-8', 'ignore'),
title.decode('utf-8', 'ignore')[:80]
))
if len(results) >= 10:
break
return results# setup.py
from setuptools import setup
from Cython.Build import cythonize
setup(
ext_modules=cythonize("search_engine.pyx"),
)# benchmark.py
import time
import statistics
from tabulate import tabulate
class NaviDuckBenchmark:
def __init__(self, naviduck_instance):
self.naviduck = naviduck_instance
self.results = {}
def run_benchmarks(self):
tests = [
("Startup Time", self.benchmark_startup),
("Simple Search", lambda: self.benchmark_search("test", "brave")),
("AI Query", lambda: self.benchmark_ai("what is python")),
("Page Load", lambda: self.benchmark_page_load("https://httpbin.org/html")),
("History Lookup", self.benchmark_history),
]
print(f"{'='*60}")
print(f"NaviDuck Performance Benchmarks")
print(f"{'='*60}")
for name, test_func in tests:
print(f"\n🔍 Testing: {name}")
times = []
for i in range(3): # Run 3 times for average
start = time.time()
test_func()
elapsed = time.time() - start
times.append(elapsed)
print(f" Run {i+1}: {elapsed:.2f}s")
avg = statistics.mean(times)
std = statistics.stdev(times) if len(times) > 1 else 0
self.results[name] = {"avg": avg, "std": std, "runs": times}
print(f" Average: {avg:.2f}s ± {std:.2f}s")
def benchmark_search(self, query, engine):
self.naviduck.search_mgr.search(query, engine)
def benchmark_ai(self, question):
self.naviduck.ai.ask(question)
def benchmark_page_load(self, url):
self.naviduck.page_loader.load_page(url, display=False)
def benchmark_history(self):
# Add some history first
for i in range(10):
self.naviduck.state.history.append({
'type': 'search',
'query': f'test {i}',
'timestamp': time.time(),
})
# Benchmark history display
self.naviduck.ui.show_history()
def print_report(self):
print(f"\n{'='*60}")
print(f"Benchmark Report")
print(f"{'='*60}")
table_data = []
for name, data in self.results.items():
table_data.append([
name,
f"{data['avg']:.2f}s",
f"±{data['std']:.2f}s",
f"{1/data['avg']:.1f}/s" if data['avg'] > 0 else "N/A"
])
print(tabulate(table_data,
headers=["Test", "Avg Time", "Std Dev", "Ops/sec"],
tablefmt="grid"))
# Performance rating
total_avg = sum(data['avg'] for data in self.results.values())
if total_avg < 5:
rating = "Excellent 🚀"
elif total_avg < 10:
rating = "Good 👍"
elif total_avg < 20:
rating = "Average ⚡"
else:
rating = "Needs optimization 🐢"
print(f"\nOverall Performance: {rating}")
print(f"Total time: {total_avg:.1f}s")# Run profiler
python -m cProfile -o profile.dat naviduck.py
# Analyze with snakeviz
pip install snakeviz
snakeviz profile.dat
# Or generate call graph
python -m gprof2dot -f pstats profile.dat | dot -Tpng -o profile.pngclass PerformanceMonitor:
def __init__(self):
self.metrics = {
'search_times': [],
'ai_times': [],
'page_load_times': [],
'memory_usage': [],
'cache_hits': 0,
'cache_misses': 0,
}
self.start_time = time.time()
def record(self, metric, value):
if metric in self.metrics:
if isinstance(self.metrics[metric], list):
self.metrics[metric].append(value)
# Keep only last 100 readings
if len(self.metrics[metric]) > 100:
self.metrics[metric] = self.metrics[metric][-100:]
else:
self.metrics[metric] += value
def print_stats(self):
print(f"\n{Colors.INFO}📊 Performance Statistics{Colors.RESET}")
print(f"{Colors.GRAY}{'─' * 40}{Colors.RESET}")
uptime = time.time() - self.start_time
print(f"Uptime: {uptime:.0f}s")
if self.metrics['search_times']:
avg_search = statistics.mean(self.metrics['search_times'])
print(f"Avg search: {avg_search:.2f}s")
if self.metrics['ai_times']:
avg_ai = statistics.mean(self.metrics['ai_times'])
print(f"Avg AI response: {avg_ai:.2f}s")
total_calls = self.metrics['cache_hits'] + self.metrics['cache_misses']
if total_calls > 0:
hit_rate = self.metrics['cache_hits'] / total_calls * 100
print(f"Cache hit rate: {hit_rate:.1f}%")def performance_audit():
"""Run quick performance checks"""
issues = []
# Check 1: Default search engine
if state.current_engine not in ['ddg_api', 'brave']:
issues.append("⚠️ Using slow default engine. Switch to 'ddg_api' or 'brave'")
# Check 2: History size
if len(state.history) > 1000:
issues.append(f"⚠️ Large history ({len(state.history)} entries). Consider clearing")
# Check 3: Cache directory exists
cache_dir = os.path.expanduser("~/.naviduck_cache")
if not os.path.exists(cache_dir):
issues.append("⚠️ Cache not enabled. Enable for faster repeat searches")
# Check 4: Tor enabled unnecessarily
if state.tor_enabled:
issues.append("⚠️ Tor enabled. Disable if not needed for better speed")
# Check 5: Many disabled engines
disabled = sum(1 for e in SEARCH_ENGINES.values() if not e['enabled'])
if disabled > 2:
issues.append(f"⚠️ {disabled} engines disabled. Consider removing unused ones")
if issues:
print(f"{Colors.WARNING}Performance Issues Found:{Colors.RESET}")
for issue in issues:
print(f" • {issue}")
return False
else:
print(f"{Colors.SUCCESS}✅ Performance configuration looks good!{Colors.RESET}")
return True-
Clear old cache:
find ~/.naviduck_cache -type f -mtime +30 -delete -
Trim history:
# Keep only last month cutoff = time.time() - (30 * 24 * 3600) state.history = [h for h in state.history if datetime.fromisoformat(h['timestamp']).timestamp() > cutoff]
-
Update search engine configurations:
# Check if engines still work for engine in SEARCH_ENGINES: test_search(engine, "test")
-
Profile and optimize:
python benchmark.py
# Add to ~/.naviduck_config.json
{
"performance_mode": true,
"default_engine": "ddg_api",
"engines": {
"ddg_api": true,
"brave": true,
"wikipedia": true,
"google": false,
"ddg": false
},
"cache_enabled": true,
"cache_ttl": 300,
"max_history": 100,
"max_bookmarks": 50,
"ai_cache_enabled": true,
"ai_timeout": 2,
"search_timeout": 3,
"page_load_timeout": 5,
"tor_enabled": false,
"use_emoji": false, # Nerd fonts render faster
"lazy_loading": true,
"compression": true
}-
Windows: Windows Terminal with:
- GPU acceleration enabled
- Cascadia Code font
- Disable animations
- UTF-8 encoding
-
Linux: Alacritty with:
- GPU backend
- JetBrains Mono font
- Scrollback limit: 10000 lines
-
Mac: iTerm2 with:
- GPU rendering enabled
- SF Mono font
- Disable transparency
# Increase file descriptor limits (Linux/Mac)
ulimit -n 65536
# Set DNS to fast servers
# Linux: /etc/resolv.conf
nameserver 1.1.1.1
nameserver 8.8.8.8
# Windows: Network settings
# Use Cloudflare (1.1.1.1) or Google (8.8.8.8) DNS| Metric | Good | Average | Needs Improvement |
|---|---|---|---|
| Search response time | < 1.5s | 1.5-3s | > 3s |
| AI response time | < 2s | 2-4s | > 4s |
| Startup time | < 1s | 1-2s | > 2s |
| Memory usage | < 50MB | 50-100MB | > 100MB |
| Cache hit rate | > 70% | 40-70% | < 40% |
| History load time | < 0.1s | 0.1-0.3s | > 0.3s |
Last updated: 12/22/2025
Performance Tips version: 3.0
Remember: The best performance improvements come from:
- Using the fastest search engine (ddg_api)
- Enabling intelligent caching
- Keeping data stores small
- Using a modern, GPU-accelerated terminal
- Disabling features you don't use
Pro Tip: Monitor performance with benchmark.py regularly and adjust settings based on your usage patterns!
Happy optimizing! 🚀🦆