pulsepal

 399class PulsePalDevice:
 400    """A class to control a Pulse Pal device on a USB serial port.
 401
 402    Creating an instance opens the serial port, exchanges a handshake
 403    with the device, verifies that its firmware is supported, reads the
 404    device properties into `PulsePalDevice.info`, and programs the
 405    device with the default parameters (see
 406    `PulsePalDevice.set_default_params`).
 407
 408    ```python
 409    from pulsepal import PulsePalDevice
 410
 411    P = PulsePalDevice("COM3")
 412    P.phase1_voltage[1] = 5
 413    P.trigger(1)
 414    P.close()
 415    ```
 416
 417    Here, replace "COM3" with Pulse Pal's USB serial port name.
 418    To view a list of available ports, use
 419    `PulsePalDevice.serialportlist()`.
 420    Pulse Pal's port may not be visible if it is connected to another
 421    instance of PulsePalDevice or an external application. Use
 422    `PulsePalDevice.serialportlist("all")` to view all ports.
 423    If you see multiple available ports, disconnect Pulse Pal's USB plug
 424    and run serialportlist() again. Notice which port disappears from the
 425    list.
 426
 427    PulsePalDevice is also a context manager, which closes the connection on
 428    exit even if an error is raised:
 429
 430    ```python
 431    with PulsePalDevice("COM3") as P:
 432        P.trigger(1)
 433    ```
 434
 435    The attributes below named after Pulse Pal parameters are lists
 436    indexed by channel number, with index 0 unused (see "Parameters"
 437    above). `print(P)` shows them all, and `PulsePalDevice.export_params`
 438    returns them as a dict.
 439    """
 440
 441    port: "serial.Serial"
 442    """The open `serial.Serial` port connected to the device."""
 443
 444    info: DeviceInfo
 445    """Properties of the connected device. See `DeviceInfo`."""
 446
 447    is_biphasic = _channel_setting("is_biphasic", """Pulse shape per channel: `False` for monophasic, `True` for biphasic.
 448
 449    Monophasic pulses use only the phase 1 parameters. Biphasic pulses
 450    follow phase 1 with `PulsePalDevice.inter_phase_interval` and then
 451    phase 2.
 452    """)
 453
 454    phase1_voltage = _channel_setting("phase1_voltage", """Voltage of the first phase of each pulse, in volts [-10, 10].""")
 455
 456    phase2_voltage = _channel_setting("phase2_voltage", """Voltage of the second phase of each pulse, in volts [-10, 10].
 457
 458    Used only when `PulsePalDevice.is_biphasic` is `True` for the channel.
 459    """)
 460
 461    resting_voltage = _channel_setting("resting_voltage", """Voltage held between pulses, in volts [-10, 10].
 462
 463    A new resting voltage reaches an idle channel's output at once. A
 464    channel playing a pulse train keeps playing it, and moves to the new
 465    resting voltage at its next transition to rest.
 466    """)
 467
 468    phase1_duration = _channel_setting("phase1_duration", """Duration of the first phase of each pulse, in seconds.
 469
 470    At least `DeviceInfo.min_pulse_width_us`.
 471    """)
 472
 473    inter_phase_interval = _channel_setting("inter_phase_interval", """Interval between the two phases of a biphasic pulse, in seconds.
 474
 475    The channel rests at `PulsePalDevice.resting_voltage` during the
 476    interval. Used only when `PulsePalDevice.is_biphasic` is `True`.
 477    """)
 478
 479    phase2_duration = _channel_setting("phase2_duration", """Duration of the second phase of each pulse, in seconds.
 480
 481    Used only when `PulsePalDevice.is_biphasic` is `True` for the channel.
 482    At least `DeviceInfo.min_pulse_width_us`.
 483    """)
 484
 485    inter_pulse_interval = _channel_setting("inter_pulse_interval", """Interval from the end of one pulse to the start of the next, in
 486    seconds. At least `DeviceInfo.min_pulse_width_us`.
 487
 488    It is not the pulse period: 1 ms pulses 49 ms apart play 20 pulses per
 489    second.
 490    """)
 491
 492    burst_duration = _channel_setting("burst_duration", """Duration of each burst of pulses, in seconds.
 493
 494    Set to `0` to disable bursts, so that pulses continue for the whole
 495    pulse train. A pulse starts only if it ends before the burst does:
 496    its first phase, or for a biphasic pulse the whole pulse, so that
 497    the end of a burst never cuts off a second phase.
 498    """)
 499
 500    inter_burst_interval = _channel_setting("inter_burst_interval", """Interval between bursts of pulses, in seconds.
 501
 502    The channel rests at `PulsePalDevice.resting_voltage` between
 503    bursts. Ignored when `PulsePalDevice.burst_duration` is `0`.
 504    """)
 505
 506    pulse_train_duration = _channel_setting("pulse_train_duration", """Total duration of the pulse train, in seconds.
 507
 508    At least `DeviceInfo.min_pulse_width_us`. The end of the train cuts
 509    short a monophasic pulse still playing. A biphasic pulse starts only
 510    if it can end by the end of the train, so that it keeps its second
 511    phase. A train's end is set when it starts: a new duration applies
 512    from the next train.
 513    """)
 514
 515    pulse_train_delay = _channel_setting("pulse_train_delay", """Delay from the trigger to the onset of the pulse train, in
 516    seconds.""")
 517
 518    link_trigger_channel1 = _channel_setting("link_trigger_channel1", """Whether each output channel is triggered by trigger channel 1.
 519
 520    `True` links the output channel to trigger channel 1, `False` unlinks
 521    it. By default no output channel is linked to either trigger channel,
 522    so a TTL starts nothing until a script links it.
 523    """)
 524
 525    link_trigger_channel2 = _channel_setting("link_trigger_channel2", """Whether each output channel is triggered by trigger channel 2.
 526
 527    `True` links the output channel to trigger channel 2, `False` unlinks
 528    it.
 529    """)
 530
 531    custom_train_id = _channel_setting("custom_train_id", """Custom pulse train played by each output channel.
 532
 533    `0` plays the parametrically defined train. `1` or higher plays the
 534    matching custom train, previously loaded with
 535    `PulsePalDevice.send_custom_pulse_train` or
 536    `PulsePalDevice.send_custom_waveform`: 1-4 on Pulse Pal 3, 1-2 on
 537    Pulse Pal 2 (`DeviceInfo.n_custom_pulse_trains`).
 538    """)
 539
 540    custom_train_target = _channel_setting("custom_train_target", """What the times of a custom train mark.
 541
 542    `"Pulses"` if each time is the onset of a pulse, `"Bursts"` if each
 543    time is the onset of a burst of pulses, which lasts
 544    `PulsePalDevice.burst_duration`. A burst whose time comes while the
 545    burst before plays is dropped, so that every burst played is on
 546    time. `"Bursts"` needs a burst duration above 0: with
 547    `PulsePalDevice.auto_sync` on, set the burst duration first, or set
 548    both in a `PulsePalDevice.batch` block.
 549    """)
 550
 551    custom_train_loop = _channel_setting("custom_train_loop", """Whether a custom train repeats.
 552
 553    `True` loops the custom train until
 554    `PulsePalDevice.pulse_train_duration` has elapsed, `False` plays it
 555    once. Each pass starts as the last pulse (or burst, for a train
 556    targeting bursts) of the pass before ends.
 557    `PulsePalDevice.continuous_loop` does not apply to custom trains.
 558    """)
 559
 560    continuous_loop = _channel_setting("continuous_loop", """Whether each output channel plays its pulse train until stopped.
 561
 562    `False` plays the pulse train once per trigger, for
 563    `PulsePalDevice.pulse_train_duration`. `True` plays it until the
 564    channel is stopped, ignoring `PulsePalDevice.pulse_train_duration`.
 565    For parametric trains only: a custom train ignores it, and plays as
 566    `PulsePalDevice.custom_train_loop` says. Requires firmware v22 or
 567    newer. The device does not report it, so
 568    `PulsePalDevice.sync_from_device` leaves it as it is.
 569    """)
 570
 571    trigger_mode = _channel_setting("trigger_mode", """Response of each trigger channel to an incoming TTL pulse.
 572
 573    Indexed by trigger channel number, 1 or 2 (index 0 is unused). The
 574    modes, also listed in `DeviceInfo.trigger_modes`, are:
 575
 576    - `"Normal"`: a rising edge starts the pulse trains of the linked
 577      output channels, and edges during a train are ignored.
 578    - `"Toggle"`: as `"Normal"`, but a rising edge during a train stops
 579      it.
 580    - `"Gated"`: the trains play only while the trigger TTL is high.
 581    - `"Param Sync"` (Pulse Pal 3 only): a TTL rising edge starts and
 582      stops nothing. It loads the parameters most recently sent by
 583      `PulsePalDevice.sync_to_device` (or at the end of a
 584      `PulsePalDevice.batch` block), and the custom trains loaded since
 585      the previous edge. This is how the next trial's parameters and
 586      custom trains are sent during the current trial and applied the
 587      instant it starts.
 588
 589    The integer codes 0-3 are accepted too.
 590
 591    An output channel that is idle at a param sync edge takes its new
 592    parameters in the timer cycle the edge is detected. One that is
 593    playing a pulse train finishes that train on the parameters it
 594    started with, and takes the new ones the moment it ends, so a train
 595    that runs past the end of a trial keeps one shape throughout and the
 596    next trigger plays a whole train with the new parameters. A channel
 597    in continuous loop mode has no train end, so it keeps its parameters
 598    until something stops it.
 599
 600    While either trigger channel is in param sync mode, **only
 601    `PulsePalDevice.sync_to_device` (and the end of a
 602    `PulsePalDevice.batch` block) and the custom train loads are held
 603    back**: with `PulsePalDevice.auto_sync` on, every assignment still
 604    programs the device immediately. So leaving param sync mode means
 605    assigning `trigger_mode` with `auto_sync` on; a trigger mode sent by
 606    `PulsePalDevice.sync_to_device` does not take effect until a sync
 607    edge arrives. A custom train loaded in param sync mode
 608    (`PulsePalDevice.send_custom_pulse_train`) replaces the old one at
 609    the edge: channels playing the old one at the edge finish their
 610    trains on it. Leaving param sync mode discards parameters still
 611    waiting for an edge, but makes a custom train still waiting current
 612    at once.
 613
 614    A param sync channel's links to output channels are ignored. To
 615    start a pulse train on the same edge, wire the TTL to the other
 616    trigger channel as well: both edges arrive in the same timer cycle,
 617    and the parameters are loaded first.
 618
 619    Connecting a new `PulsePalDevice`, and
 620    `PulsePalDevice.set_default_params`, take both trigger channels out
 621    of param sync mode, so that the default parameters reach the device
 622    instead of waiting for a TTL.
 623
 624    ```python
 625    P.trigger_mode[2] = "Param Sync"   # sent at once
 626    with P.batch():
 627        P.phase1_voltage[1] = 2.5
 628        P.phase1_duration[1] = 0.002
 629    # Stored by the device: the next rising edge on trigger channel 2 applies it
 630    ```
 631    """)
 632
 633    _CURRENT_FIRMWARE_VERSION = 22
 634    _OLDEST_FIRMWARE_SUPPORTED = 21
 635    _BAUD_RATE = 12000000  # USB serial ignores the baud rate
 636
 637    _OP_MENU_BYTE = _common.OP_MENU_BYTE
 638    _HANDSHAKE_OPCODE = _common.OP_HANDSHAKE
 639    _HANDSHAKE_DEVICE_CYCLES_OPCODE = _common.OP_HANDSHAKE_DEVICE_CYCLES
 640    _HANDSHAKE_DEVICE_CYCLES_TIMEOUT = 0.3  # Seconds to wait for a reply to op 103 before trying op 72
 641    _HANDSHAKE_RESPONSE = 75  # 'K'
 642    _DAC_BITMAX = 65535
 643    _PARAM_MESSAGE_BYTES = 178  # Length of the parameter set sent by op 93
 644    _STATUS_MESSAGE_BYTES = 29  # Length of the reply to op 101: playing channels, then 7 uint32 counters
 645    # Phase 1 and 2 durations, inter-pulse interval and train duration last at least this many
 646    # timer cycles, by hardware version, as in the MATLAB and C++ classes and the joystick menu.
 647    # Pulse Pal 2's trigger channels read their inputs once per cycle, so a one cycle pulse from
 648    # another Pulse Pal could fall between two reads. Pulse Pal 3's also latch every edge.
 649    _MIN_PULSE_CYCLES = {2: 2, 3: 1}
 650
 651    _ENDIANNESS = "<"
 652    _STRUCT_FORMATS = {
 653        "uint8": "B",
 654        "int8": "b",
 655        "uint16": "H",
 656        "int16": "h",
 657        "uint32": "I",
 658        "int32": "i",
 659    }
 660    _TYPE_RANGES = {
 661        "uint8": (0, 2**8 - 1),
 662        "int8": (-(2**7), 2**7 - 1),
 663        "uint16": (0, 2**16 - 1),
 664        "int16": (-(2**15), 2**15 - 1),
 665        "uint32": (0, 2**32 - 1),
 666        "int32": (-(2**31), 2**31 - 1),
 667    }
 668
 669    _reply_pending = None  # The op code of a command whose reply has not been read in full yet (see _write_raw())
 670    _unsent = frozenset()  # The settings assigned with auto_sync off since the parameters were last sent or read
 671
 672    def __init__(self, port_name=None, *, serial_number=None, timeout=10):
 673        """Open a connection to a Pulse Pal device.
 674
 675        Opens the serial port, exchanges the handshake, verifies the
 676        firmware version, reads the device properties into
 677        `PulsePalDevice.info`, and programs the device with the default
 678        parameters.
 679
 680        ```python
 681        P = PulsePalDevice("COM3")
 682        P = PulsePalDevice(serial_number="20345420")  # Wherever it is plugged in
 683        ```
 684
 685        Args:
 686            port_name: USB serial port for the Pulse Pal device, such as
 687                `COM3` on Windows or `/dev/ttyACM0` on Linux.
 688            serial_number: Instead of `port_name`, the device's USB serial
 689                number (`DeviceInfo.serial_number`), which finds its port.
 690                Pulse Pal 3 only: Pulse Pal 2 has none.
 691            timeout: Serial read timeout, in seconds.
 692
 693        Firmware newer than this module knows (`v22`) is used with a
 694        warning: new firmware only adds commands, so everything here still
 695        works with it. Update the package to use what is new.
 696
 697        Raises:
 698            PulsePalError: If the device does not return the expected
 699                handshake (for example, it runs Wave Pal or Synth Pal
 700                firmware), or its firmware is older than v21, or no port
 701                has the serial number given. The port is closed again
 702                before any error is raised.
 703            serial.SerialException: If the serial port cannot be opened,
 704                e.g. because another program has it open (on Linux and
 705                macOS, another session of these classes, which lock it).
 706        """
 707        self._closed = True
 708        self._gui = None
 709        self._auto_sync = True
 710        self.info = DeviceInfo()
 711        self._create_settings()
 712        port_name = _common.port_name_from(port_name, serial_number, "PulsePalDevice")
 713        self.port = serial.Serial(
 714            port_name,
 715            self._BAUD_RATE,
 716            timeout=timeout,
 717            rtscts=True,
 718            # Windows opens a port for one program only. Linux and macOS let a second program open it too, and two
 719            # sessions' bytes would then interleave; exclusive locks it (flock), so a second session that also locks
 720            # it, such as another of these classes, gets an error instead
 721            exclusive=True,
 722        )
 723        self._closed = False
 724        try:
 725            self._start_session(port_name)
 726        except BaseException:
 727            # Otherwise the port stays open until the object is garbage
 728            # collected, and a second attempt, or WavePalDevice, cannot open
 729            # it. The device is in an unknown state, so it is not sent the
 730            # disconnect op.
 731            self._close(send_disconnect=False)
 732            raise
 733
 734    def _create_settings(self):
 735        """Create the parameter lists, holding the default values (nothing is sent)."""
 736        for code, (name, _) in _OUTPUT_PARAMETERS.items():
 737            setattr(self, "_" + name, ChannelSettings(
 738                name, [_DEFAULT_OUTPUT_VALUES[name]] * 4, self, f"_apply_{name}"))
 739        self._trigger_mode = ChannelSettings(
 740            "trigger_mode", ["Normal"] * 2, self, "_apply_trigger_mode")
 741
 742    def _start_session(self, port_name):
 743        """Handshake, check the firmware, and program the defaults."""
 744        # Discard anything left unread by an earlier session (one stopped mid-reply, for example):
 745        # it would be read as the handshake's reply. A reply the device sends after this, to a
 746        # command it had not reached yet, is skipped by read_handshake_reply().
 747        self.port.reset_input_buffer()
 748        # Op 103: the session's times are in the device's own timer cycles (10 us on Pulse Pal 3). Firmware from before
 749        # October 2026, Wave Pal and Synth Pal ignore it, and answer op 72, whose times are in 50 us cycles. A device that
 750        # answers op 103 too late counts the session's times in 50 us cycles too, after the op 72 that follows; either
 751        # way op 94 reports the session's unit, which the class then uses.
 752        self._write_serial((self._OP_MENU_BYTE, self._HANDSHAKE_DEVICE_CYCLES_OPCODE), "uint8")
 753        reply = _common.read_handshake_reply(self.port, timeout=self._HANDSHAKE_DEVICE_CYCLES_TIMEOUT)
 754        if reply is None:
 755            self._write_serial((self._OP_MENU_BYTE, self._HANDSHAKE_OPCODE), "uint8")
 756            reply = _common.read_handshake_reply(self.port)
 757        if reply is None:
 758            raise PulsePalError(
 759                f"No reply to the handshake from the device on {port_name}. Is it a Pulse Pal? "
 760                f"{_common.NO_REPLY_HINT}"
 761            )
 762        handshake, firmware_version = reply
 763        if handshake != self._HANDSHAKE_RESPONSE:
 764            if handshake in _common.FIRMWARE_BY_HANDSHAKE_REPLY:
 765                raise _common.other_firmware_error(handshake, firmware_version, port_name, "Pulse Pal")
 766            raise PulsePalError(
 767                "Incorrect handshake returned. Expected "
 768                f"{self._HANDSHAKE_RESPONSE}, received {handshake}."
 769            )
 770
 771        if firmware_version < self._OLDEST_FIRMWARE_SUPPORTED:
 772            raise PulsePalError(
 773                "Old firmware detected, v"
 774                f"{firmware_version}. v{self._OLDEST_FIRMWARE_SUPPORTED} or "
 775                "newer is required."
 776            )
 777        if firmware_version > self._CURRENT_FIRMWARE_VERSION:
 778            # New firmware only adds commands (see /Firmware/PROTOCOL.md), so this class still works with it
 779            warnings.warn(
 780                f"Pulse Pal firmware v{firmware_version} is newer than this pulsepal package knows "
 781                f"(v{self._CURRENT_FIRMWARE_VERSION}). It works with it, but update the package to use what "
 782                "is new.", stacklevel=3)
 783        if firmware_version < self._CURRENT_FIRMWARE_VERSION:
 784            warnings.warn(
 785                f"Pulse Pal firmware v{firmware_version} is supported, but v"
 786                f"{self._CURRENT_FIRMWARE_VERSION} is available: see /MATLAB/FirmwareLoader.", stacklevel=3)
 787        self.info.firmware_version = firmware_version
 788
 789        if self.info.firmware_version > 21:
 790            self._write_serial((self._OP_MENU_BYTE, 94), "uint8")
 791            self.info.hardware_version = self._read_serial(1, "uint8")
 792            self.info.cycle_period_us = self._read_serial(1, "uint32")  # The session's time unit (see op 103 above)
 793            # Not 1 / (period / 1e6): at 10 us that is 99999.99999999999, and times read back from the device would
 794            # convert to 0.00010000000000000002 s instead of 0.0001 s
 795            self.info.cycle_frequency = 1000000 / self.info.cycle_period_us
 796            self.info.n_custom_pulse_trains = self._read_serial(1, "uint8")
 797            self.info.max_custom_pulses = self._read_serial(1, "uint32")
 798            self._write_serial((self._OP_MENU_BYTE, 100), "uint8")  # Circuit board revision
 799            revision = self.port.read(1)
 800            if len(revision) < 1:
 801                raise PulsePalError(
 802                    f"The Pulse Pal on {port_name} did not report its circuit board revision."
 803                )
 804            self._reply_pending = None  # Read without _read_raw() (see _write_raw())
 805            self.info.circuit_board_revision = revision[0]
 806            self._write_serial((self._OP_MENU_BYTE, _common.OP_SERIAL_NUMBER), "uint8")
 807            self.info.serial_number = _common.read_serial_number(self._read_raw)
 808        else:
 809            self.info.hardware_version = 2
 810            self.info.circuit_board_revision = 0
 811            self.info.serial_number = ""  # v21 runs on Pulse Pal 2, which has none
 812            self.info.cycle_period_us = 50
 813            self.info.cycle_frequency = 20000
 814            self.info.n_custom_pulse_trains = 2
 815            self.info.max_custom_pulses = 5000
 816        self.info.min_pulse_width_us = (
 817            self._MIN_PULSE_CYCLES[self.info.hardware_version] * self.info.cycle_period_us
 818        )
 819        if not self._has_param_sync():
 820            self.info.trigger_modes = TRIGGER_MODES[:3]
 821
 822        # Client name op + "PYTHON" in ASCII.
 823        self._write_serial(
 824            (self._OP_MENU_BYTE, 89, 80, 89, 84, 72, 79, 78),
 825            "uint8",
 826        )
 827
 828        self.set_default_params()
 829
 830    @staticmethod
 831    def serialportlist(ports_to_list="available"):
 832        """Return the names of the USB serial ports on this computer.
 833
 834        Called on the class, without connecting to a device, to find the
 835        port name to pass to `PulsePalDevice`:
 836
 837        ```python
 838        from pulsepal import PulsePalDevice
 839
 840        ports = PulsePalDevice.serialportlist()
 841        P = PulsePalDevice(ports[0])
 842        ```
 843
 844        Args:
 845            ports_to_list: `available` to list only the ports that are
 846                not already in use, or `all` to list every USB serial
 847                port. Not case sensitive.
 848
 849        Returns:
 850            Sorted list of port names, such as `["COM3", "COM7"]` on
 851            Windows or `["/dev/ttyACM0"]` on Linux.
 852
 853        Raises:
 854            PulsePalError: If `ports_to_list` is not `available` or
 855                `all`.
 856        """
 857        return _common.serialportlist(ports_to_list)
 858
 859    # ------------------------------------------------------------------
 860    # Programming the device
 861    # ------------------------------------------------------------------
 862
 863    @property
 864    def auto_sync(self):
 865        """Whether assigning a parameter programs the device at once.
 866
 867        `True` (the default): each assignment, such as
 868        `P.phase1_voltage[1] = 5`, programs the device at once, also while
 869        a trigger channel is in param sync mode. It sends the channels
 870        whose value it changes, and nothing for a channel that already
 871        holds the value assigned: the other channels keep what the device
 872        holds, e.g. values set with its joystick (read them with
 873        `PulsePalDevice.sync_from_device`).
 874
 875        `False`: assignments change only this object's copy of the
 876        parameters, checking each value, and
 877        `PulsePalDevice.sync_to_device` sends all of them in one command.
 878        Turning it back on does not send changes made meanwhile, so it
 879        raises `pulsepal.PulsePalError` while there are any: call
 880        `PulsePalDevice.sync_to_device` first to send them, or
 881        `PulsePalDevice.sync_from_device` to discard them. A
 882        `PulsePalDevice.batch` block does it all for you. `close()` warns of
 883        changes never sent.
 884        """
 885        return self._auto_sync
 886
 887    @auto_sync.setter
 888    def auto_sync(self, value):
 889        value = to_bool(value, "auto_sync")
 890        if value and self._unsent:
 891            # Turned on, auto_sync would leave them unsent, and the device playing its old parameters
 892            discard = "sync_from_device() to discard them"
 893            if "continuous_loop" in self._unsent:
 894                discard += " (all but continuous_loop, which the device does not report)"
 895            raise PulsePalError(
 896                f"Changes made with auto_sync off have not been sent to the device: {self._unsent_names()}. Call "
 897                f"sync_to_device() to send them, or {discard}, before turning auto_sync on."
 898            )
 899        self._auto_sync = value
 900
 901    def batch(self):
 902        """Change several parameters, and send them in one command.
 903
 904        Returns a context manager. In its block, assignments change only
 905        this object's copy of the parameters (as with
 906        `PulsePalDevice.auto_sync` off), and when the block ends,
 907        `PulsePalDevice.sync_to_device` sends all of them at once. If the
 908        block raises an error, nothing is sent, and the parameters go
 909        back to what they were.
 910
 911        ```python
 912        with P.batch():
 913            P.phase1_voltage = [5, 5, 2.5, 2.5]
 914            P.phase1_duration[3] = 0.002
 915        ```
 916
 917        In param sync mode, the device stores the parameters for the next
 918        rising edge on the param sync channel (see
 919        `PulsePalDevice.trigger_mode`).
 920        """
 921        return _common.batch(self)
 922
 923    def sync_to_device(self):
 924        """Program the device with this object's copy of all parameters.
 925
 926        Sends every output and trigger parameter to the device in a single
 927        command. Use it after assignments made with
 928        `PulsePalDevice.auto_sync` off; with it on, they have already
 929        reached the device.
 930
 931        ```python
 932        P.auto_sync = False
 933        P.phase1_voltage[1:5] = [5] * 4
 934        P.sync_to_device()
 935        ```
 936
 937        On Pulse Pal 3, if either trigger channel is in param sync mode
 938        (`PulsePalDevice.trigger_mode` `"Param Sync"`), the device stores
 939        the parameters instead of programming them, and loads them on the
 940        next rising edge of that channel. The device still checks every
 941        value. This is the only method whose effect is deferred that way.
 942
 943        Raises:
 944            PulsePalError: If a value is out of range for its parameter
 945                (nothing is sent), or the device does not acknowledge the
 946                command.
 947        """
 948        self._check_sync()
 949        self._send_sync()
 950
 951    def sync_from_device(self):
 952        """Read all parameters from the device into this object's copy.
 953
 954        Overwrites the parameter lists with the program currently on the
 955        device, e.g. after it has been reprogrammed from its thumb
 956        joystick. `PulsePalDevice.continuous_loop` is left as it is: the
 957        device does not report it. Changes made with
 958        `PulsePalDevice.auto_sync` off and not yet sent are discarded, but
 959        for `PulsePalDevice.continuous_loop`.
 960
 961        Requires firmware v22 or newer.
 962
 963        Raises:
 964            PulsePalError: If the connected firmware is older than v22,
 965                or the device does not return the full parameter set.
 966        """
 967        self._require_firmware(22, "sync_from_device()")
 968        for name, values in self._read_device_params().items():
 969            getattr(self, name)._store(values)
 970        self._unsent = self._unsent & {"continuous_loop"}  # Not read: a change to it is still unsent
 971
 972    def set_default_params(self):
 973        """Program the device with the default parameters.
 974
 975        The defaults are monophasic +5 V pulses of 0.1 ms, 1 ms apart
 976        (about 909 a second), for 1 second, resting at 0 V, on all four
 977        output channels; for biphasic pulses, a phase 2 of -5 V and 0.1 ms
 978        after 0.1 ms at rest. No output channel is linked to a trigger
 979        channel, so a TTL starts nothing until a script links it, and both
 980        trigger channels are in normal mode. They are the device's own
 981        defaults, which it also loads at startup, after a comm failure
 982        and after formatting its microSD card. They are sent at once,
 983        also while `PulsePalDevice.auto_sync` is off. The constructor
 984        calls it.
 985        """
 986        for name, value in _DEFAULT_OUTPUT_VALUES.items():
 987            getattr(self, name)._store([value] * 4)
 988        self._trigger_mode._store(["Normal"] * 2)
 989        if self._has_param_sync():
 990            # A device left in param sync mode would store the sync below instead of running it,
 991            # leaving the device on its old program until a TTL arrived. A trigger mode sent on
 992            # its own is not deferred that way, so both trigger channels leave param sync mode
 993            # first. See PulsePalDevice.trigger_mode.
 994            self._send_trigger_modes(["Normal"] * 2, "set_default_params()")
 995        self._send_sync()
 996
 997    def export_params(self):
 998        """Return every parameter, as a dict of plain lists.
 999
1000        Keyed by parameter name (`DeviceInfo.output_parameter_names`, then
1001        `"trigger_mode"`), with one value per channel and no unused index
1002        0. It holds only numbers, booleans and names, so it can be saved
1003        with `json` and logged with your data, to record exactly what the
1004        device played. `PulsePalDevice.import_params` programs it again.
1005
1006        ```python
1007        import json
1008        with open("trial_params.json", "w") as f:
1009            json.dump(P.export_params(), f)
1010        ```
1011        """
1012        names = [*self.info.output_parameter_names, "trigger_mode"]
1013        return {name: list(getattr(self, name))[1:] for name in names}
1014
1015    def import_params(self, params):
1016        """Program the device with parameters exported by `PulsePalDevice.export_params`.
1017
1018        All of them are checked first, and then sent in one command, as
1019        at the end of a `PulsePalDevice.batch` block: in param sync mode,
1020        the device stores them for the next sync edge. Parameters missing
1021        from `params` keep their values. A name that is not a parameter
1022        is skipped with a warning, so that parameters exported by a newer
1023        version of this package, which may have more, can still be
1024        imported; if none is, an error is raised.
1025
1026        A whole program file, as the GUIs save it, is imported too: its
1027        parameters, not its custom trains (a warning says so, if it has
1028        any).
1029
1030        ```python
1031        with open("MyProgram.json") as f:
1032            P.import_params(json.load(f))
1033        ```
1034
1035        Args:
1036            params: A dict of parameter name to one value per channel (a
1037                list, tuple or NumPy array), as `export_params` returns,
1038                or a program file's contents. The MATLAB class's names
1039                (`phase1Voltage`) are accepted too, so that the parameters
1040                its `exportParams()` returns, saved with `jsonencode`, can
1041                be imported.
1042
1043        Raises:
1044            PulsePalError: If a value is invalid, or no name in `params` is
1045                a parameter. Nothing is sent, and the parameters keep the
1046                values they had.
1047        """
1048        params = _params_to_import(params)
1049        with self.batch():
1050            for name, values in params.items():
1051                setattr(self, name, values)
1052
1053    def set_fixed_voltage(self, channels, voltage):
1054        """Set output channels to a fixed voltage.
1055
1056        Each channel holds the voltage until it is set again or until a
1057        pulse train is triggered on it.
1058
1059        ```python
1060        P.set_fixed_voltage(4, 2.5)
1061        P.set_fixed_voltage([1, 3], -1)
1062        ```
1063
1064        Args:
1065            channels: Output channel number, 1-4, or several as a list,
1066                tuple or NumPy array.
1067            voltage: Voltage to set on each of them, in volts [-10, 10].
1068
1069        Raises:
1070            PulsePalError: If a channel number is not 1-4, the voltage is
1071                outside [-10, 10] (nothing is sent), or the device does not
1072                acknowledge the command.
1073        """
1074        channels = _common.channel_numbers(channels)
1075        voltage_bits = self._volts_to_bits(voltage, "voltage")
1076        for channel in channels:
1077            self._write_serial(
1078                (self._OP_MENU_BYTE, 79, channel),
1079                "uint8",
1080                voltage_bits,
1081                "uint16",
1082            )
1083            self._read_ack("set_fixed_voltage()")
1084
1085    def set_calibration(self, channel, voltage_offset):
1086        """Calibrate the zero code of an output channel.
1087
1088        The offset is added to every voltage the channel produces, to
1089        correct for DAC offset error. It is stored in the device's
1090        EEPROM and reloaded on boot, so it only needs to be set once.
1091
1092        Requires Pulse Pal hardware v3 or newer.
1093
1094        Args:
1095            channel: Output channel number, 1-4.
1096            voltage_offset: Offset to apply, in volts [-0.1, 0.1].
1097
1098        Raises:
1099            PulsePalError: If the connected hardware is older than v3,
1100                `channel` is not 1-4, `voltage_offset` is outside
1101                [-0.1, 0.1], or the device does not acknowledge the
1102                command.
1103        """
1104        if self.info.hardware_version < 3:
1105            raise PulsePalError(
1106                "set_calibration() requires hardware v3 or newer."
1107            )
1108        if not (isinstance(channel, numbers.Integral) and channel in (1, 2, 3, 4)):
1109            raise PulsePalError("channel must be 1, 2, 3 or 4")
1110        if not (isinstance(voltage_offset, numbers.Real) and -0.1 <= voltage_offset <= 0.1):  # Also refuses NaN
1111            raise PulsePalError(
1112                "voltage_offset for zero code calibration must be in range "
1113                "[-0.1, 0.1]"
1114            )
1115        # To the nearest DAC code, halfway values to the even one, as the MATLAB class rounds it
1116        voltage_bits = int(round(voltage_offset * (1 / (20 / 65536))))
1117        self._write_serial(
1118            (self._OP_MENU_BYTE, 96, channel - 1),
1119            "uint8",
1120            voltage_bits,
1121            "int16",
1122        )
1123        self._read_ack("set_calibration()")
1124
1125    def set_screen_saver(self, enabled, timeout=1800):
1126        """Switch the device's screen saver on or off, and set its timeout.
1127
1128        With the screen saver on, the device dims its screen once it has
1129        been left alone for `timeout` seconds: no command from the
1130        computer, no rising edge on a trigger channel, and no joystick
1131        click or push. The next of these brings the screen back. Both
1132        settings are stored in the device's EEPROM and kept through power
1133        cycles; the screen saver can also be switched on and off from the
1134        device's joystick menu. A new device has it on, with 1800 s.
1135
1136        The device saves the settings once no channel is playing. Saving
1137        pauses its timer, usually for about 20 us but, once in about 2000
1138        changes, for tens of milliseconds, which would delay a trigger.
1139        Change the settings before an experiment rather than during one.
1140
1141        Requires firmware v22 or newer. Only Pulse Pal 3 has a screen
1142        saver: Pulse Pal 2 accepts `enabled=False` only.
1143
1144        ```python
1145        P.set_screen_saver(True, 300)  # dim after 5 minutes
1146        P.set_screen_saver(False)
1147        ```
1148
1149        Args:
1150            enabled: True (or 1) to switch the screen saver on, False (or
1151                0) to switch it off.
1152            timeout: Seconds without activity before the screen dims, a
1153                whole number from 1 to 65535. It is sent with every call,
1154                so leaving it out sets 1800 s.
1155
1156        Raises:
1157            PulsePalError: If the firmware is older than v22, the
1158                screen saver is switched on with hardware older than v3,
1159                a value is out of range, or the device does not
1160                acknowledge the command.
1161        """
1162        self._require_firmware(22, "set_screen_saver()")
1163        enabled = to_bool(enabled, "enabled")
1164        if not (
1165            isinstance(timeout, numbers.Real)
1166            and not isinstance(timeout, bool)
1167            and math.isfinite(timeout)
1168            and timeout == int(timeout)
1169            and 1 <= timeout <= 65535
1170        ):
1171            raise PulsePalError(
1172                "timeout must be a whole number of seconds from 1 to 65535"
1173            )
1174        if enabled and (self.info.hardware_version or 2) < 3:
1175            raise PulsePalError(
1176                "The screen saver requires hardware v3 or newer: Pulse Pal 2 "
1177                "accepts set_screen_saver(False) only."
1178            )
1179        self._write_serial(
1180            (self._OP_MENU_BYTE, 99, int(enabled)),
1181            "uint8",
1182            int(timeout),
1183            "uint16",
1184        )
1185        self._read_ack("set_screen_saver()")
1186
1187    # ------------------------------------------------------------------
1188    # Custom trains, playback and settings files
1189    # ------------------------------------------------------------------
1190
1191    def send_custom_pulse_train(
1192        self,
1193        custom_train_id,
1194        pulse_times,
1195        voltages,
1196    ):
1197        """Load a custom pulse train onto the device.
1198
1199        A custom pulse train is an arbitrary list of pulse onset times
1200        and voltages, replacing the parametric pulse voltage and onset
1201        timing. Set `PulsePalDevice.custom_train_id` on an output channel
1202        to play the train there. Each pulse takes the channel's own phase
1203        durations; a biphasic pulse's second phase is its voltage with the
1204        sign reversed. A biphasic pulse always plays whole, and a pulse
1205        whose time comes while it plays is dropped, so that every pulse
1206        played is on time.
1207
1208        ```python
1209        P.send_custom_pulse_train(
1210            2, [0, 0.2, 0.5, 1], [8, 4, -3.5, -10]
1211        )
1212        P.custom_train_id[1] = 2
1213        ```
1214
1215        Loading a train stops the output channels playing it, and their
1216        next trigger plays the new train. On Pulse Pal 3, while a trigger
1217        channel is in param sync mode (see `PulsePalDevice.trigger_mode`),
1218        the device stores the train instead, as it stores the parameters
1219        `PulsePalDevice.sync_to_device` sends: channels go on playing the
1220        old train, and the next rising edge on the param sync channel
1221        makes the new one current. A channel playing the old train at that
1222        edge finishes it, and plays the new one from its next trigger. So
1223        the next trial's custom trains can be loaded during the current
1224        trial. Leaving param sync mode makes a stored train current at
1225        once.
1226
1227        Args:
1228            custom_train_id: Custom train to load, from 1 to
1229                `DeviceInfo.n_custom_pulse_trains` (2 on Pulse Pal 2,
1230                4 on Pulse Pal 3).
1231            pulse_times: Pulse onset times, in seconds, relative to the
1232                start of the train, increasing. Each is a multiple of
1233                `DeviceInfo.min_pulse_width_us` (10 us on Pulse Pal 3,
1234                100 us on Pulse Pal 2), from 0 to
1235                `MAX_TIME`. Accepts a list, tuple or 1-D NumPy array.
1236            voltages: Voltage of each pulse, in volts [-10, 10]. Must be
1237                the same length as `pulse_times`.
1238
1239        Raises:
1240            PulsePalError: If `custom_train_id` is out of range,
1241                `pulse_times` or `voltages` is not one-dimensional (a 2-D
1242                array is refused, not flattened), they differ in length,
1243                there are no pulses, or more than
1244                `DeviceInfo.max_custom_pulses`, a pulse time is not a
1245                multiple of `DeviceInfo.min_pulse_width_us`, is out of
1246                range or is not later than the one before it, or the
1247                device does not acknowledge the command.
1248        """
1249        pulse_times_cycles = self._custom_times_to_cycles(pulse_times, "pulse_times")
1250        pulse_voltage_bits = self._volts_array_to_bits(voltages, "voltages")
1251        if len(pulse_times_cycles) != len(pulse_voltage_bits):
1252            raise PulsePalError(
1253                "pulse_times and voltages must be the same length."
1254            )
1255        self._send_custom_train(
1256            custom_train_id,
1257            pulse_times_cycles,
1258            pulse_voltage_bits,
1259            "send_custom_pulse_train()",
1260        )
1261
1262    def send_custom_waveform(
1263        self,
1264        custom_train_id,
1265        sampling_period,
1266        voltages,
1267    ):
1268        """Load an arbitrary waveform onto the device.
1269
1270        A convenience shorthand for
1271        `PulsePalDevice.send_custom_pulse_train` with evenly spaced,
1272        adjoining pulses, so that `voltages` is played as a waveform
1273        sampled every `sampling_period` seconds. It is loaded as a custom
1274        train is: see `PulsePalDevice.send_custom_pulse_train` for what
1275        happens to channels playing it, and in param sync mode.
1276
1277        Set the channel's `PulsePalDevice.phase1_duration` to
1278        `sampling_period` as well, so that each sample is held for the
1279        sampling period.
1280
1281        ```python
1282        import math
1283
1284        samples = [math.sin(i / 10.0) * 10 for i in range(1000)]
1285        P.send_custom_waveform(1, 0.001, samples)   # 1 kHz
1286        P.custom_train_id[2] = 1
1287        P.phase1_duration[2] = 0.001
1288        ```
1289
1290        Args:
1291            custom_train_id: Custom train to load, from 1 to
1292                `DeviceInfo.n_custom_pulse_trains` (2 on Pulse Pal 2,
1293                4 on Pulse Pal 3).
1294            sampling_period: In seconds: a multiple of
1295                `DeviceInfo.min_pulse_width_us` (10 us on Pulse Pal 3,
1296                100 us on Pulse Pal 2). Each voltage is
1297                held for this long.
1298            voltages: Waveform samples, in volts [-10, 10].
1299                Accepts a list, tuple or 1-D NumPy array.
1300
1301        Raises:
1302            PulsePalError: If `custom_train_id` is out of range, there
1303                are no samples, or more than
1304                `DeviceInfo.max_custom_pulses`, `sampling_period` is not a
1305                positive multiple of `DeviceInfo.min_pulse_width_us`, the
1306                last sample would start after `MAX_TIME`, or the device
1307                does not acknowledge the command.
1308        """
1309        pulse_voltage_bits = self._volts_array_to_bits(voltages, "voltages")
1310        period_cycles = self._custom_time_cycles(sampling_period, "sampling_period")
1311        if period_cycles == 0:
1312            raise PulsePalError(
1313                f"sampling_period must be a positive multiple of {self.info.min_pulse_width_us:g} us. "
1314                f"Received {sampling_period!r}."
1315            )
1316        pulse_times = period_cycles * np.arange(len(pulse_voltage_bits), dtype=np.int64)
1317        self._send_custom_train(
1318            custom_train_id,
1319            pulse_times,
1320            pulse_voltage_bits,
1321            "send_custom_waveform()",
1322        )
1323
1324    def trigger(self, channels):
1325        """Trigger output channels in software.
1326
1327        The channels start their pulse trains together, in the same timer
1328        cycle. A channel that is already playing a pulse train ignores
1329        the trigger, and `PulsePalDevice.stop` cancels a trigger that has
1330        not started its channel yet.
1331
1332        ```python
1333        P.trigger(3)          # one channel
1334        P.trigger([1, 4])     # several: a list, tuple or NumPy array
1335        ```
1336
1337        Args:
1338            channels: Output channel number, 1-4, or several as a list,
1339                tuple or NumPy array.
1340
1341        Raises:
1342            PulsePalError: If a channel number is not 1-4, or no channel
1343                is given. Nothing is triggered.
1344        """
1345        self._write_serial((self._OP_MENU_BYTE, 77, _common.channel_bits(channels)), "uint8")
1346
1347    def stop(self, channels=None):
1348        """Stop pulse trains currently playing on the device.
1349
1350        The stopped channels return to their
1351        `PulsePalDevice.resting_voltage`. A soft trigger that has not
1352        started its channel yet is cancelled too.
1353
1354        ```python
1355        P.stop()           # all output channels
1356        P.stop([1, 3, 4])  # channels 1, 3 and 4
1357        ```
1358
1359        Args:
1360            channels: Output channel number, 1-4, or several as a list,
1361                tuple or NumPy array. `None`, the default, stops all
1362                channels. Stopping some channels requires firmware v22
1363                or newer.
1364
1365        Raises:
1366            PulsePalError: If a channel number is not 1-4, or channels are
1367                given and the firmware is older than v22.
1368        """
1369        if self.info.firmware_version < 22:
1370            if channels is not None:
1371                raise PulsePalError("stop() cannot address individual channels before firmware v22")
1372            self._write_serial((self._OP_MENU_BYTE, 80), "uint8")  # Stops all channels
1373            return
1374        bits = 0x0F if channels is None else _common.channel_bits(channels)
1375        self._write_serial((self._OP_MENU_BYTE, 98, bits), "uint8")
1376
1377    def status(self):
1378        """Read which channels are playing, and the counters since the last call.
1379
1380        The counters (see `DeviceStatus`) count from the previous call, or
1381        from the device's startup, and each call clears them. Calling it
1382        before and after a trial checks that the TTLs the trial sent
1383        arrived, and started the trains they should:
1384
1385        ```python
1386        P.status()                     # Clears the counters
1387        run_trial()                    # Your code: one TTL on trigger channel 1, linked to channels 1 and 2
1388        s = P.status()
1389        if s.trigger_edges[1] != 1 or s.channel_starts[1:3] != [1, 1]:
1390            print("The trial's trains did not all start")
1391        ```
1392
1393        Requires firmware v22 or newer.
1394
1395        Returns:
1396            A `DeviceStatus`.
1397
1398        Raises:
1399            PulsePalError: If the firmware is older than v22, or the device
1400                does not reply.
1401        """
1402        self._require_firmware(22, "status()")
1403        self._write_serial((self._OP_MENU_BYTE, 101), "uint8")
1404        values = struct.unpack(f"{self._ENDIANNESS}B7I", self._read_raw(self._STATUS_MESSAGE_BYTES))
1405        return DeviceStatus(
1406            playing=[ch for ch in range(1, 5) if values[0] & (1 << (ch - 1))],
1407            trigger_edges=[None, *values[1:3]],
1408            channel_starts=[None, *values[3:7]],
1409            comm_failures=values[7],
1410        )
1411
1412    def save_program_to_microsd(self, file_name):
1413        """Save the parameters to the device's microSD card, as a program.
1414
1415        The program can be loaded later with
1416        `PulsePalDevice.load_program_from_microsd` or with the joystick
1417        menu's Load Program item. A file of the same name is replaced. To
1418        keep the parameters on this computer instead, see
1419        `PulsePalDevice.export_params`.
1420
1421        ```python
1422        P.save_program_to_microsd("MyProtocol.pps")
1423        ```
1424
1425        Args:
1426            file_name: At most 15 ASCII characters, ending in `.pps`.
1427
1428        Raises:
1429            PulsePalError: If the file name is invalid, or the device could
1430                not write the file to its microSD card. Pulse Pal 2 also
1431                refuses while a channel is playing: its card shares a bus with
1432                the outputs, so stop playback first.
1433        """
1434        self._settings_file_op(file_name, 1, "save_program_to_microsd()")
1435
1436    def load_program_from_microsd(self, file_name):
1437        """Load a program from the device's microSD card.
1438
1439        The device plays the program, and the parameters are read back
1440        into this object (see `PulsePalDevice.sync_from_device`). The
1441        program's trigger modes take effect too, and it is loaded at once,
1442        also in param sync mode.
1443
1444        Args:
1445            file_name: The program's file name, ending in `.pps`, as saved
1446                by `PulsePalDevice.save_program_to_microsd` or the joystick
1447                menu.
1448
1449        Raises:
1450            PulsePalError: If the file name is invalid, or the device could
1451                not load the program: there is no such file on its microSD
1452                card, or it is not a valid Pulse Pal program; or, on Pulse
1453                Pal 2, a channel is playing (see `save_program_to_microsd`).
1454                The device then keeps its parameters, and so does this object.
1455        """
1456        self._settings_file_op(file_name, 2, "load_program_from_microsd()")
1457
1458    def delete_program_from_microsd(self, file_name):
1459        """Delete a program from the device's microSD card.
1460
1461        Args:
1462            file_name: The program's file name, ending in `.pps`.
1463
1464        Raises:
1465            PulsePalError: If the file name is invalid, or the device could
1466                not delete the file, e.g. because there is no such file, or,
1467                on Pulse Pal 2, because a channel is playing.
1468        """
1469        self._settings_file_op(file_name, 3, "delete_program_from_microsd()")
1470
1471    def _settings_file_op(self, file_name, op_byte, context):
1472        """Save (op_byte 1), load (2) or delete (3) a settings file, with op 90."""
1473        valid = (
1474            isinstance(file_name, str)
1475            and all(32 <= ord(character) <= 126 for character in file_name)  # Printable ASCII
1476            and file_name.lower().endswith(".pps")
1477            and 4 < len(file_name) <= 15
1478        )
1479        if not valid:
1480            raise PulsePalError(
1481                f"{context}: the file name must be 1 to 11 ASCII characters followed by .pps, "
1482                f"e.g. 'Protocol1.pps'. Received {file_name!r}."
1483            )
1484        filename_bytes = file_name.encode("ascii")
1485        self._write_serial(
1486            (self._OP_MENU_BYTE, 90, op_byte, len(filename_bytes)),
1487            "uint8",
1488            list(filename_bytes),
1489            "uint8",
1490        )
1491        if self.info.firmware_version > 21:
1492            # Sent after the file operation has finished. A refused load changes nothing on the device.
1493            refusal = {
1494                1: f"Pulse Pal could not save {file_name!r} to its microSD card. Is the card in place?",
1495                2: f"Pulse Pal could not load {file_name!r}: there is no such file on its microSD card, or it is not "
1496                   "a valid Pulse Pal program. The device keeps its parameters.",
1497                3: f"Pulse Pal could not delete {file_name!r}: there is no such file on its microSD card.",
1498            }[op_byte]
1499            # Pulse Pal 2 replies 2 while a channel plays: its microSD card shares the SPI bus with the DAC
1500            busy = (f"Pulse Pal 2 does not use its microSD card while a channel is playing, because the card shares a "
1501                    f"bus with the outputs, so {file_name!r} was not {('saved', 'loaded', 'deleted')[op_byte - 1]}. "
1502                    "Stop playback first, with stop().")
1503            if op_byte == 2:
1504                busy += " The device keeps its parameters."
1505            self._read_ack(context, refusal_message=refusal, refusal_messages={2: busy})
1506        else:
1507            self._reply_pending = None  # Firmware v21 does not acknowledge op 90 (see _write_raw())
1508            if op_byte == 2:
1509                time.sleep(0.1)  # So allow time for the load
1510        if op_byte == 2:
1511            self.sync_from_device()
1512
1513    def format_microsd(self, *, confirm=True, timeout=30):
1514        """Format the device's microSD card.
1515
1516        Erases every program saved on the device and resets its
1517        parameters to the defaults. By default, asks at the console for
1518        confirmation first; a script that runs on its own, or an AI agent,
1519        passes `confirm=False`.
1520
1521        Requires Pulse Pal hardware v3 or newer.
1522
1523        Args:
1524            confirm: If `True`, ask at the console before anything is
1525                erased. If `False`, format at once.
1526            timeout: Seconds to wait for the device to report that
1527                formatting has finished.
1528
1529        Returns:
1530            `True` once the card has been formatted, or `False` if the
1531            user declines the confirmation prompt.
1532
1533        Raises:
1534            PulsePalError: If the connected hardware is older than v3, if
1535                the device reports that formatting failed, or if it does
1536                not report a result within `timeout` seconds.
1537        """
1538        if self.info.hardware_version < 3:
1539            raise PulsePalError(
1540                "format_microsd() requires hardware v3 or newer."
1541            )
1542
1543        if to_bool(confirm, "confirm"):
1544            print("*** Pulse Pal microSD Formatter ***")
1545            print("This will format Pulse Pal's microSD card,")
1546            print("erase all programs saved on the device")
1547            print("and reset all parameters to defaults.")
1548            reply = input("Do you want to continue (y/n)")
1549            if reply.strip().lower() != "y":
1550                print("Choice confirmed - microSD Card NOT formatted.")
1551                return False
1552
1553        self._write_serial((self._OP_MENU_BYTE, 97), "uint8", list(_FORMAT_KEY), "uint8")
1554
1555        # The device replies with lines of status text, the last of which
1556        # contains "!", and then a confirm byte (1 if the card was formatted,
1557        # 0 if not). The confirm byte is sent after the device has reloaded
1558        # its default parameters, so it can arrive well after the text. It
1559        # must be read here: left in the buffer, it would be taken as the
1560        # reply to the next command, and every reply after that would be
1561        # read one byte late.
1562        start = time.time()
1563        message = bytearray()
1564        flag_index = -1
1565        line_end = -1
1566
1567        while time.time() - start < timeout:
1568            n_waiting = self.port.in_waiting
1569            if n_waiting:
1570                message.extend(self.port.read(n_waiting))
1571                flag_index = message.find(b"!")
1572                if flag_index >= 0:
1573                    line_end = message.find(b"\n", flag_index)
1574                if line_end >= 0 and len(message) > line_end + 1:
1575                    break
1576            time.sleep(0.01)
1577        if line_end < 0 or len(message) <= line_end + 1:
1578            raise PulsePalError(
1579                "Pulse Pal did not report the result of formatting "
1580                f"its microSD card within {timeout} s. {_common.NO_REPLY_HINT}"
1581            )
1582        self._reply_pending = None  # The whole reply is read (see _write_raw())
1583        confirm_byte = message[line_end + 1]
1584
1585        text = bytes(message[:flag_index]).decode(
1586            "ascii", errors="replace"
1587        ).rstrip()
1588        if text:
1589            print(text)
1590
1591        # The device has loaded its own defaults. They are this class's, but the device keeps no record of
1592        # this object's other changes (continuous loop mode, for one), so the defaults are programmed, as the
1593        # MATLAB class does, and this object describes the device again.
1594        self.set_default_params()
1595        if confirm_byte != 1:
1596            raise PulsePalError(
1597                "Pulse Pal could not format its microSD card."
1598            )
1599        return True
1600
1601    def gui(self, block=None, theme=None):
1602        """Open the Pulse Pal parameter GUI, or focus an open one.
1603
1604        The GUI edits its own copy of the parameters, and loads them to
1605        the device when its 'Load to Device' button is clicked. The
1606        window closes automatically when the device is closed or
1607        deleted.
1608
1609        Calling this while the GUI is already open focuses the existing
1610        window rather than opening a second one.
1611
1612        Args:
1613            block: If `True`, the call returns when the GUI is closed. If
1614                `False`, the call returns immediately, and the host
1615                application must run the Tk event loop. If `None`, the
1616                GUI blocks only when the host does not already provide a
1617                Tk event loop, e.g. when launched from a script.
1618            theme: `"light"` or `"dark"` to select the color theme, or
1619                `None` to match the desktop theme. Passing a theme to an
1620                already-open GUI recolors it in place.
1621
1622        Returns:
1623            The `pulsepal.gui.PulsePalGUI` instance driving the window.
1624
1625        Raises:
1626            ValueError: If the theme name is not recognized.
1627        """
1628        gui = getattr(self, "_gui", None)
1629        if gui is not None and not gui.is_closed:
1630            if theme is not None:
1631                gui.set_theme(theme)
1632            gui.focus()
1633            return gui
1634
1635        from .gui import PulsePalGUI
1636
1637        gui = PulsePalGUI(self, theme=theme)
1638        self._gui = gui
1639        gui.start(block=block)
1640        return gui
1641
1642    # ------------------------------------------------------------------
1643    # Connection
1644    # ------------------------------------------------------------------
1645
1646    def close(self):
1647        """Close the connection to the device, and the GUI if open.
1648
1649        The device stops all output channels when the client disconnects,
1650        and shows its own name on its screen again. Safe to call more
1651        than once; later calls do nothing. Called automatically when
1652        leaving a `with` block and when the object is garbage collected,
1653        so pulse trains also stop when the last reference to the object
1654        goes, e.g. when a function that created it returns.
1655        """
1656        self._close(send_disconnect=True)
1657
1658    def _close(self, send_disconnect):
1659        """Close the connection. send_disconnect: tell the device first (op 81). False when the device is in
1660        an unknown state, such as after a failed handshake."""
1661        gui = getattr(self, "_gui", None)
1662        self._gui = None
1663        if gui is not None:
1664            try:
1665                gui.close()
1666            except Exception:
1667                # Cleanup must not raise; Tk may already be torn down
1668                pass
1669
1670        if getattr(self, "_closed", True):
1671            return
1672        self._closed = True
1673        if send_disconnect and self._unsent:
1674            _common.warn(f"PulsePalDevice closed with changes made with auto_sync off that were never sent to the "
1675                         f"device: {self._unsent_names()}.")
1676        self._reply_pending = None  # Nothing more is read, so a reply still to come needs no discarding
1677        try:
1678            if send_disconnect and self.port and self.port.is_open:
1679                self._write_serial((self._OP_MENU_BYTE, 81), "uint8")
1680        except Exception:
1681            pass  # The port may already be gone, e.g. the cable was unplugged
1682        finally:
1683            if self.port and self.port.is_open:
1684                self.port.close()
1685
1686    def __enter__(self):
1687        """Enter a `with` block, returning the connected device."""
1688        return self
1689
1690    def __exit__(self, exc_type, exc_value, traceback):
1691        """Disconnect and close the port when leaving a `with` block.
1692
1693        Returns:
1694            `False`, so any exception raised in the block propagates.
1695        """
1696        self.close()
1697        return False
1698
1699    def __del__(self):
1700        """Disconnect and close the port when the object is collected."""
1701        try:
1702            self.close()
1703        except Exception:
1704            # Destructors should not raise; the serial object may already be
1705            # gone during interpreter shutdown.
1706            pass
1707
1708    def __setattr__(self, name, value):
1709        """Refuse a public attribute that is not a setting, such as a misspelled one: it would program nothing (see
1710        _common.refuse_unknown_attribute())."""
1711        if name[0] != "_":
1712            _common.refuse_unknown_attribute(self, name, _SETTING_NAMES)
1713        super().__setattr__(name, value)
1714
1715    def __repr__(self):
1716        """Describe the device and its parameters, e.g. with `print(P)`."""
1717        port = getattr(getattr(self, "port", None), "port", None)
1718        lines = [
1719            f"PulsePalDevice on {port} (Pulse Pal {self.info.hardware_version}, "
1720            f"firmware v{self.info.firmware_version})",
1721            f"auto_sync: {self._auto_sync}",
1722        ]
1723        for name in self.info.output_parameter_names:
1724            lines.append(f"{name}: {list(getattr(self, name))}")
1725        lines.append(f"trigger_mode: {list(self._trigger_mode)}")
1726        return "\n".join(lines)
1727
1728    # ------------------------------------------------------------------
1729    # Internals: parameters
1730    # ------------------------------------------------------------------
1731
1732    def _has_param_sync(self):
1733        """Param sync mode is on Pulse Pal 3, with firmware v22 or newer."""
1734        return (self.info.hardware_version or 2) > 2 and (self.info.firmware_version or 0) > 21
1735
1736    def _normalize_output_value(self, code, value, channel):
1737        """Check a value for an output parameter, and return it as the parameter's list holds it:
1738        a time as the device plays it, rounded to its timer cycle.
1739
1740        Raises PulsePalError, before anything is sent, for a value the device cannot play. The
1741        device refuses most of them too, but it also resets them (see validateOutputParams() in
1742        /Firmware/PulsePal/USBOps.ino), so the local copy would describe a different program.
1743        """
1744        name, kind = _OUTPUT_PARAMETERS[code]
1745        label = f"{name} on channel {channel}"
1746        if kind == "bool":
1747            value = to_bool(value, label)
1748            if code == _CONTINUOUS_LOOP_CODE and value and self.info.firmware_version < 22:
1749                raise PulsePalError("continuous_loop requires firmware v22 or newer.")
1750            return value
1751        if kind == "volts":
1752            self._volts_to_bits(value, label)
1753            return float(value)
1754        if kind in ("time", "pulse_time"):
1755            cycles = self._seconds_to_cycles(value, label)
1756            # Checked in whole cycles, as the device receives it: 100 * 1e-6 is just under 0.0001,
1757            # but is exactly 10 cycles of 10 us
1758            if kind == "pulse_time" and cycles < self._min_pulse_cycles():
1759                raise PulsePalError(
1760                    f"{label} must be at least {self._min_pulse_seconds()} s "
1761                    f"({self._min_pulse_cycles_text()} of the device's {self.info.cycle_period_us} us "
1762                    f"timer), so that trigger channels can detect the pulses. Received {value!r}."
1763                )
1764            return self._cycles_to_seconds(cycles)  # The time the device plays
1765        if kind == "train_id":
1766            self._check_whole_number(value, label, 0, self.info.n_custom_pulse_trains)
1767            return int(value)
1768        # "target"
1769        return to_name(value, CUSTOM_TRAIN_TARGETS, label, codes={"Pulses": 0, "Bursts": 1})
1770
1771    def _normalize_trigger_mode(self, value):
1772        modes = self.info.trigger_modes
1773        return to_name(value, modes, "trigger_mode", codes={mode: TRIGGER_MODES.index(mode) for mode in modes})
1774
1775    def _encode_output_param(self, code, values):
1776        """Convert output parameter values to device units.
1777
1778        Returns the converted values and the datatype the device reads for this
1779        parameter code: DAC codes for voltages, hardware timer cycles for times,
1780        and bytes for everything else.
1781        """
1782        name, kind = _OUTPUT_PARAMETERS[code]
1783        if kind == "volts":
1784            return [self._volts_to_bits(v, name) for v in values], "uint16"
1785        if kind in ("time", "pulse_time"):
1786            return [self._seconds_to_cycles(v, name) for v in values], "uint32"
1787        if kind == "target":
1788            return [CUSTOM_TRAIN_TARGETS.index(v) for v in values], "uint8"
1789        return [int(v) for v in values], "uint8"
1790
1791    def _apply_output_param(self, code, values):
1792        """Check new values for all four channels of an output parameter, and with auto_sync on,
1793        program the channels whose value changes. Returns them as the parameter's list holds them."""
1794        values = [self._normalize_output_value(code, v, ch) for ch, v in enumerate(values, start=1)]
1795        name = _OUTPUT_PARAMETERS[code][0]
1796        if not self._auto_sync:
1797            self._unsent = self._unsent | {name}  # Until sync_to_device() sends it (see auto_sync)
1798        else:
1799            if code in (_BURST_DURATION_CODE, _CUSTOM_TRAIN_TARGET_CODE):
1800                # Sent alone, so it must suit the other parameter as the device holds it. In a batch() block, or
1801                # with auto_sync off, both are checked together by sync_to_device().
1802                new = {name: values}
1803                self._check_custom_bursts(new.get("custom_train_target", list(self._custom_train_target)[1:]),
1804                                          new.get("burst_duration", list(self._burst_duration)[1:]))
1805            # Compared as the device receives them: 5 and 5.00001 V are the same DAC code
1806            old_data, _ = self._encode_output_param(code, list(getattr(self, name))[1:])
1807            new_data, _ = self._encode_output_param(code, values)
1808            channels = [ch for ch in range(1, 5) if old_data[ch - 1] != new_data[ch - 1]]
1809            self._send_output_param(code, values, channels, f"setting {name}")
1810        return values
1811
1812    def _send_output_param(self, code, values, channels, context):
1813        """Program one output parameter on some output channels: op 91 if all four change, op 74 for each otherwise.
1814
1815        The other channels are left as the device holds them. Sending all four (op 91) for a change to one put this
1816        object's copy of the others back on the device, undoing what the joystick menu, or a program loaded with it,
1817        had set there.
1818        """
1819        if self.info.firmware_version > 21 and len(channels) == 4:
1820            data, datatype = self._encode_output_param(code, values)
1821            self._write_serial((self._OP_MENU_BYTE, 91, code), "uint8", data, datatype)
1822            self._read_ack(context, on_refusal=lambda: self._refresh_output_param(code, channels))
1823            return
1824        for channel in channels:
1825            # A refusal reads back every channel this assignment changes, so that the copy describes the device for
1826            # those already sent too
1827            self._send_output_param_to_channel(code, channel, values[channel - 1], context, refresh_channels=channels)
1828
1829    def _send_output_param_to_channel(self, code, channel, value, context, refresh_channels=None):
1830        """Program one output parameter on one channel (op 74)."""
1831        if code == _CONTINUOUS_LOOP_CODE and self.info.firmware_version < 22:
1832            return  # Firmware v21 has no parameter 18, and _normalize_output_value() lets only False through
1833        data, datatype = self._encode_output_param(code, [value])
1834        self._write_serial((self._OP_MENU_BYTE, 74, code, channel), "uint8", data, datatype)
1835        self._read_ack(context, on_refusal=lambda: self._refresh_output_param(code, refresh_channels or (channel,)))
1836
1837    def _apply_trigger_mode(self, values):
1838        """Check new trigger modes, and with auto_sync on, program the trigger channels whose mode changes."""
1839        modes = [self._normalize_trigger_mode(v) for v in values]
1840        if not self._auto_sync:
1841            self._unsent = self._unsent | {"trigger_mode"}
1842        else:
1843            channels = [ch for ch in (1, 2) if modes[ch - 1] != self._trigger_mode[ch]]
1844            if len(channels) == 2:
1845                self._send_trigger_modes(modes, "setting trigger_mode")
1846            elif channels:  # One: op 74, which leaves the other trigger channel as the device holds it
1847                channel = channels[0]
1848                self._write_serial((self._OP_MENU_BYTE, 74, _TRIGGER_MODE_CODE, channel,
1849                                    TRIGGER_MODES.index(modes[channel - 1])), "uint8")
1850                self._read_ack("setting trigger_mode", on_refusal=self._refresh_trigger_mode)
1851        return modes
1852
1853    def _send_trigger_modes(self, modes, context):
1854        """Program both trigger channels' modes: op 91, or op 74 per channel on firmware v21."""
1855        codes = [TRIGGER_MODES.index(mode) for mode in modes]
1856        if self.info.firmware_version > 21:
1857            self._write_serial((self._OP_MENU_BYTE, 91, _TRIGGER_MODE_CODE, *codes), "uint8")
1858            self._read_ack(context, on_refusal=self._refresh_trigger_mode)
1859        else:
1860            for channel, mode_code in enumerate(codes, start=1):
1861                self._write_serial((self._OP_MENU_BYTE, 74, _TRIGGER_MODE_CODE, channel, mode_code), "uint8")
1862                self._read_ack(context)
1863
1864    def _snapshot(self):
1865        """All parameters, and which are unsent, for batch() to restore if its block fails."""
1866        names = [*self.info.output_parameter_names, "trigger_mode"]
1867        return {name: list(getattr(self, name))[1:] for name in names}, self._unsent
1868
1869    def _restore(self, snapshot):
1870        params, self._unsent = snapshot
1871        for name, values in params.items():
1872            getattr(self, name)._store(values)
1873
1874    def _unsent_names(self):
1875        """The settings assigned with auto_sync off and not yet sent, for messages, in parameter code order."""
1876        return ", ".join(name for name in _SETTING_NAMES if name in self._unsent)
1877
1878    def _check_sync(self):
1879        """Check the whole local copy before sync_to_device() sends it."""
1880        for code, (name, _) in _OUTPUT_PARAMETERS.items():
1881            values = getattr(self, name)
1882            for ch in range(1, 5):
1883                self._normalize_output_value(code, values[ch], ch)
1884        for ch in (1, 2):
1885            self._normalize_trigger_mode(self._trigger_mode[ch])
1886        self._check_custom_bursts(list(self._custom_train_target)[1:], list(self._burst_duration)[1:])
1887
1888    @staticmethod
1889    def _check_custom_bursts(targets, burst_durations):
1890        """Refuse a custom train targeting bursts on a channel with no burst duration.
1891
1892        Its times are burst onsets, and a burst of no duration plays one timer cycle of a pulse at each: a
1893        program that cannot be what was meant. (The device once played one pulse and then left the channel
1894        "playing", ignoring triggers until it was stopped.)
1895        """
1896        for channel, (target, burst) in enumerate(zip(targets, burst_durations), start=1):
1897            if target == "Bursts" and burst == 0:
1898                raise PulsePalError(
1899                    f"Output channel {channel}: custom_train_target 'Bursts' makes the custom train's times burst "
1900                    "onsets, so burst_duration must be above 0. Set burst_duration first, or set both in a batch() "
1901                    "block."
1902                )
1903
1904    def _send_sync(self):
1905        """Send every parameter in one command: op 92 on firmware v22 or newer, op 73 on v21."""
1906        self._unsent = frozenset()  # All are sent below
1907        if self.info.firmware_version > 21:
1908            self._sync_all_params()
1909        else:
1910            self._sync_all_params_legacy()
1911        self._read_ack("sync_to_device()", on_refusal=self._refresh_after_refused_sync)
1912
1913    def _device_bytes(self, name):
1914        """The byte values of a byte parameter for channels 1-4, as the device reads them."""
1915        values = list(getattr(self, name))[1:]
1916        if name == "custom_train_target":
1917            return [CUSTOM_TRAIN_TARGETS.index(v) for v in values]
1918        return [int(v) for v in values]
1919
1920    def _sync_all_params(self):
1921        """Send all parameters using the packed sync op (firmware v22+).
1922
1923        Values are grouped by width so that the whole program travels as
1924        one uint32 block, one uint16 block and one uint8 block. See op 92
1925        in /Firmware/PROTOCOL.md.
1926        """
1927        time_values = []
1928        for name in _TIME_PARAMETERS:
1929            time_values.extend(self._seconds_to_cycles(v) for v in list(getattr(self, name))[1:])
1930        voltage_values = []
1931        for name in _VOLTAGE_PARAMETERS:
1932            voltage_values.extend(self._volts_to_bits(v) for v in list(getattr(self, name))[1:])
1933        single_byte_values = []
1934        for name in ("is_biphasic", "custom_train_id", "custom_train_target", "custom_train_loop",
1935                     "continuous_loop", "link_trigger_channel1", "link_trigger_channel2"):
1936            single_byte_values.extend(self._device_bytes(name))
1937        single_byte_values.extend(TRIGGER_MODES.index(mode) for mode in list(self._trigger_mode)[1:])
1938
1939        self._write_serial(
1940            (self._OP_MENU_BYTE, 92),
1941            "uint8",
1942            time_values,
1943            "uint32",
1944            voltage_values,
1945            "uint16",
1946            single_byte_values,
1947            "uint8",
1948        )
1949
1950    def _sync_all_params_legacy(self):
1951        """Send all parameters using the legacy sync op (firmware v21).
1952
1953        Equivalent to `_sync_all_params`, but lays the program out
1954        channel by channel as the older firmware expects. Continuous
1955        loop mode is not part of it.
1956        """
1957        program_values_32 = []
1958        program_values_16 = []
1959        program_values_8 = []
1960        for ch in range(1, 5):
1961            program_values_32.extend(self._seconds_to_cycles(getattr(self, name)[ch]) for name in _TIME_PARAMETERS)
1962        for ch in range(1, 5):
1963            program_values_16.extend(self._volts_to_bits(getattr(self, name)[ch]) for name in _VOLTAGE_PARAMETERS)
1964        byte_values = {name: self._device_bytes(name) for name in
1965                       ("is_biphasic", "custom_train_id", "custom_train_target", "custom_train_loop")}
1966        for ch in range(4):
1967            program_values_8.extend(byte_values[name][ch] for name in byte_values)
1968        links = self._device_bytes("link_trigger_channel1") + self._device_bytes("link_trigger_channel2")
1969        modes = [TRIGGER_MODES.index(mode) for mode in list(self._trigger_mode)[1:]]
1970
1971        self._write_serial(
1972            (self._OP_MENU_BYTE, 73),
1973            "uint8",
1974            program_values_32,
1975            "uint32",
1976            program_values_16,
1977            "uint16",
1978            program_values_8 + links + modes,
1979            "uint8",
1980        )
1981
1982    def _read_device_params(self):
1983        """Read the device's parameters (op 93).
1984
1985        Returns a dict of parameter values for channels 1-4 (trigger
1986        channels 1-2 for trigger_mode), as the parameter lists hold them.
1987        Continuous loop mode is not part of op 93, so `continuous_loop` is
1988        not included.
1989        """
1990        self._write_serial((self._OP_MENU_BYTE, 93), "uint8")
1991        # The device sends the whole parameter set as one message, so read it in one go and
1992        # unpack it here. See "Op codes", op 93, in /Firmware/PROTOCOL.md.
1993        message = self._read_raw(self._PARAM_MESSAGE_BYTES)
1994        values = struct.unpack(f"{self._ENDIANNESS}32I12H26B", message)
1995        params = {}
1996        for index, name in enumerate(_TIME_PARAMETERS):
1997            params[name] = [self._cycles_to_seconds(x) for x in values[index * 4:index * 4 + 4]]
1998        for index, name in enumerate(_VOLTAGE_PARAMETERS):
1999            params[name] = [self._bits_to_volts(x) for x in values[32 + index * 4:32 + index * 4 + 4]]
2000        for index, name in enumerate(("is_biphasic", "custom_train_id", "custom_train_target",
2001                                      "custom_train_loop", "link_trigger_channel1", "link_trigger_channel2")):
2002            codes = values[44 + index * 4:44 + index * 4 + 4]
2003            if name == "custom_train_id":
2004                params[name] = list(codes)
2005            elif name == "custom_train_target":
2006                params[name] = [CUSTOM_TRAIN_TARGETS[min(code, 1)] for code in codes]
2007            else:
2008                params[name] = [bool(code) for code in codes]
2009        params["trigger_mode"] = [TRIGGER_MODES[min(code, 3)] for code in values[68:70]]
2010        return params
2011
2012    def _refresh_output_param(self, code, channels):
2013        """Read an output parameter back from the device into the local copy, for these channels.
2014
2015        Called after the device refused a value for them, which it resets. The parameter's other
2016        channels, and other parameters, keep their local values, which may hold edits not yet synced.
2017        """
2018        name = _OUTPUT_PARAMETERS[code][0]
2019        if self.info.firmware_version < 22 or code == _CONTINUOUS_LOOP_CODE:
2020            return  # No op 93 to read it with, or not in op 93
2021        device_values = self._read_device_params()[name]
2022        settings = getattr(self, name)
2023        local_values = list(settings)[1:]
2024        for ch in channels:
2025            local_values[ch - 1] = device_values[ch - 1]
2026        settings._store(local_values)
2027
2028    def _refresh_trigger_mode(self):
2029        """Read the trigger modes back from the device, after the device refused a new one."""
2030        if self.info.firmware_version > 21:
2031            self._trigger_mode._store(self._read_device_params()["trigger_mode"])
2032
2033    def _refresh_after_refused_sync(self):
2034        """Read the parameters back after the device refused a sync_to_device() set.
2035
2036        The device programs the set with the refused values reset, so the local copy is read back.
2037        In param sync mode the device stores the set instead, and op 93 would return the parameters
2038        playing now, so the local copy is left as it is.
2039        """
2040        if self.info.firmware_version > 21 and "Param Sync" not in list(self._trigger_mode)[1:]:
2041            self.sync_from_device()
2042
2043    def _check_whole_number(self, value, name, low, high):
2044        """Raise PulsePalError unless value is a whole number from low to high."""
2045        valid = (
2046            isinstance(value, numbers.Real)
2047            and not isinstance(value, (bool, np.bool_))
2048            and math.isfinite(value)
2049            and value == int(value)
2050            and low <= value <= high
2051        )
2052        if not valid:
2053            raise PulsePalError(
2054                f"{name} must be a whole number from {low} to {high}. Received {value!r}."
2055            )
2056
2057    def _min_pulse_cycles(self):
2058        """The shortest pulse phase, interval or train, in timer cycles: DeviceInfo.min_pulse_width_us. See
2059        _MIN_PULSE_CYCLES."""
2060        return int(round(self.info.min_pulse_width_us / self.info.cycle_period_us))
2061
2062    def _min_pulse_cycles_text(self):
2063        """_min_pulse_cycles() for messages: "1 cycle", "2 cycles"."""
2064        cycles = self._min_pulse_cycles()
2065        return f"{cycles} cycle" if cycles == 1 else f"{cycles} cycles"
2066
2067    def _min_pulse_seconds(self):
2068        """The shortest pulse phase, interval or train, in seconds: DeviceInfo.min_pulse_width_us."""
2069        return self._cycles_to_seconds(self._min_pulse_cycles())
2070
2071    def _require_firmware(self, minimum_version, context):
2072        """Raise unless the device firmware is new enough for an op."""
2073        if (
2074            self.info.firmware_version is None
2075            or self.info.firmware_version < minimum_version
2076        ):
2077            raise PulsePalError(
2078                f"{context} requires firmware v{minimum_version} or newer. "
2079                f"Detected firmware is v{self.info.firmware_version}."
2080            )
2081
2082    # ------------------------------------------------------------------
2083    # Internals: custom trains
2084    # ------------------------------------------------------------------
2085
2086    def _send_custom_train(self, custom_train_id, pulse_times_cycles,
2087                           pulse_voltage_bits, context):
2088        """Send a custom pulse train, already converted to device units (NumPy integer arrays).
2089
2090        Uses op 95 on firmware v22 or newer, and the legacy ops 75 and 76
2091        (trains 1 and 2 only) on firmware v21.
2092        """
2093        n_trains = self.info.n_custom_pulse_trains
2094        valid_id = (
2095            isinstance(custom_train_id, numbers.Integral)
2096            and not isinstance(custom_train_id, (bool, np.bool_))
2097            and 1 <= custom_train_id <= n_trains
2098        )
2099        if not valid_id:
2100            raise PulsePalError(
2101                f"{context}: custom_train_id must be an integer from 1 to "
2102                f"{n_trains}. Received {custom_train_id!r}."
2103            )
2104        train_id = int(custom_train_id)
2105        n_pulses = len(pulse_times_cycles)
2106        if n_pulses == 0:
2107            # The device refuses an empty train too: it would hold no pulse times to play
2108            raise PulsePalError(
2109                f"{context}: a custom train needs at least one pulse. To play nothing on a channel, set its "
2110                "custom_train_id to a train that holds what it should play, or do not trigger it."
2111            )
2112        if n_pulses > self.info.max_custom_pulses:
2113            raise PulsePalError(
2114                f"{context}: {n_pulses} pulses were given. Pulse Pal can "
2115                f"store up to {self.info.max_custom_pulses} pulses per "
2116                "custom train."
2117            )
2118        # Times are multiples of _min_pulse_cycles() (see _custom_times_to_cycles()), so increasing times are far
2119        # enough apart for a trigger channel to detect each pulse.
2120        too_close = np.flatnonzero(np.diff(pulse_times_cycles) < self._min_pulse_cycles())
2121        if too_close.size:
2122            i = int(too_close[0]) + 1
2123            raise PulsePalError(
2124                f"{context}: pulse times must increase, by at least "
2125                f"{self._min_pulse_seconds()} s ({self._min_pulse_cycles_text()} "
2126                f"of the device's {self.info.cycle_period_us} us "
2127                f"timer). Pulse {i + 1} is at "
2128                f"{self._cycles_to_seconds(pulse_times_cycles[i])} s, "
2129                f"and pulse {i} is at "
2130                f"{self._cycles_to_seconds(pulse_times_cycles[i - 1])} s."
2131            )
2132        if pulse_times_cycles[-1] > self._max_time_cycles():
2133            raise PulsePalError(
2134                f"{context}: the last pulse starts at {self._cycles_to_seconds(pulse_times_cycles[-1])} s. "
2135                f"Pulse times must be at most {self.info.max_time} s."
2136            )
2137
2138        if self.info.firmware_version > 21:
2139            header = bytes([self._OP_MENU_BYTE, 95, train_id - 1])
2140        else:
2141            header = bytes([self._OP_MENU_BYTE, 74 + train_id])
2142        self._write_raw(
2143            header + struct.pack("<I", n_pulses)
2144            + np.asarray(pulse_times_cycles).astype("<u4").tobytes()
2145            + np.asarray(pulse_voltage_bits).astype("<u2").tobytes()
2146        )
2147        self._read_ack(context)
2148
2149    def _number_array(self, values, name, description):
2150        """Return values (a scalar, a list, a tuple or a 1-D NumPy array) as a float64 array, refusing booleans,
2151        strings and anything else that is not a number, as the scalar conversions do.
2152
2153        An array of two or more dimensions, or a list of lists, is refused rather than flattened: one value per
2154        pulse is a list, and a 2-D one is a mistake, such as both rows of a stereo waveform. (The MATLAB class, which
2155        took a matrix's longest side as its pulse count and sent all of it, sent the rest to the device as commands.)
2156        """
2157        if isinstance(values, np.ndarray):
2158            self._check_one_dimensional(values, name)
2159            if values.dtype.kind in "iuf":
2160                return values.ravel().astype(float)
2161        items = self._as_list(values)
2162        refused_types = tuple(kind for kind in {type(item) for item in items}  # Each type once: fast for long lists
2163                              if issubclass(kind, (bool, np.bool_, str, bytes)))
2164        if refused_types:
2165            item = next(item for item in items if isinstance(item, refused_types))
2166            raise PulsePalError(f"{name} must be {description}. Received {item!r}.")
2167        try:
2168            array = np.asarray(items, dtype=float)
2169        except (TypeError, ValueError):
2170            for item in items:
2171                try:
2172                    float(item)
2173                except (TypeError, ValueError):
2174                    raise PulsePalError(f"{name} must be {description}. Received {item!r}.") from None
2175            raise
2176        self._check_one_dimensional(array, name)  # A list of lists
2177        return array
2178
2179    @staticmethod
2180    def _check_one_dimensional(array, name):
2181        if array.ndim > 1:
2182            raise PulsePalError(
2183                f"{name} must be one-dimensional, one value per pulse: a list, a tuple or a 1-D NumPy array. "
2184                f"Received an array of shape {array.shape}."
2185            )
2186
2187    @staticmethod
2188    def _first(values, refused):
2189        """The first value at a True position of the boolean array `refused`, for an error message."""
2190        i = int(np.flatnonzero(refused)[0])
2191        item = values.ravel()[i] if isinstance(values, np.ndarray) else PulsePalDevice._as_list(values)[i]
2192        return item.item() if isinstance(item, np.generic) else item
2193
2194    def _custom_times_to_cycles(self, values, name):
2195        """Convert custom pulse times to timer cycles, all at once: _custom_time_cycles() for each of them.
2196
2197        Each must be a time the device can play, and a whole number of DeviceInfo.min_pulse_width_us to the nearest
2198        microsecond, as in the MATLAB class. Returns an int64 array.
2199        """
2200        seconds = self._number_array(values, name, "a time of 0 s or more")
2201        refused = ~(np.isfinite(seconds) & (seconds >= 0))
2202        if refused.any():
2203            raise PulsePalError(f"{name} must be a time of 0 s or more. Received {self._first(values, refused)!r}.")
2204        # The same arithmetic as _seconds_to_cycles() and _custom_time_cycles(): NumPy's round() also takes a
2205        # halfway value to the even integer
2206        refused = np.round(seconds * float(self.info.cycle_frequency)) > self._max_time_cycles()
2207        if refused.any():
2208            raise PulsePalError(
2209                f"{name} must be at most {self.info.max_time} s, the longest time the device's screen "
2210                f"shows. Received {self._first(values, refused)!r}."
2211            )
2212        microseconds = np.round(seconds * 1e6)
2213        step_us = int(self.info.min_pulse_width_us)
2214        refused = np.mod(microseconds, step_us) != 0
2215        if refused.any():
2216            raise PulsePalError(
2217                f"{name} must be multiples of {step_us} us ({step_us / 1e6:g} s). "
2218                f"Received {self._first(values, refused)!r}."
2219            )
2220        return (microseconds // int(self.info.cycle_period_us)).astype(np.int64)
2221
2222    def _volts_array_to_bits(self, values, name):
2223        """Convert voltages to DAC codes, all at once: _volts_to_bits() for each of them. Returns an int64 array."""
2224        volts = self._number_array(values, name, "in [-10, 10] V")
2225        refused = ~((volts >= -10) & (volts <= 10))  # Also refuses NaN
2226        if refused.any():
2227            raise PulsePalError(f"{name} must be in [-10, 10] V. Received {self._first(values, refused)!r}.")
2228        return np.round((volts + 10) / 20 * self._DAC_BITMAX).astype(np.int64)
2229
2230    # ------------------------------------------------------------------
2231    # Internals: serial port and units
2232    # ------------------------------------------------------------------
2233
2234    def _write_serial(self, *args):
2235        """Write one or more data/type pairs to the serial port, in one write."""
2236        if len(args) % 2 != 0:
2237            raise PulsePalError(
2238                "Serial writes require data/type argument pairs."
2239            )
2240
2241        payload = bytearray()
2242        for i in range(0, len(args), 2):
2243            payload.extend(self._pack_values(args[i], args[i + 1]))
2244        self._write_raw(bytes(payload))
2245
2246    def _write_raw(self, payload):
2247        """Write a command to the serial port, in one write.
2248
2249        A command whose reply was not read in full, because the read timed out or was interrupted (Ctrl-C), leaves
2250        that reply, or the rest of it, to arrive later. Read as the reply to the next command, it put every reply
2251        after it out of step: status() reported channels playing that were not, and assignments the device took
2252        raised refusals. So it is discarded first (_discard_lost_reply()). Then this command's reply, if it has one,
2253        is pending until it has been read: _read_raw() marks it read, and a method that reads a reply otherwise
2254        (format_microsd()) sets _reply_pending back to None itself.
2255        """
2256        if self._reply_pending is not None:
2257            self._discard_lost_reply()
2258        if payload[1] not in _OPS_WITHOUT_TRACKED_REPLY:
2259            self._reply_pending = payload[1]  # Before the write: a Ctrl-C can come as soon as it returns
2260        bytes_written = self.port.write(payload)
2261        if bytes_written != len(payload):
2262            raise PulsePalError(
2263                f"Wrote {bytes_written} byte(s), expected to write "
2264                f"{len(payload)} byte(s)."
2265            )
2266
2267    def _discard_lost_reply(self):
2268        """Discard what arrives of a reply that was not read in full (see _write_raw()), and bring this object back
2269        in step with the device.
2270
2271        The command may have changed the parameters, unknown to this object: an assignment interrupted after its
2272        write, for one, which the device took and this object did not. So after one of _PARAMETER_OPS, they are read
2273        back (op 93), on firmware v22 or newer and outside param sync mode, where op 93 returns the parameters playing
2274        rather than a set stored for the next sync edge.
2275        """
2276        op = self._reply_pending
2277        self._reply_pending = None
2278        n_late = len(_common.read_until_quiet(self.port))
2279        read_back = (op in _PARAMETER_OPS and self.info.firmware_version > 21
2280                     and "Param Sync" not in list(self._trigger_mode)[1:])
2281        _common.warn_lost_reply(f"op {op}", n_late,
2282                                ", and the parameters were read back from the device" if read_back else "")
2283        if read_back:
2284            try:
2285                self.sync_from_device()
2286            except PulsePalError as exc:
2287                raise PulsePalError(
2288                    f"Pulse Pal did not send its parameters, read back after the reply to an earlier command was "
2289                    f"lost. {_common.NO_REPLY_HINT}"
2290                ) from exc
2291
2292    def _read_raw(self, n_bytes):
2293        """Read exactly n_bytes from the serial port: the whole reply to the command last sent, which is then no
2294        longer pending (see _write_raw()). Only connecting reads a reply in more than one call."""
2295        message_bytes = self.port.read(n_bytes)
2296        if len(message_bytes) < n_bytes:
2297            raise PulsePalError(
2298                f"Pulse Pal sent {len(message_bytes)} of the {n_bytes} byte(s) of its reply within "
2299                f"{self.port.timeout} s. {_common.NO_REPLY_HINT}"
2300            )
2301        self._reply_pending = None
2302        return message_bytes
2303
2304    def _read_serial(self, n_values, datatype):
2305        """Read values from the serial port and unpack them with struct."""
2306        fmt = self._STRUCT_FORMATS[datatype]
2307        n_values = int(n_values)
2308        n_bytes = n_values * struct.calcsize(fmt)
2309        message_bytes = self._read_raw(n_bytes)
2310
2311        values = struct.unpack(
2312            f"{self._ENDIANNESS}{n_values}{fmt}",
2313            message_bytes,
2314        )
2315        if n_values == 1:
2316            return values[0]
2317        return list(values)
2318
2319    def _read_ack(self, context, on_refusal=None, refusal_message=None, refusal_messages=None):
2320        """Read a one-byte acknowledgement from the device.
2321
2322        The device replies 1 if it executed the command, or 0 if it rejected
2323        the command because a value was out of range. On a 0, on_refusal()
2324        is called before the error is raised: the device resets a value it
2325        refuses, so callers use it to read the value back into the local copy.
2326        refusal_message replaces the error's usual message, for a command
2327        refused for another reason. refusal_messages maps other replies to
2328        their own message, e.g. op 90's 2.
2329        """
2330        try:
2331            acknowledgement = self._read_serial(1, "uint8")
2332        except PulsePalError as exc:
2333            raise PulsePalError(
2334                f"Pulse Pal did not confirm {context}: no reply arrived within {self.port.timeout} s. "
2335                f"{_common.NO_REPLY_HINT}"
2336            ) from exc
2337        if acknowledgement != 1:
2338            refusal = PulsePalError(
2339                (refusal_messages or {}).get(acknowledgement) or
2340                refusal_message or
2341                f"Pulse Pal rejected the command sent by {context}. "
2342                "This usually means that a channel number, parameter code or "
2343                "value was out of range for the connected device."
2344            )
2345            if on_refusal is not None:
2346                try:
2347                    on_refusal()
2348                except PulsePalError as exc:
2349                    raise refusal from exc
2350            raise refusal
2351
2352    def _pack_values(self, values, datatype):
2353        """Pack scalar, list/tuple, or NumPy array values into bytes."""
2354        fmt = self._STRUCT_FORMATS[datatype]
2355        values_list = self._as_list(values)
2356        min_value, max_value = self._TYPE_RANGES[datatype]
2357        normalized = []
2358        for value in values_list:
2359            value = int(value)
2360            if not min_value <= value <= max_value:
2361                raise PulsePalError(
2362                    f"Value {value} is out of range for {datatype} "
2363                    f"({min_value} to {max_value})."
2364                )
2365            normalized.append(value)
2366        return struct.pack(f"{self._ENDIANNESS}{len(normalized)}{fmt}", *normalized)
2367
2368    @staticmethod
2369    def _as_list(values):
2370        """Return values as a flat list, wrapping scalars in one."""
2371        if isinstance(values, np.ndarray):
2372            return values.ravel().tolist()
2373        if isinstance(values, (bytes, bytearray, str)):
2374            return [values]
2375        if isinstance(values, numbers.Number):
2376            return [values]
2377        try:
2378            return list(values)
2379        except TypeError:
2380            return [values]
2381
2382    def _volts_to_bits(self, value, name="voltage"):
2383        """Convert -10 V to +10 V to the corresponding DAC code.
2384
2385        Rounds to the nearest code, halves to even (round()), as the MATLAB and C++ classes do.
2386        Raises PulsePalError for a voltage outside [-10, 10] or not a number: clamping it would
2387        play a different voltage.
2388        """
2389        try:
2390            volts = float(value)
2391        except (TypeError, ValueError):
2392            volts = float("nan")
2393        if isinstance(value, (bool, np.bool_, str)) or not -10 <= volts <= 10:  # Also refuses NaN
2394            raise PulsePalError(f"{name} must be in [-10, 10] V. Received {value!r}.")
2395        return int(round((volts + 10) / 20 * self._DAC_BITMAX))
2396
2397    def _bits_to_volts(self, value):
2398        """Convert a DAC code to volts, snapping to a clean value within 1 LSB."""
2399        raw_volts = (float(value) / self._DAC_BITMAX * 20) - 10
2400        lsb_volts = 20.0 / self._DAC_BITMAX
2401        # The nearest 3 decimal number (e.g. 5.000, 4.255), if the code is within 1 LSB of it
2402        clean_volts = round(raw_volts, 3)
2403        if abs(raw_volts - clean_volts) <= lsb_volts:
2404            return clean_volts
2405        return round(raw_volts, 4)
2406
2407    def _seconds_to_cycles(self, value, name="time"):
2408        """Convert seconds to the corresponding refresh-cycle count.
2409
2410        Rounds to the nearest cycle, halves to even (round()), as the MATLAB and C++ classes do:
2411        125 us, 2.5 cycles, is 2 cycles in all three. Raises PulsePalError for a time that is
2412        negative, not a number, or longer than DeviceInfo.max_time.
2413        """
2414        try:
2415            seconds = float(value)
2416        except (TypeError, ValueError):
2417            seconds = float("nan")
2418        if isinstance(value, (bool, np.bool_, str)) or not (math.isfinite(seconds) and seconds >= 0):
2419            raise PulsePalError(f"{name} must be a time of 0 s or more. Received {value!r}.")
2420        cycles = int(round(seconds * float(self.info.cycle_frequency)))
2421        if cycles > self._max_time_cycles():
2422            raise PulsePalError(
2423                f"{name} must be at most {self.info.max_time} s, the longest time the device's screen "
2424                f"shows. Received {value!r}."
2425            )
2426        return cycles
2427
2428    def _max_time_cycles(self):
2429        """DeviceInfo.max_time in timer cycles: 999999990 on a 10 us timer, 199999998 on a 50 us timer."""
2430        return int(round(self.info.max_time * float(self.info.cycle_frequency)))
2431
2432    def _custom_time_cycles(self, value, name):
2433        """Convert a custom pulse time or sampling period to timer cycles.
2434
2435        It must be a whole number of DeviceInfo.min_pulse_width_us, to the nearest microsecond,
2436        as in the MATLAB class. A time between two steps is refused, not rounded
2437        as the time parameters are, so that a train plays as it was written.
2438        """
2439        self._seconds_to_cycles(value, name)  # Checks that it is a time the device can play
2440        microseconds = int(round(float(value) * 1e6))
2441        step_us = int(self.info.min_pulse_width_us)
2442        if microseconds % step_us:
2443            raise PulsePalError(
2444                f"{name} must be multiples of {step_us} us ({step_us / 1e6:g} s). Received {value!r}."
2445            )
2446        return microseconds // int(self.info.cycle_period_us)
2447
2448    def _cycles_to_seconds(self, value):
2449        """Convert hardware timer cycle counts to seconds."""
2450        return float(value) / float(self.info.cycle_frequency)