Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_update_nr_samples(history: History):
history.store_initial_data(None, {}, {}, {}, ["m0"], "", "", "")
pops = history.get_all_populations()
assert 0 == pops[pops['t'] == History.PRE_TIME]['samples'].values
history.update_nr_samples(History.PRE_TIME, 43)
pops = history.get_all_populations()
assert 43 == pops[pops['t'] == History.PRE_TIME]['samples'].values
def test_create_db():
# temporary file name
file_ = tempfile.mkstemp(suffix=".db")[1]
# should work just fine though file empty
pyabc.History("sqlite:///" + file_, create=False)
# delete file and check we cannot create a History object
os.remove(file_)
with pytest.raises(ValueError):
pyabc.History("sqlite:///" + file_, create=False)
def test_model_name_load_single_with_pop(history_uninitialized: History):
h = history_uninitialized
model_names = ["m1"]
h.store_initial_data(0, {}, {}, {}, model_names, "", "", "")
particle_list = [
Particle(m=0,
parameter=Parameter({"a": 23, "b": 12}),
weight=.2,
accepted_sum_stats=[{"ss": .1}],
accepted_distances=[.1])]
h.append_population(0, 42, Population(particle_list), 2, model_names)
h2 = History(h.db)
model_names_loaded = h2.model_names()
assert model_names == model_names_loaded
def make_hist():
h = History("sqlite:///" + path)
h.store_initial_data(0, {}, {}, {}, model_names, "", "", "")
return h
def history(request):
# Test in-memory and filesystem based database
if request.param == "file":
this_path = "/" + path()
elif request.param == "memory":
this_path = ""
else:
raise Exception(f"Bad database type for testing: {request.param}")
model_names = ["fake_name_{}".format(k) for k in range(50)]
h = History("sqlite://" + this_path)
h.store_initial_data(0, {}, {}, {}, model_names,
"", "", '{"name": "pop_strategy_str_test"}')
yield h
if request.param == "file":
try:
os.remove(this_path)
except FileNotFoundError:
pass
def test_model_name_load(history_uninitialized: History):
h = history_uninitialized
model_names = ["m1", "m2", "m3"]
h.store_initial_data(0, {}, {}, {}, model_names, "", "", "")
h2 = History(h.db)
model_names_loaded = h2.model_names()
assert model_names == model_names_loaded
def test_create_db():
# temporary file name
file_ = tempfile.mkstemp(suffix=".db")[1]
# should work just fine though file empty
pyabc.History("sqlite:///" + file_, create=False)
# delete file and check we cannot create a History object
os.remove(file_)
with pytest.raises(ValueError):
pyabc.History("sqlite:///" + file_, create=False)
try:
mp0 = mp.p[0]
except KeyError:
mp0 = 0
try:
mp1 = mp.p[1]
except KeyError:
mp1 = 0
assert abs(mp0 - p1_expected) + abs(mp1 - p2_expected) < np.inf
# check that sampler only did nr_particles samples in first round
pops = history.get_all_populations()
# since we had calibration (of epsilon), check that was saved
pre_evals = pops[pops['t'] == History.PRE_TIME]['samples'].values
assert pre_evals >= pop_size.nr_particles
# our samplers should not have overhead in calibration, except batching
batch_size = sampler.batch_size if hasattr(sampler, 'batch_size') else 1
max_expected = pop_size.nr_particles + batch_size - 1
if pre_evals > max_expected:
# Violations have been observed occasionally for the redis server
# due to runtime conditions with the increase of the evaluations
# counter. This could be overcome, but as it usually only happens
# for low-runtime models, this should not be a problem. Thus, only
# print a warning here.
logger.warning(
f"Had {pre_evals} simulations in the calibration iteration, "
f"but a maximum of {max_expected} would have been sufficient for "
"""
Created on Mon Nov 21 15:01:26 2016
@author: emmanuel
"""
import style
import pyabc
import scipy as sp
import pandas as pd
sm = style.make(output="test.npz",
input="/home/emmanuel/abc/data/raw/toy:modes=2.db",
wildcards=["4"])
history = pyabc.History("sqlite:///" + sm.input[0], 23, ["sdf"])
history.id = 1
transition = pyabc.MultivariateNormalTransition()
MAX_SIZE = 10
NR_PLOT_POINTS = 50
TX, TY = sp.meshgrid(sp.linspace(-MAX_SIZE, MAX_SIZE, NR_PLOT_POINTS),
sp.linspace(-MAX_SIZE, MAX_SIZE, NR_PLOT_POINTS))
TXY = sp.stack((TX, TY), axis=2)
fitted_kdes = []
ts = []
def run_app(db, debug, port):
db = os.path.expanduser(db)
history = History("sqlite:///" + db)
app.config["HISTORY"] = history
app.run(debug=debug, port=port)