writer example test fails
@pithan I cannot reproduce this locally:
__________________________ test_external_hdf5_writer ___________________________
alias_session_scans_listener = {}
alias_session = <bliss.common.session.Session object at 0x7f44e835bb10>
dummy_acq_device = <conftest.dummy_acq_device.<locals>.DummyAcqDeviceFactory object at 0x7f44eb367e10>
@pytest.mark.writer
def test_external_hdf5_writer(
alias_session_scans_listener, alias_session, dummy_acq_device
):
env_dict = alias_session.env_dict
scan_stack = alias_session_scans_listener
lima_sim = env_dict["lima_simulator"]
## a simple scan
s1 = scans.ascan(env_dict["robyy"], 0, 1, 3, .1, lima_sim)
## a scan with multiple top masters
chain = AcquisitionChain()
master1 = timer.SoftwareTimerMaster(0.1, npoints=2, name="timer1")
diode_sim = alias_session.config.get("diode")
diode_device = SamplingCounterAcquisitionSlave(diode_sim, count_time=0.1)
master2 = timer.SoftwareTimerMaster(0.001, npoints=50, name="timer2")
lima_master = LimaAcquisitionMaster(lima_sim, acq_nb_frames=1, acq_expo_time=0.0005)
# note: dummy device has 2 channels: pi and nb
dummy_device = dummy_acq_device.get(None, name="dummy_device", npoints=1)
chain.add(lima_master, dummy_device)
chain.add(master2, lima_master)
chain.add(master1, diode_device)
master1.terminator = False
s2 = Scan(chain, "test", save=True)
s2.run()
diode2 = alias_session.config.get("diode2")
# TEST NOT RELIALE when excuting kill ... has to be fixed
### test scan with undefined number of points
s3 = scans.timescan(.05, diode2, run=False)
gevent.sleep(.2)
## just to see if there is no event created before the scan runs...
scan_task = gevent.spawn(s3.run)
# todo add synchronisatin once !1594 is merged
# for now .. incrase sleep time
gevent.sleep(1)
try:
scan_task.kill(KeyboardInterrupt)
except:
assert scan_task.ready()
# just until the test above is reliable
# ~ s3 = scans.timescan(.05, diode2, npoints=1)
## scan with counter that exports individual samples (SamplingMode.Samples)
scan5_a = scans.loopscan(5, 0.1, alias_session.config.get("diode9"), save=True)
## artifical scan that forces different length of datasets in SamplingMode.Samples
from bliss.common.counter import SoftCounter, SamplingMode
from bliss.common.soft_axis import SoftAxis
class A:
def __init__(self):
self.val = 0
self.i = 0
def read(self):
gevent.sleep((self.val % 5 + 1) * 0.002)
self.i += 1
return self.i
@property
def position(self):
return self.val
@position.setter
def position(self, val):
self.val = val
self.i = 0
a = A()
ax = SoftAxis("test-sample-pos", a)
c_samp = SoftCounter(a, "read", name="test-samp", mode=SamplingMode.SAMPLES)
scan5_b = scans.ascan(ax, 1, 9, 9, .1, c_samp)
# a group entry
s1 = scans.loopscan(3, .1, diode_sim)
s2 = scans.loopscan(3, .05, diode_sim)
g = Group(s1, s2)
##wait for all scan entries
external_writer_file = s1.scan_info["filename"].replace(".", "_external.")
bliss_writer_file = s1.scan_info["filename"]
# check that external writer has at least started to procces all scans
for i in range(0, 20):
if len(h5py.File(external_writer_file).keys()) < 6:
print("##### external writer not done yet")
gevent.sleep(1)
# check that all scans have been finalized
for i in range(0, 20):
if len(scan_stack) > 0:
print("##### waiting for finalization")
gevent.sleep(1)
## check if external file is the same as the one of bliss writer for simple scan
external_writer = h5todict(external_writer_file)["2_ascan"]
bliss_writer = h5todict(bliss_writer_file)["2_ascan"]
> deep_compare(external_writer, bliss_writer)
tests/scans/test_external_hdf5_writer.py:190:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
d = {'proposal': {}, 'sample': {}, 'sample_description': {}, 'technique': {}}
u = {'nxwriter': {}, 'proposal': {}, 'sample': {}, 'sample_description': {}, ...}
def deep_compare(d, u):
"""using logic of deep update used here to compare two dicts
"""
stack = [(d, u)]
while stack:
d, u = stack.pop(0)
> assert len(d) == len(u)
E assert 4 == 5
E -4
E +5
tests/scans/test_external_hdf5_writer.py:43: AssertionError