from .trgen import TrgenPort
from .trgen_pin import TrgenPin, TrgenLevel, GPIODirection
from .instruction import istrActiveFor, istrUnactiveFor, istrRepeat, istrEnd, istrNotAdmissible
import socket
import time
CMD_PACKET_PROGRAM = 0x01
CMD_PACKET_START = 0x02
CMD_SET_GPIO = 0x03
CMD_REQ_IMPL = 0x04
CMD_REQ_STATUS = 0x05
CMD_SET_LEVEL = 0x06
CMD_REQ_GPIO = 0x07
CMD_REQ_LEVEL = 0x08
CMD_STOP_TRGEN = 0x09
CMD_SET_THR_VALUE = 0x0A
CMD_GET_THR_VALUE = 0x0B
CMD_START_CALIBRATION = 0x0C
[docs]
class LevelConfig:
"""
Class to configure trigger levels.
Example:
.. code-block:: python
from trgenpy import TrgenClient, TrgenPin
client = TrgenClient()
client.connect()
# Configura NS0 e SA1 come HIGH, GPIO0 come LOW
config = LevelConfig().high(TrgenPin.NS0, TrgenPin.SA1).low(TrgenPin.GPIO0).build()
client.set_level(config)
"""
def __init__(self):
self.levels = {}
def high(self, *pins):
for p in pins:
self.levels[p] = TrgenLevel.HIGH
return self
def low(self, *pins):
for p in pins:
self.levels[p] = TrgenLevel.LOW
return self
def build(self):
return self.encode_levels(self.levels)
[docs]
@staticmethod
def encode_levels(levels: dict[TrgenPin, TrgenLevel]) -> int:
"""
levels: dizionario {pin: TrgenLevel.HIGH/LOW}
ritorna: bitmask da passare a client.set_level()
"""
mask = 0
for pin, level in levels.items():
if level == TrgenLevel.HIGH:
mask |= (1 << pin)
return mask
[docs]
class DirectionConfig:
"""
Class to configure GPIO directions.
Example:
.. code-block:: python
config = DirectionConfig().output(TrgenPin.GPIO0, TrgenPin.GPIO1).input(TrgenPin.GPIO2).build()
client.set_gpio_direction(config)
"""
def __init__(self):
self.levels = {}
def output(self, *pins):
self._check_gpio(pins)
for p in pins:
self.levels[p] = GPIODirection.OUT
return self
def input(self, *pins):
self._check_gpio(pins)
for p in pins:
self.levels[p] = GPIODirection.IN
return self
def build(self):
return self.encode_direction(self.levels)
[docs]
@staticmethod
def encode_direction(levels: dict[TrgenPin, GPIODirection]) -> int:
"""
Encodes GPIO directions into a bitmask.
* 1: Output
* 0: Input
Args:
levels (dict): Dictionary mapping TrgenPin to GPIODirection (IN/OUT).
Returns:
int: Bitmask to pass to client.set_gpio_direction()
Example:
.. code-block:: python
levels = {
TrgenPin.GPIO0: GPIODirection.OUT,
TrgenPin.GPIO1: GPIODirection.IN,
TrgenPin.GPIO2: GPIODirection.OUT,
}
mask = DirectionConfig.encode_direction(levels)
"""
mask = 0
for pin, level in levels.items():
if level == GPIODirection.OUT:
mask |= (1 << pin)
return mask
@staticmethod
def _check_gpio(pins):
valid_gpio = {TrgenPin.GPIO0, TrgenPin.GPIO1, TrgenPin.GPIO2, TrgenPin.GPIO3,
TrgenPin.GPIO4, TrgenPin.GPIO5, TrgenPin.GPIO6, TrgenPin.GPIO7}
for pin in pins:
if pin not in valid_gpio:
raise InvalidPinError(f"Pin {pin} is not a GPIO pin. Allowed: GPIO0-GPIO7")
[docs]
class InvalidPinError(Exception):
"""Raised when a non-GPIO pin is used for GPIO configuration."""
pass
[docs]
class InvalidAckError(Exception):
"""Raised when an invalid ACK response is received from the Trgen."""
def __init__(self, expected, received):
super().__init__(f"Expected ACK{expected}, got '{received}'")
[docs]
class TrgenClient:
"""
Client used to communicate with TrgenPort Devices via TCP/IP sockets.
Attributes:
ip (str): IP address of the device, default is '192.168.123.1'
port (int): Communication port, default is 4242.
timeout (float): Timeout for the connection in seconds.
default_trigger_duration_us (int): Default trigger duration in microseconds.
Raises:
InvalidAckError: If the ACK response is invalid.
AckFormatError: If the ACK response is malformed.
TimeoutError: If the connection times out.
Example:
.. code-block:: python
from trgenpy import TrgenClient, TrgenPin
client = TrgenClient()
"""
def __init__(self, ip='192.168.123.1', port=4242, timeout=2.0, default_trigger_duration_us=20):
"""
Initialize the client for the Trgen.
Args:
ip (str): IP address of the device, default is '192.168.123.1'
port (int): Communication port, default is 4242.
timeout (float): Timeout for the connection in seconds.
default_trigger_duration_us (int): Default trigger duration in microseconds.
"""
self.ip = ip
self.port = port
self.timeout = timeout
self.default_trigger_duration_us = default_trigger_duration_us
self._impl = None
self._memory_length = 32
[docs]
def createTrgenPort(self, trigger_id: TrgenPin, instruction_set = None) -> TrgenPort:
"""
Create a TrgenPort object for the specified trigger ID.
Args:
trigger_id (TrgenPin): ID of the trigger to create.
Raises:
ValueError: If trigger_id is not a valid TrgenPin value.
Returns:
TrgenPort: The created TrgenPort object.
Example:
.. code-block:: python
client = TrgenClient()
trgen_port = client.createTrgenPort(TrgenPin.NS0)
print(trgen_port)
"""
# Validate trigger_id
if isinstance(trigger_id, TrgenPin):
pin_value = trigger_id.value
elif isinstance(trigger_id, int):
# Backward compatibility for integer IDs
valid_pin_ids = set(pin.value for pin in TrgenPin)
if trigger_id not in valid_pin_ids:
available_pins = ", ".join([f"{pin.name}({pin.value})" for pin in TrgenPin])
raise ValueError(
f"Invalid pin ID {trigger_id}. Available pins: {available_pins}"
)
pin_value = trigger_id
else:
raise TypeError(f"trigger_id must be TrgenPin or int, got {type(trigger_id).__name__}")
return TrgenPort(pin_value, self._memory_length)
# Connect to Trgen and get the TrgenPort Implementation
[docs]
def connect(self):
"""
Connect to the Trgen and retrieve its implementation details.
Raises:
InvalidAckError: If the ACK response is invalid.
AckFormatError: If the ACK response is malformed.
TimeoutError: If the connection times out.
"""
try:
self._impl = self.getImplementation()
self._memory_length = self._impl.memory_length
except InvalidAckError as e:
print(f"⚠️ ACK sbagliato: {e}")
except AckFormatError as e:
print(f"⚠️ ACK malformato: {e}")
except TimeoutError as e:
print(f"⏱️ Timeout: {e}")
# Get Device Availability
[docs]
def is_available(self):
"""
Verifica se il dispositivo TrgenPort è raggiungibile.
Returns:
bool: True se il dispositivo risponde, False altrimenti.
"""
try:
with socket.create_connection((self.ip, self.port), timeout=self.timeout) as sock:
return True
except (socket.timeout, ConnectionRefusedError, OSError):
return False
[docs]
def disable_trigger(self, trgenport):
"""
Disable a specific trgenport by sending a default program that ends immediately.
"""
packet_id = CMD_PACKET_PROGRAM | (trgenport.id << 24)
payload = [0] * self._memory_length # Default program (no-op)
return self.__send_packet(packet_id, payload)
# Start all triggers activities
[docs]
def start(self):
"""
Start the execution of triggers on the device.
"""
# Invia CMD_START (0x02) senza payload
return self.__send_packet(CMD_PACKET_START)
# Stop all triggers activities
[docs]
def stop(self):
"""
Stop the execution of triggers on the device.
"""
# Invia CMD_STOP (0x09) senza payload
return self.__send_packet(CMD_STOP_TRGEN)
# Send program command for single trgenport
[docs]
def writeTrgenMemory(self, trgenport: TrgenPort):
"""
Send the memory bank of one pre-programmed trigger Pin to the TrgenPort device.
Args:
trgenport (TrgenPort): Object to send.
Raises:
TypeError: If trgenport is not a TrgenPort instance.
ValueError: If trgenport.id is not a valid TrgenPin value.
Returns:
None
Example:
.. code-block:: python
trgen_port = TrgenPort(id=0, memory_length=32)
# Populate trgen_port.memory with instructions...
client.writeTrgenMemory(trgen_port)
"""
if not isinstance(trgenport, TrgenPort):
raise TypeError(f"Expected TrgenPort instance, got {type(trgenport).__name__}")
# Validate pin ID against TrgenPin enum values
valid_pin_ids = set(pin.value for pin in TrgenPin)
if trgenport.id not in valid_pin_ids:
valid_range = f"0-{max(valid_pin_ids)}" if valid_pin_ids else "none"
available_pins = ", ".join([f"{pin.name}({pin.value})" for pin in TrgenPin])
raise ValueError(
f"Invalid pin ID {trgenport.id}. Valid range: {valid_range}. "
f"Available pins: {available_pins}"
)
# Additional check: verify ID is within hardware implementation limits
if self._impl is not None:
max_supported_pins = self._impl.total_triggers
if trgenport.id >= max_supported_pins:
raise ValueError(
f"Pin ID {trgenport.id} exceeds hardware limits. "
f"This device supports pins 0-{max_supported_pins-1}"
)
packet_id = CMD_PACKET_PROGRAM | (trgenport.id << 24)
return self.__send_packet(packet_id, trgenport.memory)
# Set the Polarity Level for single TrgenPort
[docs]
def set_level(self, level_mask):
"""
Set the trigger polarity (bitmask: 1 = active high, 0 = active low)
Args:
level_mask (int): Bitmask representing the desired levels for each trigger pin.
Raises:
ValueError: If level_mask is not in the range 0-0x03FFFFFF (26 bits).
ValueError: If level_mask is not in the range 0-0x03FFFFFF (26 bits).
Example:
.. code-block:: python
client = TrgenClient()
level_mask = 0b00000000000000000000000011 # NS0 and NS1 HIGH, others LOW
client.set_level(level_mask)
"""
packet_id = CMD_SET_LEVEL
payload = [level_mask]
return self.__send_packet(packet_id, payload)
[docs]
def get_level(self):
"""
Get the trigger active polarity
"""
ack = self.__send_packet(CMD_REQ_LEVEL)
return self.__parse_ack_value(ack, expected_id=CMD_REQ_LEVEL)
# Get Trgen status
[docs]
def get_status(self):
"""
Get current State for all triggers
"""
ack = self.__send_packet(CMD_REQ_STATUS)
return self.__parse_ack_value(ack, expected_id=CMD_REQ_STATUS)
[docs]
def set_gpio_direction(self, gpio_mask):
"""
Set the direction (bitmask da 0 a 7) for GPIO
Args:
gpio_mask (int): Bitmask representing the desired GPIO directions (1=output, 0=input).
Raises:
ValueError: If gpio_mask is not in the range 0-0xFF (8 bits).
Returns:
None
Example:
.. code-block:: python
client = TrgenClient()
client.set_gpio_direction(0b00001111) # GPIO0-3 ON, GPIO4-7 OFF
"""
packet_id = CMD_SET_GPIO
payload = [gpio_mask]
return self.__send_packet(packet_id, payload)
[docs]
def get_gpio_direction(self):
"""
Get current direction for GPIO
"""
ack = self.__send_packet(CMD_REQ_GPIO)
return self.__parse_ack_value(ack, expected_id=CMD_REQ_GPIO)
[docs]
def getImplementation(self):
"""
Get current implementation for Trgen
Returns:
TrgenImplementation: Implementation details of the connected Trgen.
Raises:
InvalidAckError: If the ACK response is invalid.
AckFormatError: If the ACK response is malformed.
TimeoutError: If the connection times out.
"""
ack = self.__send_packet(CMD_REQ_IMPL) #
value = self.__parse_ack_value(ack, expected_id=0x04)
from .implementation import TrgenImplementation
impl = TrgenImplementation.from_packed_value(value)
print(f"[TRGEN] Config: ns={impl.ns_num}, sa={impl.sa_num}, "
f"bnco={impl.bnco_num}, bnci={impl.bnci_num}, "
f"gpio={impl.gpio_num}, mtml={impl.mtml} → memory_length={impl.memory_length}")
return impl
[docs]
def setDefaultDuration(self, duration_us: int):
"""
Set the default trigger duration in microseconds.
Args:
duration_us (int): Duration in microseconds
"""
if duration_us <= 0:
raise ValueError("Duration must be positive")
self.default_trigger_duration_us = duration_us
[docs]
def getDefaultDuration(self) -> int:
"""
Get the current default trigger duration in microseconds.
Returns:
int: Default duration in microseconds
"""
return self.default_trigger_duration_us
[docs]
def getTresholdValue(self, photod_num: int):
"""
Get the threshold value for a specific photodiode.
Sends a ``trgen_thr_t`` struct with ``photod_num`` set; the device
responds with a ``uint32_t`` where:
* bits 0-15 → threshold value
* bit 16 → trigger flag
Args:
photod_num (int): Photodiode index (0-based).
Returns:
dict: ``{'photod_num': int, 'threshold': int, 'trigger': bool}``
Example:
.. code-block:: python
result = client.getTresholdValue(0)
print(result['threshold'], result['trigger'])
"""
import struct
# Packed struct layout (C): uint8_t photod_num | uint16_t threshold | bool trigger
# threshold and trigger fields are don't-care on the request side
payload_bytes = struct.pack('!BHB', photod_num, 0, 0)
ack = self.__send_packet(CMD_GET_THR_VALUE, payload_bytes)
value32bit = self.__parse_ack_value(ack, expected_id=CMD_GET_THR_VALUE)
#return {
# 'photod_num': photod_num,
# 'threshold': value32bit & 0xFFFF,
# 'trigger': bool(value32bit & 0x10000),
#}
return value32bit & 0xFFFF
[docs]
def setTresholdValue(self, photod_num: int, threshold: int):
"""
Set the threshold value for a specific photodiode.
Sends a ``trgen_thr_t`` struct with ``photod_num`` and ``threshold``
in network byte order (equivalent to C ``htons()``).
Args:
photod_num (int): Photodiode index (0-based).
threshold (int): Threshold value to set (uint16).
Example:
.. code-block:: python
client.setTresholdValue(0, 512)
"""
import struct
# threshold is stored in network byte order (big-endian) as per C htons()
payload_bytes = struct.pack('!BHB', photod_num, threshold, 0)
self.__send_packet(CMD_SET_THR_VALUE, payload_bytes)
[docs]
def startCalibration(self, count: int = 1):
"""
Start the calibration process and read back the calibrated threshold values.
Sends ``CMD_START_CALIBRATION`` with no payload, waits 100 ms for the
hardware to complete the calibration, then reads back the threshold
value for each photodiode via ``getTresholdValue``.
Args:
count (int): Number of photodiodes to read back after calibration.
Defaults to 1.
Returns:
list[dict]: List of ``{'photod_num', 'threshold', 'trigger'}`` dicts,
one per photodiode.
Example:
.. code-block:: python
results = client.startCalibration(count=4)
for r in results:
print(r['photod_num'], r['threshold'], r['trigger'])
"""
import time
self.__send_packet(CMD_START_CALIBRATION)
# Wait 100 ms for the hardware to complete calibration (mirrors C usleep(100000))
time.sleep(0.1)
results = []
for i in range(count):
result = self.getTresholdValue(i)
results.append(result)
return results
def __send_packet(self, packet_id, payload=None):
"""
Send a packet to the TrgenPort device.
Args:
packet_id (int): The ID of the packet to send.
payload (list or bytes, optional): The payload data to include in the packet.
If a list, each element is packed as a little-endian uint32.
If bytes/bytearray, sent as-is (used for packed structs).
Raises:
ConnectionError: If the connection to the TrgenPort device fails.
TimeoutError: If the request times out.
"""
try:
if payload is None:
payload_bytes = b''
elif isinstance(payload, (bytes, bytearray)):
payload_bytes = bytes(payload)
else:
from struct import pack
payload_bytes = b''.join(pack('<I', w) for w in payload)
from struct import pack
from .crc import compute_crc32
header = pack('<I', packet_id)
raw = header + payload_bytes
crc = pack('<I', compute_crc32(raw))
packet = raw + crc
with socket.create_connection((self.ip, self.port), timeout=self.timeout) as sock:
sock.sendall(packet)
try:
response = sock.recv(64)
return response.decode(errors='ignore')
except socket.timeout:
raise TimeoutError(f"No ACK received for packet 0x{packet_id:02X}")
except socket.timeout:
raise TimeoutError(f"⏱️ Connection timed out towards Trgen [{self.ip}:{self.port}]")
except OSError as e:
raise ConnectionError(f"🔌 Connection failed towards Trgen [{self.ip}:{self.port}] – {e.strerror or str(e)}")
def __parse_ack_value(self, ack_str, expected_id):
"""
Parse the ACK response from the TrgenPort device.
Args:
ack_str (str): The ACK response string.
expected_id (int): The expected packet ID.
Raises:
InvalidAckError: If the ACK response is invalid.
AckFormatError: If the ACK response is malformed.
"""
if not ack_str.startswith(f"ACK{expected_id}"):
raise InvalidAckError(f"Unexpected ACK: '{ack_str}'")
parts = ack_str.strip().split(".")
if len(parts) != 2:
raise AckFormatError(f"ACK format invalid: '{ack_str}'")
return int(parts[1])
[docs]
def resetTrgenPort(self, trgenport):
"""
Reset a specific trigger to a safe state.
"""
trgenport.setInstruction(0, istrEnd())
for i in range(1, self._memory_length):
trgenport.setInstruction(i,istrNotAdmissible())
self.writeTrgenMemory(trgenport)
[docs]
def resetAllTrgenPorts(self):
"""
Reset all triggers to a safe state.
"""
self.__reset_all_gpio()
self.__reset_all_sa()
self.__reset_all_ns()
self.__reset_all_bnc()
self.__reset_all_pd()
def __reset_all_pd(self):
"""
Reset all PD triggers to a safe state.
"""
pdPinoutMap = [
TrgenPin.PD,
]
for id in pdPinoutMap:
pd = self.createTrgenPort(id)
pd.setInstruction(0, istrEnd())
for i in range(1, self._memory_length):
pd.setInstruction(i,istrNotAdmissible())
self.writeTrgenMemory(pd)
def __reset_all_gpio(self):
"""
Reset all GPIO triggers to a safe state.
"""
gpioPinoutMap = [
TrgenPin.GPIO0,
TrgenPin.GPIO1,
TrgenPin.GPIO2,
TrgenPin.GPIO3,
TrgenPin.GPIO4,
TrgenPin.GPIO5,
TrgenPin.GPIO6,
TrgenPin.GPIO7
]
for id in gpioPinoutMap:
gpio = self.createTrgenPort(id)
gpio.setInstruction(0, istrEnd())
for i in range(1, self._memory_length):
gpio.setInstruction(i,istrNotAdmissible())
self.writeTrgenMemory(gpio)
def __reset_all_sa(self):
"""
Reset all Synamaps triggers to a safe state.
"""
synampsPinoutMap = [
TrgenPin.SA0,
TrgenPin.SA1,
TrgenPin.SA2,
TrgenPin.SA3,
TrgenPin.SA4,
TrgenPin.SA5,
TrgenPin.SA6,
TrgenPin.SA7
]
for id in synampsPinoutMap:
sa = self.createTrgenPort(id)
sa.setInstruction(0, istrEnd())
for i in range(1, self._memory_length):
sa.setInstruction(i,istrNotAdmissible())
self.writeTrgenMemory(sa)
def __reset_all_ns(self):
"""
Reset all NeuroScan triggers to a safe state.
"""
neuroscanPinoutMap = [
TrgenPin.NS0,
TrgenPin.NS1,
TrgenPin.NS2,
TrgenPin.NS3,
TrgenPin.NS4,
TrgenPin.NS5,
TrgenPin.NS6,
TrgenPin.NS7
]
for id in neuroscanPinoutMap:
ns = self.createTrgenPort(id)
ns.setInstruction(0, istrEnd())
for i in range(1, self._memory_length):
ns.setInstruction(i,istrNotAdmissible())
self.writeTrgenMemory(ns)
def __reset_all_bnc(self):
"""
Reset all BNCO triggers to a safe state.
"""
bncoPinoutMap = [
TrgenPin.BNCO,
TrgenPin.BNCI
]
for id in bncoPinoutMap:
bnco = self.createTrgenPort(id)
bnco.setInstruction(0, istrEnd())
for i in range(1, self._memory_length):
bnco.setInstruction(i,istrNotAdmissible())
self.writeTrgenMemory(bnco)
# Prende in input un oggetto TrgenPort e lo programma
# con un impulso di default di durata configurabile (default 20µs)
[docs]
def programDefaultTrigger(self, trgenport, us=None):
"""
Program one trigger (by Id) with a default pulse.
Args:
trgenport (TrgenPort): TrgenPort object to program.
us (int, optional): duration of the pulse in microseconds.
If None, uses the client's default duration.
"""
# Use client's default duration if not specified
if us is None:
us = self.default_trigger_duration_us
# TODO check if not Only Input (BNCI or GPIO Inputs)
trgenport.setInstruction(0, istrActiveFor(us))
trgenport.setInstruction(1, istrUnactiveFor(3))
trgenport.setInstruction(2, istrEnd())
for i in range(3, self._memory_length):
trgenport.setInstruction(i, istrNotAdmissible())
# Invio del trigger
self.writeTrgenMemory(trgenport)
# decode and send out a marker to all ports
# You can also choose to send individually to:
# - NeuroScan 25Pin Serial connector
# - Synamps 15Pin Serial connector
# - GPIO 8Pin DIN connector
#
#
[docs]
def sendMarker(self, markerNS=0, markerSA=0, markerGPIO=0, LSB=False,stop=True, autoStart=True):
"""
Send a marker to NS, SA and/or GPIO connectors.
Args:
markerNS (int, optional): Marker for NeuroScan.
markerSA (int, optional): Marker for Synamps.
markerGPIO (int, optional): Marker for GPIO.
LSB (bool, optional): If True, use the least significant bit first.
stop (bool, optional): If True, stop the trigger sequence after sending.
autoStart (bool, optional): If True, start the trigger sequence automatically.
"""
if markerNS and markerSA and markerGPIO == None:
return
neuroscanMap = [
TrgenPin.NS0,
TrgenPin.NS1,
TrgenPin.NS2,
TrgenPin.NS3,
TrgenPin.NS4,
TrgenPin.NS5,
TrgenPin.NS6,
TrgenPin.NS7
]
synampsMap = [
TrgenPin.SA0,
TrgenPin.SA1,
TrgenPin.SA2,
TrgenPin.SA3,
TrgenPin.SA4,
TrgenPin.SA5,
TrgenPin.SA6,
TrgenPin.SA7
]
gpioMap = [
TrgenPin.GPIO0,
TrgenPin.GPIO1,
TrgenPin.GPIO2,
TrgenPin.GPIO3,
TrgenPin.GPIO4,
TrgenPin.GPIO5,
TrgenPin.GPIO6,
TrgenPin.GPIO7,
]
self.__reset_all_ns()
self.__reset_all_sa()
self.__reset_all_gpio()
self.__reset_all_bnc()
self.__reset_all_pd()
maskNS = list(format(markerNS, 'b').zfill(8))
maskSA = list(format(markerSA, 'b').zfill(8))
maskGPIO = list(format(markerGPIO, 'b').zfill(8))
if LSB == False:
maskNS = maskNS[::-1]
maskSA = maskSA[::-1]
maskGPIO = maskGPIO[::-1]
for idx, i in enumerate(maskNS):
if maskNS[idx] == '1':
if(markerNS != None):
nsx = self.createTrgenPort(neuroscanMap[idx])
self.programDefaultTrigger(nsx)
for idx, i in enumerate(maskGPIO):
if maskGPIO[idx] == '1':
if(markerGPIO != None):
sax = self.createTrgenPort(gpioMap[idx])
self.programDefaultTrigger(sax)
for idx, i in enumerate(maskSA):
if maskSA[idx] == '1':
if(markerSA != None):
sax = self.createTrgenPort(synampsMap[idx])
self.programDefaultTrigger(sax)
# Avvio sequenza
if autoStart:
self.start()
if stop == True:
# Stop alla sequenza del trigger
self.stop()
[docs]
def sendTrigger(self,trgenPinList = None,autoStart=True):
"""
Send a custom trigger signal out of the BNC (BNCO).
Args:
triggerList (list of TrgenPin): List of :class:`TrgenPin` to be triggered.
"""
try:
# Verifica connessione prima di procedere
if not self.is_available():
raise ConnectionError(f"TrgenPort device not reachable at {self.ip}:{self.port}")
if trgenPinList != None:
if not isinstance(trgenPinList, list):
raise ValueError("triggerList must be a list of TrgenPin")
if len(trgenPinList) == 0:
raise ValueError("triggerList cannot be empty")
for tid in trgenPinList:
if not isinstance(tid, TrgenPin):
raise ValueError("triggerList must contain only TrgenPin values")
if TrgenPin.BNCI in trgenPinList:
raise ValueError("BNCI is input only and cannot be used in triggerList")
# reset all triggers
self.resetAllTrgenPorts()
for tid in trgenPinList:
# create trigger
tr = self.createTrgenPort(tid)
# set default program for each one
self.programDefaultTrigger(tr)
else:
# create trigger for default BNCO
tr = self.createTrgenPort(TrgenPin.BNCO)
self.programDefaultTrigger(tr)
# start
if autoStart:
self.start()
# Calcola il delay basato sulla durata del trigger + margine
trigger_duration_ms = self.default_trigger_duration_us / 1000.0
delay_time = max(0.001, trigger_duration_ms + 0.005) # Minimo 1ms, o durata trigger + 5ms
import time
time.sleep(delay_time)
except Exception as e:
print(f"❌ Errore durante l'invio del trigger: {e}")
finally:
print("✅ Trigger sequence completed")
return True
[docs]
def callbackCustomTrigger(self, ne=False, trgenPinList=None, inputPin=None,customInstructions = None):
"""
Send a custom trigger signal out of the BNC (BNCO).
Args:
triggerList (list of TrgenPin): List of :class:`TrgenPin` to be triggered, when None uses BNCO.
inputPin (TrgenPin): ID del pin di input (BNCI o GPIO)
ne (bool): Se True, attende fronte negativo; se False, fronte positivo
customInstructions (list of Instruction, optional): List of custom instructions to set for each trigger.
Raises:
ValueError: If triggerList is not a list of TrgenPin or is empty.
ValueError: If inputPin is not a valid TrgenPin.
Example:
.. code-block:: python
from trgenpy import TrgenClient, TrgenPin, istrActiveFor, istrUnactiveFor, istrEnd
client = TrgenClient()
client.connect()
client.callbackCustomTrigger(
ne=False,
trgenPinList=[TrgenPin.NS0, TrgenPin.SA1],
inputPin=TrgenPin.BNCI,
customInstructions=[
istrActiveFor(50),
istrUnactiveFor(50),
istrEnd()
]
)
"""
# Configure the GPIO direction if inputPin is of GPIO type
from .instruction import istrWaitNE, istrWaitPE, istrActiveFor, istrUnactiveFor, istrEnd, istrNotAdmissible
edge = ne
#if not isinstance(trgenPinList, list):
# raise ValueError("triggerList must be a list of TrgenPin")
#if len(trgenPinList) == 0:
# raise ValueError("triggerList cannot be empty")
if trgenPinList != None and isinstance(trgenPinList, list):
for tid in trgenPinList:
if not isinstance(tid, TrgenPin):
raise ValueError("triggerList must contain only TrgenPin values")
if TrgenPin.BNCI in trgenPinList:
raise ValueError("BNCI is input only and cannot be used in triggerList")
# GPIO validation
valid_gpio = {TrgenPin.GPIO0, TrgenPin.GPIO1, TrgenPin.GPIO2, TrgenPin.GPIO3,
TrgenPin.GPIO4, TrgenPin.GPIO5, TrgenPin.GPIO6, TrgenPin.GPIO7}
if inputPin in valid_gpio and inputPin in trgenPinList:
print("❌ GPIO cannot be used as both input and output simultaneously")
return
if trgenPinList == None or len(trgenPinList) == 0:
bnco = self.createTrgenPort(TrgenPin.BNCO)
# Configure instructions
if edge:
bnco.setInstruction(0, istrWaitNE(inputPin))
else:
bnco.setInstruction(0, istrWaitPE(inputPin))
if customInstructions is None or len(customInstructions) == 0:
bnco.setInstruction(1, istrActiveFor(20))
bnco.setInstruction(2, istrUnactiveFor(20))
bnco.setInstruction(3, istrEnd())
else:
# Use custom instructions
for i, instruction in enumerate(customInstructions):
if i < self._memory_length:
bnco.setInstruction(i+1, instruction)
# Fill the rest if necessary
for i in range(len(customInstructions), self._memory_length):
bnco.setInstruction(i+1, istrNotAdmissible())
# Program the trigger
self.writeTrgenMemory(bnco)
else:
for tid in trgenPinList:
cedge = edge
if tid == TrgenPin.PD:
cedge = not edge
# create trigger
tr = self.createTrgenPort(tid)
# set default program for each one
# self.programDefaultTrigger(tr)
# Configure instructions
if cedge:
tr.setInstruction(0, istrWaitNE(inputPin))
else:
tr.setInstruction(0, istrWaitPE(inputPin))
if customInstructions is None or len(customInstructions) == 0:
tr.setInstruction(1, istrActiveFor(20))
tr.setInstruction(2, istrUnactiveFor(20))
tr.setInstruction(3, istrEnd())
else:
# Use custom instructions
for i, instruction in enumerate(customInstructions):
if i < self._memory_length:
tr.setInstruction(i+1, instruction)
# Fill the rest if necessary
for i in range(len(customInstructions), self._memory_length):
tr.setInstruction(i+1, istrNotAdmissible())
# Program the trigger
self.writeTrgenMemory(tr)
[docs]
def callbackMarker(self, markerNS=0, markerSA=0, markerGPIO=0, LSB=False, inputPin=None, ne=False):
"""
Callback for marker to NS, SA and/or GPIO connectors based on input trigger.
Args:
markerNS (int, optional): Marker for NeuroScan.
markerSA (int, optional): Marker for Synamps.
markerGPIO (int, optional): Marker for GPIO.
inputPin (TrgenPin): ID del pin di input (BNCI o GPIO)
LSB (bool, optional): If True, use the least significant bit first.
ne (bool): Se True, attende fronte negativo; se False, fronte positivo
Raises:
ValueError: If inputPin is not a valid TrgenPin.
Example:
.. code-block:: python
from trgenpy import TrgenClient, TrgenPin
client = TrgenClient()
client.connect()
client.callbackMarker(
markerNS=5,
markerSA=3,
markerGPIO=12,
inputPin=TrgenPin.BNCI,
LSB=True,
ne=False
)
"""
from .instruction import istrWaitNE, istrWaitPE, istrActiveFor, istrUnactiveFor, istrEnd, istrNotAdmissible
edge = ne
# PIN validation
valid_gpio = {TrgenPin.GPIO0, TrgenPin.GPIO1, TrgenPin.GPIO2, TrgenPin.GPIO3,
TrgenPin.GPIO4, TrgenPin.GPIO5, TrgenPin.GPIO6, TrgenPin.GPIO7}
if inputPin in valid_gpio and markerGPIO != None:
print("❌ GPIO cannot be used as both input and output simultaneously")
return
if inputPin in valid_gpio and markerGPIO != 0:
print("❌ GPIO cannot be used as both input and output simultaneously")
return
# Configure the GPIO direction if inputPin is of GPIO type
if inputPin in valid_gpio:
gpio_direction_config = DirectionConfig().input(inputPin).build()
self.set_gpio_direction(gpio_direction_config)
if inputPin is None:
print("❌ Pin di input non specificato")
return
if inputPin != TrgenPin.BNCI and inputPin != TrgenPin.PD and inputPin not in valid_gpio:
print(f"❌ Pin di input non valido: {inputPin}. Solo BNCI, PD e GPIO sono supportati")
return
if inputPin == TrgenPin.PD:
edge = not edge
if markerNS and markerSA and markerGPIO == None:
return
if markerNS and markerSA and markerGPIO == 0:
return
neuroscanMap = [
TrgenPin.NS0,
TrgenPin.NS1,
TrgenPin.NS2,
TrgenPin.NS3,
TrgenPin.NS4,
TrgenPin.NS5,
TrgenPin.NS6,
TrgenPin.NS7
]
synampsMap = [
TrgenPin.SA0,
TrgenPin.SA1,
TrgenPin.SA2,
TrgenPin.SA3,
TrgenPin.SA4,
TrgenPin.SA5,
TrgenPin.SA6,
TrgenPin.SA7
]
gpioMap = [
TrgenPin.GPIO0,
TrgenPin.GPIO1,
TrgenPin.GPIO2,
TrgenPin.GPIO3,
TrgenPin.GPIO4,
TrgenPin.GPIO5,
TrgenPin.GPIO6,
TrgenPin.GPIO7,
]
self.resetAllTrgenPorts()
# convert markers to bitmasks
maskNS = list(format(markerNS, 'b').zfill(8))
maskSA = list(format(markerSA, 'b').zfill(8))
maskGPIO = list(format(markerGPIO, 'b').zfill(8))
if LSB == False:
maskNS = maskNS[::-1]
maskSA = maskSA[::-1]
maskGPIO = maskGPIO[::-1]
for idx, i in enumerate(maskNS):
if maskNS[idx] == '1':
if(markerNS != None):
nsx = self.createTrgenPort(neuroscanMap[idx])
if edge:
nsx.setInstruction(0, istrWaitNE(inputPin))
else:
nsx.setInstruction(0, istrWaitPE(inputPin))
nsx.setInstruction(1, istrActiveFor(20))
nsx.setInstruction(2, istrUnactiveFor(20))
nsx.setInstruction(3, istrEnd())
for i in range(4, self._memory_length):
nsx.setInstruction(i, istrNotAdmissible())
# Programma il trigger NSx
self.writeTrgenMemory(nsx)
for idx, i in enumerate(maskGPIO):
if maskGPIO[idx] == '1':
if(markerGPIO != None):
gpiox = self.createTrgenPort(gpioMap[idx])
if edge:
gpiox.setInstruction(0, istrWaitNE(inputPin))
else:
gpiox.setInstruction(0, istrWaitPE(inputPin))
gpiox.setInstruction(1, istrActiveFor(20))
gpiox.setInstruction(2, istrUnactiveFor(20))
gpiox.setInstruction(3, istrEnd())
for i in range(4, self._memory_length):
gpiox.setInstruction(i, istrNotAdmissible())
# Programma il trigger GPIOx
self.writeTrgenMemory(gpiox)
for idx, i in enumerate(maskSA):
if maskSA[idx] == '1':
if(markerSA != None):
sax = self.createTrgenPort(synampsMap[idx])
if edge:
sax.setInstruction(0, istrWaitNE(inputPin))
else:
sax.setInstruction(0, istrWaitPE(inputPin))
sax.setInstruction(1, istrActiveFor(20))
sax.setInstruction(2, istrUnactiveFor(20))
sax.setInstruction(3, istrEnd())
for i in range(4, self._memory_length):
sax.setInstruction(i, istrNotAdmissible())
# Program the SAx trigger
self.writeTrgenMemory(sax)
[docs]
def setTrgenPortInstructions(self, trgenport: TrgenPort, instruction_set):
"""
Set a complete instruction set to a TrgenPort.
This method automatically handles:
- Setting each instruction in sequence
- Adding istrNotAdmissible() for remaining memory slots
- Memory bounds validation
Args:
trgenport (TrgenPort): The TrgenPort to program.
instruction_set (List[int]): List of instructions to set.
Must end with istrEnd().
Raises:
ValueError: If instruction_set is empty, too long, or doesn't end with istrEnd().
TypeError: If trgenport is not a TrgenPort instance.
Example:
.. code-block:: python
instruction_set = [
istrWaitPE(TrgenPin.BNCI),
istrActiveFor(50),
istrUnactiveFor(10),
istrEnd()
]
bnco = client.createTrgenPort(TrgenPin.BNCO)
client.setTrgenPortInstructions(bnco, instruction_set)
client.writeTrgenMemory(bnco)
"""
from .trgen import TrgenPort
from .instruction import istrNotAdmissible, istrEnd
# Validate inputs
if not isinstance(trgenport, TrgenPort):
raise TypeError(f"Expected TrgenPort instance, got {type(trgenport).__name__}")
if instruction_set is None or len(instruction_set) == 0:
raise ValueError("Instruction set cannot be empty")
# Check memory bounds
if len(instruction_set) > self._memory_length:
raise ValueError(
f"Instruction set too long: {len(instruction_set)} instructions, "
f"but memory length is {self._memory_length}"
)
# Validate that instruction set ends with istrEnd()
if instruction_set[-1] != istrEnd():
raise ValueError("Instruction set must end with istrEnd()")
# Set instructions sequentially
for i, instruction in enumerate(instruction_set):
trgenport.setInstruction(i, instruction)
# Fill remaining memory slots with istrNotAdmissible()
for i in range(len(instruction_set), self._memory_length):
trgenport.setInstruction(i, istrNotAdmissible())
print(f"✅ Set {len(instruction_set)} instructions for {trgenport.type} port (ID: {trgenport.id})")
print(f" Filled remaining {self._memory_length - len(instruction_set)} slots with istrNotAdmissable()")