import numpy as np import sounddevice as sd import ctypes import time import sys import os import wave import threading ADC_CHANNELS = [1, 3, 5, 7, 9, 11, 13] ADC_RANGE = 6 ADC_CONVERSION_TIME = 1 ADC_SINGLE_ENDED = 0 SPIKE_THRESHOLD_UV = { 1: -8000.0, 2: 0.0, 3: 180000.0, 4: -200.0, 5: 20.0, 6: -800.0, 7: 60.0, } SPIKE_COOLDOWN_S = 1.0 AMBIENT_V_MIN = 0.0 AMBIENT_V_MAX = 300.0 AMBIENT_MAX_VOL = 0.6 AMBIENT_SMOOTHING = 0.1 WAV_DIR = os.path.expanduser("Libby/Desktop/hyphae_audios") SPIKE_WAV_NAMES = { 1: "1.wav", 2: "2.wav", 3: "3.wav", 4: "4.wav", 5: "5.wav", 6: "6.wav", 7: "7.wav", } AMBIENT_WAV_NAMES = { 1: "ambient_metal.wav", 2: "ambient_strings.wav", 3: "ambient_violin.wav", } SPIKE_WAV_VOLUME = 0.8 SAMPLE_RATE = 44100 BLOCK_SIZE = 1024 RANGE_MV = { 0: 2500.0, 1: 1250.0, 2: 625.0, 3: 313.0, 4: 156.0, 5: 78.0, 6: 39.0, } ADC_MAX_VALUE = 2**23 active_wavs = [] active_wavs_lock = threading.Lock() ambient_loops = {} ambient_lock = threading.Lock() last_spike_time = [0.0] * 7 def load_wav(path): if not os.path.exists(path): return None with wave.open(path, 'rb') as wf: n_channels = wf.getnchannels() sampwidth = wf.getsampwidth() framerate = wf.getframerate() n_frames = wf.getnframes() raw = wf.readframes(n_frames) if sampwidth == 2: data = np.frombuffer(raw, dtype=np.int16).astype(np.float32) / 32768.0 elif sampwidth == 3: data = np.zeros(n_frames * n_channels, dtype=np.float32) for j in range(n_frames * n_channels): b = raw[j*3:(j+1)*3] val = int.from_bytes(b, byteorder='little', signed=True) data[j] = val / 8388608.0 elif sampwidth == 4: data = np.frombuffer(raw, dtype=np.int32).astype(np.float32) / 2147483648.0 elif sampwidth == 1: data = (np.frombuffer(raw, dtype=np.uint8).astype(np.float32) - 128.0) / 128.0 else: return None if n_channels == 2: data = (data[0::2] + data[1::2]) / 2.0 elif n_channels > 2: data = data[0::n_channels] if framerate != SAMPLE_RATE: duration = len(data) / framerate n_new = int(duration * SAMPLE_RATE) x_old = np.linspace(0, 1, len(data)) x_new = np.linspace(0, 1, n_new) data = np.interp(x_new, x_old, data) dur = len(data) / SAMPLE_RATE print(f" Loaded {os.path.basename(path)} ({dur:.1f}s)") return data def audio_callback(outdata, frames, time_info, status): """Mix spike wavs + ambient loops.""" signal = np.zeros(frames, dtype=np.float32) with active_wavs_lock: finished = [] for i, layer in enumerate(active_wavs): samples, pos = layer remaining = len(samples) - pos if remaining <= 0: finished.append(i) continue chunk_len = min(frames, remaining) signal[:chunk_len] += SPIKE_WAV_VOLUME * samples[pos:pos + chunk_len] layer[1] = pos + chunk_len if pos + chunk_len >= len(samples): finished.append(i) for i in reversed(finished): active_wavs.pop(i) with ambient_lock: for key, amb in ambient_loops.items(): samples = amb["samples"] pos = amb["position"] amb["current_vol"] += (amb["target_vol"] - amb["current_vol"]) * AMBIENT_SMOOTHING vol = amb["current_vol"] if vol < 0.001: amb["position"] = (pos + frames) % len(samples) continue remaining = len(samples) - pos if remaining >= frames: signal[:frames] += vol * samples[pos:pos + frames] amb["position"] = pos + frames else: signal[:remaining] += vol * samples[pos:] leftover = frames - remaining signal[remaining:remaining + leftover] += vol * samples[:leftover] amb["position"] = leftover amb["position"] = amb["position"] % len(samples) signal = np.clip(signal, -1.0, 1.0) outdata[:, 0] = signal def adc_count_to_uv(adc_value): range_mv = RANGE_MV[ADC_RANGE] voltage_mv = (adc_value / ADC_MAX_VALUE) * range_mv return voltage_mv * 1000.0 def voltage_to_ambient_vol(v_uv): """Map absolute voltage to ambient volume.""" v_abs = abs(v_uv) if v_abs <= AMBIENT_V_MIN: return 0.0 if v_abs >= AMBIENT_V_MAX: return AMBIENT_MAX_VOL normalized = (v_abs - AMBIENT_V_MIN) / (AMBIENT_V_MAX - AMBIENT_V_MIN) return normalized * AMBIENT_MAX_VOL def main(): print("=" * 60) print(" MYCELIUM SONIFICATION — 7 ELECTRODES") print("=" * 60) print(f" Channels: {ADC_CHANNELS}") print(f" Range: ±{RANGE_MV[ADC_RANGE]} mV") print(f" Spike thresh: ±{SPIKE_THRESHOLD_UV} µV") print(f" Spike cool: {SPIKE_COOLDOWN_S}s") print(f" Wav dir: {WAV_DIR}") print(f" Ambient: E1→metal, E2→strings, E3→violin") print("=" * 60) print("\\n Loading spike wav files...") spike_wavs = {} for electrode_idx in range(1, 8): name = SPIKE_WAV_NAMES[electrode_idx] data = load_wav(os.path.join(WAV_DIR, name)) if data is not None: spike_wavs[electrode_idx] = data print("\\n Loading ambient wav files...") for electrode_idx, name in AMBIENT_WAV_NAMES.items(): data = load_wav(os.path.join(WAV_DIR, name)) if data is not None: ambient_loops[electrode_idx] = { "samples": data, "position": 0, "current_vol": 0.0, "target_vol": 0.0, } if not spike_wavs and not ambient_loops: print(" ERROR: No wav files loaded.") sys.exit(1) print(f"\\n Spike wavs: {len(spike_wavs)}, Ambient loops: {len(ambient_loops)}") try: from picosdk.picohrdl import picohrdl as hrdl except ImportError: print(" ERROR: picosdk not installed.") sys.exit(1) print("\\n Opening ADC-24...") chandle = hrdl.HRDLOpenUnit() print(f" Handle: {chandle}") if chandle <= 0: sys.exit(1) hrdl.HRDLSetMains(chandle, 1) print(" Configuring channels...") for ch in ADC_CHANNELS: status_ch = hrdl.HRDLSetAnalogInChannel( chandle, ctypes.c_int16(ch), ctypes.c_int16(1), ctypes.c_int16(ADC_RANGE), ctypes.c_int16(ADC_SINGLE_ENDED), ) stream = sd.OutputStream( samplerate=SAMPLE_RATE, blocksize=BLOCK_SIZE, channels=1, dtype='float32', callback=audio_callback, ) stream.start() reading_count = 0 try: while True: voltages = [] now = time.time() for i, ch in enumerate(ADC_CHANNELS): electrode_num = i + 1 overflow = ctypes.c_int16(0) value = ctypes.c_int32(0) status_read = hrdl.HRDLGetSingleValue( chandle, ctypes.c_int16(ch), ctypes.c_int16(ADC_RANGE), ctypes.c_int16(ADC_CONVERSION_TIME), ctypes.c_int16(ADC_SINGLE_ENDED), ctypes.byref(overflow), ctypes.byref(value), ) if status_read == 1: v_uv = adc_count_to_uv(value.value) voltages.append(v_uv) if abs(v_uv) > SPIKE_THRESHOLD_UV[electrode_num]: if now - last_spike_time[i] >= SPIKE_COOLDOWN_S: if electrode_num in spike_wavs: with active_wavs_lock: active_wavs.append([spike_wavs[electrode_num], 0]) last_spike_time[i] = now if electrode_num in ambient_loops: with ambient_lock: ambient_loops[electrode_num]["target_vol"] = voltage_to_ambient_vol(v_uv) else: voltages.append(float('nan')) reading_count += 1 v_strs = [f"{v:7.1f}" if not np.isnan(v) else " err " for v in voltages] spike_indicators = "" for i, v in enumerate(voltages): electrode_num = i + 1 if not np.isnan(v) and abs(v) > SPIKE_THRESHOLD_UV[electrode_num]: spike_indicators += f" E{i+1}!" amb_vols = "" with ambient_lock: for key in sorted(ambient_loops.keys()): amb_vols += f" a{key}:{ambient_loops[key]['current_vol']:.2f}" print(f"\\r E1:{voltages[0]:8.1f} E2:{voltages[1]:8.1f} E3:{voltages[2]:8.1f} E4:{voltages[3]:8.1f} E5:{voltages[4]:8.1f} E6:{voltages[5]:8.1f} E7:{voltages[6]:8.1f} µV", end='', flush=True) except KeyboardInterrupt: print("\\n\\n Stopping...") finally: stream.stop() stream.close() hrdl.HRDLCloseUnit(chandle) print(" Done.") if __name__ == "__main__": main()