#!/usr/bin/env python3 """ Solar Seeing Monitor - Universal Version (Linux/Windows/macOS) Microchip PIC18F14K50 HID Communication """ import hid import time import struct from datetime import datetime import math import sys import platform import os # ============================================================================ # OS-SPECIFIC CONFIGURATION # ============================================================================ class OSConfig: """OS-specific configuration and optimizations""" def __init__(self): self.os_name = platform.system() self.is_windows = self.os_name == 'Windows' self.is_linux = self.os_name == 'Linux' self.is_macos = self.os_name == 'Darwin' # Configure console encoding for Windows if self.is_windows: self._configure_windows_console() # Set optimal sleep times self.read_sleep = 0.001 if self.is_windows else 0.01 self.loop_sleep = 2.0 # Set default log path if self.is_windows: self.log_dir = os.path.join(os.environ.get('USERPROFILE', 'C:\\'), 'solar_data') else: self.log_dir = os.path.join(os.path.expanduser('~'), 'solar_data') def _configure_windows_console(self): """Configure Windows console for UTF-8 output""" try: import io sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8') sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8') # Enable ANSI colors on Windows 10+ if hasattr(sys.stdout, 'reconfigure'): sys.stdout.reconfigure(encoding='utf-8') os.system('') # Enable ANSI escape sequences except Exception: pass def print_system_info(self): """Display system information""" print("=" * 80) print(f"System Information:") print(f" OS : {self.os_name} {platform.release()}") print(f" Python : {platform.python_version()}") print(f" Architecture : {platform.machine()}") print(f" Read delay : {self.read_sleep*1000:.1f} ms") print(f" Log directory : {self.log_dir}") print("=" * 80 + "\n") # ============================================================================ # MAIN CLASS # ============================================================================ class MicrochipHIDSolar: def __init__(self, vendor_id=0x04d8, product_id=0x003f): self.vendor_id = vendor_id self.product_id = product_id self.device = None self.packet_size = 65 self.os_config = OSConfig() # System parameters self.system_info = { 'rshunt': 0.0, 'gain_filtered': 1.0, 'gain_unfiltered': 1.0, 'offset_volts': 0.0 } # Constants self.ONE_ADU_VOLTS = 4.096 / 65535.0 self.SUN_DIAMETER_ARCSEC = 1900.0 self.WAVELENGTH_M = 550e-9 # 550 nm def connect(self): """Connect to device with OS-specific optimizations""" try: self.device = hid.device() self.device.open(self.vendor_id, self.product_id) # Set non-blocking mode (may not be supported on all platforms) try: self.device.set_nonblocking(1) except Exception: print("⚠ Non-blocking mode not available (normal on some OS)") manufacturer = self.device.get_manufacturer_string() product = self.device.get_product_string() serial = self.device.get_serial_number_string() print(f"✓ Connected to: {manufacturer} - {product}") if serial: print(f" Serial: {serial}") print(f" Platform: {self.os_config.os_name}\n") return True except Exception as e: print(f"✗ Connection error: {e}") if self.os_config.is_linux: print("\n Linux troubleshooting:") print(" 1. Check udev rules: /etc/udev/rules.d/99-microchip.rules") print(" 2. Reload rules: sudo udevadm control --reload-rules") print(" 3. Or run with: sudo python3 script.py") elif self.os_config.is_windows: print("\n Windows troubleshooting:") print(" 1. Check Device Manager for HID device") print(" 2. Try: pip install --upgrade hidapi") print(" 3. Reconnect USB device") return False def disconnect(self): """Disconnect""" if self.device: self.device.close() print("\n✓ Disconnected") def send_command(self, command, data=None): """Send command""" try: packet = [0x00, command] + (data if data else []) packet += [0x00] * (self.packet_size - len(packet)) bytes_written = self.device.write(packet) return bytes_written > 0 except Exception as e: print(f"✗ Send error: {e}") return False def read_response(self, timeout_ms=500): """Read response with OS-optimized timing""" try: start_time = time.time() timeout_sec = timeout_ms / 1000.0 while (time.time() - start_time) < timeout_sec: response = self.device.read(self.packet_size) if response: return list(response) time.sleep(self.os_config.read_sleep) return None except Exception as e: print(f"✗ Read error: {e}") return None def decode_system_info(self, buffer_out): """ Decode system information from buffer Structure (starting at index 1): [1-2] : Major/Minor version [3] : Channel to read [4] : LED enable [5-8] : Serial number (Big Endian) [9-12] : Rshunt (float) [13-16]: Gain filtered (float) [17-20]: Gain unfiltered (float) [21-24]: Offset volts (float) """ try: # Versions (bytes 1-2) major_version = buffer_out[1] minor_version = buffer_out[2] # Channel to read (byte 3) channel_to_read = buffer_out[3] # LED Enable (byte 4) led_enable = buffer_out[4] > 0 # Serial (bytes 5-8) - Big Endian serial = struct.unpack('>I', bytes([ buffer_out[5], # as4 buffer_out[6], # as3 buffer_out[7], # as2 buffer_out[8] # as1 ]))[0] # Rshunt (bytes 9-12) rshunt = struct.unpack('>f', bytes([ buffer_out[9], buffer_out[10], buffer_out[11], buffer_out[12] ]))[0] # Gain filtered (bytes 13-16) gain_filtered = struct.unpack('>f', bytes([ buffer_out[13], buffer_out[14], buffer_out[15], buffer_out[16] ]))[0] # Gain unfiltered (bytes 17-20) gain_unfiltered = struct.unpack('>f', bytes([ buffer_out[17], buffer_out[18], buffer_out[19], buffer_out[20] ]))[0] # Offset volts (bytes 21-24) offset_volts = struct.unpack('>f', bytes([ buffer_out[21], buffer_out[22], buffer_out[23], buffer_out[24] ]))[0] return { 'major_version': major_version, 'minor_version': minor_version, 'channel_to_read': channel_to_read, 'led_enable': led_enable, 'serial': serial, 'rshunt': rshunt, 'gain_filtered': gain_filtered, 'gain_unfiltered': gain_unfiltered, 'offset_volts': offset_volts } except Exception as e: print(f"✗ Decode system info error: {e}") return None def decode_measurement_data(self, buffer_out): """ Decode measurement data from buffer Structure (starting at index 1): [1-4] : MeanADU_NF (float) [5-8] : EC_ADU_NF (float) [9-12] : MeanADU_F (float) [13-16]: EC_ADU_F (float) [17-20]: Nb_Samples (int) [21-24]: TempInsideBox (float) """ try: # MeanADU_NF (bytes 1-4) mean_adu_nf = struct.unpack('>f', bytes([ buffer_out[1], buffer_out[2], buffer_out[3], buffer_out[4] ]))[0] # EC_ADU_NF (bytes 5-8) ec_adu_nf = struct.unpack('>f', bytes([ buffer_out[5], buffer_out[6], buffer_out[7], buffer_out[8] ]))[0] # MeanADU_F (bytes 9-12) mean_adu_f = struct.unpack('>f', bytes([ buffer_out[9], buffer_out[10], buffer_out[11], buffer_out[12] ]))[0] # EC_ADU_F (bytes 13-16) ec_adu_f = struct.unpack('>f', bytes([ buffer_out[13], buffer_out[14], buffer_out[15], buffer_out[16] ]))[0] # nb_samples (bytes 17-20) nb_samples = struct.unpack('>I', bytes([ buffer_out[17], buffer_out[18], buffer_out[19], buffer_out[20] ]))[0] # TempInsideBox (bytes 25-29) temp_inside_box = struct.unpack('>f', bytes([ buffer_out[25], buffer_out[26], buffer_out[27], buffer_out[28] ]))[0] # Calculate derived values valid_seeing = (ec_adu_nf > 0 and mean_adu_nf > 0) if valid_seeing: # Variance calculation variance = ec_adu_nf - (mean_adu_nf ** 2) if variance < 0: variance = 0 # Seeing calculation theta_0_rad = math.sqrt(variance) * self.ONE_ADU_VOLTS / mean_adu_nf seeing_rad = 0.98 * self.WAVELENGTH_M / theta_0_rad if theta_0_rad > 0 else 0 seeing_arcsec = seeing_rad * 206265.0 # Fried parameter r0 r0 = 0.98 * self.WAVELENGTH_M / seeing_rad * 1000 if seeing_rad > 0 else 0 # in mm else: seeing_arcsec = 0 r0 = 0 return { 'mean_adu_nf': mean_adu_nf, 'ec_adu_nf': ec_adu_nf, 'mean_adu_f': mean_adu_f, 'ec_adu_f': ec_adu_f, 'nb_samples': nb_samples, 'temp_inside_box': temp_inside_box, 'seeing_arcsec': seeing_arcsec, 'r0': r0, 'valid_seeing': valid_seeing } except Exception as e: print(f"✗ Decode measurement error: {e}") return None def get_system_info(self, command=0x80): """Request and decode system information""" if not self.send_command(command): return None time.sleep(0.1) response = self.read_response() if not response: return None sys_info = self.decode_system_info(response) if sys_info: # Store for later calculations self.system_info['rshunt'] = sys_info['rshunt'] self.system_info['gain_filtered'] = sys_info['gain_filtered'] self.system_info['gain_unfiltered'] = sys_info['gain_unfiltered'] self.system_info['offset_volts'] = sys_info['offset_volts'] return sys_info def get_measurement_data(self, command=0x81): """Request and decode measurement data""" if not self.send_command(command): return None time.sleep(0.1) response = self.read_response() if not response: return None return self.decode_measurement_data(response) def display_system_info(self, info): """Display system information""" print("=" * 80) print("SYSTEM INFORMATION") print("=" * 80) print(f"Firmware version : {info['major_version']}.{info['minor_version']}") print(f"Serial number : {info['serial']}") print(f"Channel to read : {info['channel_to_read']}") print(f"LED enable : {'YES' if info['led_enable'] else 'NO'}") print(f"\nCalibration:") print(f" Rshunt : {info['rshunt']:.6f} Ω") print(f" Gain filtered : {info['gain_filtered']:.6f}") print(f" Gain unfiltered : {info['gain_unfiltered']:.6f}") print(f" Offset volts : {info['offset_volts']:.6f} V") print("=" * 80 + "\n") def display_measurement(self, data): """Display measurement data""" timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") print("─" * 80) print(f"Timestamp: {timestamp}") print("─" * 80) print(f"Raw ADU values:") print(f" Mean ADU (NF) : {data['mean_adu_nf']:10.2f}") print(f" EC ADU (NF) : {data['ec_adu_nf']:10.2f}") print(f" Mean ADU (F) : {data['mean_adu_f']:10.2f}") print(f" EC ADU (F) : {data['ec_adu_f']:10.2f}") print(f"\nDerived values:") print(f"Sample number : {data['nb_samples']:10d}") print(f"Seeing (FWHM) : {data['seeing_arcsec']:10.2f} arcsec {'✓' if data['valid_seeing'] else '✗'}") print(f"Fried parameter R₀ : {data['r0']:10.2f} mm") print(f"Box temperature : {data['temp_inside_box']:10.2f} °C") if data['valid_seeing']: if data['seeing_arcsec'] < 1.0: quality = "EXCELLENT" elif data['seeing_arcsec'] < 2.0: quality = "GOOD" elif data['seeing_arcsec'] < 3.0: quality = "AVERAGE" else: quality = "POOR" print(f"\nAtmospheric quality : {quality}") print("=" * 80 + "\n") def save_to_log(self, data, filename=None): """Save measurement to log file (OS-specific path)""" try: if filename is None: os.makedirs(self.os_config.log_dir, exist_ok=True) filename = os.path.join( self.os_config.log_dir, f"solar_log_{datetime.now().strftime('%Y%m%d')}.csv" ) # Create file with header if it doesn't exist file_exists = os.path.isfile(filename) with open(filename, 'a', encoding='utf-8') as f: if not file_exists: f.write("Timestamp,MeanADU_NF,EC_ADU_NF,MeanADU_F,EC_ADU_F,") f.write("nb_samples,TempBox_C,Seeing_arcsec,R0_mm,Valid\n") timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") f.write(f"{timestamp},") f.write(f"{data['mean_adu_nf']:.2f},") f.write(f"{data['ec_adu_nf']:.2f},") f.write(f"{data['mean_adu_f']:.2f},") f.write(f"{data['ec_adu_f']:.2f},") f.write(f"{data['nb_samples']},") f.write(f"{data['temp_inside_box']:.2f},") f.write(f"{data['seeing_arcsec']:.2f},") f.write(f"{data['r0']:.2f},") f.write(f"{1 if data['valid_seeing'] else 0}\n") return True except Exception as e: print(f"✗ Log save error: {e}") return False # ============================================================================ # MAIN PROGRAM # ============================================================================ def main(): """Main program with OS detection""" # Create instance and show OS info solar = MicrochipHIDSolar() solar.os_config.print_system_info() # Connect to device if not solar.connect(): print("\n✗ Unable to connect. Exiting.") return 1 try: # Read system parameters print("→ Reading system parameters...") sys_info = solar.get_system_info() if not sys_info: print("✗ Unable to read system parameters") return 1 solar.display_system_info(sys_info) # Ask for logging enable_logging = input("Enable data logging? (y/n): ").lower() == 'y' if enable_logging: print(f"✓ Logging enabled to: {solar.os_config.log_dir}\n") # Measurement loop print("Starting measurements (Ctrl+C to stop)...\n") counter = 0 while True: print(f"[Measurement #{counter}]") # Get measurement (command 0x81) data = solar.get_measurement_data(command=0x81) if data: solar.display_measurement(data) # Save to log if enabled if enable_logging: if solar.save_to_log(data): print("✓ Saved to log") # Summary print(f"→ Seeing={data['seeing_arcsec']:.2f}\" | " f"Sample_nb={data['nb_samples']} | " f"T={data['temp_inside_box']:.1f}°C | " f"R₀={data['r0']:.0f}mm") else: print("← No data received") print() counter += 1 time.sleep(solar.os_config.loop_sleep) except KeyboardInterrupt: print("\n✓ Stop requested by user") return 0 except Exception as e: print(f"\n✗ Unexpected error: {e}") import traceback traceback.print_exc() return 1 finally: solar.disconnect() if __name__ == "__main__": sys.exit(main())