Newer
Older
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
import logging
logger = logging.getLogger("tester")
sh = logging.StreamHandler()
sh.setFormatter(logging.Formatter("{name}-{levelname:8}: {message}", style='{'))
logger.addHandler(sh)
logger.setLevel(logging.DEBUG)
# Put root folder in path
import sys, os
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
# test import
logger.info("Testing import")
import ArchiveExtractor
# it should have been automatically initialized, check that extractors are set
logger.info("Testing auto init")
logger.debug(ArchiveExtractor._Extractors)
if None in ArchiveExtractor._Extractors:
raise RuntimeError("ArchiveExtractor does not seems properly initialized")
# Test init()
logger.info("Testing init")
ArchiveExtractor.init(loglevel='debug')
###############################################################################
# Test findattr
logger.info("Testing findattr()")
attrs = ArchiveExtractor.findattr("*dg*dcct*current*")
logger.debug(attrs)
if len(attrs) < 1:
raise RuntimeError("Failed to get attributes with findattr")
attrs = ArchiveExtractor.findattr("*dg*dcct*current*", db='T')
logger.debug(attrs)
if len(attrs) < 1:
raise RuntimeError("Failed to get attributes with findattr")
###############################################################################
# Test infoattr
logger.info("Testing infoattr()")
info = ArchiveExtractor.infoattr(attrs[0], db='T')
logger.debug(info)
###############################################################################
# Test extractions
attr = ArchiveExtractor.findattr("ans/dg*dcct*current*")[0]
logger.info("Testing extract() ; scalar, nearest, timedelta")
val = ArchiveExtractor.extract(attr, "0h")
logger.debug(val)
if val is None:
raise RuntimeError("Could not perform extraction")
logger.info("Testing extract() ; scalar, between, precise date and timedelta")
val = ArchiveExtractor.extract(attr, "1h", "2023-12-13-00:30", method='between')
logger.debug(val)
if val is None:
raise RuntimeError("Could not perform extraction")
logger.info("Testing extract() ; scalar, nearest, specific date")
# Test several formats
for fmt in [
"2023-08",
"2024-01-10",
"2024-01-10-12:00",
]:
logger.debug(fmt)
val = ArchiveExtractor.extract(attr, fmt)
logger.debug(val)
if val is None:
raise RuntimeError("Could not perform extraction")
logger.info("Testing extract() ; dict, nearest, specific date")
val = ArchiveExtractor.extract({"attr":attr, "attr2":attr}, "2023-06")
logger.debug(val)
logger.info("Testing extract() ; list, nearest, specific date")
val = ArchiveExtractor.extract(ArchiveExtractor.findattr("dg*dcct*current"), "2023-06")
logger.debug(val)
logger.info("Testing extract() ; scalar, between, timedelta")
val = ArchiveExtractor.extract(attr, "3h", method='between')
logger.debug(val)
if val is None:
raise RuntimeError("Could not perform extraction")
logger.info("Testing extract() ; scalar, between, precise date")
val = ArchiveExtractor.extract(attr, "2023-12-13-00:30", "2023-12-13-01:30", method='between')
logger.debug(val)
if val is None:
raise RuntimeError("Could not perform extraction")
logger.info("Testing extract() ; spectrum, nearest, precise date")
val = ArchiveExtractor.extract('ANS/DG/BPM-MANAGER/zRefOrbit', "2023-12-13-00:30")
logger.debug(val)
if val is None:
raise RuntimeError("Could not perform extraction")
logger.info("Testing extract() ; scalar boolean, between, precise date")
val = ArchiveExtractor.extract("ans/dg/fofb-watcher/fofbrunning_x", "2024-01-29-00:30", "2024-01-29-15:30", method='between')
logger.debug(val)
if val is None:
raise RuntimeError("Could not perform extraction")