Skip to content

Commit 3e9e136

Browse files
cboulaycboulay
committed
Update DIN Python script and docs
1 parent b85e714 commit 3e9e136

2 files changed

Lines changed: 177 additions & 88 deletions

File tree

‎README.md‎

Lines changed: 34 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -306,45 +306,51 @@ eeg_sample, timestamp = eeg_inlet.pull_sample()
306306

307307
## Digital Inputs (DIN)
308308

309-
The EEG stream includes a `DIN` channel as the last channel, containing the raw 16-bit digital input value from the amplifier's digital I/O port.
309+
The DIN values are published on a **separate LSL stream** (not as a channel in the EEG stream). The stream is event-driven: a sample is only pushed when the DIN value changes.
310310

311-
### Channel Details
311+
### Stream Details
312312

313-
- **Label**: `DIN`
314-
- **Type**: `DIN`
315-
- **Unit**: `uint16`
316-
- **Position**: Last channel in the stream (after EEG and any Physio16 channels)
313+
- **Name**: `EGI NetAmp <amp_id>_DIN`
314+
- **Type**: `Markers`
315+
- **Format**: `int32`
316+
- **Rate**: Irregular (event-driven, only on change)
317+
- **Channels**: 1 (`DIN`)
317318
- **Value range**: 0-65535 (0x0000-0xFFFF)
319+
- **Idle value**: 0 (all inputs off)
318320

319-
### Accessing Individual Bits
321+
The amplifier's DIN lines are **active-low** with internal pull-ups. The application inverts the raw value so that `0` = idle and a set bit = active input.
320322

321-
The DIN value is transmitted as a float (or int32 in native format), but all 16-bit values are exactly preserved. To access individual bits:
323+
### Accessing Individual Bits
322324

323325
```python
324326
import pylsl
325327

326-
streams = pylsl.resolve_byprop('type', 'EEG', timeout=5)
327-
inlet = pylsl.StreamInlet(streams[0])
328-
329-
sample, timestamp = inlet.pull_sample()
330-
din_value = int(sample[-1]) # Last channel, convert to integer
331-
332-
# Extract individual bits (DIN1-DIN16)
333-
din1 = (din_value >> 0) & 1 # Bit 0
334-
din2 = (din_value >> 1) & 1 # Bit 1
335-
din3 = (din_value >> 2) & 1 # Bit 2
336-
# ... etc
337-
338-
# Or extract all 16 bits
339-
bits = [(din_value >> i) & 1 for i in range(16)]
340-
print(f"DIN1-16: {bits}")
328+
# Resolve the DIN stream (separate from EEG)
329+
streams = pylsl.resolve_byprop('type', 'Markers', timeout=5)
330+
din_streams = [s for s in streams if s.name().endswith('_DIN')]
331+
inlet = pylsl.StreamInlet(din_streams[0])
332+
333+
# Pull a DIN change event
334+
sample, timestamp = inlet.pull_sample(timeout=5.0)
335+
if sample:
336+
din_value = int(sample[0])
337+
338+
# Extract individual bits (DIN1-DIN16)
339+
din1 = (din_value >> 0) & 1 # Bit 0
340+
din2 = (din_value >> 1) & 1 # Bit 1
341+
din3 = (din_value >> 2) & 1 # Bit 2
342+
# ... etc
343+
344+
# Or extract all 16 bits
345+
bits = [(din_value >> i) & 1 for i in range(16)]
346+
print(f"DIN1-16: {bits}")
341347
```
342348

343349
### Timing Considerations
344350

345-
The amplifier's internal DIN ADC samples at a fixed 1 kHz rate, regardless of the EEG sample rate:
351+
The amplifier's internal DIN ADC samples at a fixed 1 kHz rate, regardless of the EEG sample rate. The DIN stream only emits a sample when the value changes, so the effective timing resolution depends on the EEG sample rate (which determines how often the DIN register is read):
346352

347-
| EEG Sample Rate | DIN Behavior | Effective DIN Resolution |
353+
| EEG Sample Rate | DIN Poll Rate | Effective DIN Resolution |
348354
|-----------------|--------------|--------------------------|
349355
| 250 Hz | Decimated (1 in 4 samples) | 4 ms |
350356
| 500 Hz | Decimated (1 in 2 samples) | 2 ms |
@@ -356,7 +362,9 @@ The amplifier's internal DIN ADC samples at a fixed 1 kHz rate, regardless of th
356362

357363
### Hardware Notes
358364

359-
The NA400/NA410 amplifiers have a 16-bit digital I/O port. By default, all bits are configured for input. The `cmd_SetDigitalInOutDirection` command can configure specific bits for output if needed (consult EGI documentation).
365+
The NA400/NA410 amplifiers have a 16-bit digital I/O port with active-low inputs (internal pull-ups). When no trigger device is connected, all lines are pulled high and the DIN stream reports `0` (idle). Grounding a line activates the corresponding bit.
366+
367+
The `cmd_SetDigitalInOutDirection` command can configure specific bits for output if needed (consult EGI documentation).
360368

361369
# Acknowledgements
362370
This application was written to behave near-identically to the BCI2000 AmpServer module that was originally created by EGI.

‎scripts/din_serial_test.py‎

Lines changed: 143 additions & 62 deletions
Original file line numberDiff line numberDiff line change
@@ -1,90 +1,171 @@
11
#!/usr/bin/env python3
22
"""
3-
Test script to probe DIN inputs via USB-serial connection.
4-
Run this while monitoring DIN output from EGIAmpServerCLI.
3+
DIN test script: toggle USB-serial output lines while monitoring the DIN LSL stream.
4+
5+
This uses DTR, RTS, and TX-break as 3 independent binary outputs on a standard
6+
USB-serial adapter. Depending on how your cable maps DB-9 pins to the amplifier's
7+
DIN connector, some or all of these may register as DIN bit changes.
8+
9+
DB-9 output pins:
10+
Pin 4: DTR (pyserial: ser.dtr)
11+
Pin 7: RTS (pyserial: ser.rts)
12+
Pin 3: TX (pyserial: ser.break_condition -- held low when True)
13+
14+
Note: RS-232 voltage levels may not be compatible with the amp's TTL-level DIN
15+
inputs. If no bits change, voltage level mismatch is likely.
16+
17+
Requirements:
18+
pip install pyserial pylsl
519
"""
620

7-
import serial
21+
import argparse
22+
import threading
823
import time
9-
import sys
1024

11-
PORT = "/dev/cu.usbserial-210"
12-
BAUD = 9600
25+
import pylsl
26+
import serial
27+
28+
29+
def monitor_din(stop_event: threading.Event, stream_name: str = ""):
30+
"""Background thread: resolve DIN marker stream and print changes."""
31+
inlet = None
32+
33+
while not stop_event.is_set():
34+
if inlet is None:
35+
print("[DIN] Resolving DIN stream (type='Markers', name='*_DIN')...")
36+
streams = pylsl.resolve_byprop("type", "Markers", timeout=5)
37+
din_streams = [s for s in streams if s.name().endswith("_DIN")]
38+
if stream_name:
39+
din_streams = [s for s in din_streams if stream_name in s.name()]
40+
if not din_streams:
41+
names = [s.name() for s in streams]
42+
print(f"[DIN] Not found yet (available Markers: {names}). "
43+
"Is the amp linked/streaming? Retrying...")
44+
continue
45+
46+
info = din_streams[0]
47+
print(f"[DIN] Connected to '{info.name()}'")
48+
print("[DIN] This stream only sends samples on DIN value *changes*.")
49+
print("[DIN] Waiting for events...\n")
50+
inlet = pylsl.StreamInlet(info, max_buflen=1)
51+
52+
sample, ts = inlet.pull_sample(timeout=0.5)
53+
if sample is None:
54+
continue
55+
din = int(sample[0])
56+
bits = f"{din:016b}"
57+
active = [i + 1 for i in range(16) if (din >> i) & 1]
58+
active_str = f" active: DIN {','.join(map(str, active))}" if active else ""
59+
print(f"[DIN] value={din:5d} bits={bits} (0x{din:04X}){active_str}")
60+
1361

1462
def main():
15-
print(f"Opening {PORT} at {BAUD} baud...")
16-
ser = serial.Serial(PORT, BAUD, timeout=1)
17-
18-
print("\nSerial port opened. Current control line states:")
19-
print(f" DTR: {ser.dtr}")
20-
print(f" RTS: {ser.rts}")
21-
print(f" CTS: {ser.cts}")
22-
print(f" DSR: {ser.dsr}")
23-
print(f" CD: {ser.cd}")
24-
print(f" RI: {ser.ri}")
25-
26-
print("\n--- Interactive Mode ---")
63+
parser = argparse.ArgumentParser(description="Test DIN inputs via USB-serial adapter")
64+
parser.add_argument("--port", default="/dev/cu.usbserial-210",
65+
help="Serial port (default: /dev/cu.usbserial-210)")
66+
parser.add_argument("--stream", default="",
67+
help="LSL stream name filter (default: match any DIN stream)")
68+
args = parser.parse_args()
69+
70+
stop_event = threading.Event()
71+
72+
# Start DIN monitor in background
73+
din_thread = threading.Thread(target=monitor_din, args=(stop_event, args.stream),
74+
daemon=True)
75+
din_thread.start()
76+
77+
# Open serial port
78+
print(f"[SER] Opening {args.port}...")
79+
try:
80+
ser = serial.Serial(args.port, 9600, timeout=1)
81+
except serial.SerialException as e:
82+
print(f"[SER] Failed to open port: {e}")
83+
stop_event.set()
84+
return
85+
86+
# Start with all outputs in their 'idle' state
87+
ser.dtr = False
88+
ser.rts = False
89+
ser.break_condition = False
90+
time.sleep(0.5)
91+
92+
print(f"\n[SER] Port opened. Output pins:")
93+
print(f" DTR (pin 4): {'HIGH' if ser.dtr else 'LOW'}")
94+
print(f" RTS (pin 7): {'HIGH' if ser.rts else 'LOW'}")
95+
print(f" TX (pin 3): IDLE (high)")
96+
print()
2797
print("Commands:")
28-
print(" d - Toggle DTR")
29-
print(" r - Toggle RTS")
30-
print(" b - Toggle BREAK")
31-
print(" 0-9 - Send that digit as a byte")
32-
print(" s - Send 0x00")
33-
print(" f - Send 0xFF")
34-
print(" a - Send all bytes 0x00-0xFF rapidly")
35-
print(" t - Send test pattern (0x55, 0xAA)")
36-
print(" p - Print control line states")
37-
print(" q - Quit")
98+
print(" d - Toggle DTR")
99+
print(" r - Toggle RTS")
100+
print(" b - Toggle TX break (holds TX low)")
101+
print(" 1 - All outputs ON (DTR=1, RTS=1, break=1)")
102+
print(" 0 - All outputs OFF (DTR=0, RTS=0, break=0)")
103+
print(" scan - Toggle each output one at a time with pauses")
104+
print(" p - Print current output states")
105+
print(" q - Quit")
106+
print()
107+
print("Watch the [DIN] lines above for bit changes.")
38108
print()
39109

40110
break_state = False
41111

42-
while True:
112+
while not stop_event.is_set():
43113
try:
44114
cmd = input("> ").strip().lower()
45-
except EOFError:
115+
except (EOFError, KeyboardInterrupt):
46116
break
47117

48-
if cmd == 'q':
118+
if cmd == "q":
49119
break
50-
elif cmd == 'd':
120+
elif cmd == "d":
51121
ser.dtr = not ser.dtr
52-
print(f"DTR = {ser.dtr}")
53-
elif cmd == 'r':
122+
print(f"[SER] DTR = {ser.dtr}")
123+
elif cmd == "r":
54124
ser.rts = not ser.rts
55-
print(f"RTS = {ser.rts}")
56-
elif cmd == 'b':
125+
print(f"[SER] RTS = {ser.rts}")
126+
elif cmd == "b":
57127
break_state = not break_state
58128
ser.break_condition = break_state
59-
print(f"BREAK = {break_state}")
60-
elif cmd == 'p':
61-
print(f" DTR: {ser.dtr}, RTS: {ser.rts}")
62-
print(f" CTS: {ser.cts}, DSR: {ser.dsr}, CD: {ser.cd}, RI: {ser.ri}")
63-
elif cmd == 's':
64-
ser.write(b'\x00')
65-
print("Sent 0x00")
66-
elif cmd == 'f':
67-
ser.write(b'\xff')
68-
print("Sent 0xFF")
69-
elif cmd == 'a':
70-
print("Sending 0x00-0xFF...")
71-
for i in range(256):
72-
ser.write(bytes([i]))
73-
time.sleep(0.01)
74-
print("Done")
75-
elif cmd == 't':
76-
ser.write(b'\x55\xaa')
77-
print("Sent 0x55, 0xAA")
78-
elif cmd.isdigit():
79-
ser.write(cmd.encode())
80-
print(f"Sent '{cmd}' (0x{ord(cmd):02x})")
129+
print(f"[SER] TX break = {break_state}")
130+
elif cmd == "1":
131+
ser.dtr = True
132+
ser.rts = True
133+
break_state = True
134+
ser.break_condition = True
135+
print("[SER] All ON: DTR=1, RTS=1, break=1")
136+
elif cmd == "0":
137+
ser.dtr = False
138+
ser.rts = False
139+
break_state = False
140+
ser.break_condition = False
141+
print("[SER] All OFF: DTR=0, RTS=0, break=0")
142+
elif cmd == "scan":
143+
print("[SER] Scanning each output (2s per toggle)...")
144+
for name, on, off in [
145+
("DTR", lambda: setattr(ser, 'dtr', True),
146+
lambda: setattr(ser, 'dtr', False)),
147+
("RTS", lambda: setattr(ser, 'rts', True),
148+
lambda: setattr(ser, 'rts', False)),
149+
("TX break", lambda: setattr(ser, 'break_condition', True),
150+
lambda: setattr(ser, 'break_condition', False)),
151+
]:
152+
print(f"[SER] {name} ON")
153+
on()
154+
time.sleep(2)
155+
print(f"[SER] {name} OFF")
156+
off()
157+
time.sleep(2)
158+
break_state = False
159+
print("[SER] Scan complete.")
160+
elif cmd == "p":
161+
print(f"[SER] DTR={ser.dtr}, RTS={ser.rts}, break={break_state}")
81162
elif cmd:
82-
# Send arbitrary string
83-
ser.write(cmd.encode())
84-
print(f"Sent: {cmd!r}")
163+
print(f"Unknown command: {cmd}")
85164

165+
stop_event.set()
86166
ser.close()
87-
print("Closed.")
167+
print("Done.")
168+
88169

89170
if __name__ == "__main__":
90171
main()

0 commit comments

Comments
 (0)