1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
|
"""
api_layers/protocols/scpi.py
SCPI (Standard Commands for Programmable Instruments) protocol layer.
Each channel maps to a query string. The layer sends each query in
sequence and parses the numeric response.
Terminator: LF (\\n) by default; instruments also accept CR+LF.
Response parsing strips unit suffixes — "+3.14159 V" → 3.14159.
Example channel setup:
SCPIChannel("V1", "Voltage", query="MEAS:VOLT?", unit="V")
SCPIChannel("I1", "Current", query="MEAS:CURR?", unit="A")
SCPIChannel("T1", "Temp", query="SENS:TEMP:DATA?", unit="°C")
"""
import math
import re
from dataclasses import dataclass, field
from typing import Dict, List, Optional
from api_layers.protocols.base_protocol import BaseProtocol
@dataclass
class SCPIChannel:
channel_id: str
name: str
query: str # e.g. "MEAS:VOLT?" or "MEAS:VOLT? (@1)"
unit: str = ""
scale: float = 1.0
write_cmd: str = "" # e.g. "VOLT {value}" — set to send writes
class SCPILayer(BaseProtocol):
def __init__(
self,
port: str,
baud: int = 9600,
channels: List[SCPIChannel] = None,
poll_interval: float = 0.2,
simulate: bool = True,
):
super().__init__(port, baud, poll_interval, simulate)
self.channels = channels or []
# ── Protocol ──────────────────────────────────────────────────────────
def _poll(self) -> Dict[str, float]:
result: Dict[str, float] = {}
for ch in self.channels:
if not ch.query.strip():
continue
try:
self._ser.write(f"{ch.query.strip()}\n".encode())
resp = self._ser.readline().decode(errors="replace").strip()
val = _parse_numeric(resp)
if val is not None:
result[ch.channel_id] = val * ch.scale
except Exception:
pass
return result
def _simulate(self, t: float) -> Dict[str, float]:
return {
ch.channel_id: math.sin(t * (0.3 + i * 0.2)) * 5.0 + i * 2.0
for i, ch in enumerate(self.channels)
}
def write(self, channel_id: str, value) -> bool:
ch = next((c for c in self.channels if c.channel_id == channel_id), None)
if ch is None or not ch.write_cmd:
return False
try:
cmd = ch.write_cmd.format(value=value)
self._ser.write(f"{cmd}\n".encode())
return True
except Exception:
return False
def query_idn(self) -> str:
"""Send *IDN? and return instrument identification string."""
if not self._ser:
return "(not connected)"
try:
self._ser.write(b"*IDN?\n")
return self._ser.readline().decode(errors="replace").strip()
except Exception as e:
return f"(error: {e})"
# ── Helpers ───────────────────────────────────────────────────────────────────
_NUM_RE = re.compile(r"[+-]?\d+\.?\d*(?:[eE][+-]?\d+)?")
def _parse_numeric(resp: str) -> Optional[float]:
"""Extract first float from SCPI response, ignoring unit suffixes."""
m = _NUM_RE.search(resp)
return float(m.group()) if m else None
|
