diff --git a/README.md b/README.md index 7e3a80c..4568082 100644 --- a/README.md +++ b/README.md @@ -9,6 +9,7 @@ Refer to the [wiki](https://github.com/diglet48/restim/wiki) for help. * Stereostim (three-phase only) and other audio-based devices (Mk312, 2B, ...) * FOC-Stim * NeoDK (coming soon) +* Coyote 3.0 (coming soon, experimental) ## Main features diff --git a/designer/device_wizard/coyote_waveform_select.ui b/designer/device_wizard/coyote_waveform_select.ui new file mode 100644 index 0000000..aa61a18 --- /dev/null +++ b/designer/device_wizard/coyote_waveform_select.ui @@ -0,0 +1,54 @@ + + + WizardPageCoyote + + + + 0 + 0 + 611 + 497 + + + + WizardPage + + + + + + Three-phase + + + + + + + <html><head/><body> + <p>A = left<br/>B = right<br/>C = neutral</p> + <p>Connect A- and B- to a shared common electrode (e.g. a conductive rubber loop).</p> + </body></html> + + + true + + + + + + + Qt::Vertical + + + + 20 + 40 + + + + + + + + + diff --git a/designer/device_wizard/type_select.ui b/designer/device_wizard/type_select.ui index db8a6e3..1e692d0 100644 --- a/designer/device_wizard/type_select.ui +++ b/designer/device_wizard/type_select.ui @@ -34,7 +34,21 @@ + + + + NeoStim + + + + + + Coyote 3 + + + + Qt::Orientation::Vertical @@ -47,15 +61,8 @@ - - - - NeoStim - - - - + \ No newline at end of file diff --git a/designer/mainwindow.ui b/designer/mainwindow.ui index 7a2cdff..9103160 100644 --- a/designer/mainwindow.ui +++ b/designer/mainwindow.ui @@ -254,6 +254,11 @@ Carrier settings + + + Coyote + + Pulse settings @@ -486,6 +491,12 @@
qt_ui/three_phase_settings_widget.h
1 + + CoyoteSettingsWidget + QWidget +
qt_ui/coyote_settings_widget.h
+ 1 + PulseSettingsWidget QWidget diff --git a/designer/preferencesdialog.ui b/designer/preferencesdialog.ui index 64b6d88..6aa95b8 100644 --- a/designer/preferencesdialog.ui +++ b/designer/preferencesdialog.ui @@ -612,6 +612,208 @@ + + + Coyote + + + + + + Device + + + + + + Channel A Limit + + + + + + + 0 + + + 200 + + + + + + + Channel B Limit + + + + + + + 0 + + + 200 + + + + + + + Channel A Freq Balance + + + + + + + 0 + + + 255 + + + + + + + Channel B Freq Balance + + + + + + + 0 + + + 255 + + + + + + + Channel A Intensity Balance + + + + + + + 0 + + + 255 + + + + + + + Channel B Intensity Balance + + + + + + + 0 + + + 255 + + + + + + + + + + Algorithm + + + + + + Max Intensity Change per Pulse (%) + + + + + + + 1 + + + 0.0 + + + 100.0 + + + 0.1 + + + 3.0 + + + + + + + + + + Display + + + + + + Graph Window (s) + + + + + + + 1 + + + 0.1 + + + 10.0 + + + 0.1 + + + 3.0 + + + + + + + Debug Logging + + + + + + + + + + + + + + + Media sync diff --git a/device/coyote/algorithm.py b/device/coyote/algorithm.py new file mode 100644 index 0000000..e1bf088 --- /dev/null +++ b/device/coyote/algorithm.py @@ -0,0 +1,282 @@ +""" +DG-LAB Coyote 3.0 E-Stim Algorithm Implementation + +EXPERIMENTAL: Best-effort adaptation of restim's funscript-based algorithms to the Coyote 3.0's +hardware square pulse generator. This implementation attempts to simulate smooth parameter changes +and modulation on hardware that fundamentally outputs discrete square pulses. + +Protocol Specification (Chinese): +https://github.com/DG-LAB-OPENSOURCE/DG-LAB-OPENSOURCE/blob/main/coyote/v3/README_V3.md + +Hardware Specifications & Limitations: +-------------------------------------- +- Two independent channels (A and B) +- Square pulse generator only (no smooth waveforms like continuous audio devices) +- Pulse parameters: + * Intensity: 0-100% + * Duration: 5-240ms (sent as "waveform frequency" parameter in protocol, despite the name) + - Spec documents 10-240ms range, but hardware appears to support down to 5ms + - Spec also provides an optional extended mapping from input values 10-1000 → output 10-240 + * Relationship: frequency_hz = 1000 / duration_ms (simple inverse) + * Effective frequency range: ~4.17Hz (240ms) to 200Hz (5ms) + * This algorithm works in Hz internally, then converts to duration_ms when sending to device +- Protocol: B0 command contains 4 pulses per packet (20 bytes total) +- Spec recommends ~100ms update rate, but this implementation uses adaptive scheduling + (sends next packet at 80% of current packet duration for seamless transitions) +- Device repeats last packet until new one arrives +- Invalid waveform data causes device to discard entire 4-pulse packet for that channel +- Channel strength range: 0-200 (separate from pulse intensity 0-100%) +- Balance parameters (BF command): frequency balance and intensity balance affect perceived output + +Key Limitations: +- Cannot produce smooth continuous waveforms - only discrete square pulses +- Limited frequency range compared to audio-based devices +- Packet-based protocol requires continuous streaming (no gaps or device repeats last packet) +- No native envelope/modulation support - must be simulated via parameter variations +- Hardware balance parameters (frequency/intensity) affect output independently +- Each pulse duration is quantized to integer milliseconds + +Algorithm Overview (Best-Effort Approach): +------------------------------------------ +- Maps funscript pulse_frequency to channel-specific frequency ranges +- Applies optional jitter (pulse_interval_random) to pulse timing +- Adds zero-mean micro-texture via pulse_width modulation to simulate smoothness +- Smooths intensity transitions based on pulse_rise_time +- Maintains pulse queues (750ms horizon) for continuous output +- Uses barycentric mapping for three-phase position diagram intensity control +- Adaptive packet scheduling (80% of packet duration) for seamless output + +Each channel maintains an independent pulse queue. The algorithm attempts to create perceptually +smooth sensations by varying pulse duration and intensity, but results will differ significantly +from audio-based continuous algorithms due to fundamental hardware limitations. +""" + +from __future__ import annotations + +import logging +import time +from dataclasses import dataclass +from typing import Dict, List, Optional, Tuple + +from device.coyote.channel_controller import ChannelController +from device.coyote.channel_state import ChannelState +from device.coyote.common import normalize, split_seconds, volume_at +from device.coyote.config import PulseTuning, load_pulse_tuning +from device.coyote.constants import PULSES_PER_PACKET +from device.coyote.pulse_generator import PulseGenerator +from device.coyote.types import CoyotePulse, CoyotePulses +from stim_math.axis import AbstractMediaSync +from stim_math.audio_gen.params import CoyoteAlgorithmParams, SafetyParams +from stim_math.audio_gen.various import ThreePhasePosition +from stim_math.threephase import ThreePhaseCenterCalibration + +logger = logging.getLogger("restim.coyote") + + +@dataclass +class ChannelPipeline: + name: str + generator: PulseGenerator + controller: ChannelController + state: ChannelState + + +class CoyoteAlgorithm: + def __init__( + self, + media: AbstractMediaSync, + params: CoyoteAlgorithmParams, + safety_limits: SafetyParams, + carrier_freq_limits: Tuple[float, float], + pulse_freq_limits: Tuple[float, float], + pulse_width_limits: Tuple[float, float], + pulse_rise_time_limits: Tuple[float, float], + tuning: Optional[PulseTuning] = None, + ) -> None: + self.media = media + self.params = params + self.safety_limits = safety_limits + self._carrier_limits = carrier_freq_limits + self._pulse_rise_time_limits = pulse_rise_time_limits # retained for API compatibility + self.tuning = tuning or load_pulse_tuning() + + self.position = ThreePhasePosition(params.position, params.transform) + + channels: List[ChannelPipeline] = [] + for name, channel_params in (("A", params.channel_a), ("B", params.channel_b)): + generator = PulseGenerator(name, params, channel_params, carrier_freq_limits, pulse_freq_limits, pulse_width_limits, self.tuning) + controller = ChannelController(name, media, params, generator, self._positional_intensity, self.tuning) + state = ChannelState() + channels.append(ChannelPipeline(name, generator, controller, state)) + self._channels: Tuple[ChannelPipeline, ...] = tuple(channels) + + self._last_update_time: Optional[float] = None + self.next_update_time: float = 0.0 + self._start_time: Optional[float] = None + + def generate_packet(self, current_time: float) -> Optional[CoyotePulses]: + if self._last_update_time is None: + self._last_update_time = current_time + + delta_ms = max(0.0, (current_time - self._last_update_time) * 1000.0) + self._last_update_time = current_time + + self._advance_state(current_time, delta_ms) + + if not self._needs_packet(): + self._schedule_from_remaining(current_time) + return None + + for channel in self._channels: + channel.controller.fill_queue(current_time) + + packet_map: Dict[str, List[CoyotePulse]] = {} + duration_map: Dict[str, int] = {} + for channel in self._channels: + pulses = channel.controller.next_packet() + channel.state.load_packet(current_time, pulses) + packet_map[channel.name] = pulses + duration_map[channel.name] = sum(p.duration for p in pulses) + + pulses_a = packet_map.get("A", []) + pulses_b = packet_map.get("B", []) + duration_a_ms = duration_map.get("A", 0) + duration_b_ms = duration_map.get("B", 0) + + durations = [duration for duration in duration_map.values() if duration > 0] + if not durations: + durations = [1] + min_duration_ms = max(1, min(durations)) + self.next_update_time = current_time + (min_duration_ms / 1000.0) * self.tuning.packet_margin + + self._log_packet(current_time, pulses_a, pulses_b, duration_a_ms, duration_b_ms) + + return CoyotePulses(pulses_a, pulses_b) + + def get_next_update_time(self) -> float: + return self.next_update_time + + def _needs_packet(self) -> bool: + queue_low = any(not channel.controller.has_pulses(PULSES_PER_PACKET) for channel in self._channels) + ready = any(channel.state.ready() for channel in self._channels) + return ready or queue_low + + def _advance_state(self, current_time: float, delta_ms: float) -> None: + for channel in self._channels: + channel.state.advance(delta_ms) + + delta_s = delta_ms / 1000.0 + if delta_s <= 0: + return + + carrier_hz = float(self.params.carrier_frequency.interpolate(current_time)) + carrier_norm = normalize(carrier_hz, self._carrier_limits) + texture_speed = self.tuning.texture_min_hz + (self.tuning.texture_max_hz - self.tuning.texture_min_hz) * carrier_norm + + for channel in self._channels: + channel.generator.advance_phase(texture_speed, delta_s) + + def _schedule_from_remaining(self, current_time: float) -> None: + remaining = min(channel.state.remaining_ms() for channel in self._channels) + self.next_update_time = current_time + (remaining / 1000.0) * self.tuning.packet_margin + + def _positional_intensity(self, time_s: float, volume: float) -> Tuple[int, int]: + alpha, beta = self.position.get_position(time_s) + + w_left = max(0.0, (beta + 1.0) / 2.0) + w_right = max(0.0, (1.0 - beta) / 2.0) + w_neutral = max(0.0, alpha) + + total = w_left + w_right + w_neutral + if total > 0: + w_left /= total + w_right /= total + w_neutral /= total + else: + w_left = w_right = w_neutral = 0.0 + + center_db = float(self.params.calibrate.center.last_value()) + scale = ThreePhaseCenterCalibration(center_db).get_scale(alpha, beta) + + intensity_a = int((w_left + w_neutral) * volume * scale * 100.0) + intensity_b = int((w_right + w_neutral) * volume * scale * 100.0) + return intensity_a, intensity_b + + def _log_packet( + self, + current_time: float, + pulses_a: List[CoyotePulse], + pulses_b: List[CoyotePulse], + duration_a_ms: int, + duration_b_ms: int, + ) -> None: + if not logger.isEnabledFor(logging.DEBUG): + return + + alpha, beta = self.position.get_position(current_time) + comps = self._display_time_components(current_time) + media_type = self._media_type() + volume = volume_at(self.media, self.params.volume, current_time) + + lines = [ + "=" * 72, + f"Packet Generated @ {comps[0]:02}:{comps[1]:02}:{comps[2]:02}.{comps[3]:03} [{media_type}]", + "=" * 72, + f"Position: alpha={alpha:+.2f}, beta={beta:+.2f}, volume={volume:.0%}", + "", + f"Channel A: duration={duration_a_ms:.0f} ms", + ] + + for idx, pulse in enumerate(pulses_a, 1): + lines.append(f" Pulse {idx}: {pulse.duration} ms @ {pulse.frequency} Hz ({pulse.intensity}%)") + + lines.extend(["", f"Channel B: duration={duration_b_ms:.0f} ms"]) + for idx, pulse in enumerate(pulses_b, 1): + lines.append(f" Pulse {idx}: {pulse.duration} ms @ {pulse.frequency} Hz ({pulse.intensity}%)") + + next_ms = max(0.0, (self.next_update_time - current_time) * 1000.0) + lines.extend( + [ + "", + f"Next update: {next_ms:.0f} ms " + f"(packet_dur_a={duration_a_ms:.0f} ms, packet_dur_b={duration_b_ms:.0f} ms, margin={self.tuning.packet_margin:.0%})", + "=" * 72, + "", + ] + ) + + logger.debug("\n".join(lines)) + + def _media_type(self) -> str: + media_type = getattr(self.media, "media_type", None) + if media_type: + return str(media_type) + class_name = self.media.__class__.__name__.lower() + if class_name.startswith("internal"): + return "internal" + if "vlc" in class_name: + return "vlc" + if "mpv" in class_name: + return "mpv" + return class_name + + def _display_time_components(self, current_time: float): + media_type = getattr(self.media, "media_type", None) + if media_type and str(media_type).lower() != "internal": + mapper = getattr(self.media, "map_timestamp", None) + if callable(mapper) and self.media.is_playing(): + try: + rel_time = mapper(time.time()) + if rel_time is not None and rel_time >= 0: + return split_seconds(rel_time) + except Exception: # pragma: no cover - defensive + pass + + if media_type and str(media_type).lower() == "internal": + now = time.localtime() + millis = int((time.time() - int(time.time())) * 1000) + return now.tm_hour, now.tm_min, now.tm_sec, millis + + if self._start_time is None: + self._start_time = current_time + return split_seconds(current_time - self._start_time) diff --git a/device/coyote/channel_controller.py b/device/coyote/channel_controller.py new file mode 100644 index 0000000..94da5b6 --- /dev/null +++ b/device/coyote/channel_controller.py @@ -0,0 +1,153 @@ +from __future__ import annotations + +import logging +from collections import deque +from typing import Callable, Deque, List, Tuple + +from device.coyote.common import clamp, volume_at +from device.coyote.config import PulseTuning +from device.coyote.constants import MIN_PULSE_DURATION_MS, PULSES_PER_PACKET +from device.coyote.pulse_generator import PulseDebug, PulseGenerator +from device.coyote.types import CoyotePulse +from stim_math.axis import AbstractMediaSync +from stim_math.audio_gen.params import CoyoteAlgorithmParams + +logger = logging.getLogger("restim.coyote") + + +class ChannelController: + """Maintains a rolling queue of pulses for a single hardware channel.""" + + def __init__( + self, + name: str, + media: AbstractMediaSync, + params: CoyoteAlgorithmParams, + generator: PulseGenerator, + positional_intensity_fn: Callable[[float, float], Tuple[int, int]], + tuning: PulseTuning, + ) -> None: + self._name = name + self._media = media + self._params = params + self._generator = generator + self._positional_intensity_fn = positional_intensity_fn + self._tuning = tuning + + self._queue: Deque[CoyotePulse] = deque() + self._queued_ms = 0.0 + + self._last_intensity: float | None = None + self._last_time: float | None = None + self._max_change_per_pulse = float(self._params.max_intensity_change_per_pulse.get()) + + def has_pulses(self, count: int) -> bool: + return len(self._queue) >= count + + def queue_duration_ms(self) -> float: + return self._queued_ms + + def fill_queue(self, now_s: float) -> None: + horizon_end = now_s + self._tuning.queue_horizon_s + coverage_s = self._queued_ms / 1000.0 + end_time = now_s + coverage_s + + added: List[CoyotePulse] = [] + seq_index = 0 + while end_time < horizon_end or len(self._queue) < PULSES_PER_PACKET: + pulse_time = end_time + pulse, debug = self._generate_pulse(pulse_time, seq_index) + self._queue.append(pulse) + added.append(pulse) + self._queued_ms += pulse.duration + end_time += pulse.duration / 1000.0 + seq_index += 1 + + if added and logger.isEnabledFor(logging.DEBUG): + durations = [p.duration for p in added] + freqs = [p.frequency for p in added] + logger.debug( + "[%s] queued %d pulses (dur %d-%d ms, freq %d-%d Hz)", + self._name, + len(added), + min(durations), + max(durations), + min(freqs), + max(freqs), + ) + + def next_packet(self) -> List[CoyotePulse]: + packet: List[CoyotePulse] = [] + while len(packet) < PULSES_PER_PACKET: + if self._queue: + pulse = self._queue.popleft() + self._queued_ms = max(0.0, self._queued_ms - pulse.duration) + packet.append(pulse) + else: + pulse = CoyotePulse(frequency=0, intensity=0, duration=MIN_PULSE_DURATION_MS) + packet.append(pulse) + return packet + + def _generate_pulse(self, time_s: float, seq_index: int) -> Tuple[CoyotePulse, PulseDebug]: + volume = volume_at(self._media, self._params.volume, time_s) + intensity_a, intensity_b = self._positional_intensity_fn(time_s, volume) + target = float(intensity_a if self._name == "A" else intensity_b) + smoothed = self._smooth_intensity(target, time_s) + pulse, debug = self._generator.create_pulse(time_s, int(round(smoothed)), seq_index) + + if logger.isEnabledFor(logging.DEBUG): + logger.debug( + ( + " [%s] pulse #%d: freq_raw=%.2f Hz, freq_norm=%.2f, " + "freq_mapped=%.2f Hz, freq_limits=(%.1f-%.1f) Hz | " + "base_dur=%.2f ms, dur_limits=(%d-%d) ms, width_norm=%.2f, jitter=%.0f%% | " + "texture_mode=%s, texture_up=%.2f ms, texture_dn=%.2f ms, " + "texture_used=%.2f ms | desired=%.2f ms, residual=%.2f ms | " + "result: dur=%d ms, freq=%d Hz, intensity=%d%%" + ), + self._name, + debug.sequence_index, + debug.raw_frequency_hz, + debug.normalised_frequency, + debug.mapped_frequency_hz, + debug.frequency_limits[0], + debug.frequency_limits[1], + debug.base_duration_ms, + debug.duration_limits[0], + debug.duration_limits[1], + debug.width_normalised, + debug.jitter_fraction * 100.0, + debug.texture_mode, + debug.texture_headroom_up_ms, + debug.texture_headroom_down_ms, + debug.texture_applied_ms, + debug.desired_duration_ms, + debug.residual_ms, + pulse.duration, + pulse.frequency, + pulse.intensity, + ) + + return pulse, debug + + def _smooth_intensity(self, target: float, time_s: float) -> float: + carrier_hz = float(self._params.carrier_frequency.interpolate(time_s)) + rise_cycles = float(self._params.pulse_rise_time.interpolate(time_s)) + tau_s = rise_cycles / carrier_hz if carrier_hz > 0 else 0.0 + + last_value = self._last_intensity + last_time = self._last_time + + if last_value is None or tau_s <= 0 or last_time is None: + result = target + else: + dt = max(0.0, time_s - last_time) + allowed = (dt / tau_s) * 100.0 if tau_s > 0 else float("inf") + if self._max_change_per_pulse > 0: + allowed = min(allowed, self._max_change_per_pulse) + delta = clamp(target - last_value, -allowed, allowed) + result = last_value + delta + + self._last_intensity = result + self._last_time = time_s + return result diff --git a/device/coyote/channel_state.py b/device/coyote/channel_state.py new file mode 100644 index 0000000..3bb8529 --- /dev/null +++ b/device/coyote/channel_state.py @@ -0,0 +1,46 @@ +from __future__ import annotations + +from collections import deque +from typing import Deque, Iterable, List + +from device.coyote.types import CoyotePulse + + +class ChannelState: + """Tracks playback progress for the last packet issued to a channel.""" + + def __init__(self) -> None: + self._current_packet: Deque[CoyotePulse] = deque() + self._elapsed_ms = 0.0 + self._total_ms = 0.0 + self._start_time_s = 0.0 + self._finish_time_s = 0.0 + + def load_packet(self, start_time_s: float, packet: Iterable[CoyotePulse]) -> None: + pulses = list(packet) + self._current_packet = deque(pulses) + self._total_ms = float(sum(p.duration for p in pulses)) + self._elapsed_ms = 0.0 + self._start_time_s = start_time_s + self._finish_time_s = start_time_s + (self._total_ms / 1000.0 if self._total_ms else 0.0) + + def advance(self, delta_ms: float) -> None: + if delta_ms <= 0: + return + self._elapsed_ms += delta_ms + + def remaining_ms(self) -> float: + if self._total_ms == 0: + return 0.0 + return max(0.0, self._total_ms - self._elapsed_ms) + + def ready(self) -> bool: + return self.remaining_ms() <= 0.0 + + @property + def finish_time_s(self) -> float: + return self._finish_time_s + + @property + def current_packet(self) -> List[CoyotePulse]: + return list(self._current_packet) diff --git a/device/coyote/common.py b/device/coyote/common.py new file mode 100644 index 0000000..80a2b7a --- /dev/null +++ b/device/coyote/common.py @@ -0,0 +1,42 @@ +from __future__ import annotations + +from typing import Tuple + +from stim_math.audio_gen.params import VolumeParams +from stim_math.axis import AbstractMediaSync + + +def clamp(value: float, lower: float, upper: float) -> float: + if lower > upper: + lower, upper = upper, lower + return max(lower, min(value, upper)) + + +def normalize(value: float, bounds: Tuple[float, float]) -> float: + low, high = bounds + if high <= low: + return 0.0 + return clamp((value - low) / (high - low), 0.0, 1.0) + + +def volume_at(media: AbstractMediaSync, volume: VolumeParams, time_s: float) -> float: + if not media.is_playing(): + return 0.0 + + master = clamp(float(volume.master.last_value()), 0.0, 1.0) + api = clamp(float(volume.api.interpolate(time_s)), 0.0, 1.0) + inactivity = clamp(float(volume.inactivity.last_value()), 0.0, 1.0) + external = clamp(float(volume.external.last_value()), 0.0, 1.0) + + if inactivity == 0: + inactivity = 1.0 + + return master * api * inactivity * external + + +def split_seconds(seconds: float) -> Tuple[int, int, int, int]: + hours = int(seconds // 3600) + minutes = int((seconds % 3600) // 60) + secs = int(seconds % 60) + millis = int((seconds - int(seconds)) * 1000) + return hours, minutes, secs, millis diff --git a/device/coyote/config.py b/device/coyote/config.py new file mode 100644 index 0000000..25498db --- /dev/null +++ b/device/coyote/config.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +from dataclasses import dataclass + +from qt_ui import settings + + +def _clamp(value: float, lower: float, upper: float) -> float: + return max(lower, min(value, upper)) + + +@dataclass(frozen=True) +class PulseTuning: + queue_horizon_s: float + packet_margin: float + texture_min_hz: float + texture_max_hz: float + texture_depth_fraction: float + jitter_limit_fraction: float + residual_bound: float + + @classmethod + def from_settings(cls) -> "PulseTuning": + queue_horizon = max(0.1, float(settings.coyote_queue_horizon_seconds.get())) + margin = _clamp(float(settings.coyote_packet_margin.get()), 0.1, 1.0) + + texture_min = max(0.0, float(settings.coyote_texture_min_hz.get())) + texture_max = max(texture_min + 1e-6, float(settings.coyote_texture_max_hz.get())) + + depth = _clamp(float(settings.coyote_texture_depth_fraction.get()), 0.0, 1.0) + jitter_limit = _clamp(float(settings.coyote_jitter_limit_fraction.get()), 0.0, 1.0) + residual = max(0.0, float(settings.coyote_residual_bound.get())) + + return cls( + queue_horizon_s=queue_horizon, + packet_margin=margin, + texture_min_hz=texture_min, + texture_max_hz=texture_max, + texture_depth_fraction=depth, + jitter_limit_fraction=jitter_limit, + residual_bound=residual, + ) + + +def load_pulse_tuning() -> PulseTuning: + """Helper for consumers that do not need to customise tuning.""" + return PulseTuning.from_settings() diff --git a/device/coyote/constants.py b/device/coyote/constants.py new file mode 100644 index 0000000..51c7862 --- /dev/null +++ b/device/coyote/constants.py @@ -0,0 +1,35 @@ +# BLE device identification +DEVICE_NAME = "47L121000" + +# Coyote hardware timing and derived limits +# Pulse duration limits in milliseconds +MIN_PULSE_DURATION_MS = 5 +MAX_PULSE_DURATION_MS = 240 + +# Derived hardware frequency limits (Hz) +HARDWARE_MAX_FREQ_HZ = 1000.0 / MIN_PULSE_DURATION_MS # ~200 Hz +HARDWARE_MIN_FREQ_HZ = 1000.0 / MAX_PULSE_DURATION_MS # ~4.17 Hz + +# Packet behaviour +PULSES_PER_PACKET = 4 + +# BLE / Protocol constants +LOG_PREFIX = "[Coyote]" +BATTERY_SERVICE_UUID = "0000180A-0000-1000-8000-00805f9b34fb" +MAIN_SERVICE_UUID = "0000180C-0000-1000-8000-00805f9b34fb" +WRITE_CHAR_UUID = "0000150A-0000-1000-8000-00805f9b34fb" +NOTIFY_CHAR_UUID = "0000150B-0000-1000-8000-00805f9b34fb" +BATTERY_CHAR_UUID = "00001500-0000-1000-8000-00805f9b34fb" + +CMD_B0 = 0xB0 +CMD_POWER_UPDATE = 0xB1 +CMD_ACK = 0x51 +CMD_ACTIVE_POWER = 0x53 + +INTERP_ABSOLUTE_SET = 0b11 +INTERP_NO_CHANGE = 0b00 +SEQUENCE_MODULO = 16 # 4-bit sequence number wraps at 16 +B0_NO_PULSES_PAD_BYTES = 16 + +# Connection behavior +SCAN_RETRY_SECONDS = 5 diff --git a/device/coyote/device.py b/device/coyote/device.py new file mode 100644 index 0000000..5820a74 --- /dev/null +++ b/device/coyote/device.py @@ -0,0 +1,441 @@ +import asyncio +import logging +from typing import Optional +import time +import threading + +from bleak import BleakClient, BleakScanner +from device.output_device import OutputDevice + +from PySide6.QtCore import QObject, Signal +from device.coyote.constants import ( + LOG_PREFIX, + WRITE_CHAR_UUID, + NOTIFY_CHAR_UUID, + BATTERY_CHAR_UUID, + CMD_B0, + CMD_POWER_UPDATE, + CMD_ACK, + CMD_ACTIVE_POWER, + INTERP_ABSOLUTE_SET, + INTERP_NO_CHANGE, + SEQUENCE_MODULO, + B0_NO_PULSES_PAD_BYTES, + PULSES_PER_PACKET, + SCAN_RETRY_SECONDS, +) +from device.coyote.types import CoyoteParams, CoyotePulse, CoyotePulses, CoyoteStrengths, ConnectionStage +from device.coyote.algorithm import CoyoteAlgorithm + +logger = logging.getLogger('restim.coyote') + +class CoyoteDevice(OutputDevice, QObject): + parameters: CoyoteParams = None + connection_status_changed = Signal(bool, str) # Connected, Stage + battery_level_changed = Signal(int) + parameters_changed = Signal() + power_levels_changed = Signal(CoyoteStrengths) + pulse_sent = Signal(CoyotePulses) + + def __init__(self, device_name: str): + OutputDevice.__init__(self) + QObject.__init__(self) + self.device_name = device_name + self.client: Optional[BleakClient] = None + self.algorithm: Optional[CoyoteAlgorithm] = None + self.running = False + self.connection_stage = ConnectionStage.DISCONNECTED + self.strengths = CoyoteStrengths(channel_a=0, channel_b=0) + self.battery_level = 100 + self.parameters = None + self._event_loop = None + self.sequence_number = 1 + + # Start connection process + self._start_connection_loop() + + def _start_connection_loop(self): + """Start the connection process in a separate thread""" + loop = asyncio.new_event_loop() + self._event_loop = loop + asyncio.set_event_loop(loop) + + def run_loop(): + logger.info(f"{LOG_PREFIX} Starting asyncio loop thread") + loop.run_until_complete(self._connection_loop()) + loop.run_forever() + + threading.Thread(target=run_loop, daemon=True).start() + + async def _connection_loop(self): + """Main connection loop that runs the state machine""" + logger.info(f"{LOG_PREFIX} Starting connection loop") + prev_stage = self.connection_stage + + attempt_counter = 0 + + while True: + try: + # Check if client is still connected + if (self.connection_stage == ConnectionStage.CONNECTED and + (not self.client or not self.client.is_connected)): + logger.warning(f"{LOG_PREFIX} Device disconnected unexpectedly") + await self.disconnect() + continue + + if self.connection_stage == ConnectionStage.DISCONNECTED: + logger.info(f"{LOG_PREFIX} Starting connection process") + self.connection_stage = ConnectionStage.SCANNING + + elif self.connection_stage == ConnectionStage.SCANNING: + if await self._scan_for_device(): + attempt_counter = 0 + logger.info(f"{LOG_PREFIX} Device found, connecting...") + self.connection_stage = ConnectionStage.CONNECTING + else: + attempt_counter += 1 + logger.info(f"{LOG_PREFIX} Device not found (attempt {attempt_counter}); retrying in {SCAN_RETRY_SECONDS} seconds...") + await asyncio.sleep(SCAN_RETRY_SECONDS) + + elif self.connection_stage == ConnectionStage.CONNECTING: + try: + await self.client.connect() + logger.info(f"{LOG_PREFIX} Connected, discovering services...") + self.connection_stage = ConnectionStage.SERVICE_DISCOVERY + except Exception as e: + logger.error(f"{LOG_PREFIX} Connection failed: {e}") + await self.disconnect() + + elif self.connection_stage == ConnectionStage.SERVICE_DISCOVERY: + services = self.client.services.services + if len(services) > 0: + logger.info(f"{LOG_PREFIX} Services discovered ({len(services)}), subscribing to battery...") + self.connection_stage = ConnectionStage.BATTERY_SUBSCRIBE + else: + logger.error(f"{LOG_PREFIX} Service discovery failed") + await self.disconnect() + + elif self.connection_stage == ConnectionStage.BATTERY_SUBSCRIBE: + if await self._subscribe_to_notifications(BATTERY_CHAR_UUID): + logger.info(f"{LOG_PREFIX} Battery subscribed, subscribing to status...") + self.connection_stage = ConnectionStage.STATUS_SUBSCRIBE + else: + logger.error(f"{LOG_PREFIX} Battery subscription failed") + await self.disconnect() + + elif self.connection_stage == ConnectionStage.STATUS_SUBSCRIBE: + if await self._subscribe_to_notifications(NOTIFY_CHAR_UUID): + logger.info(f"{LOG_PREFIX} Status subscribed, syncing parameters...") + self.connection_stage = ConnectionStage.SYNC_PARAMETERS + else: + logger.error(f"{LOG_PREFIX} Status subscription failed") + await self.disconnect() + + elif self.connection_stage == ConnectionStage.SYNC_PARAMETERS: + if await self._send_parameters(): + logger.info(f"{LOG_PREFIX} Parameters synced, connection complete") + # TODO: wait for ACK so we know device is ready + self.connection_stage = ConnectionStage.CONNECTED + else: + logger.error(f"{LOG_PREFIX} Parameter sync failed") + await self.disconnect() + + elif self.connection_stage == ConnectionStage.CONNECTED: + # Just maintain the connection + await asyncio.sleep(1) + + # Emit signal when connection status changes + if prev_stage != self.connection_stage: + is_connected = self.connection_stage == ConnectionStage.CONNECTED + self.connection_status_changed.emit(is_connected, self.connection_stage) + prev_stage = self.connection_stage + + except Exception as e: + logger.error(f"{LOG_PREFIX} Connection loop error: {e}") + # raise e + await self.disconnect() + + # Small delay between iterations + await asyncio.sleep(0.1) + + def start_updates(self, algorithm: Optional[any]): + logger.info(f"{LOG_PREFIX} start_updates called") + self.algorithm = algorithm + self.running = True + + future = None + if self._event_loop: + logger.info(f"{LOG_PREFIX} scheduling update_loop in event loop") + future = asyncio.run_coroutine_threadsafe(self.update_loop(), self._event_loop) + else: + logger.error(f"{LOG_PREFIX} No event loop present!") + + if future: + logger.info(f"{LOG_PREFIX} Future scheduled") + else: + logger.warning(f"{LOG_PREFIX} Update loop not scheduled") + + def stop_updates(self): + """Stop the update loop but maintain connection""" + logger.info(f"{LOG_PREFIX} Stopping updates") + self.running = False + self.algorithm = None + + async def _handle_battery_notification(self, sender, data: bytearray): + """Handle battery level notifications""" + battery_level = data[0] + + logger.info(f"{LOG_PREFIX} Battery level notification received: {battery_level}%") + + self.battery_level = battery_level + self.battery_level_changed.emit(battery_level) + + async def _handle_status_notification(self, sender, data: bytearray): + """Handle incoming status notifications from the device.""" + + if not data: + logger.warning(f"{LOG_PREFIX} Received empty status notification") + return + + # if len(data) != 4: + # logger.warning(f"Unexpected notification length: {len(data)} - {list(data)}") + # return + + command_id = data[0] + sequence_number = data[1] + power_a = data[2] + power_b = data[3] + + if command_id == CMD_POWER_UPDATE: + logger.info(f"{LOG_PREFIX} Power level update (seq={sequence_number}) - Channel A: {power_a}, Channel B: {power_b}") + self.strengths.channel_a = power_a + self.strengths.channel_b = power_b + self.power_levels_changed.emit(self.strengths) + + elif command_id == CMD_ACK: + logger.debug(f"{LOG_PREFIX} Command acknowledged (seq={sequence_number})") + + elif command_id == CMD_ACTIVE_POWER: + if len(data) < 4: + logger.warning(f"{LOG_PREFIX} Malformed active power notification: {list(data)}") + return + + power_a = data[2] + power_b = data[3] + + logger.info(f"{LOG_PREFIX} Active power update - Channel A: {power_a}, Channel B: {power_b}") + + # self.strengths.channel_a = power_a + # self.strengths.channel_b = power_b + # self.power_levels_changed.emit(self.strengths) + + # if len(data) > 4: + # extra = data[4:] + # logger.warning(f"Extra fields in 0x53 notification (undocumented): {list(extra)}") + + else: + logger.warning(f"{LOG_PREFIX} Unknown notification type: 0x{command_id:02X} (seq={sequence_number})") + logger.warning(f"{LOG_PREFIX} Raw notification: {list(data)}") + + async def _send_parameters(self): + """Send device parameters""" + logger.info( + f"{LOG_PREFIX} Syncing parameters - " + f"Limits: A={self.parameters.channel_a_limit}, B={self.parameters.channel_b_limit}, " + f"Freq Balance: A={self.parameters.channel_a_freq_balance}, B={self.parameters.channel_b_freq_balance}, " + f"Intensity Balance: A={self.parameters.channel_a_intensity_balance}, B={self.parameters.channel_b_intensity_balance}" + ) + + command = bytes([ + 0xBF, # Does this command produce an ACK? Only if the seq nibble is > 0 + self.parameters.channel_a_limit, + self.parameters.channel_b_limit, + self.parameters.channel_a_freq_balance, + self.parameters.channel_b_freq_balance, + self.parameters.channel_a_intensity_balance, + self.parameters.channel_b_intensity_balance + ]) + + try: + await self.client.write_gatt_char(WRITE_CHAR_UUID, command) + return True + except Exception as e: + logger.error(f"{LOG_PREFIX} Failed to sync parameters: {e}") + return False + + async def _subscribe_to_notifications(self, char_uuid: str) -> bool: + """Subscribe to notifications for a characteristic""" + try: + char = self.client.services.get_characteristic(char_uuid) + if not char: + logger.error(f"{LOG_PREFIX} Characteristic {char_uuid} not found") + return False + + await self.client.start_notify(char_uuid, + self._handle_battery_notification if char_uuid == BATTERY_CHAR_UUID + else self._handle_status_notification) + return True + except Exception as e: + logger.error(f"{LOG_PREFIX} Failed to subscribe to {char_uuid}: {e}") + return False + + async def _scan_for_device(self): + """Scan for Coyote device""" + try: + logger.info(f"{LOG_PREFIX} Scanning for device: {self.device_name}") + device = await BleakScanner.find_device_by_name(self.device_name) + if device: + logger.info(f"{LOG_PREFIX} Found device: {device.name} ({device.address})") + self.client = BleakClient(device) + self.connection_stage = ConnectionStage.CONNECTING + return True + else: + logger.info(f"{LOG_PREFIX} No BLE advertisement for {self.device_name} detected during scan window") + return False + except Exception as e: + logger.error(f"{LOG_PREFIX} Scan error: {e}") + await self.disconnect() + return False + + async def send_command(self, + strengths: Optional[CoyoteStrengths] = None, + pulses: Optional[CoyotePulses] = None): + """ + Send strength update and/or pulse pattern command to device. + + Args: + strengths: Optional strength update for channels A and B + pulses: Optional pulse patterns for channels A and B + """ + + if pulses: + self.pulse_sent.emit(pulses) + + if not self.client or not self.client.is_connected: + # logger.warning("Attempted to send command while disconnected") + + # Optimistic update for offline testing + if strengths: + self.strengths.channel_a = strengths.channel_a + self.strengths.channel_b = strengths.channel_b + + return + + if not strengths and not pulses: + logger.warning(f"{LOG_PREFIX} send_command called with no data") + return + + # Determine strength interpretation (default absolute set if new strength provided) + if strengths: + interp_a = INTERP_ABSOLUTE_SET # Absolute set for Channel A + interp_b = INTERP_ABSOLUTE_SET # Absolute set for Channel B + else: + interp_a = INTERP_NO_CHANGE # No change + interp_b = INTERP_NO_CHANGE # No change + + # Pack sequence number + interpretation into 1 byte (upper 4 = seq, lower 4 = interp) + request_ack = not pulses + control_byte = ((self.sequence_number if request_ack else 0) << 4) | (interp_a << 2) | interp_b + + # Build base command (B0 packet structure) + command = bytearray([ + CMD_B0, # Command ID + control_byte, # Combined seq + interpretation + strengths.channel_a if strengths else 0, + strengths.channel_b if strengths else 0, + ]) + + # Append pulse data if provided (waveform duration (aka frequency) + intensity) + if pulses: + command.extend([a.duration for a in pulses.channel_a]) + command.extend([a.intensity for a in pulses.channel_a]) + command.extend([b.duration for b in pulses.channel_b]) + command.extend([b.intensity for b in pulses.channel_b]) + else: + command.extend([0] * B0_NO_PULSES_PAD_BYTES) # No pulses = zero padding + + # Log what we're sending + if logger.isEnabledFor(logging.DEBUG): + logger.debug(f"{LOG_PREFIX} Sending command (seq={self.sequence_number}):") + + if pulses: + pulses_a = "\n".join( + f" Pulse {i+1}: Freq={pulse.frequency} Hz, Intensity={pulse.intensity}" + for i, pulse in enumerate(pulses.channel_a) + ) + pulses_b = "\n".join( + f" Pulse {i+1}: Freq={pulse.frequency} Hz, Intensity={pulse.intensity}" + for i, pulse in enumerate(pulses.channel_b) + ) + + logger.debug( + f"{LOG_PREFIX} Channel A ({self.strengths.channel_a}):\n{pulses_a}\n" + f"{LOG_PREFIX} Channel B ({self.strengths.channel_b}):\n{pulses_b}" + ) + + # Send the final command + try: + await self.client.write_gatt_char(WRITE_CHAR_UUID, command) + self.sequence_number = (self.sequence_number + 1) % SEQUENCE_MODULO # Wrap seq at 4 bits (0-15) + except Exception as e: + logger.error(f"{LOG_PREFIX} Failed to send command: {e}") + + async def disconnect(self): + """Disconnect from device""" + logger.info(f"{LOG_PREFIX} Disconnecting from device") + + if self.client: + self.running = False + + # Send zero pulses to turn off outputs + zero_pulses = CoyotePulses( + channel_a=[CoyotePulse(frequency=0, intensity=0, duration=0)] * PULSES_PER_PACKET, + channel_b=[CoyotePulse(frequency=0, intensity=0, duration=0)] * PULSES_PER_PACKET + ) + await self.send_command(pulses=zero_pulses) + await self.client.disconnect() + self.client = None + self.connection_stage = ConnectionStage.DISCONNECTED + + async def update_loop(self): + logger.info(f"{LOG_PREFIX} Starting update loop, running={self.running}, algorithm={self.algorithm}") + + try: + logger.info(f"{LOG_PREFIX} Update loop started, running={self.running}") + + while self.running: + try: + if not self.algorithm: + logger.warning(f"{LOG_PREFIX} Algorithm not yet set") + await asyncio.sleep(0.1) + continue + + current_time = time.time() + # Only log when a packet is actually generated and sent + if current_time >= self.algorithm.next_update_time: + pulses = self.algorithm.generate_packet(current_time) + if pulses is not None: + await self.send_command(pulses=pulses) + # Check if algorithm still exists after generate_packet() + if self.algorithm: + sleep_time = max(0.001, self.algorithm.next_update_time - time.time()) + else: + sleep_time = 0.01 + else: + sleep_time = 0.01 + + await asyncio.sleep(sleep_time) + + except Exception as inner_e: + logger.exception(f"{LOG_PREFIX} Exception inside update loop iteration: {inner_e}") + await asyncio.sleep(0.1) # prevent tight-crash-loop + + except Exception as outer_e: + logger.exception(f"{LOG_PREFIX} Fatal exception in update_loop: {outer_e}") + + finally: + logger.info(f"{LOG_PREFIX} Update loop stopped") + + def is_connected_and_running(self) -> bool: + return (self.connection_stage == ConnectionStage.CONNECTED and + self.client and self.client.is_connected) diff --git a/device/coyote/pulse_generator.py b/device/coyote/pulse_generator.py new file mode 100644 index 0000000..1de5eb7 --- /dev/null +++ b/device/coyote/pulse_generator.py @@ -0,0 +1,205 @@ +from __future__ import annotations + +import logging +import math +import random +from dataclasses import dataclass +from typing import Tuple + +from device.coyote.common import clamp, normalize +from device.coyote.config import PulseTuning +from device.coyote.constants import ( + HARDWARE_MAX_FREQ_HZ, + HARDWARE_MIN_FREQ_HZ, + MAX_PULSE_DURATION_MS, + MIN_PULSE_DURATION_MS, +) +from device.coyote.types import CoyotePulse +from stim_math.audio_gen.params import CoyoteAlgorithmParams, CoyoteChannelParams + +logger = logging.getLogger("restim.coyote") + + +@dataclass +class TextureInfo: + offset_ms: float + mode: str + headroom_up_ms: float + headroom_down_ms: float + + +@dataclass +class PulseDebug: + sequence_index: int + raw_frequency_hz: float + normalised_frequency: float + mapped_frequency_hz: float + frequency_limits: Tuple[float, float] + base_duration_ms: float + duration_limits: Tuple[int, int] + jitter_fraction: float + jitter_factor: float + width_normalised: float + texture_mode: str + texture_headroom_up_ms: float + texture_headroom_down_ms: float + texture_applied_ms: float + desired_duration_ms: float + residual_ms: float + + +class PulseGenerator: + """Builds hardware-friendly pulses for a single Coyote channel.""" + + def __init__( + self, + name: str, + params: CoyoteAlgorithmParams, + channel_params: CoyoteChannelParams, + carrier_freq_limits: Tuple[float, float], + pulse_freq_limits: Tuple[float, float], + pulse_width_limits: Tuple[float, float], + tuning: PulseTuning, + ) -> None: + self.name = name + self.params = params + self.channel_params = channel_params + self._carrier_limits = carrier_freq_limits + self._pulse_freq_limits = pulse_freq_limits + self._pulse_width_limits = pulse_width_limits + self._tuning = tuning + + self._phase = 0.0 + self._residual_ms = 0.0 + + @property + def carrier_limits(self) -> Tuple[float, float]: + return self._carrier_limits + + def advance_phase(self, texture_speed_hz: float, delta_time_s: float) -> None: + if delta_time_s <= 0 or texture_speed_hz <= 0: + return + phase_delta = delta_time_s * texture_speed_hz * 2 * math.pi + self._phase = (self._phase + phase_delta) % (2 * math.pi) + + def create_pulse(self, time_s: float, intensity: int, sequence_index: int) -> Tuple[CoyotePulse, PulseDebug]: + min_freq, max_freq = self._channel_frequency_window() + duration_limits = self._duration_limits(min_freq, max_freq) + + raw_frequency = float(self.params.pulse_frequency.interpolate(time_s)) + normalised = normalize(raw_frequency, self._pulse_freq_limits) + mapped_frequency = min_freq + (max_freq - min_freq) * normalised + if mapped_frequency <= 0: + mapped_frequency = 1000.0 / duration_limits[1] + base_duration = 1000.0 / mapped_frequency + + jitter_fraction = clamp( + float(self.params.pulse_interval_random.interpolate(time_s)), + 0.0, + self._tuning.jitter_limit_fraction, + ) + jitter_factor = 1.0 + random.uniform(-jitter_fraction, jitter_fraction) + + width_normalised = self._pulse_width_normalised(time_s) + texture_info = self._texture_offset(base_duration, width_normalised, min_freq, max_freq) + + desired_ms = base_duration * jitter_factor + texture_info.offset_ms + duration, residual = self._apply_residual(desired_ms) + duration, clamped = self._clamp_duration(duration, duration_limits) + if clamped: + residual = 0.0 + + final_duration = max(MIN_PULSE_DURATION_MS, duration) + final_frequency = int(max(1, round(1000.0 / final_duration))) + final_intensity = int(clamp(intensity, 0, 100)) + + debug = PulseDebug( + sequence_index=sequence_index, + raw_frequency_hz=raw_frequency, + normalised_frequency=normalised, + mapped_frequency_hz=mapped_frequency, + frequency_limits=(min_freq, max_freq), + base_duration_ms=base_duration, + duration_limits=duration_limits, + jitter_fraction=jitter_fraction, + jitter_factor=jitter_factor, + width_normalised=width_normalised, + texture_mode=texture_info.mode, + texture_headroom_up_ms=texture_info.headroom_up_ms, + texture_headroom_down_ms=texture_info.headroom_down_ms, + texture_applied_ms=texture_info.offset_ms, + desired_duration_ms=desired_ms, + residual_ms=residual, + ) + + return CoyotePulse(duration=final_duration, intensity=final_intensity, frequency=final_frequency), debug + + def _channel_frequency_window(self) -> Tuple[float, float]: + minimum = max(float(self.channel_params.minimum_frequency.get()), HARDWARE_MIN_FREQ_HZ) + maximum = min(float(self.channel_params.maximum_frequency.get()), HARDWARE_MAX_FREQ_HZ) + if minimum >= maximum: + return HARDWARE_MIN_FREQ_HZ, HARDWARE_MAX_FREQ_HZ + return minimum, maximum + + def _pulse_width_normalised(self, time_s: float) -> float: + raw = float(self.params.pulse_width.interpolate(time_s)) + low, high = self._pulse_width_limits + if high <= low: + return 0.0 + return clamp((raw - low) / (high - low), 0.0, 1.0) + + def _texture_offset( + self, + base_duration: float, + width_norm: float, + min_freq: float, + max_freq: float, + ) -> TextureInfo: + if width_norm <= 0 or self._tuning.texture_depth_fraction <= 0: + return TextureInfo(offset_ms=0.0, mode="none", headroom_up_ms=0.0, headroom_down_ms=0.0) + + min_duration = 1000.0 / max_freq + max_duration = 1000.0 / min_freq + + up_headroom = max(0.0, max_duration - base_duration) * self._tuning.texture_depth_fraction * width_norm + down_headroom = max(0.0, base_duration - min_duration) * self._tuning.texture_depth_fraction * width_norm + + if up_headroom > 1e-6 and down_headroom > 1e-6: + amplitude = min(up_headroom, down_headroom) + offset = amplitude * math.sin(self._phase) + return TextureInfo(offset_ms=offset, mode="sym", headroom_up_ms=up_headroom, headroom_down_ms=down_headroom) + + sine = math.sin(self._phase) + rectified = abs(sine) - 2.0 / math.pi + + if up_headroom > 1e-6: + offset = up_headroom * rectified + return TextureInfo(offset_ms=offset, mode="up", headroom_up_ms=up_headroom, headroom_down_ms=down_headroom) + if down_headroom > 1e-6: + offset = -down_headroom * rectified + return TextureInfo(offset_ms=offset, mode="down", headroom_up_ms=up_headroom, headroom_down_ms=down_headroom) + return TextureInfo(offset_ms=0.0, mode="none", headroom_up_ms=up_headroom, headroom_down_ms=down_headroom) + + def _apply_residual(self, desired_ms: float) -> Tuple[int, float]: + accum = desired_ms + self._residual_ms + rounded = int(round(accum)) + residual = accum - rounded + bound = self._tuning.residual_bound + residual = clamp(residual, -bound, bound) + self._residual_ms = residual + return max(1, rounded), residual + + def _duration_limits(self, min_freq: float, max_freq: float) -> Tuple[int, int]: + minimum = max(MIN_PULSE_DURATION_MS, int(round(1000.0 / max_freq))) + maximum = min(MAX_PULSE_DURATION_MS, int(round(1000.0 / min_freq))) + if minimum > maximum: + return MIN_PULSE_DURATION_MS, MAX_PULSE_DURATION_MS + return minimum, maximum + + def _clamp_duration(self, duration_ms: int, limits: Tuple[int, int]) -> Tuple[int, bool]: + low, high = limits + clamped_duration = int(clamp(duration_ms, low, high)) + clamped = clamped_duration != duration_ms + if clamped: + self._residual_ms = 0.0 + return clamped_duration, clamped diff --git a/device/coyote/types.py b/device/coyote/types.py new file mode 100644 index 0000000..bdbc2dd --- /dev/null +++ b/device/coyote/types.py @@ -0,0 +1,54 @@ +from dataclasses import dataclass + + +class ConnectionStage: + DISCONNECTED = "Disconnected" + SCANNING = "Scanning for device..." + CONNECTING = "Connecting..." + SERVICE_DISCOVERY = "Discovering services..." + BATTERY_SUBSCRIBE = "Setting up battery notifications..." + STATUS_SUBSCRIBE = "Setting up status notifications..." + SYNC_PARAMETERS = "Syncing parameters..." + CONNECTED = "Connected" + + +@dataclass +class CoyoteParams: + """ + Represents configurable parameters for the Coyote device + channel_a_limit: 0-200 power limit for channel A + channel_b_limit: 0-200 power limit for channel B + channel_a_freq_balance: 0-255 frequency balance for channel A + channel_b_freq_balance: 0-255 frequency balance for channel B + channel_a_intensity_balance: 0-255 intensity balance for channel A + channel_b_intensity_balance: 0-255 intensity balance for channel B + """ + channel_a_limit: int + channel_b_limit: int + channel_a_freq_balance: int + channel_b_freq_balance: int + channel_a_intensity_balance: int + channel_b_intensity_balance: int + + +@dataclass +class CoyotePulse: + frequency: int # Calculated from duration: 1000/duration_ms, range ~4-200 Hz + intensity: int # 0-100 + duration: int # 5-240ms (spec says 10-240, but 5ms works) + + +@dataclass +class CoyotePulses: + channel_a: list[CoyotePulse] # Exactly 4 pulses + channel_b: list[CoyotePulse] # Exactly 4 pulses + + def duration() -> int: + return 0 + + +@dataclass +class CoyoteStrengths: + """Represents channel strength (volume) settings""" + channel_a: int # 0-100 + channel_b: int # 0-100 diff --git a/qt_ui/algorithm_factory.py b/qt_ui/algorithm_factory.py index 33dd584..cbf8d15 100644 --- a/qt_ui/algorithm_factory.py +++ b/qt_ui/algorithm_factory.py @@ -3,6 +3,7 @@ from device.focstim.fourphase_algorithm import FOCStimFourphaseAlgorithm from device.neostim.algorithm import NeoStimAlgorithm +from device.coyote.algorithm import CoyoteAlgorithm from qt_ui.device_wizard.enums import DeviceConfiguration, DeviceType, WaveformType from stim_math.audio_gen.base_classes import AudioGenerationAlgorithm from device.focstim.threephase_algorithm import FOCStimThreephaseAlgorithm @@ -34,7 +35,7 @@ def __init__(self, mainwindow, self.load_funscripts = load_funscripts self.create_for_bake = create_for_bake - def create_algorithm(self, device: DeviceConfiguration) -> AudioGenerationAlgorithm | NeoStimAlgorithm: + def create_algorithm(self, device: DeviceConfiguration) -> AudioGenerationAlgorithm | NeoStimAlgorithm | CoyoteAlgorithm: if device.device_type == DeviceType.AUDIO_THREE_PHASE: if device.waveform_type == WaveformType.CONTINUOUS: return self.create_3phase_continuous(device) @@ -50,6 +51,8 @@ def create_algorithm(self, device: DeviceConfiguration) -> AudioGenerationAlgori return self.create_focstim_4phase_pulsebased(device) elif device.device_type == DeviceType.NEOSTIM_THREE_PHASE: return self.create_neostim(device) + elif device.device_type == DeviceType.COYOTE_THREE_PHASE: + return self.create_coyote(device) else: raise RuntimeError('unknown device type') @@ -241,6 +244,59 @@ def create_neostim(self, device: DeviceConfiguration) -> NeoStimAlgorithm: ), ) return algorithm + + def create_coyote(self, device: DeviceConfiguration) -> AudioGenerationAlgorithm: + # Get frequency limits from kit + carrier_freq_limits = self.kit.limits_for_axis(AxisEnum.CARRIER_FREQUENCY) + pulse_freq_limits = self.kit.limits_for_axis(AxisEnum.PULSE_FREQUENCY) + pulse_width_limits = self.kit.limits_for_axis(AxisEnum.PULSE_WIDTH) + pulse_rise_time_limits = self.kit.limits_for_axis(AxisEnum.PULSE_RISE_TIME) + + # Create the algorithm + algorithm = CoyoteAlgorithm( + self.media_sync, + CoyoteAlgorithmParams( + position=ThreephasePositionParams( + self.get_axis_alpha(), + self.get_axis_beta(), + ), + transform=self.mainwindow.tab_threephase.transform_params, + calibrate=self.mainwindow.tab_threephase.calibrate_params, + volume=VolumeParams( + api=self.get_axis_volume_api(), + master=self.get_axis_volume_master(), + inactivity=self.get_axis_volume_inactivity(), + external=self.get_axis_volume_external(), + ), + carrier_frequency=self.get_axis_pulse_carrier_frequency(), + pulse_frequency=self.get_axis_pulse_frequency(), + pulse_width=self.get_axis_pulse_width(), + pulse_interval_random=self.get_axis_pulse_interval_random(), + pulse_rise_time=self.get_axis_pulse_rise_time(), + max_intensity_change_per_pulse=settings.coyote_max_intensity_change_per_pulse, + channel_a=CoyoteChannelParams( + minimum_frequency=settings.coyote_channel_a_freq_min, + maximum_frequency=settings.coyote_channel_a_freq_max, + maximum_strength=settings.coyote_channel_a_strength_max, + vibration=self.get_axis_vib1_all() + ), + channel_b=CoyoteChannelParams( + minimum_frequency=settings.coyote_channel_b_freq_min, + maximum_frequency=settings.coyote_channel_b_freq_max, + maximum_strength=settings.coyote_channel_b_strength_max, + vibration=self.get_axis_vib2_all() + ) + ), + safety_limits=SafetyParams( + device.min_frequency, + device.max_frequency, + ), + carrier_freq_limits=carrier_freq_limits, + pulse_freq_limits=pulse_freq_limits, + pulse_width_limits=pulse_width_limits, + pulse_rise_time_limits=pulse_rise_time_limits + ) + return algorithm def get_axis_alpha(self): return self.get_axis_from_script_mapping(AxisEnum.POSITION_ALPHA) or self.mainwindow.alpha @@ -401,13 +457,13 @@ def get_axis_neostim_switch_time(self): def get_axis_neostim_debug(self): return self.mainwindow.tab_neostim.axis_debug - def get_axis_from_script_mapping(self, axis: AxisEnum) -> AbstractAxis | None: + def get_axis_from_script_mapping(self, axis: AxisEnum, limits: Optional[(int, int)] = None) -> AbstractAxis | None: if not self.load_funscripts: return None funscript_item = self.script_mapping.get_config_for_axis(axis) if funscript_item: - limit_min, limit_max = self.kit.limits_for_axis(axis) + limit_min, limit_max = limits or self.kit.limits_for_axis(axis) # TODO: not very memory efficient if multiple algorithms reference the same script. # but worst-case it only wastes a few MB or so... return create_precomputed_axis(funscript_item.script.x, diff --git a/qt_ui/coyote_settings_widget.py b/qt_ui/coyote_settings_widget.py new file mode 100644 index 0000000..c166652 --- /dev/null +++ b/qt_ui/coyote_settings_widget.py @@ -0,0 +1,699 @@ +import asyncio +import logging +import time +from dataclasses import dataclass +from typing import Dict, Optional +from PySide6 import QtWidgets +from PySide6.QtWidgets import (QWidget, QVBoxLayout, QLabel, QSlider, QHBoxLayout, + QGraphicsView, QGraphicsScene, QGraphicsLineItem, QSpinBox, + QGraphicsRectItem, QToolTip, QGraphicsEllipseItem) +from PySide6.QtCore import QSettings, Qt, QTimer +from PySide6.QtGui import QPen, QColor, QBrush, QPainterPath +from device.coyote.device import CoyoteDevice, CoyotePulse, CoyotePulses, CoyoteStrengths +from qt_ui import settings + +class CoyoteSettingsWidget(QtWidgets.QWidget): + def __init__(self, parent=None): + super().__init__(parent) + self.device: Optional[CoyoteDevice] = None + self.channel_controls: Dict[str, ChannelControl] = {} + self.coyote_logger = logging.getLogger('restim.coyote') + self._base_log_level = self.coyote_logger.getEffectiveLevel() + self.graph_window = settings.coyote_graph_window + self.setupUi() + self.apply_debug_logging(settings.coyote_debug_logging.get()) + + def setupUi(self): + self.setLayout(QVBoxLayout()) + + self.label_connection_status = QLabel("Device: Disconnected") + self.label_connection_stage = QLabel("Stage: Waiting") + self.label_battery_level = QLabel("Battery: —") + status_layout = QHBoxLayout() + status_layout.addWidget(self.label_connection_status) + status_layout.addWidget(self.label_connection_stage) + status_layout.addWidget(self.label_battery_level) + self.layout().addLayout(status_layout) + + configs = ( + ChannelConfig( + channel_id='A', + freq_min_setting=settings.coyote_channel_a_freq_min, + freq_max_setting=settings.coyote_channel_a_freq_max, + strength_max_setting=settings.coyote_channel_a_strength_max, + ), + ChannelConfig( + channel_id='B', + freq_min_setting=settings.coyote_channel_b_freq_min, + freq_max_setting=settings.coyote_channel_b_freq_max, + strength_max_setting=settings.coyote_channel_b_strength_max, + ), + ) + + for config in configs: + control = ChannelControl(self, config) + self.channel_controls[config.channel_id] = control + self.layout().addLayout(control.build_ui()) + control.reset_volume() + + def setup_device(self, device: CoyoteDevice): + self.device = device + + self.device.connection_status_changed.connect(self.on_connection_status_changed) + self.device.battery_level_changed.connect(self.on_battery_level_changed) + self.device.parameters_changed.connect(self.on_parameters_changed) + self.device.power_levels_changed.connect(self.on_power_levels_changed) + self.device.pulse_sent.connect(self.on_pulse_sent) + + for control in self.channel_controls.values(): + control.reset_volume() + + if device.strengths: + for control in self.channel_controls.values(): + control.update_from_device(device.strengths) + + def update_channel_strength(self, control: 'ChannelControl', value: int): + if not self.device or not self.device._event_loop: + return + + strengths = control.with_strength(self.device.strengths, value) + + asyncio.run_coroutine_threadsafe( + self.device.send_command(strengths), + self.device._event_loop + ) + + self.device.strengths = strengths + + def on_connection_status_changed(self, connected: bool, stage: str = None): + self.label_connection_status.setText("Device: Connected" if connected else "Device: Disconnected") + if stage: + normalized_stage = stage.strip() + if connected and normalized_stage.lower() == "connected": + stage_text = "Ready" + else: + stage_text = normalized_stage + self.label_connection_stage.setText(f"Stage: {stage_text}") + else: + self.label_connection_stage.setText("Stage: —") + + def on_battery_level_changed(self, level: int): + self.label_battery_level.setText(f"Battery: {level}%") + + def on_parameters_changed(self): + pass + + def on_power_levels_changed(self, strengths: CoyoteStrengths): + for control in self.channel_controls.values(): + control.update_from_device(strengths) + + def on_pulse_sent(self, pulses: CoyotePulses): + if not self.device: + return + + for control in self.channel_controls.values(): + control.apply_pulses(pulses, self.device.strengths) + + def apply_debug_logging(self, enabled: bool): + new_level = logging.DEBUG if enabled else logging.INFO + self.coyote_logger.setLevel(new_level) + +@dataclass(frozen=True) +class ChannelConfig: + channel_id: str + freq_min_setting: settings.Setting + freq_max_setting: settings.Setting + strength_max_setting: settings.Setting + +class ChannelControl: + def __init__(self, parent: 'CoyoteSettingsWidget', config: ChannelConfig): + self.parent = parent + self.config = config + + self.freq_min: Optional[QSpinBox] = None + self.freq_max: Optional[QSpinBox] = None + self.strength_max: Optional[QSpinBox] = None + self.volume_slider: Optional[QSlider] = None + self.volume_label: Optional[QLabel] = None + self.pulse_graph: Optional[PulseGraphContainer] = None + self.stats_label: Optional[QLabel] = None + + @property + def channel_id(self) -> str: + return self.config.channel_id + + @property + def _is_channel_a(self) -> bool: + return self.channel_id.upper() == 'A' + + def build_ui(self) -> QHBoxLayout: + layout = QHBoxLayout() + + left = QVBoxLayout() + left.addWidget(QLabel(f"Channel {self.channel_id}")) + + freq_min_layout = QHBoxLayout() + self.freq_min = QSpinBox() + self.freq_min.setRange(10, 500) + self.freq_min.setSingleStep(10) + self.freq_min.setValue(self.config.freq_min_setting.get()) + self.freq_min.valueChanged.connect(self.on_freq_min_changed) + freq_min_layout.addWidget(QLabel("Min Freq (Hz)")) + freq_min_layout.addWidget(self.freq_min) + left.addLayout(freq_min_layout) + + freq_max_layout = QHBoxLayout() + self.freq_max = QSpinBox() + self.freq_max.setRange(10, 500) + self.freq_max.setSingleStep(10) + self.freq_max.setValue(self.config.freq_max_setting.get()) + self.freq_max.valueChanged.connect(self.on_freq_max_changed) + freq_max_layout.addWidget(QLabel("Max Freq (Hz)")) + freq_max_layout.addWidget(self.freq_max) + left.addLayout(freq_max_layout) + + strength_layout = QHBoxLayout() + strength_layout.addWidget(QLabel("Max Strength")) + self.strength_max = QSpinBox() + self.strength_max.setRange(1, 200) + self.strength_max.setSingleStep(1) + self.strength_max.setValue(self.config.strength_max_setting.get()) + self.strength_max.valueChanged.connect(self.on_strength_max_changed) + strength_layout.addWidget(self.strength_max) + left.addLayout(strength_layout) + + layout.addLayout(left) + + self.pulse_graph = PulseGraphContainer(self.parent.graph_window, self.freq_min, self.freq_max) + self.pulse_graph.plot.setMinimumHeight(100) + + graph_column = QVBoxLayout() + graph_column.addWidget(self.pulse_graph) + + self.stats_label = QLabel("Intensity: 0%\nFrequency: 0 Hz") + self.stats_label.setAlignment(Qt.AlignHCenter) + self.pulse_graph.attach_stats_label(self.stats_label) + graph_column.addWidget(self.stats_label) + + layout.addLayout(graph_column) + + volume_layout = QVBoxLayout() + self.volume_slider = QSlider(Qt.Vertical) + self.volume_slider.setRange(0, self.config.strength_max_setting.get()) + self.volume_slider.valueChanged.connect(self.on_volume_changed) + self.volume_label = QLabel() + self.volume_label.setAlignment(Qt.AlignHCenter) + volume_layout.addWidget(self.volume_slider) + volume_layout.addWidget(self.volume_label) + layout.addLayout(volume_layout) + + self.update_volume_label(self.volume_slider.value()) + return layout + + def reset_volume(self): + self.set_strength_from_device(0) + + def select_strength(self, strengths: CoyoteStrengths) -> int: + return strengths.channel_a if self._is_channel_a else strengths.channel_b + + def with_strength(self, strengths: CoyoteStrengths, value: int) -> CoyoteStrengths: + if self._is_channel_a: + return CoyoteStrengths(channel_a=value, channel_b=strengths.channel_b) + return CoyoteStrengths(channel_a=strengths.channel_a, channel_b=value) + + def extract_pulses(self, pulses: CoyotePulses) -> list[CoyotePulse]: + return pulses.channel_a if self._is_channel_a else pulses.channel_b + + def update_from_device(self, strengths: CoyoteStrengths): + self.set_strength_from_device(self.select_strength(strengths)) + + def apply_pulses(self, pulses: CoyotePulses, strengths: CoyoteStrengths): + channel_pulses = self.extract_pulses(pulses) + if not channel_pulses: + return + self.handle_pulses(channel_pulses, self.select_strength(strengths)) + + def on_volume_changed(self, value: int): + self.update_volume_label(value) + self.parent.update_channel_strength(self, value) + + def update_volume_label(self, value: int): + max_strength = max(1, self.config.strength_max_setting.get()) + percentage = int((value / max_strength) * 100) + self.volume_label.setText(f"{value} ({percentage}%)") + + def set_strength_from_device(self, value: int): + if self.volume_slider is None: + return + self.volume_slider.blockSignals(True) + self.volume_slider.setValue(value) + self.volume_slider.blockSignals(False) + self.update_volume_label(value) + + def on_strength_max_changed(self, value: int): + self.config.strength_max_setting.set(value) + + current_value = self.volume_slider.value() if self.volume_slider else 0 + if self.volume_slider: + self.volume_slider.blockSignals(True) + self.volume_slider.setRange(0, value) + clamped_value = min(current_value, value) + self.volume_slider.setValue(clamped_value) + self.volume_slider.blockSignals(False) + self.update_volume_label(clamped_value) + current_value = clamped_value + + self.parent.update_channel_strength(self, current_value) + + def on_freq_min_changed(self, value: int): + if self.freq_min is None or self.freq_max is None: + return + + corrected = value + if value >= self.freq_max.value(): + corrected = max(self.freq_max.value() - self.freq_min.singleStep(), self.freq_min.minimum()) + if corrected != value: + self.freq_min.blockSignals(True) + self.freq_min.setValue(corrected) + self.freq_min.blockSignals(False) + self.config.freq_min_setting.set(corrected) + + def on_freq_max_changed(self, value: int): + if self.freq_min is None or self.freq_max is None: + return + + corrected = value + if value <= self.freq_min.value(): + corrected = min(self.freq_min.value() + self.freq_max.singleStep(), self.freq_max.maximum()) + if corrected != value: + self.freq_max.blockSignals(True) + self.freq_max.setValue(corrected) + self.freq_max.blockSignals(False) + self.config.freq_max_setting.set(corrected) + + def handle_pulses(self, pulses: list[CoyotePulse], strength: int): + if not self.pulse_graph or not pulses: + return + + channel_limit = self.config.strength_max_setting.get() + for pulse in pulses: + self.pulse_graph.add_pulse( + frequency=pulse.frequency, + intensity=pulse.intensity, + duration=pulse.duration, + current_strength=strength, + channel_limit=channel_limit, + ) + +class PulseGraphContainer(QWidget): + def __init__(self, window_seconds: settings.Setting, freq_min: QSpinBox, freq_max: QSpinBox, *args, **kwargs): + super().__init__(*args, **kwargs) + # Store frequency range controls + self.freq_min = freq_min + self.freq_max = freq_max + + # Initialize entries list to store CoyotePulse objects + self.entries = [] + + # Time window for stats display (in seconds) + self.stats_window = window_seconds + + # Create layout + self.layout = QVBoxLayout(self) + + # Create plot widget + self.plot = PulseGraph(window_seconds, *args, **kwargs) + self.layout.addWidget(self.plot) + + # Optional stats label managed by parent component + self.stats_label: Optional[QLabel] = None + + def attach_stats_label(self, label: QLabel): + self.stats_label = label + self.stats_label.setText("Intensity: 0%\nFrequency: 0 Hz") + + def get_frequency_range_text(self, entries) -> str: + """Get the frequency range text from the given entries.""" + if not entries: + return "N/A" + frequencies = [entry.frequency for entry in entries] + avg_frequency = sum(frequencies) / len(frequencies) + min_freq = min(frequencies) + max_freq = max(frequencies) + + # If min, max, and average are all the same, just show the single value + if min_freq == max_freq == round(avg_frequency): + return f"{int(avg_frequency)} Hz" + # If min and max differ, show average with range + return f"{avg_frequency:.0f} Hz ({min_freq} – {max_freq})" + + def format_intensity_text(self, intensities) -> str: + """Format intensity text with smart range display.""" + if not intensities: + return "N/A" + avg_intensity = sum(intensities) / len(intensities) + min_intensity = min(intensities) + max_intensity = max(intensities) + + # If min, max, and average are all the same, just show the single value + if min_intensity == max_intensity == round(avg_intensity): + return f"{int(avg_intensity)}%" + # If min and max differ, show average with range + return f"{avg_intensity:.0f}% ({min_intensity} – {max_intensity})" + + def clean_old_entries(self): + """Remove entries outside the time window""" + current_time = time.time() + stats_window = self.stats_window.get() + self.entries = [e for e in self.entries if current_time - e.timestamp <= stats_window] + + def update_label_text(self): + # Clean up old entries + self.clean_old_entries() + + # Calculate stats using pulses from the time window + recent_entries = self.entries + + # Get frequency range text + freq_text = self.get_frequency_range_text(recent_entries) + + # Get intensity range + intensities = [entry.intensity for entry in recent_entries] + intensity_text = self.format_intensity_text(intensities) + + if self.stats_label: + self.stats_label.setText(f"Intensity: {intensity_text}\nFrequency: {freq_text}") + + def add_pulse(self, frequency, intensity, duration, current_strength, channel_limit): + # Calculate effective intensity after applying current strength + effective_intensity = intensity * (current_strength / 100) + + # For zero intensity pulses, still create them but with zero intensity + # This shows empty space in the graph + + # Create a CoyotePulse object + pulse = CoyotePulse( + frequency=frequency, + intensity=intensity, + duration=duration + ) + + # Add timestamp for time-window filtering + pulse.timestamp = time.time() + + # Store pulse data + self.entries.append(pulse) + + self.update_label_text() + + # Update the plot - even zero intensity pulses are sent through for visualization + self.plot.add_pulse(pulse, effective_intensity, channel_limit) + +class PulseGraph(QWidget): + def __init__(self, window_seconds: settings.Setting, parent=None): + super().__init__(parent) + self.setLayout(QVBoxLayout()) + + self.view = QGraphicsView() + self.scene = QGraphicsScene() + self.view.setScene(self.scene) + + # Completely disable scrolling and user interaction + self.view.setHorizontalScrollBarPolicy(Qt.ScrollBarAlwaysOff) + self.view.setVerticalScrollBarPolicy(Qt.ScrollBarAlwaysOff) + self.view.setInteractive(True) # Enable interaction for tooltips + self.view.setDragMode(QGraphicsView.NoDrag) + self.view.setTransformationAnchor(QGraphicsView.NoAnchor) + self.view.setResizeAnchor(QGraphicsView.NoAnchor) + self.view.setViewportUpdateMode(QGraphicsView.FullViewportUpdate) + + # Prevent wheel events + self.view.wheelEvent = lambda event: None + + self.layout().addWidget(self.view) + + # Configuration for time window (in seconds) + self.time_window = window_seconds + + # Store pulses for visualization + self.pulses = [] + self.channel_limit = 100 # Default channel limit + + # Packet tracking for FIFO visualization + self.current_packet_index = 0 # Which 4-pulse packet is currently active + self.last_packet_time = 0 # When the last packet was received + self.pulse_fingerprints = {} # Track pulse fingerprints to avoid duplicates + + # Initialize the scene size + self.updateSceneRect() + + self.timer = QTimer() + self.timer.timeout.connect(self.refresh) + self.timer.start(50) + + # Colors for visualization + self.pulse_color = QColor(0, 255, 0, 200) # Semi-transparent lime + self.pulse_border_color = QColor("darkgreen") + + # Time scaling factor - how many pixels per ms of duration + self.time_scale_factor = 0.5 # pixels per ms + + def resizeEvent(self, event): + """Handle resize events by updating the scene rectangle""" + super().resizeEvent(event) + self.updateSceneRect() + # Force a refresh after resize + self.refresh() + + def updateSceneRect(self): + """Update the scene rectangle to match the view size""" + if self.view: + width = self.view.viewport().width() + height = self.view.viewport().height() + self.view.setSceneRect(0, 0, width, height) + + def get_pulse_fingerprint(self, pulse: CoyotePulse) -> str: + """Generate a fingerprint for a pulse to detect duplicates""" + return f"{pulse.frequency}_{pulse.intensity}_{pulse.duration}" + + def clean_old_pulses(self): + """Remove pulses outside the time window""" + current_time = time.time() + time_window = self.time_window.get() + self.pulses = [p for p in self.pulses if current_time - p.timestamp <= time_window] + + # Also clean up old fingerprints + for fingerprint, timestamp in list(self.pulse_fingerprints.items()): + if current_time - timestamp > time_window: + self.pulse_fingerprints.pop(fingerprint) + + def add_pulse(self, pulse: CoyotePulse, applied_intensity: float, channel_limit: int): + """Add a new pulse to the visualization""" + # Don't skip zero intensity pulses, but display them differently + self.channel_limit = channel_limit + + # Generate a fingerprint for this pulse + fingerprint = self.get_pulse_fingerprint(pulse) + + # Check if this pulse is from a new packet + current_time = time.time() + is_new_packet = len(self.pulses) % 4 == 0 or current_time - self.last_packet_time > 0.2 + + # If we've seen this exact pulse recently and it's not a new packet, skip it + if fingerprint in self.pulse_fingerprints and not is_new_packet: + # Only add if it's been more than 1 second since we last saw this pulse + last_seen_time = self.pulse_fingerprints[fingerprint] + if current_time - last_seen_time < 1.0: + return # Skip this pulse, it's a duplicate + + # Update fingerprint timestamp + self.pulse_fingerprints[fingerprint] = current_time + + # If it's a new packet, increment the packet index + if is_new_packet: + self.current_packet_index += 1 + self.last_packet_time = current_time + + # Store the CoyotePulse with additional metadata + pulse_copy = CoyotePulse( + frequency=pulse.frequency, + intensity=pulse.intensity, + duration=pulse.duration + ) + + # Add additional attributes to the pulse + pulse_copy.applied_intensity = applied_intensity + pulse_copy.packet_index = self.current_packet_index + pulse_copy.timestamp = current_time + + # Add the pulse + self.pulses.append(pulse_copy) + + # Clean up old pulses that are outside our time window + self.clean_old_pulses() + + def refresh(self): + """Redraw the pulse visualization""" + self.scene.clear() + + # Always ensure we're using the current viewport size + self.updateSceneRect() + + width = self.view.viewport().width() + height = self.view.viewport().height() + + # Clean up old pulses again (in case the timer fired without any new pulses added) + self.clean_old_pulses() + + if not self.pulses: + return + + # Sort pulses by timestamp so they display in chronological order + sorted_pulses = sorted(self.pulses, key=lambda p: p.timestamp) + + # Find the maximum intensity in current visible pulses + max_intensity = max(pulse.applied_intensity for pulse in sorted_pulses) + # Use either the channel limit or the current max intensity, whichever is larger + scale_max = max(max_intensity, self.channel_limit) + + # Get the time span of the visible pulses + now = time.time() + time_window = self.time_window.get() + oldest_time = now - time_window + newest_time = now + time_span_sec = time_window + + # Calculate total width available for all pulses + usable_width = width - 10 # Leave small margin on right side + + # Scale based on the time window, not the pulse count + # This ensures consistent scaling regardless of pulse frequency + time_scale = usable_width / (time_span_sec * 1000) # Convert to ms + + # Group pulses by packet for continuous display + pulses_by_packet = {} + for pulse in sorted_pulses: + packet_idx = pulse.packet_index + if packet_idx not in pulses_by_packet: + pulses_by_packet[packet_idx] = [] + pulses_by_packet[packet_idx].append(pulse) + + # Get sorted list of packet indices + packet_indices = sorted(pulses_by_packet.keys()) + + # Draw each packet's pulses as a continuous sequence + for i, packet_idx in enumerate(packet_indices): + packet_pulses = sorted(pulses_by_packet[packet_idx], key=lambda p: p.timestamp) + + # Determine the time range this packet covers + if i < len(packet_indices) - 1: + # This packet runs until the next packet starts + next_packet_idx = packet_indices[i + 1] + next_packet_start = min(p.timestamp for p in pulses_by_packet[next_packet_idx]) + packet_end_time = next_packet_start + else: + # This is the last packet, it runs until now + packet_end_time = now + + # Calculate packet colors + packet_color = QColor(0, 255, 0, 200) if packet_idx % 2 == 0 else QColor(100, 255, 100, 200) + + # Draw each pulse in this packet + for j, pulse in enumerate(packet_pulses): + # Calculate time positions + pulse_start_time = pulse.timestamp + + # For continuity, calculate the end time: + if j < len(packet_pulses) - 1: + # If there's another pulse in this packet, it extends to that pulse + pulse_end_time = packet_pulses[j + 1].timestamp + else: + # If this is the last pulse in the packet, it extends to the packet end + pulse_end_time = packet_end_time + + # Ensure we're within the visible time window + pulse_start_time = max(pulse_start_time, oldest_time) + pulse_end_time = min(pulse_end_time, newest_time) + + # Calculate positions and dimensions + time_position_start = (pulse_start_time - oldest_time) / time_span_sec + time_position_end = (pulse_end_time - oldest_time) / time_span_sec + + x_start = 5 + (time_position_start * usable_width) + x_end = 5 + (time_position_end * usable_width) + rect_width = max(2, x_end - x_start) # Ensure minimum width + + # Calculate height based on intensity (always define rect_height) + height_ratio = pulse.applied_intensity / scale_max if scale_max > 0 else 0 + rect_height = height * height_ratio + + # For zero-intensity pulses, still show something to indicate timing + if pulse.applied_intensity <= 0: + # Draw a thin line or empty rectangle to show timing without intensity + empty_rect = QGraphicsRectItem( + x_start, height - 2, # Just a thin line at the bottom + rect_width, 2 + ) + empty_rect.setPen(QPen(QColor(100, 100, 100, 100), 1)) # Very light gray + empty_rect.setBrush(QBrush(QColor(100, 100, 100, 50))) # Almost transparent + self.scene.addItem(empty_rect) + else: + # Create rectangle for the pulse + rect = PulseRectItem( + x_start, height - rect_height, # x, y (bottom-aligned) + rect_width, rect_height, # width, height + pulse # pass pulse data for tooltip + ) + + rect.setPen(QPen(self.pulse_border_color, 1)) + rect.setBrush(QBrush(packet_color)) + + # Add rectangle to scene + self.scene.addItem(rect) + + # Draw frequency tick marks for visualization + if pulse.frequency > 0 and rect_width > 10: + # Number of ticks based on frequency (higher frequency = more ticks) + num_ticks = min(max(2, int(pulse.frequency / 20)), 8) # 2-8 ticks + + tick_spacing = rect_width / (num_ticks + 1) + tick_height = rect_height * 0.4 # 40% of rectangle height + + for t in range(1, num_ticks + 1): + tick_x = x_start + (t * tick_spacing) + tick_y = height - rect_height + + # Draw tick mark + tick = QGraphicsLineItem( + tick_x, tick_y, # Start at top of rectangle + tick_x, tick_y + tick_height # Go down + ) + tick.setPen(QPen(QColor("white"), 1)) + self.scene.addItem(tick) + +class PulseRectItem(QGraphicsRectItem): + def __init__(self, x, y, width, height, pulse): + super().__init__(x, y, width, height) + self.pulse = pulse + self.setAcceptHoverEvents(True) + + def hoverEnterEvent(self, event): + # Show tooltip with pulse information + freq = self.pulse.frequency + intensity = self.pulse.intensity + duration = self.pulse.duration + + tooltip_text = f"Frequency: {freq} Hz\nIntensity: {intensity}%\nDuration: {duration} ms" + QToolTip.showText(event.screenPos(), tooltip_text) + + # Change appearance on hover + current_pen = self.pen() + current_pen.setWidth(2) # Make border thicker + self.setPen(current_pen) + + def hoverLeaveEvent(self, event): + # Restore original appearance + current_pen = self.pen() + current_pen.setWidth(1) # Restore original border width + self.setPen(current_pen) + diff --git a/qt_ui/device_wizard/coyote_waveform_select.py b/qt_ui/device_wizard/coyote_waveform_select.py new file mode 100644 index 0000000..ea0b8f3 --- /dev/null +++ b/qt_ui/device_wizard/coyote_waveform_select.py @@ -0,0 +1,16 @@ +from PySide6.QtWidgets import QWizardPage + +from qt_ui.device_wizard.coyote_waveform_select_ui import Ui_WizardPageCoyote + + +class WizardPageCoyoteWaveformSelect(QWizardPage, Ui_WizardPageCoyote): + def __init__(self, parent=None): + super().__init__(parent) + self.setupUi(self) + + self.three_phase_radio.toggled.connect(self.completeChanged) + + def isComplete(self) -> bool: + return any([ + self.three_phase_radio.isChecked() and self.three_phase_radio.isEnabled(), + ]) diff --git a/qt_ui/device_wizard/coyote_waveform_select_ui.py b/qt_ui/device_wizard/coyote_waveform_select_ui.py new file mode 100644 index 0000000..3c8f2bd --- /dev/null +++ b/qt_ui/device_wizard/coyote_waveform_select_ui.py @@ -0,0 +1,57 @@ +# -*- coding: utf-8 -*- + +################################################################################ +## Form generated from reading UI file 'coyote_waveform_select.ui' +## +## Created by: Qt User Interface Compiler version 6.9.0 +## +## WARNING! All changes made in this file will be lost when recompiling UI file! +################################################################################ + +from PySide6.QtCore import (QCoreApplication, QDate, QDateTime, QLocale, + QMetaObject, QObject, QPoint, QRect, + QSize, QTime, QUrl, Qt) +from PySide6.QtGui import (QBrush, QColor, QConicalGradient, QCursor, + QFont, QFontDatabase, QGradient, QIcon, + QImage, QKeySequence, QLinearGradient, QPainter, + QPalette, QPixmap, QRadialGradient, QTransform) +from PySide6.QtWidgets import (QApplication, QLabel, QRadioButton, QSizePolicy, + QSpacerItem, QVBoxLayout, QWidget, QWizardPage) + +class Ui_WizardPageCoyote(object): + def setupUi(self, WizardPageCoyote): + if not WizardPageCoyote.objectName(): + WizardPageCoyote.setObjectName(u"WizardPageCoyote") + WizardPageCoyote.resize(611, 497) + self.verticalLayout = QVBoxLayout(WizardPageCoyote) + self.verticalLayout.setObjectName(u"verticalLayout") + self.three_phase_radio = QRadioButton(WizardPageCoyote) + self.three_phase_radio.setObjectName(u"three_phase_radio") + + self.verticalLayout.addWidget(self.three_phase_radio) + + self.label = QLabel(WizardPageCoyote) + self.label.setObjectName(u"label") + self.label.setWordWrap(True) + + self.verticalLayout.addWidget(self.label) + + self.verticalSpacer = QSpacerItem(20, 40, QSizePolicy.Policy.Minimum, QSizePolicy.Policy.Expanding) + + self.verticalLayout.addItem(self.verticalSpacer) + + + self.retranslateUi(WizardPageCoyote) + + QMetaObject.connectSlotsByName(WizardPageCoyote) + # setupUi + + def retranslateUi(self, WizardPageCoyote): + WizardPageCoyote.setWindowTitle(QCoreApplication.translate("WizardPageCoyote", u"WizardPage", None)) + self.three_phase_radio.setText(QCoreApplication.translate("WizardPageCoyote", u"Three-phase", None)) + self.label.setText(QCoreApplication.translate("WizardPageCoyote", u"\n" +"

A = left
B = right
C = neutral

\n" +"

Connect A- and B- to a shared common electrode (e.g. a conductive rubber loop).

\n" +" ", None)) + # retranslateUi + diff --git a/qt_ui/device_wizard/enums.py b/qt_ui/device_wizard/enums.py index 4544286..9bbf21b 100644 --- a/qt_ui/device_wizard/enums.py +++ b/qt_ui/device_wizard/enums.py @@ -12,6 +12,7 @@ class DeviceType(Enum): FOCSTIM_THREE_PHASE = 5 NEOSTIM_THREE_PHASE = 6 FOCSTIM_FOUR_PHASE = 7 + COYOTE_THREE_PHASE = 8 class WaveformType(Enum): @@ -30,7 +31,7 @@ class DeviceConfiguration: def save(self): settings.device_config_device_type.set(self.device_type.value) - if self.device_type in (DeviceType.AUDIO_THREE_PHASE, DeviceType.FOCSTIM_THREE_PHASE): + if self.device_type in (DeviceType.AUDIO_THREE_PHASE, DeviceType.FOCSTIM_THREE_PHASE, DeviceType.COYOTE_THREE_PHASE): settings.device_config_waveform_type.set(self.waveform_type.value) settings.device_config_min_freq.set(self.min_frequency) settings.device_config_max_freq.set(self.max_frequency) diff --git a/qt_ui/device_wizard/type_select.py b/qt_ui/device_wizard/type_select.py index 81f63e5..4b50d23 100644 --- a/qt_ui/device_wizard/type_select.py +++ b/qt_ui/device_wizard/type_select.py @@ -11,10 +11,12 @@ def __init__(self, parent=None): self.audio_based_radio.toggled.connect(self.completeChanged) self.focstim_radio.toggled.connect(self.completeChanged) self.neostim_radio.toggled.connect(self.completeChanged) + self.coyote_radio.toggled.connect(self.completeChanged) def isComplete(self) -> bool: return any([ self.audio_based_radio.isChecked(), self.focstim_radio.isChecked(), - self.neostim_radio.isChecked() + self.neostim_radio.isChecked(), + self.coyote_radio.isChecked() ]) diff --git a/qt_ui/device_wizard/type_select_ui.py b/qt_ui/device_wizard/type_select_ui.py index 12ccc6a..f6ddbe1 100644 --- a/qt_ui/device_wizard/type_select_ui.py +++ b/qt_ui/device_wizard/type_select_ui.py @@ -36,15 +36,20 @@ def setupUi(self, WizardPageDeviceType): self.formLayout.setWidget(1, QFormLayout.LabelRole, self.focstim_radio) - self.verticalSpacer = QSpacerItem(20, 40, QSizePolicy.Policy.Minimum, QSizePolicy.Policy.Expanding) - - self.formLayout.setItem(3, QFormLayout.LabelRole, self.verticalSpacer) - self.neostim_radio = QRadioButton(WizardPageDeviceType) self.neostim_radio.setObjectName(u"neostim_radio") self.formLayout.setWidget(2, QFormLayout.LabelRole, self.neostim_radio) + self.coyote_radio = QRadioButton(WizardPageDeviceType) + self.coyote_radio.setObjectName(u"coyote_radio") + + self.formLayout.setWidget(3, QFormLayout.LabelRole, self.coyote_radio) + + self.verticalSpacer = QSpacerItem(20, 40, QSizePolicy.Policy.Minimum, QSizePolicy.Policy.Expanding) + + self.formLayout.setItem(4, QFormLayout.LabelRole, self.verticalSpacer) + self.retranslateUi(WizardPageDeviceType) @@ -57,5 +62,6 @@ def retranslateUi(self, WizardPageDeviceType): self.audio_based_radio.setText(QCoreApplication.translate("WizardPageDeviceType", u"Audio-based three-phase", None)) self.focstim_radio.setText(QCoreApplication.translate("WizardPageDeviceType", u"FOC-Stim", None)) self.neostim_radio.setText(QCoreApplication.translate("WizardPageDeviceType", u"NeoStim", None)) + self.coyote_radio.setText(QCoreApplication.translate("WizardPageDeviceType", u"Coyote 3", None)) # retranslateUi diff --git a/qt_ui/device_wizard/wizard.py b/qt_ui/device_wizard/wizard.py index 862e9c8..159a50d 100644 --- a/qt_ui/device_wizard/wizard.py +++ b/qt_ui/device_wizard/wizard.py @@ -9,7 +9,10 @@ from qt_ui.device_wizard.waveform_select import WizardPageWaveformType from qt_ui.device_wizard.safety_limits import WizardPageSafetyLimits from qt_ui.device_wizard.neostim_waveform_select import WizardPageNeoStimWaveformSelect +from qt_ui.device_wizard.coyote_waveform_select import WizardPageCoyoteWaveformSelect from qt_ui.device_wizard.enums import DeviceType, WaveformType, DeviceConfiguration +from qt_ui.settings import device_config_waveform_amplitude_amps +from device.coyote import constants as coyote_constants logger = logging.getLogger('restim.device_wizard') @@ -22,6 +25,7 @@ class WizardPage(Enum): Page_limits_foc = 6 Page_neostim_waveform = 4 Page_focstim_waveform = 5 + Page_coyote_waveform = 7 class DeviceSelectionWizard(QWizard): @@ -48,6 +52,9 @@ def __init__(self, parent=None): self.setPage(WizardPage.Page_neostim_waveform.value, self.page_neostim_waveform_select) self.page_focstim_waveform_select = WizardPageFocStimWaveformSelect() self.setPage(WizardPage.Page_focstim_waveform.value, self.page_focstim_waveform_select) + self.page_coyote_waveform_select = WizardPageCoyoteWaveformSelect() + self.page_coyote_waveform_select.setFinalPage(True) + self.setPage(WizardPage.Page_coyote_waveform.value, self.page_coyote_waveform_select) self.set_configuration(DeviceConfiguration.from_settings()) @@ -76,6 +83,8 @@ def nextId(self): return WizardPage.Page_focstim_waveform.value elif self.page_device_type.neostim_radio.isChecked(): return WizardPage.Page_neostim_waveform.value + elif self.page_device_type.coyote_radio.isChecked(): + return WizardPage.Page_coyote_waveform.value else: raise RuntimeError("unknown device type") @@ -96,6 +105,8 @@ def validateCurrentPage(self) -> bool: pass elif self.page_device_type.neostim_radio.isChecked(): pass + elif self.page_device_type.coyote_radio.isChecked(): + pass return super(DeviceSelectionWizard, self).validateCurrentPage() @@ -146,6 +157,14 @@ def get_configuration(self) -> DeviceConfiguration: None, None, None ) + elif self.page_device_type.coyote_radio.isChecked(): + return DeviceConfiguration( + DeviceType.COYOTE_THREE_PHASE, + WaveformType.PULSE_BASED, + coyote_constants.HARDWARE_MIN_FREQ_HZ, + coyote_constants.HARDWARE_MAX_FREQ_HZ, + None + ) else: assert(False) @@ -160,6 +179,8 @@ def set_configuration(self, config: DeviceConfiguration): self.page_focstim_waveform_select.four_phase_radio.setChecked(True) if config.device_type == DeviceType.NEOSTIM_THREE_PHASE: self.page_device_type.neostim_radio.setChecked(True) + if config.device_type == DeviceType.COYOTE_THREE_PHASE: + self.page_device_type.coyote_radio.setChecked(True) self.page_waveform_type.continuous_radio.setChecked(config.waveform_type == WaveformType.CONTINUOUS) self.page_waveform_type.pulse_based_radio.setChecked(config.waveform_type == WaveformType.PULSE_BASED) diff --git a/qt_ui/main_window_ui.py b/qt_ui/main_window_ui.py index 3649d62..e8d229f 100644 --- a/qt_ui/main_window_ui.py +++ b/qt_ui/main_window_ui.py @@ -23,6 +23,7 @@ from qt_ui.ab_test_widget import ABTestWidget from qt_ui.carrier_settings_widget import CarrierSettingsWidget +from qt_ui.coyote_settings_widget import CoyoteSettingsWidget from qt_ui.four_phase_settings_widget import FourPhaseSettingsWidget from qt_ui.media_settings_widget import MediaSettingsWidget from qt_ui.neostim_settings_widget import NeoStimSettingsWidget @@ -205,6 +206,9 @@ def setupUi(self, MainWindow): self.tab_carrier = CarrierSettingsWidget() self.tab_carrier.setObjectName(u"tab_carrier") self.tabWidget.addTab(self.tab_carrier, "") + self.tab_coyote = CoyoteSettingsWidget() + self.tab_coyote.setObjectName(u"tab_coyote") + self.tabWidget.addTab(self.tab_coyote, "") self.tab_pulse_settings = PulseSettingsWidget() self.tab_pulse_settings.setObjectName(u"tab_pulse_settings") self.tabWidget.addTab(self.tab_pulse_settings, "") @@ -308,6 +312,7 @@ def retranslateUi(self, MainWindow): self.tabWidget.setTabText(self.tabWidget.indexOf(self.tab_threephase), QCoreApplication.translate("MainWindow", u"3-phase", None)) self.tabWidget.setTabText(self.tabWidget.indexOf(self.tab_fourphase), QCoreApplication.translate("MainWindow", u"4-phase", None)) self.tabWidget.setTabText(self.tabWidget.indexOf(self.tab_carrier), QCoreApplication.translate("MainWindow", u"Carrier settings", None)) + self.tabWidget.setTabText(self.tabWidget.indexOf(self.tab_coyote), QCoreApplication.translate("MainWindow", u"Coyote", None)) self.tabWidget.setTabText(self.tabWidget.indexOf(self.tab_pulse_settings), QCoreApplication.translate("MainWindow", u"Pulse settings", None)) self.tabWidget.setTabText(self.tabWidget.indexOf(self.tab_neostim), QCoreApplication.translate("MainWindow", u"NeoStim", None)) self.tabWidget.setTabText(self.tabWidget.indexOf(self.tab_a_b_testing), QCoreApplication.translate("MainWindow", u"A/B testing", None)) diff --git a/qt_ui/mainwindow.py b/qt_ui/mainwindow.py index 6adf27c..7db5aab 100644 --- a/qt_ui/mainwindow.py +++ b/qt_ui/mainwindow.py @@ -31,6 +31,8 @@ from qt_ui.models.funscript_kit import FunscriptKitModel from device.focstim.proto_device import FOCStimProtoDevice from device.neostim.neostim_device import NeoStim +from device.coyote.device import CoyoteDevice, CoyoteParams +from device.coyote.constants import DEVICE_NAME from qt_ui.widgets.icon_with_connection_status import IconWithConnectionStatus from stim_math.axis import create_temporal_axis @@ -349,7 +351,8 @@ def set_visible(widget, state): self.tab_vibrate, self.tab_details, self.tab_a_b_testing, - self.tab_neostim} + self.tab_neostim, + self.tab_coyote} visible = {self.tab_threephase, self.tab_volume, self.tab_vibrate, self.tab_details} @@ -372,6 +375,9 @@ def set_visible(widget, state): if config.device_type == DeviceType.NEOSTIM_THREE_PHASE: visible |= {self.tab_neostim} visible -= {self.tab_vibrate, self.tab_details} + if config.device_type == DeviceType.COYOTE_THREE_PHASE: + visible |= {self.tab_coyote, self.tab_pulse_settings} + visible -= {self.tab_vibrate} for tab in all_tabs: set_visible(tab, tab in visible) @@ -388,7 +394,7 @@ def set_visible(widget, state): self.tcode_command_router.set_carrier_axis(self.tab_pulse_settings.axis_carrier_frequency) # populate motion generator and patterns combobox - if config.device_type in (DeviceType.AUDIO_THREE_PHASE, DeviceType.NEOSTIM_THREE_PHASE, DeviceType.FOCSTIM_THREE_PHASE): + if config.device_type in (DeviceType.AUDIO_THREE_PHASE, DeviceType.NEOSTIM_THREE_PHASE, DeviceType.FOCSTIM_THREE_PHASE, DeviceType.COYOTE_THREE_PHASE): self.motion_3.set_enable(True) self.motion_4.set_enable(False) self.stackedWidget_visual.setCurrentIndex( @@ -401,6 +407,20 @@ def set_visible(widget, state): self.stackedWidget_visual.setCurrentIndex( self.stackedWidget_visual.indexOf(self.page_fourphase) ) + + if config.device_type == DeviceType.COYOTE_THREE_PHASE: + self.output_device = CoyoteDevice(DEVICE_NAME) + self.output_device.parameters = CoyoteParams( + channel_a_limit=qt_ui.settings.coyote_channel_a_limit.get(), + channel_b_limit=qt_ui.settings.coyote_channel_b_limit.get(), + channel_a_freq_balance=qt_ui.settings.coyote_channel_a_freq_balance.get(), + channel_b_freq_balance=qt_ui.settings.coyote_channel_b_freq_balance.get(), + channel_a_intensity_balance=qt_ui.settings.coyote_channel_a_intensity_balance.get(), + channel_b_intensity_balance=qt_ui.settings.coyote_channel_b_intensity_balance.get() + ) + self.tab_coyote.setup_device(self.output_device) + + self.refresh_pattern_combobox() self.refresh_pattern_combobox() @@ -416,10 +436,11 @@ def signal_start_stop(self): self.signal_stop(PlayState.STOPPED) def signal_start(self): - assert self.output_device is None - self.autostart_timer.stop() device = DeviceConfiguration.from_settings() + + assert (self.output_device is None or device.device_type == DeviceType.COYOTE_THREE_PHASE) + algorithm_factory = AlgorithmFactory( self, FunscriptKitModel.load_from_settings(), @@ -474,13 +495,25 @@ def signal_start(self): self.playstate = PlayState.PLAYING self.tab_volume.set_play_state(self.playstate) self.refresh_play_button_icon() + elif device.device_type == DeviceType.COYOTE_THREE_PHASE: + if not self.output_device: + logger.warning("Coyote device is no longer initialized") + return + + self.output_device.start_updates(algorithm) + self.playstate = PlayState.PLAYING + self.refresh_play_button_icon() else: raise RuntimeError("Unknown device type") def signal_stop(self, new_playstate: PlayState = PlayState.STOPPED): + """Stop signal generation.""" if self.output_device is not None: - self.output_device.stop() - self.output_device = None + if isinstance(self.output_device, CoyoteDevice): + self.output_device.stop_updates() # Only stop sending to device + else: + self.output_device.stop() # Other devices may need full stop + self.output_device = None self.playstate = new_playstate self.tab_volume.set_play_state(self.playstate) self.refresh_play_button_icon() @@ -551,7 +584,7 @@ def refresh_pattern_combobox(self): config = DeviceConfiguration.from_settings() currently_selected_text = self.comboBox_patternSelect.currentText() - if config.device_type in (DeviceType.AUDIO_THREE_PHASE, DeviceType.NEOSTIM_THREE_PHASE, DeviceType.FOCSTIM_THREE_PHASE): + if config.device_type in (DeviceType.AUDIO_THREE_PHASE, DeviceType.NEOSTIM_THREE_PHASE, DeviceType.FOCSTIM_THREE_PHASE, DeviceType.COYOTE_THREE_PHASE): self.comboBox_patternSelect.clear() for pattern in self.motion_3.patterns: self.comboBox_patternSelect.addItem(pattern.name(), pattern) diff --git a/qt_ui/preferences_dialog.py b/qt_ui/preferences_dialog.py index 2a4edab..c8dee7e 100644 --- a/qt_ui/preferences_dialog.py +++ b/qt_ui/preferences_dialog.py @@ -24,6 +24,9 @@ def __init__(self, parent=None): self.tabWidget.setCurrentIndex(0) + self._coyote_logger = logging.getLogger('restim.coyote') + self._coyote_default_log_level = self._coyote_logger.getEffectiveLevel() + # Initialize pattern service and cache pattern data immediately self.pattern_service = PatternControlService() self._cached_patterns = None @@ -151,6 +154,19 @@ def loadSettings(self): # neostim settings self.neostim_port.setCurrentIndex(self.neostim_port.findData(qt_ui.settings.neostim_serial_port.get())) + # Coyote 3 + self.coyote_channel_a_limit.setValue(qt_ui.settings.coyote_channel_a_limit.get()) + self.coyote_channel_b_limit.setValue(qt_ui.settings.coyote_channel_b_limit.get()) + self.coyote_channel_a_freq_balance.setValue(qt_ui.settings.coyote_channel_a_freq_balance.get()) + self.coyote_channel_b_freq_balance.setValue(qt_ui.settings.coyote_channel_b_freq_balance.get()) + self.coyote_channel_a_intensity_balance.setValue(qt_ui.settings.coyote_channel_a_intensity_balance.get()) + self.coyote_channel_b_intensity_balance.setValue(qt_ui.settings.coyote_channel_b_intensity_balance.get()) + self.coyote_max_intensity_change_per_pulse.setValue( + qt_ui.settings.coyote_max_intensity_change_per_pulse.get() + ) + self.coyote_graph_window.setValue(qt_ui.settings.coyote_graph_window.get()) + self.coyote_debug_logging.setChecked(qt_ui.settings.coyote_debug_logging.get()) + # media sync settings self.mpc_address.setText(qt_ui.settings.media_sync_mpc_address.get()) self.heresphere_address.setText(qt_ui.settings.media_sync_heresphere_address.get()) @@ -169,6 +185,8 @@ def loadSettings(self): # refresh pattern preferences (just reload checkboxes from settings) self.refresh_pattern_preferences() + self.apply_coyote_logging() + def repopulate_audio_devices(self): self.audio_output_device.clear() default_audio_output_device_name = qt_ui.settings.audio_output_device.get() @@ -313,6 +331,19 @@ def saveSettings(self): # neoStim qt_ui.settings.neostim_serial_port.set(str(self.neostim_port.currentData())) + # Coyote 3 + qt_ui.settings.coyote_channel_a_limit.set(self.coyote_channel_a_limit.value()) + qt_ui.settings.coyote_channel_b_limit.set(self.coyote_channel_b_limit.value()) + qt_ui.settings.coyote_channel_a_freq_balance.set(self.coyote_channel_a_freq_balance.value()) + qt_ui.settings.coyote_channel_b_freq_balance.set(self.coyote_channel_b_freq_balance.value()) + qt_ui.settings.coyote_channel_a_intensity_balance.set(self.coyote_channel_a_intensity_balance.value()) + qt_ui.settings.coyote_channel_b_intensity_balance.set(self.coyote_channel_b_intensity_balance.value()) + qt_ui.settings.coyote_max_intensity_change_per_pulse.set( + self.coyote_max_intensity_change_per_pulse.value() + ) + qt_ui.settings.coyote_graph_window.set(self.coyote_graph_window.value()) + qt_ui.settings.coyote_debug_logging.set(self.coyote_debug_logging.isChecked()) + # media sync settings qt_ui.settings.media_sync_mpc_address.set(self.mpc_address.text()) qt_ui.settings.media_sync_heresphere_address.set(self.heresphere_address.text()) @@ -340,6 +371,13 @@ def saveSettings(self): if was_enabled != is_enabled: self.pattern_service.set_pattern_enabled(pattern_name, is_enabled) + self.apply_coyote_logging() + + def apply_coyote_logging(self): + enabled = qt_ui.settings.coyote_debug_logging.get() + new_level = logging.DEBUG if enabled else logging.INFO + self._coyote_logger.setLevel(new_level) + def funscript_reset_defaults(self): self.tableView.model().reset_to_defaults() @@ -417,4 +455,3 @@ def disable_all_patterns(self): widget = self.patterns_table.cellWidget(row, 1) if isinstance(widget, QCheckBox): widget.setChecked(False) - diff --git a/qt_ui/preferences_dialog_ui.py b/qt_ui/preferences_dialog_ui.py index 8768ac4..cc99980 100644 --- a/qt_ui/preferences_dialog_ui.py +++ b/qt_ui/preferences_dialog_ui.py @@ -314,11 +314,6 @@ def setupUi(self, PreferencesDialog): self.groupBox_10.setObjectName(u"groupBox_10") self.gridLayout_7 = QGridLayout(self.groupBox_10) self.gridLayout_7.setObjectName(u"gridLayout_7") - self.label_22 = QLabel(self.groupBox_10) - self.label_22.setObjectName(u"label_22") - - self.gridLayout_7.addWidget(self.label_22, 3, 0, 1, 1) - self.focstim_ssid = QLineEdit(self.groupBox_10) self.focstim_ssid.setObjectName(u"focstim_ssid") @@ -334,11 +329,6 @@ def setupUi(self, PreferencesDialog): self.gridLayout_7.addWidget(self.label_21, 1, 0, 1, 1) - self.focstim_read_ip = QToolButton(self.groupBox_10) - self.focstim_read_ip.setObjectName(u"focstim_read_ip") - - self.gridLayout_7.addWidget(self.focstim_read_ip, 3, 2, 1, 1) - self.focstim_password = QLineEdit(self.groupBox_10) self.focstim_password.setObjectName(u"focstim_password") @@ -352,7 +342,17 @@ def setupUi(self, PreferencesDialog): self.focstim_ip = QLineEdit(self.groupBox_10) self.focstim_ip.setObjectName(u"focstim_ip") - self.gridLayout_7.addWidget(self.focstim_ip, 3, 1, 1, 1) + self.gridLayout_7.addWidget(self.focstim_ip, 4, 1, 1, 1) + + self.label_22 = QLabel(self.groupBox_10) + self.label_22.setObjectName(u"label_22") + + self.gridLayout_7.addWidget(self.label_22, 4, 0, 1, 1) + + self.focstim_read_ip = QToolButton(self.groupBox_10) + self.focstim_read_ip.setObjectName(u"focstim_read_ip") + + self.gridLayout_7.addWidget(self.focstim_read_ip, 4, 2, 1, 1) self.verticalLayout_5.addWidget(self.groupBox_10) @@ -361,15 +361,15 @@ def setupUi(self, PreferencesDialog): self.groupBox_9.setObjectName(u"groupBox_9") self.formLayout_8 = QFormLayout(self.groupBox_9) self.formLayout_8.setObjectName(u"formLayout_8") - self.label_18 = QLabel(self.groupBox_9) - self.label_18.setObjectName(u"label_18") + self.label_15 = QLabel(self.groupBox_9) + self.label_15.setObjectName(u"label_15") - self.formLayout_8.setWidget(2, QFormLayout.ItemRole.LabelRole, self.label_18) + self.formLayout_8.setWidget(0, QFormLayout.ItemRole.LabelRole, self.label_15) - self.focstim_dump_notifications = QCheckBox(self.groupBox_9) - self.focstim_dump_notifications.setObjectName(u"focstim_dump_notifications") + self.focstim_use_teleplot = QCheckBox(self.groupBox_9) + self.focstim_use_teleplot.setObjectName(u"focstim_use_teleplot") - self.formLayout_8.setWidget(2, QFormLayout.ItemRole.FieldRole, self.focstim_dump_notifications) + self.formLayout_8.setWidget(0, QFormLayout.ItemRole.FieldRole, self.focstim_use_teleplot) self.label_16 = QLabel(self.groupBox_9) self.label_16.setObjectName(u"label_16") @@ -381,15 +381,15 @@ def setupUi(self, PreferencesDialog): self.formLayout_8.setWidget(1, QFormLayout.ItemRole.FieldRole, self.focstim_teleplot_prefix) - self.label_15 = QLabel(self.groupBox_9) - self.label_15.setObjectName(u"label_15") + self.label_18 = QLabel(self.groupBox_9) + self.label_18.setObjectName(u"label_18") - self.formLayout_8.setWidget(0, QFormLayout.ItemRole.LabelRole, self.label_15) + self.formLayout_8.setWidget(2, QFormLayout.ItemRole.LabelRole, self.label_18) - self.focstim_use_teleplot = QCheckBox(self.groupBox_9) - self.focstim_use_teleplot.setObjectName(u"focstim_use_teleplot") + self.focstim_dump_notifications = QCheckBox(self.groupBox_9) + self.focstim_dump_notifications.setObjectName(u"focstim_dump_notifications") - self.formLayout_8.setWidget(0, QFormLayout.ItemRole.FieldRole, self.focstim_use_teleplot) + self.formLayout_8.setWidget(2, QFormLayout.ItemRole.FieldRole, self.focstim_dump_notifications) self.verticalLayout_5.addWidget(self.groupBox_9) @@ -432,6 +432,144 @@ def setupUi(self, PreferencesDialog): self.verticalLayout_8.addItem(self.verticalSpacer_5) self.tabWidget.addTab(self.tab_neostim, "") + self.tab_coyote = QWidget() + self.tab_coyote.setObjectName(u"tab_coyote") + self.verticalLayout_coyote = QVBoxLayout(self.tab_coyote) + self.verticalLayout_coyote.setObjectName(u"verticalLayout_coyote") + self.groupBox_coyote_params = QGroupBox(self.tab_coyote) + self.groupBox_coyote_params.setObjectName(u"groupBox_coyote_params") + self.formLayout_coyote_params = QFormLayout(self.groupBox_coyote_params) + self.formLayout_coyote_params.setObjectName(u"formLayout_coyote_params") + self.label_coyote_channel_a_limit = QLabel(self.groupBox_coyote_params) + self.label_coyote_channel_a_limit.setObjectName(u"label_coyote_channel_a_limit") + + self.formLayout_coyote_params.setWidget(0, QFormLayout.ItemRole.LabelRole, self.label_coyote_channel_a_limit) + + self.coyote_channel_a_limit = QSpinBox(self.groupBox_coyote_params) + self.coyote_channel_a_limit.setObjectName(u"coyote_channel_a_limit") + self.coyote_channel_a_limit.setMinimum(0) + self.coyote_channel_a_limit.setMaximum(200) + + self.formLayout_coyote_params.setWidget(0, QFormLayout.ItemRole.FieldRole, self.coyote_channel_a_limit) + + self.label_coyote_channel_b_limit = QLabel(self.groupBox_coyote_params) + self.label_coyote_channel_b_limit.setObjectName(u"label_coyote_channel_b_limit") + + self.formLayout_coyote_params.setWidget(1, QFormLayout.ItemRole.LabelRole, self.label_coyote_channel_b_limit) + + self.coyote_channel_b_limit = QSpinBox(self.groupBox_coyote_params) + self.coyote_channel_b_limit.setObjectName(u"coyote_channel_b_limit") + self.coyote_channel_b_limit.setMinimum(0) + self.coyote_channel_b_limit.setMaximum(200) + + self.formLayout_coyote_params.setWidget(1, QFormLayout.ItemRole.FieldRole, self.coyote_channel_b_limit) + + self.label_coyote_channel_a_freq_balance = QLabel(self.groupBox_coyote_params) + self.label_coyote_channel_a_freq_balance.setObjectName(u"label_coyote_channel_a_freq_balance") + + self.formLayout_coyote_params.setWidget(2, QFormLayout.ItemRole.LabelRole, self.label_coyote_channel_a_freq_balance) + + self.coyote_channel_a_freq_balance = QSpinBox(self.groupBox_coyote_params) + self.coyote_channel_a_freq_balance.setObjectName(u"coyote_channel_a_freq_balance") + self.coyote_channel_a_freq_balance.setMinimum(0) + self.coyote_channel_a_freq_balance.setMaximum(255) + + self.formLayout_coyote_params.setWidget(2, QFormLayout.ItemRole.FieldRole, self.coyote_channel_a_freq_balance) + + self.label_coyote_channel_b_freq_balance = QLabel(self.groupBox_coyote_params) + self.label_coyote_channel_b_freq_balance.setObjectName(u"label_coyote_channel_b_freq_balance") + + self.formLayout_coyote_params.setWidget(3, QFormLayout.ItemRole.LabelRole, self.label_coyote_channel_b_freq_balance) + + self.coyote_channel_b_freq_balance = QSpinBox(self.groupBox_coyote_params) + self.coyote_channel_b_freq_balance.setObjectName(u"coyote_channel_b_freq_balance") + self.coyote_channel_b_freq_balance.setMinimum(0) + self.coyote_channel_b_freq_balance.setMaximum(255) + + self.formLayout_coyote_params.setWidget(3, QFormLayout.ItemRole.FieldRole, self.coyote_channel_b_freq_balance) + + self.label_coyote_channel_a_intensity_balance = QLabel(self.groupBox_coyote_params) + self.label_coyote_channel_a_intensity_balance.setObjectName(u"label_coyote_channel_a_intensity_balance") + + self.formLayout_coyote_params.setWidget(4, QFormLayout.ItemRole.LabelRole, self.label_coyote_channel_a_intensity_balance) + + self.coyote_channel_a_intensity_balance = QSpinBox(self.groupBox_coyote_params) + self.coyote_channel_a_intensity_balance.setObjectName(u"coyote_channel_a_intensity_balance") + self.coyote_channel_a_intensity_balance.setMinimum(0) + self.coyote_channel_a_intensity_balance.setMaximum(255) + + self.formLayout_coyote_params.setWidget(4, QFormLayout.ItemRole.FieldRole, self.coyote_channel_a_intensity_balance) + + self.label_coyote_channel_b_intensity_balance = QLabel(self.groupBox_coyote_params) + self.label_coyote_channel_b_intensity_balance.setObjectName(u"label_coyote_channel_b_intensity_balance") + + self.formLayout_coyote_params.setWidget(5, QFormLayout.ItemRole.LabelRole, self.label_coyote_channel_b_intensity_balance) + + self.coyote_channel_b_intensity_balance = QSpinBox(self.groupBox_coyote_params) + self.coyote_channel_b_intensity_balance.setObjectName(u"coyote_channel_b_intensity_balance") + self.coyote_channel_b_intensity_balance.setMinimum(0) + self.coyote_channel_b_intensity_balance.setMaximum(255) + + self.formLayout_coyote_params.setWidget(5, QFormLayout.ItemRole.FieldRole, self.coyote_channel_b_intensity_balance) + + + self.verticalLayout_coyote.addWidget(self.groupBox_coyote_params) + + self.groupBox_coyote_algorithm = QGroupBox(self.tab_coyote) + self.groupBox_coyote_algorithm.setObjectName(u"groupBox_coyote_algorithm") + self.formLayout_coyote_algorithm = QFormLayout(self.groupBox_coyote_algorithm) + self.formLayout_coyote_algorithm.setObjectName(u"formLayout_coyote_algorithm") + self.label_coyote_max_intensity_change_per_pulse = QLabel(self.groupBox_coyote_algorithm) + self.label_coyote_max_intensity_change_per_pulse.setObjectName(u"label_coyote_max_intensity_change_per_pulse") + + self.formLayout_coyote_algorithm.setWidget(0, QFormLayout.ItemRole.LabelRole, self.label_coyote_max_intensity_change_per_pulse) + + self.coyote_max_intensity_change_per_pulse = QDoubleSpinBox(self.groupBox_coyote_algorithm) + self.coyote_max_intensity_change_per_pulse.setObjectName(u"coyote_max_intensity_change_per_pulse") + self.coyote_max_intensity_change_per_pulse.setDecimals(1) + self.coyote_max_intensity_change_per_pulse.setMinimum(0.000000000000000) + self.coyote_max_intensity_change_per_pulse.setMaximum(100.000000000000000) + self.coyote_max_intensity_change_per_pulse.setSingleStep(0.100000000000000) + self.coyote_max_intensity_change_per_pulse.setValue(3.000000000000000) + + self.formLayout_coyote_algorithm.setWidget(0, QFormLayout.ItemRole.FieldRole, self.coyote_max_intensity_change_per_pulse) + + + self.verticalLayout_coyote.addWidget(self.groupBox_coyote_algorithm) + + self.groupBox_coyote_display = QGroupBox(self.tab_coyote) + self.groupBox_coyote_display.setObjectName(u"groupBox_coyote_display") + self.formLayout_coyote_display = QFormLayout(self.groupBox_coyote_display) + self.formLayout_coyote_display.setObjectName(u"formLayout_coyote_display") + self.label_coyote_graph_window = QLabel(self.groupBox_coyote_display) + self.label_coyote_graph_window.setObjectName(u"label_coyote_graph_window") + + self.formLayout_coyote_display.setWidget(0, QFormLayout.ItemRole.LabelRole, self.label_coyote_graph_window) + + self.coyote_graph_window = QDoubleSpinBox(self.groupBox_coyote_display) + self.coyote_graph_window.setObjectName(u"coyote_graph_window") + self.coyote_graph_window.setDecimals(1) + self.coyote_graph_window.setMinimum(0.100000000000000) + self.coyote_graph_window.setMaximum(10.000000000000000) + self.coyote_graph_window.setSingleStep(0.100000000000000) + self.coyote_graph_window.setValue(3.000000000000000) + + self.formLayout_coyote_display.setWidget(0, QFormLayout.ItemRole.FieldRole, self.coyote_graph_window) + + self.label_coyote_debug_logging = QLabel(self.groupBox_coyote_display) + self.label_coyote_debug_logging.setObjectName(u"label_coyote_debug_logging") + + self.formLayout_coyote_display.setWidget(1, QFormLayout.ItemRole.LabelRole, self.label_coyote_debug_logging) + + self.coyote_debug_logging = QCheckBox(self.groupBox_coyote_display) + self.coyote_debug_logging.setObjectName(u"coyote_debug_logging") + + self.formLayout_coyote_display.setWidget(1, QFormLayout.ItemRole.FieldRole, self.coyote_debug_logging) + + + self.verticalLayout_coyote.addWidget(self.groupBox_coyote_display) + + self.tabWidget.addTab(self.tab_coyote, "") self.tab_media_settings = QWidget() self.tab_media_settings.setObjectName(u"tab_media_settings") self.verticalLayout_6 = QVBoxLayout(self.tab_media_settings) @@ -690,9 +828,7 @@ def setupUi(self, PreferencesDialog): QWidget.setTabOrder(self.focstim_refresh_serial_devices, self.focstim_ssid) QWidget.setTabOrder(self.focstim_ssid, self.focstim_password) QWidget.setTabOrder(self.focstim_password, self.focstim_sync) - QWidget.setTabOrder(self.focstim_sync, self.focstim_ip) - QWidget.setTabOrder(self.focstim_ip, self.focstim_read_ip) - QWidget.setTabOrder(self.focstim_read_ip, self.focstim_use_teleplot) + QWidget.setTabOrder(self.focstim_sync, self.focstim_use_teleplot) QWidget.setTabOrder(self.focstim_use_teleplot, self.focstim_teleplot_prefix) QWidget.setTabOrder(self.focstim_teleplot_prefix, self.focstim_dump_notifications) QWidget.setTabOrder(self.focstim_dump_notifications, self.tcp_port) @@ -794,25 +930,39 @@ def retranslateUi(self, PreferencesDialog): self.label_14.setText(QCoreApplication.translate("PreferencesDialog", u"Serial port", None)) self.focstim_refresh_serial_devices.setText(QCoreApplication.translate("PreferencesDialog", u"Refresh", None)) self.groupBox_10.setTitle(QCoreApplication.translate("PreferencesDialog", u"Network", None)) - self.label_22.setText(QCoreApplication.translate("PreferencesDialog", u"IP", None)) - self.focstim_sync.setText(QCoreApplication.translate("PreferencesDialog", u"Sync with device", None)) + self.focstim_sync.setText(QCoreApplication.translate("PreferencesDialog", u"Upload ssid/password", None)) self.label_21.setText(QCoreApplication.translate("PreferencesDialog", u"Password", None)) - self.focstim_read_ip.setText(QCoreApplication.translate("PreferencesDialog", u"Read from device", None)) self.label_20.setText(QCoreApplication.translate("PreferencesDialog", u"SSID", None)) + self.label_22.setText(QCoreApplication.translate("PreferencesDialog", u"IP", None)) + self.focstim_read_ip.setText(QCoreApplication.translate("PreferencesDialog", u"Read from device", None)) self.groupBox_9.setTitle(QCoreApplication.translate("PreferencesDialog", u"Advanced", None)) - self.label_18.setText(QCoreApplication.translate("PreferencesDialog", u"Dump notifications to file", None)) - self.focstim_dump_notifications.setText("") + self.label_15.setText(QCoreApplication.translate("PreferencesDialog", u"Use teleplot", None)) + self.focstim_use_teleplot.setText("") #if QT_CONFIG(tooltip) self.label_16.setToolTip(QCoreApplication.translate("PreferencesDialog", u"Useful if you have multiple FOC-Stim boxes", None)) #endif // QT_CONFIG(tooltip) self.label_16.setText(QCoreApplication.translate("PreferencesDialog", u"teleplot prefix (?)", None)) - self.label_15.setText(QCoreApplication.translate("PreferencesDialog", u"Use teleplot", None)) - self.focstim_use_teleplot.setText("") + self.label_18.setText(QCoreApplication.translate("PreferencesDialog", u"Dump notifications to file", None)) + self.focstim_dump_notifications.setText("") self.tabWidget.setTabText(self.tabWidget.indexOf(self.tab_foc), QCoreApplication.translate("PreferencesDialog", u"FOC-Stim", None)) self.groupBox_4.setTitle(QCoreApplication.translate("PreferencesDialog", u"NeoStim", None)) self.neostim_refresh_serial_devices.setText(QCoreApplication.translate("PreferencesDialog", u"Refresh", None)) self.label_17.setText(QCoreApplication.translate("PreferencesDialog", u"Serial port", None)) self.tabWidget.setTabText(self.tabWidget.indexOf(self.tab_neostim), QCoreApplication.translate("PreferencesDialog", u"NeoStim", None)) + self.groupBox_coyote_params.setTitle(QCoreApplication.translate("PreferencesDialog", u"Device", None)) + self.label_coyote_channel_a_limit.setText(QCoreApplication.translate("PreferencesDialog", u"Channel A Limit", None)) + self.label_coyote_channel_b_limit.setText(QCoreApplication.translate("PreferencesDialog", u"Channel B Limit", None)) + self.label_coyote_channel_a_freq_balance.setText(QCoreApplication.translate("PreferencesDialog", u"Channel A Freq Balance", None)) + self.label_coyote_channel_b_freq_balance.setText(QCoreApplication.translate("PreferencesDialog", u"Channel B Freq Balance", None)) + self.label_coyote_channel_a_intensity_balance.setText(QCoreApplication.translate("PreferencesDialog", u"Channel A Intensity Balance", None)) + self.label_coyote_channel_b_intensity_balance.setText(QCoreApplication.translate("PreferencesDialog", u"Channel B Intensity Balance", None)) + self.groupBox_coyote_algorithm.setTitle(QCoreApplication.translate("PreferencesDialog", u"Algorithm", None)) + self.label_coyote_max_intensity_change_per_pulse.setText(QCoreApplication.translate("PreferencesDialog", u"Max Intensity Change per Pulse (%)", None)) + self.groupBox_coyote_display.setTitle(QCoreApplication.translate("PreferencesDialog", u"Display", None)) + self.label_coyote_graph_window.setText(QCoreApplication.translate("PreferencesDialog", u"Graph Window (s)", None)) + self.label_coyote_debug_logging.setText(QCoreApplication.translate("PreferencesDialog", u"Debug Logging", None)) + self.coyote_debug_logging.setText("") + self.tabWidget.setTabText(self.tabWidget.indexOf(self.tab_coyote), QCoreApplication.translate("PreferencesDialog", u"Coyote", None)) self.groupBox_3.setTitle(QCoreApplication.translate("PreferencesDialog", u"MPC-HC", None)) self.label_31.setText(QCoreApplication.translate("PreferencesDialog", u"address:port", None)) self.mpc_reload.setText(QCoreApplication.translate("PreferencesDialog", u"...", None)) diff --git a/qt_ui/settings.py b/qt_ui/settings.py index 1ffbcbe..21ff739 100644 --- a/qt_ui/settings.py +++ b/qt_ui/settings.py @@ -166,6 +166,29 @@ def set(self, value): neostim_serial_port = Setting("neostim/serial_port", '', str) +coyote_channel_a_limit = Setting("coyote/channel_a_limit", 200, int) +coyote_channel_b_limit = Setting("coyote/channel_b_limit", 200, int) +coyote_channel_a_freq_balance = Setting("coyote/channel_a_freq_balance", 160, int) +coyote_channel_b_freq_balance = Setting("coyote/channel_b_freq_balance", 160, int) +coyote_channel_a_intensity_balance = Setting("coyote/channel_a_intensity_balance", 0, int) +coyote_channel_b_intensity_balance = Setting("coyote/channel_b_intensity_balance", 0, int) +coyote_channel_a_strength_max = Setting("coyote/channel_a_strength_max", 50, int) +coyote_channel_a_freq_min = Setting("coyote/channel_a_freq_min", 70, int) +coyote_channel_a_freq_max = Setting("coyote/channel_a_freq_max", 100, int) +coyote_channel_b_strength_max = Setting("coyote/channel_b_strength_max", 50, int) +coyote_channel_b_freq_min = Setting("coyote/channel_b_freq_min", 30, int) +coyote_channel_b_freq_max = Setting("coyote/channel_b_freq_max", 60, int) +coyote_max_intensity_change_per_pulse = Setting("coyote/max_intensity_change_per_pulse", 1.0, float) +coyote_debug_logging = Setting("coyote/debug_logging", False, bool) +coyote_graph_window = Setting("coyote/graph_window", 3.0, float) +coyote_queue_horizon_seconds = Setting("coyote/queue_horizon_seconds", 0.75, float) +coyote_packet_margin = Setting("coyote/packet_margin", 0.8, float) +coyote_texture_min_hz = Setting("coyote/texture_min_hz", 0.5, float) +coyote_texture_max_hz = Setting("coyote/texture_max_hz", 5.0, float) +coyote_texture_depth_fraction = Setting("coyote/texture_depth_fraction", 0.5, float) +coyote_jitter_limit_fraction = Setting("coyote/jitter_limit_fraction", 0.5, float) +coyote_residual_bound = Setting("coyote/residual_bound", 0.49, float) + # Pattern preferences - we'll store this as a JSON string and convert to dict import json diff --git a/requirements.txt b/requirements.txt index b810596..13a7abb 100644 --- a/requirements.txt +++ b/requirements.txt @@ -9,4 +9,5 @@ msdparser # protoletariat # dev only protobuf>=6.0.0 pystream-protobuf -stm32loader @ git+https://github.com/diglet48/stm32loader@feat/device-table \ No newline at end of file +bleak # Coyote only +stm32loader @ git+https://github.com/diglet48/stm32loader@feat/device-table diff --git a/stim_math/audio_gen/params.py b/stim_math/audio_gen/params.py index a84c1aa..8234170 100644 --- a/stim_math/audio_gen/params.py +++ b/stim_math/audio_gen/params.py @@ -170,6 +170,34 @@ class NeoStimParams: debug: AbstractAxis # NeoStimDebugSettings + +from qt_ui import settings + +@dataclass +class CoyoteChannelParams: + minimum_frequency: settings.Setting + maximum_frequency: settings.Setting + maximum_strength: settings.Setting + vibration: VibrationParams # TODO: modulate channel A/B freq + +@dataclass +class CoyoteAlgorithmParams: + position: ThreephasePositionParams + transform: ThreephasePositionTransformParams + calibrate: ThreephaseCalibrationParams + volume: VolumeParams + carrier_frequency: AbstractAxis # Hz + pulse_frequency: AbstractAxis # Hz + pulse_width: AbstractAxis # carrier cycles + pulse_interval_random: AbstractAxis + pulse_rise_time: AbstractAxis + max_intensity_change_per_pulse: settings.Setting + + channel_a: CoyoteChannelParams + channel_b: CoyoteChannelParams + + + @dataclass class SafetyParams: minimum_carrier_frequency: float