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
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
|
"""
Provide basic class to build an operational scope
Created by Eugeniy E. Mikhailov 2021/11/29
"""
import numpy as np
from qolab.hardware.scpi import SCPIinstr
from qolab.hardware.basic import BasicInstrument
from qolab.data.trace import TraceSetSameX
import time
import logging
logging.basicConfig(
format="%(asctime)s %(levelname)8s %(name)s: %(message)s",
datefmt="%m/%d/%Y %H:%M:%S",
)
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
def calcSparsingAndNumPoints(availableNpnts=None, maxRequiredPoints=None):
"""Calculate sparcing and number of sparced points.
Parameters
----------
availableNpnts: int or None (throws error)
Number of available points. If set to None exit with error
maxRequiredPoints: int or None (throws error)
number of requested points after decimation.
If availableNpnts< maxRequiredPoints*2,
decimation is impossible and we will get up to factor of 2 more
than requested.
Return
------
(sparsing, Npnts, availableNpnts, maxRequiredPoints)
"""
if availableNpnts is None:
raise ValueError("Invalid availableNpnts value, must be int.")
if maxRequiredPoints is None:
raise ValueError("Invalid maxRequiredPoints value, must be int.")
if availableNpnts <= maxRequiredPoints * 2:
Npnts = availableNpnts
sparsing = 1
else:
sparsing = int(np.floor(availableNpnts / maxRequiredPoints))
Npnts = int(np.floor(availableNpnts / sparsing))
return (sparsing, Npnts, availableNpnts, maxRequiredPoints)
class Scope(BasicInstrument):
"""Minimal class to implement a scope.
Intended to be used as a parent for hardware aware scopes.
Provide a minimal set of methods to be implemented by a scope.
"""
def __init__(self, *args, **kwds):
BasicInstrument.__init__(self, *args, **kwds)
self.config["Device type"] = "Scope"
self.config["Device model"] = "Generic Scope Without Hardware interface"
self.config["FnamePrefix"] = "scope"
self.numberOfChannels = 0
# deviceProperties must have 'get' and preferably 'set' methods available,
# i.e. 'SampleRate' needs getSampleRate() and love to have setSampleRate(value)
# they will be used to obtain config and set device according to it
self.deviceProperties.update(
{"SampleRate", "TimePerDiv", "TrigDelay", "TriggerMode", "Roll", "Run"}
)
# same is applied to channelProperties
# but we need setter/getter with channel number
# i.e. VoltsPerDiv must provide
# getChanVoltsPerDiv(chNum) and setSampleRate(chNum, value)
self.channelProperties = {
"VoltsPerDiv",
"VoltageOffset",
}
def getTrace(
self, chNum, availableNpnts=None, maxRequiredPoints=None, decimate=True
):
# Should work with minimal arguments list
# but might be faster if parameters provided: less IO requests
# old_trg_mode = self.getTriggerMode()
# self.setTriggerMode('STOP'); # to get synchronous channels
raise NotImplementedError("getTrace function is not implemented")
# if old_trg_mode != "STOP":
# short speed up here with this check
# self.setTriggerMode(old_trg_mode)
def getTriggerMode(self):
# we expect NORM, AUTO, SINGLE
raise NotImplementedError("getTriggerMode function is not implemented")
def setTriggerMode(self, mode):
# we expect NORM, AUTO, SINGLE
raise NotImplementedError("setTriggerMode function is not implemented")
def getRun(self):
"""Is acquisition running or stopped."""
raise NotImplementedError("getRun function is not implemented")
def setRun(self, val):
"""Either enable run or stop the acquisition."""
raise NotImplementedError("setRun function is not implemented")
def _waitUntillStop(self, timeout=0.1):
"""Wait until scope in the stop state.
Just because we ask for a scope to stop, does not mean
that it is stopped. It can still wait for a trigger or untill
the time span is filled.
Parameter
---------
timeout : float
timeout in seconds, default is 1 second
"""
starttime = time.time()
deadline = starttime + timeout
while time.time() < deadline:
if self.getRun():
time.sleep(0.010)
else:
logger.debug(f"Scope stopped within {time.time()-starttime} seconds.")
return
logger.warning(
f"Scope did not reach STOP state within {timeout=} sec, try to increase it."
)
def getAllTraces(self, availableNpnts=None, maxRequiredPoints=None, decimate=True):
allTraces = TraceSetSameX("scope traces")
allTraces.config["tags"]["DAQ"] = self.getConfig()
old_run_status = self.getRun()
if old_run_status: # avoid unnecessary status change
self.setRun(False) # stop if currently running
self._waitUntillStop()
# to get synchronous channels
for chNum in range(1, self.numberOfChannels + 1):
allTraces.addTrace(
self.getTrace(
chNum,
availableNpnts=availableNpnts,
maxRequiredPoints=maxRequiredPoints,
decimate=decimate,
)
)
# restore scope to the before acquisition mode
if old_run_status: # avoid unnecessary status change
self.setRun(old_run_status) # start running if it was old run state
return allTraces
def plot(self, **kwargs):
allTraces = self.getAllTraces(**kwargs)
allTraces.plot()
def save(
self,
fname=None,
item_format="e",
availableNpnts=None,
maxRequiredPoints=None,
decimate=True,
extension="dat",
):
allTraces = self.getAllTraces(
availableNpnts=availableNpnts,
maxRequiredPoints=maxRequiredPoints,
decimate=decimate,
)
allTraces.config["item_format"] = item_format
if fname is None:
fname = self.getNextDataFile(extension=extension)
allTraces.save(fname)
print(f"Data saved to: {fname}")
return fname
class ScopeSCPI(SCPIinstr, Scope):
"""SCPI aware scope.
Use as a parent for a hardware aware scope classes.
Example
-------
>>> rm = pyvisa.ResourceManager()
>>> ScopeSCPI(rm.open_resource('TCPIP::192.168.0.2::INSTR'))
"""
def __init__(self, resource, *args, **kwds):
SCPIinstr.__init__(self, resource)
Scope.__init__(self, *args, **kwds)
self.config["DeviceId"] = str.strip(self.idn)
|
