Coverage for src / monte_neo / cli / menu / hardware.py: 0%

1"""Hardware configuration workflow.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING 

6 

7import questionary 

8from rich.console import Console 

9from rich.panel import Panel 

10from rich.table import Table 

11 

12from monte_neo.cli.styles import CUSTOM_STYLE 

13 

14if TYPE_CHECKING: 

15 from monte_neo.cli.menu.main import InteractiveMenu 

16 

17console = Console() 

18 

19 

20def hardware_workflow(menu: InteractiveMenu) -> None: 

21 """Hardware configuration menu.""" 

22 console.print("\n[bold cyan]🔧 Hardware Configuration[/]\n") 

23 

24 # Display current hardware info 

25 _show_hardware_info(menu) 

26 

27 choices = [ 

28 {"name": f"🚀 Use GPU ({'✅' if menu._use_gpu else '❌'})", "value": "toggle_gpu"}, 

29 {"name": f"💎 GPU Precision ({menu._gpu_precision})", "value": "gpu_precision"}, 

30 {"name": f"⚡ Metal Driver ({menu._metal_driver.upper()})", "value": "metal_driver"}, 

31 {"name": f"🎯 MC Pass Threshold ({menu._mc_pass_threshold:.0%})", "value": "mc_threshold"}, 

32 {"name": "📊 Benchmark Hardware", "value": "benchmark"}, 

33 {"name": "🔙 Back", "value": "back"}, 

34 ] 

35 

36 choice = questionary.select("Select hardware option:", choices=choices, style=CUSTOM_STYLE).ask() 

37 

38 if choice == "toggle_gpu": 

39 menu._use_gpu = not menu._use_gpu 

40 status = "enabled" if menu._use_gpu else "disabled" 

41 console.print(f"[green]GPU acceleration {status}[/]") 

42 

43 elif choice == "gpu_precision": 

44 prec_choices = [ 

45 {"name": "float32 (Standard - Best compatibility)", "value": "float32"}, 

46 {"name": "float16 (Faster - Good performance)", "value": "float16"}, 

47 {"name": "float8_e4m3 (Extreme - 4x memory bandwidth)", "value": "float8_e4m3"}, 

48 {"name": "float8_e5m2 (Extreme - 4x memory bandwidth)", "value": "float8_e5m2"}, 

49 ] 

50 val = questionary.select("Select GPU precision:", choices=prec_choices, style=CUSTOM_STYLE).ask() 

51 if val: 

52 menu._gpu_precision = val 

53 console.print(f"[green]GPU precision set to {val}[/]") 

54 

55 elif choice == "metal_driver": 

56 if not menu._use_gpu: 

57 console.print("[yellow]⚠️ Enable GPU first to use Metal drivers[/]") 

58 else: 

59 driver_choices = [ 

60 {"name": "Auto-Select (Recommended)", "value": "auto"}, 

61 {"name": "Clang C++ (Optimized)", "value": "cpp"}, 

62 {"name": "Objective-C++ (Native)", "value": "objc"}, 

63 {"name": "Apple Swift (Modern)", "value": "swift"}, 

64 ] 

65 val = questionary.select("Select Metal driver:", choices=driver_choices, style=CUSTOM_STYLE).ask() 

66 if val: 

67 menu._metal_driver = val 

68 # Update config object and save 

69 menu.config.metal_driver = val 

70 from monte_neo.utils.config import save_config 

71 save_config(menu.config, "config.yaml") 

72 console.print(f"[green]Metal driver set to {val.upper()} and saved to config.yaml[/]") 

73 

74 elif choice == "mc_threshold": 

75 val = questionary.text("MC Pass Threshold (0.0-1.0):", default=str(menu._mc_pass_threshold)).ask() 

76 if val: 

77 menu._mc_pass_threshold = float(val) 

78 console.print(f"[green]MC pass threshold set to {menu._mc_pass_threshold:.0%}[/]") 

79 

80 elif choice == "benchmark": 

81 _run_hardware_benchmark(menu) 

82 

83 

84def _show_hardware_info(menu: InteractiveMenu) -> None: 

85 """Display current hardware configuration.""" 

86 

87 # Check system capabilities 

88 try: 

89 import mlx.core as mx 

90 mlx_available = True 

91 device_info = mx.get_default_device() if hasattr(mx, 'get_default_device') else "Unknown" 

92 except ImportError: 

93 mlx_available = False 

94 device_info = "Not available" 

95 

96 try: 

97 import Metal 

98 metal_available = Metal.MTLCreateSystemDefaultDevice() is not None 

99 if metal_available: 

100 device = Metal.MTLCreateSystemDefaultDevice() 

101 metal_info = f"{device.name()} - {device.maxThreadgroupMemoryLength()//1024}KB shared memory" 

102 else: 

103 metal_info = "No Metal device" 

104 except ImportError: 

105 metal_available = False 

106 metal_info = "Metal framework not available" 

107 

108 # Create hardware info table 

109 table = Table(title="Hardware Configuration") 

110 table.add_column("Component", style="cyan") 

111 table.add_column("Status", style="green") 

112 table.add_column("Details", style="dim") 

113 

114 table.add_row("GPU Acceleration", "✅ Enabled" if menu._use_gpu else "❌ Disabled", 

115 f"MLX: {'Available' if mlx_available else 'Not available'}") 

116 table.add_row("GPU Precision", menu._gpu_precision, 

117 "Memory bandwidth: 4x with float8" if menu._gpu_precision.startswith("float8") else "Standard") 

118 table.add_row("Metal Driver", menu._metal_driver.upper(), 

119 metal_info if metal_available else "Metal not available") 

120 table.add_row("MC Pass Threshold", f"{menu._mc_pass_threshold:.0%}", 

121 "Production readiness threshold") 

122 

123 console.print(Panel(table, title="Current Configuration", border_style="blue")) 

124 

125 

126def _run_hardware_benchmark(menu: InteractiveMenu) -> None: 

127 """Run hardware performance benchmark.""" 

128 console.print("\n[bold yellow]🏃 Running hardware benchmark...[/]\n") 

129 

130 try: 

131 from monte_neo.core.acceleration.float8 import Float8Encoder 

132 

133 # Test float8 performance 

134 test_sizes = [1000, 10000, 100000] 

135 

136 table = Table(title="Float8 Performance Benchmark") 

137 table.add_column("Array Size", style="cyan") 

138 table.add_column("float32 Time (ms)", justify="right") 

139 table.add_column("float8 Time (ms)", justify="right") 

140 table.add_column("Speedup", justify="right") 

141 table.add_column("Memory Saved", justify="right") 

142 

143 import time 

144 

145 import numpy as np 

146 

147 for size in test_sizes: 

148 # Create test data 

149 test_data = np.random.randn(size).astype(np.float32) 

150 

151 # Test float32 operations 

152 start_time = time.time() 

153 result_f32 = test_data * 2.0 + 1.0 

154 f32_time = (time.time() - start_time) * 1000 

155 

156 # Test float8 operations 

157 encoder = Float8Encoder("e4m3") 

158 start_time = time.time() 

159 packed = encoder.encode_array(test_data) 

160 unpacked = encoder.decode_array(packed) 

161 result_f8 = unpacked * 2.0 + 1.0 

162 f8_time = (time.time() - start_time) * 1000 

163 

164 # Calculate metrics 

165 speedup = f32_time / f8_time if f8_time > 0 else 0 

166 memory_saved = "75%" # 8/32 = 0.25, so 75% saved 

167 

168 table.add_row( 

169 f"{size:,}", 

170 f"{f32_time:.2f}", 

171 f"{f8_time:.2f}", 

172 f"{speedup:.2f}x", 

173 memory_saved 

174 ) 

175 

176 console.print(table) 

177 console.print("\n[green]✅ Benchmark completed![/]") 

178 

179 except Exception as e: 

180 console.print(f"[red]❌ Benchmark failed: {e}[/]") 

181 console.print("[yellow]💡 Make sure all dependencies are installed for float8 support[/]")