Newer
Older
#!/usr/Local/pyroot/PyTangoRoot/bin/python
"""
Python module for extracting attribute from Arhive Extractor Device.
Includes a Command Line Interface.
Can be imported as is to use function in user script.
"""
import logging
import datetime
import numpy as np
import PyTango as tango
__version__ = "1.0.1"
##########################################################################
""" Commodity variables """
# Extractor date format for GetAttDataBetweenDates
DBDFMT = "%Y-%m-%d %H:%M:%S"
# Extractor date format for GetNearestValue
DBDFMT2 = "%d-%m-%Y %H:%M:%S"
# Vectorized fromtimestamp function
ArrayTimeStampToDatetime = np.vectorize(datetime.datetime.fromtimestamp)
# Max number of point per extraction chunks
Nmax = 100000
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
##########################################################################
def __init__(
self,
ExtractorKind='H', ExtractorNumber=2,
ExtractorPath=None,
logger=logging.getLogger("ArchiveExtractor")
):
"""
Constructor function
Parameters
----------
ExtractorKind: char
Either 'H' or 'T' for HDB or TDB.
ExtractorNumber: int
Number of the archive extractor instance to use.
ExtractorPath: string
Tango path to the extractor.
If this argument is given, it takes precedence over ExtractorKind and ExtractorNumber.
logger: logging.Logger
Logger object to use
Return
------
ArchiveExtractor
"""
# Get logger
self.logger = logger
#######################################################
# Select Extractor
if ExtractorPath is None:
self.extractor = tango.DeviceProxy(
"archiving/%sDBExtractor/%d"%(ExtractorKind, ExtractorNumber)
)
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
else:
self.extractor = tango.DeviceProxy(ExtractorPath)
self.extractor.set_timeout_millis(3000)
self.logger.debug("Archive Extractor %s used."%self.extractor.name())
##---------------------------------------------------------------------------##
@staticmethod
def dateparse(datestr):
"""
Convenient function to parse date strings.
Global format is %Y-%m-%d-%H:%M:%S and it can be reduced to be less precise.
Parameters
---------
datestr : string
Date as a string, format %Y-%m-%d-%H:%M:%S or less precise.
Exceptions
----------
ValueError
If the parsing failed.
Returns
-------
date : datetime.datetime
Parsed date
"""
# This gives all format that will be tried, in order.
# Stop on first parse success. Raise error if none succeed.
fmt = [
"%Y-%m-%d-%H:%M:%S",
"%Y-%m-%d-%H:%M",
"%Y-%m-%d-%H",
"%Y-%m-%d",
"%Y-%m",
]
date = None
for f in fmt:
try:
date = datetime.datetime.strptime(datestr, f)
except ValueError:
continue
else:
break
else:
raise ValueError("Could not parse argument to a date")
##---------------------------------------------------------------------------##
def evalPoints(
self,
attribute,
dateStart,
dateStop,
):
"""
Evaluate the number of points for the attribute on the date range.
Also checks for its presence.
Parameters
----------
attribute : String
Name of the attribute. Full Tango name i.e. "test/dg/panda/current".
dateStart : datetime.datetime
Start date for extraction.
dateStop : datetime.datetime
Stop date for extraction.
Default is now (datetime.datetime.now())
Exceptions
----------
ValueError
The attribute is not found in the database.
NotImplemented
The archive mode returned by the DB is not handled.
Return
------
N: int
Number of points on the date range.
##---------------------------------------------------------------------------##
dateStart,
dateStop=datetime.datetime.now(),
):
"""
Query attribute data from an archiver database, get all points between dates.
Use ExtractBetweenDates.
Parameters
----------
attr : String
Name of the attribute. Full Tango name i.e. "test/dg/panda/current".
dateStart : datetime.datetime, string
Start date for extraction. If string, it will be parsed.
dateStop : datetime.datetime, string
Stop date for extraction. If string, it will be parsed.
Default is now (datetime.datetime.now())
Exceptions
----------
ValueError
The attribute is not found in the database.
Returns
-------
[date, value] : array
date : numpy.ndarray of datetime.datime objects
Dates of the values
value : numpy.ndarray
Archived values
"""
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
# Parse date if it is string
if type(dateStart) is str:
dateStart = self.dateparse(dateStart)
if type(dateStop) is str:
dateStop = self.dateparse(dateStop)
# Check that the attribute is in the database
self.logger.debug("Check that %s is archived."%attribute)
if not self.extractor.IsArchived(attribute):
self.logger.error("Attribute '%s' is not archived in DB %s"%(attribute, extractor))
raise ValueError("Attribute '%s' is not archived in DB %s"%(attribute, extractor))
# Get its sampling period in seconds
req=self.extractor.GetArchivingMode(attribute)
self.logger.debug("GetArchivingMode: "+str(req))
if req[0] == "MODE_P":
samplingPeriod = int(req[1])*10**-3
self.logger.debug("Attribute is sampled every %g seconds"%samplingPeriod)
elif req[0] == "MODE_EVT":
self.logger.warning("Attribute is archived on event. Chunks of data are sized with an estimated datarate of 0.1Hz")
samplingPeriod = 10
else:
self.logger.error("Archive mode not implemented in this script")
raise NotImplemented("Archive mode not implemented in this script")
# Get the number of points
N=self.extractor.GetAttDataBetweenDatesCount([
attribute,
dateStart.strftime(DBDFMT2),
dateStop.strftime(DBDFMT2)
])
self.logger.debug("On the period, there is %d entries"%N)
# If data chunk is too much, we need to cut it
if N > self.Nmax:
dt = datetime.timedelta(seconds=samplingPeriod)*self.Nmax
dt = (dateStop-dateStart)/(N//self.Nmax)
cdates = [dateStart]
while cdates[-1] < dateStop:
cdates.append(cdates[-1]+dt)
cdates[-1] = dateStop
self.logger.debug("Cutting access to %d little chunks of time, %s each."%(len(cdates)-1, dt))
else:
cdates=[dateStart, dateStop]
# Arrays to hold every chunks
value = []
date = []
# For each date chunk
for i_d in range(len(cdates)-1):
# Make retrieval request
self.logger.debug("Perform ExtractBetweenDates (%s, %s, %s)"%(
cdates[i_d].strftime(DBDFMT),
cdates[i_d+1].strftime(DBDFMT))
)
_date, _value = self.extractor.ExtractBetweenDates([
cdates[i_d].strftime(DBDFMT),
cdates[i_d+1].strftime(DBDFMT)
])
# Transform to datetime - value arrays
_value = np.asarray(_value, dtype=float)
if len(_date) > 0:
_date = ArrayTimeStampToDatetime(_date/1000.0)
value.append(_value)
date.append(_date)
self.logger.debug("Concatenate chunks")
value = np.concatenate(value)
date = np.concatenate(date)
self.logger.debug("Extraction done for %s."%attribute)
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
return [date, value]
##---------------------------------------------------------------------------##
def query_ADB_BetweenDates_MinMaxMean(
attr,
dateStart,
dateStop=datetime.datetime.now(),
timeinterval=datetime.timedelta(seconds=60),
extractor="archiving/TDBExtractor/4"):
"""
Query attribute data from archiver database.
Divide the time range in time intervals.
Get min, max and mean value on each time interval.
The date stamp is in the middle of the interval.
Parameters
----------
attr : String
Name of the attribute. Full Tango name i.e. "test/dg/panda/current".
dateStart : datetime.datetime
Start date for extraction.
dateStop : datetime.datetime
Stop date for extraction.
Default is now (datetime.datetime.now())
timeinterval : datetime.timedelta
Interval time to divide the time range in chunks.
Default is 1 minute.
extractor : String
Name of the DB Extractor device.
Default is "archiving/TDBExtractor/4"
Exceptions
----------
ValueError
The attribute is not found in the database.
Returns
-------
[date, value] : array
date : numpy.ndarray of datetime.datime objects
Dates of the values
value : numpy.ndarray
Archived values
"""
# TEMP Dev not finished
logger.error("Feature not implemented yet.")
return
# Device Proxy to DB
logger.debug("Instantiate proxy to %s"%extractor)
ADB = tango.DeviceProxy(extractor)
# Give the DB extractor 3 seconds timeout
ADB.set_timeout_millis(3000)
# Check that the attribute is in the database
logger.debug("Check that %s is archived."%attr)
if not ADB.IsArchived(attr):
logger.error("Attribute '%s' is not archived in DB %s"%(attr, extractor))
raise ValueError("Attribute '%s' is not archived in DB %s"%(attr, extractor))
# Cut data range in time chunks
cdates = [dateStart]
while cdates[-1] < dateStop:
cdates.append(cdates[-1]+timeinterval)
cdates[-1] = dateStop
logger.debug("Cutting time range to %d chunks of time, %s each."%(len(cdates)-1, dt))
# Prepare arrays
value_min = np.empty(len(cdates-1))
value_max = np.empty(len(cdates-1))
value_mean = np.empty(len(cdates-1))
# For each time chunk
for i_d in range(len(cdates)-1):
# Make requests
logger.debug("Perform GetAttDataMaxBetweenDates (%s, %s, %s)"%(
attr,
cdates[i_d].strftime(DBDFMT),
cdates[i_d+1].strftime(DBDFMT))
)
ADB.GetAttDataMaxBetweenDates([
attr,
cdates[i_d].strftime(DBDFMT),
cdates[i_d+1].strftime(DBDFMT)
])
##---------------------------------------------------------------------------##
def query_ADB_NearestValue(attr,
dates,
extractor="archiving/TDBExtractor/4"):
"""
Query attribute data from an archiver database, get nearest points from dates.
Use GetNearestValue and perform multiple calls.
For each date in dates, it read the closest sampled value.
Return the real dates of the samples.
Parameters
----------
attr : String
Name of the attribute. Full Tango name i.e. "test/dg/panda/current".
dates : numpy.ndarray of datetime.datetime
Dates for extraction.
extractor : String
Name of the DB Extractor device.
Default is "archiving/TDBExtractor/4"
Exceptions
----------
ValueError
The attribute is not found in the database.
Returns
-------
[realdate, value] : array
realdate : numpy.ndarray of datetime.datime objects
Dates of the values
value : numpy.ndarray
Archived values
# Device Proxy to DB
ADB = tango.DeviceProxy(extractor)
# Give the DB extractor 3 seconds timeout
ADB.set_timeout_millis(3000)
# Check that the attribute is in the database
if not ADB.IsArchived(attr):
raise ValueError("Attribute '%s' is not archived in DB %s"%(attr, extractor))
# Prepare arrays
value = np.empty(len(dates), dtype=float)
realdate = np.empty(len(dates), dtype=object)
# Loop on dates
for i in range(len(dates)):
# Make retrieval
answ = ADB.GetNearestValue([attr, dates[i].strftime(DBDFMT2)])
answ = answ.split(";")
realdate[i] = datetime.datetime.fromtimestamp(int(answ[0])/1000)
value[i] = answ[1]
return [realdate, value]