lattice/pyerrors
1
0
Fork 0
mirror of https://github.com/fjosw/pyerrors.git synced 2025-05-14 19:43:41 +02:00
Code Issues Projects Releases Packages Wiki Activity

[Fix] extended tests and corrected a small bug in the previous commit

Browse source
This commit is contained in:
Simon Kuberski 2025-02-24 12:48:13 +01:00
parent f2fb69a79d
commit 1501c3aef0
3 changed files with 52 additions and 40 deletions
Download patch file Download diff file Expand all files Collapse all files

4
pyerrors/input/json.py
View file

@ -133,10 +133,10 @@ def create_json_string(ol, description='', indent=1):
names = [] names = []
idl = [] idl = []
for key, value in obs.idl.items(): for key, value in obs.idl.items():
samples.append([np.nan] * len(value)) samples.append(np.array([np.nan] * len(value)))
names.append(key) names.append(key)
idl.append(value) idl.append(value)
my_obs = Obs(np.array(samples), names, idl, means=[np.nan for n in names]) my_obs = Obs(samples, names, idl, means=[np.nan for n in names])
my_obs._value = np.nan my_obs._value = np.nan
my_obs._covobs = obs._covobs my_obs._covobs = obs._covobs
for name in obs._covobs: for name in obs._covobs:

82
tests/json_io_test.py
View file

@ -12,7 +12,7 @@ def test_jsonio():
o = pe.pseudo_Obs(1.0, .2, 'one') o = pe.pseudo_Obs(1.0, .2, 'one')
o2 = pe.pseudo_Obs(0.5, .1, 'two|r1') o2 = pe.pseudo_Obs(0.5, .1, 'two|r1')
o3 = pe.pseudo_Obs(0.5, .1, 'two|r2') o3 = pe.pseudo_Obs(0.5, .1, 'two|r2')
o4 = pe.merge_obs([o2, o3]) o4 = pe.merge_obs([o2, o3, pe.pseudo_Obs(0.5, .1, 'two|r3', samples=3221)])
otag = 'This has been merged!' otag = 'This has been merged!'
o4.tag = otag o4.tag = otag
do = o - .2 * o4 do = o - .2 * o4
@ -101,8 +101,8 @@ def test_json_string_reconstruction():
def test_json_corr_io(): def test_json_corr_io():
my_list = [pe.Obs([np.random.normal(1.0, 0.1, 100)], ['ens1']) for o in range(8)] my_list = [pe.Obs([np.random.normal(1.0, 0.1, 100), np.random.normal(1.0, 0.1, 321)], ['ens1|r1', 'ens1|r2'], idl=[range(1, 201, 2), range(321)]) for o in range(8)]
rw_list = pe.reweight(pe.Obs([np.random.normal(1.0, 0.1, 100)], ['ens1']), my_list) rw_list = pe.reweight(pe.Obs([np.random.normal(1.0, 0.1, 100), np.random.normal(1.0, 0.1, 321)], ['ens1|r1', 'ens1|r2'], idl=[range(1, 201, 2), range(321)]), my_list)
for obs_list in [my_list, rw_list]: for obs_list in [my_list, rw_list]:
for tag in [None, "test"]: for tag in [None, "test"]:
@ -111,40 +111,51 @@ def test_json_corr_io():
for corr_tag in [None, 'my_Corr_tag']: for corr_tag in [None, 'my_Corr_tag']:
for prange in [None, [3, 6]]: for prange in [None, [3, 6]]:
for gap in [False, True]: for gap in [False, True]:
my_corr = pe.Corr(obs_list, padding=[pad, pad], prange=prange) for mult in [1., pe.cov_Obs([12.22, 1.21], [.212**2, .11**2], 'renorm')[0]]:
my_corr.tag = corr_tag my_corr = mult * pe.Corr(obs_list, padding=[pad, pad], prange=prange)
if gap: my_corr.tag = corr_tag
my_corr.content[4] = None if gap:
pe.input.json.dump_to_json(my_corr, 'corr') my_corr.content[4] = None
recover = pe.input.json.load_json('corr') pe.input.json.dump_to_json(my_corr, 'corr')
os.remove('corr.json.gz') recover = pe.input.json.load_json('corr')
assert np.all([o.is_zero() for o in [x for x in (my_corr - recover) if x is not None]]) os.remove('corr.json.gz')
for index, entry in enumerate(my_corr): assert np.all([o.is_zero() for o in [x for x in (my_corr - recover) if x is not None]])
if entry is None: for index, entry in enumerate(my_corr):
assert recover[index] is None if entry is None:
assert my_corr.tag == recover.tag assert recover[index] is None
assert my_corr.prange == recover.prange assert my_corr.tag == recover.tag
assert my_corr.reweighted == recover.reweighted assert my_corr.prange == recover.prange
assert my_corr.reweighted == recover.reweighted
def test_json_corr_2d_io(): def test_json_corr_2d_io():
obs_list = [np.array([[pe.pseudo_Obs(1.0 + i, 0.1 * i, 'test'), pe.pseudo_Obs(0.0, 0.1 * i, 'test')], [pe.pseudo_Obs(0.0, 0.1 * i, 'test'), pe.pseudo_Obs(1.0 + i, 0.1 * i, 'test')]]) for i in range(4)] obs_list = [np.array([
[
pe.merge_obs([pe.pseudo_Obs(1.0 + i, 0.1 * i, 'test|r2'), pe.pseudo_Obs(1.0 + i, 0.1 * i, 'test|r1', samples=321)]),
pe.merge_obs([pe.pseudo_Obs(0.0, 0.1 * i, 'test|r2'), pe.pseudo_Obs(0.0, 0.1 * i, 'test|r1', samples=321)]),
],
[
pe.merge_obs([pe.pseudo_Obs(0.0, 0.1 * i, 'test|r2'), pe.pseudo_Obs(0.0, 0.1 * i, 'test|r1', samples=321),]),
pe.merge_obs([pe.pseudo_Obs(1.0 + i, 0.1 * i, 'test|r2'), pe.pseudo_Obs(1.0 + i, 0.1 * i, 'test|r1', samples=321)]),
],
]) for i in range(4)]
for tag in [None, "test"]: for tag in [None, "test"]:
obs_list[3][0, 1].tag = tag obs_list[3][0, 1].tag = tag
for padding in [0, 1]: for padding in [0, 1]:
for prange in [None, [3, 6]]: for prange in [None, [3, 6]]:
my_corr = pe.Corr(obs_list, padding=[padding, padding], prange=prange) for mult in [1., pe.cov_Obs([12.22, 1.21], [.212**2, .11**2], 'renorm')[0]]:
my_corr.tag = tag my_corr = mult * pe.Corr(obs_list, padding=[padding, padding], prange=prange)
pe.input.json.dump_to_json(my_corr, 'corr') my_corr.tag = tag
recover = pe.input.json.load_json('corr') pe.input.json.dump_to_json(my_corr, 'corr')
os.remove('corr.json.gz') recover = pe.input.json.load_json('corr')
assert np.all([np.all([o.is_zero() for o in q]) for q in [x.ravel() for x in (my_corr - recover) if x is not None]]) os.remove('corr.json.gz')
for index, entry in enumerate(my_corr): assert np.all([np.all([o.is_zero() for o in q]) for q in [x.ravel() for x in (my_corr - recover) if x is not None]])
if entry is None: for index, entry in enumerate(my_corr):
assert recover[index] is None if entry is None:
assert my_corr.tag == recover.tag assert recover[index] is None
assert my_corr.prange == recover.prange assert my_corr.tag == recover.tag
assert my_corr.prange == recover.prange
def test_json_dict_io(): def test_json_dict_io():
@ -211,6 +222,7 @@ def test_json_dict_io():
'd': pe.pseudo_Obs(.01, .001, 'testd', samples=10) * pe.cov_Obs(1, .01, 'cov1'), 'd': pe.pseudo_Obs(.01, .001, 'testd', samples=10) * pe.cov_Obs(1, .01, 'cov1'),
'se': None, 'se': None,
'sf': 1.2, 'sf': 1.2,
'k': pe.cov_Obs(.1, .001**2, 'cov') * pe.merge_obs([pe.pseudo_Obs(1.0, 0.1, 'test|r2'), pe.pseudo_Obs(1.0, 0.1, 'test|r1', samples=321)]),
} }
} }
@ -314,7 +326,7 @@ def test_dobsio():
o2 = pe.pseudo_Obs(0.5, .1, 'two|r1') o2 = pe.pseudo_Obs(0.5, .1, 'two|r1')
o3 = pe.pseudo_Obs(0.5, .1, 'two|r2') o3 = pe.pseudo_Obs(0.5, .1, 'two|r2')
o4 = pe.merge_obs([o2, o3]) o4 = pe.merge_obs([o2, o3, pe.pseudo_Obs(0.5, .1, 'two|r3', samples=3221)])
otag = 'This has been merged!' otag = 'This has been merged!'
o4.tag = otag o4.tag = otag
do = o - .2 * o4 do = o - .2 * o4
@ -328,7 +340,7 @@ def test_dobsio():
o5 /= co2[0] o5 /= co2[0]
o5.tag = 2 * otag o5.tag = 2 * otag
tt1 = pe.Obs([np.random.rand(100), np.random.rand(100)], ['t|r1', 't|r2'], idl=[range(2, 202, 2), range(22, 222, 2)]) tt1 = pe.Obs([np.random.rand(100), np.random.rand(102)], ['t|r1', 't|r2'], idl=[range(2, 202, 2), range(22, 226, 2)])
tt3 = pe.Obs([np.random.rand(102)], ['qe|r1']) tt3 = pe.Obs([np.random.rand(102)], ['qe|r1'])
tt = tt1 + tt3 tt = tt1 + tt3
@ -363,9 +375,9 @@ def test_dobsio():
def test_reconstruct_non_linear_r_obs(tmp_path): def test_reconstruct_non_linear_r_obs(tmp_path):
to = ( to = (
pe.Obs([np.random.rand(500), np.random.rand(500)], pe.Obs([np.random.rand(500), np.random.rand(1200)],
["e|r1", "e|r2", ], ["e|r1", "e|r2", ],
idl=[range(1, 501), range(0, 500)]) idl=[range(1, 501), range(0, 1200)])
+ pe.Obs([np.random.rand(111)], ["my_new_ensemble_54^£$|8'[@124435%6^7&()~#"], idl=[range(1, 999, 9)]) + pe.Obs([np.random.rand(111)], ["my_new_ensemble_54^£$|8'[@124435%6^7&()~#"], idl=[range(1, 999, 9)])
) )
to = np.log(to ** 2) / to to = np.log(to ** 2) / to
@ -376,9 +388,9 @@ def test_reconstruct_non_linear_r_obs(tmp_path):
def test_reconstruct_non_linear_r_obs_list(tmp_path): def test_reconstruct_non_linear_r_obs_list(tmp_path):
to = ( to = (
pe.Obs([np.random.rand(500), np.random.rand(500)], pe.Obs([np.random.rand(500), np.random.rand(1200)],
["e|r1", "e|r2", ], ["e|r1", "e|r2", ],
idl=[range(1, 501), range(0, 500)]) idl=[range(1, 501), range(0, 1200)])
+ pe.Obs([np.random.rand(111)], ["my_new_ensemble_54^£$|8'[@124435%6^7&()~#"], idl=[range(1, 999, 9)]) + pe.Obs([np.random.rand(111)], ["my_new_ensemble_54^£$|8'[@124435%6^7&()~#"], idl=[range(1, 999, 9)])
) )
to = np.log(to ** 2) / to to = np.log(to ** 2) / to

6
tests/obs_test.py
View file

@ -507,11 +507,11 @@ def test_reweighting():
def test_merge_obs(): def test_merge_obs():
my_obs1 = pe.Obs([np.random.rand(100)], ['t|1']) my_obs1 = pe.Obs([np.random.normal(1, .1, 100)], ['t|1'])
my_obs2 = pe.Obs([np.random.rand(100)], ['t|2'], idl=[range(1, 200, 2)]) my_obs2 = pe.Obs([np.random.normal(1, .1, 100)], ['t|2'], idl=[range(1, 200, 2)])
merged = pe.merge_obs([my_obs1, my_obs2]) merged = pe.merge_obs([my_obs1, my_obs2])
diff = merged - (my_obs2 + my_obs1) / 2 diff = merged - (my_obs2 + my_obs1) / 2
assert diff.value == 0 assert np.isclose(0, diff.value, atol=1e-16)
with pytest.raises(ValueError): with pytest.raises(ValueError):
pe.merge_obs([my_obs1, my_obs1]) pe.merge_obs([my_obs1, my_obs1])
my_covobs = pe.cov_Obs(1.0, 0.003, 'cov') my_covobs = pe.cov_Obs(1.0, 0.003, 'cov')