Adding tuner

setup.py

@@ -3,16 +3,18 @@
 requirements = [
     "dill",
     "dataclasses",
-    "python-varname",
+    "varname",
     "tabulate",
     "numpy",
+    "lyncs_utils",
 ]
 
 extras = {
     "graph": [
         "graphviz",
     ],
     "test": ["pytest", "pytest-cov"],
+    "optuna": ["optuna"],
 }
 
 setup(

test/test_optuna.py

@@ -0,0 +1,122 @@
+from tuneit import *
+import numpy
+from tuneit.tools.time import Time, default_timer
+from tuneit.tools.optuna import OptunaSampler
+import pytest
+import optuna
+from optuna.trial import Trial
+from optuna.trial import create_trial
+from optuna.distributions import CategoricalDistribution
+from optuna.trial import TrialState
+from tuneit.finalize import HighLevel
+from optuna.study import Study
+from hashlib import md5
+from dill import dumps
+
+
+def test_optuna_sampler():
+    # simple example to use in tests
+    # building a graph with variables for sorting (preprocessing) and searching to be tuned:
+
+    @alternatives(
+        mergesort=lambda a: numpy.sort(a, kind="mergesort"),
+        heapsort=lambda a: numpy.sort(a, kind="heapsort"),
+        timsort=lambda a: numpy.array(sorted(a)),
+    )
+    def preprocessing(array):
+        res = numpy.sort(array)
+        return res
+
+    @alternatives(
+        indices=lambda a, b: [i for i, x in enumerate(a.tolist()) if x == b][0],
+        array_search=lambda a, b: numpy.where(a == b)[0][0],
+        binary_search=lambda a, b: numpy.searchsorted(a, b),
+    )
+    def searching(array, element):
+        l = array.tolist()
+        index = l.index(element)
+        return index
+
+    element = 65
+    result = searching(
+        preprocessing(numpy.random.randint(1000, size=(10000))), element
+    )  # input size: 10 000, type: integers
+    fz = finalize(result)
+
+    callback_function = lambda fnc: Time(default_timer(fnc))
+    obj_A = OptunaSampler(
+        fz, callback=callback_function, storage="sqlite:///example.db"
+    )
+    assert isinstance(obj_A, OptunaSampler)
+    assert isinstance(obj_A.tunable, HighLevel)
+    assert not bool(
+        obj_A.compute_kwargs
+    )  # at the moment no kwargs are used for the compute function, so compute_kwargs must be empty
+    assert callable(obj_A.callback)
+    assert obj_A.n_trials > 0
+
+    # test get_attributes function
+    assert obj_A.get_attributes()["callback"] == obj_A.callback
+
+    # test get_study function
+    study = obj_A.get_study()
+    assert isinstance(study, Study)
+    assert study.study_name == md5(dumps(obj_A.get_attributes())).hexdigest()
+    assert [*study.user_attrs.keys()] == ["callback"]
+    obj_B = OptunaSampler(
+        fz, callback=callback_function, storage="sqlite:///example.db", n_trials=10
+    )
+    assert obj_A.compute() == finalize(result).compute()  # test compute function
+    assert obj_B.compute() == finalize(result).compute()
+    name_A = obj_A.get_study().study_name
+    name_B = obj_B.get_study().study_name
+    assert name_A == name_B
+    assert len(study.trials) >= 11
+
+    # test _call_wrapper function
+    assert obj_A._call_wrapper(obj_A.tunable) == finalize(result).compute()
+
+    # test objective function
+    tid = study._storage.create_new_trial(study._study_id)
+    trial = Trial(study, trial_id=tid)
+    assert isinstance(obj_A.objective(trial), Time)
+
+    # test get_next_trial
+    temp = obj_A.get_next_trial(trial)
+    assert len(temp.fixed_variables) == 2
+    var_A = temp.get_variable("preprocessing")
+    var_B = temp.get_variable("searching")
+    assert var_A.fixed and var_B.fixed
+
+    # test get_suggetions
+    selected_options = obj_A.get_suggestions(trial)
+    assert len(selected_options) == len(fz.variables)
+    assert [*selected_options.values()][0] in preprocessing.keys()
+    assert [*selected_options.values()][1] in searching.keys()
+
+    # test get_var_args function
+    assert OptunaSampler.get_var_args("categorical", ["a", "b", "c"]) == (
+        tuple(["a", "b", "c"]),
+    )
+    args = OptunaSampler.get_var_args("discrete_uniform", range(100))
+    assert len(args) == 3
+    assert args[0] == min(range(100))
+    assert args[1] == max(range(100))
+    assert args[2] == 1
+
+    # test deduce_type function
+    assert OptunaSampler.deduce_type(range(100)) == "discrete_uniform"
+    assert OptunaSampler.deduce_type([1, 2, 17, 25]) == "categorical"
+    assert OptunaSampler.deduce_type(preprocessing.values()) == "categorical"
+
+    # test best_params
+    best = obj_A.best_params()
+    assert len(best.keys()) == len(fz.variables)
+    assert [*best.values()][0] in preprocessing.keys()
+    assert [*best.values()][1] in searching.keys()
+
+    optuna.delete_study(name_A, storage="sqlite:///example.db")
+
+
+# If I want to delete the study and start a new one the next time I run the tests
+# optuna.delete_study(name_B, storage="sqlite:///example.db")

test/test_sampler.py

@@ -0,0 +1,74 @@
+from tuneit import *
+import numpy
+from tuneit.tools.time import Time, default_timer
+from tuneit.tools.base import Sampler
+from tuneit.finalize import HighLevel
+import itertools
+
+
+def test_sampler():
+    # simple example to use in tests
+    # building a graph with variables for sorting (preprocessing) and searching to be tuned:
+
+    @alternatives(
+        mergesort=lambda a: numpy.sort(a, kind="mergesort"),
+        heapsort=lambda a: numpy.sort(a, kind="heapsort"),
+        timsort=lambda a: numpy.array(sorted(a)),
+    )
+    def preprocessing(array):
+        res = numpy.sort(array)
+        return res
+
+    @alternatives(
+        indices=lambda a, b: [i for i, x in enumerate(a.tolist()) if x == b][0],
+        array_search=lambda a, b: numpy.where(a == b)[0][0],
+        binary_search=lambda a, b: numpy.searchsorted(a, b),
+    )
+    def searching(array, element):
+        l = array.tolist()
+        index = l.index(element)
+        return index
+
+    element = 65
+    result = searching(preprocessing(numpy.random.randint(1000, size=(10000))), element)
+    fz = finalize(result)
+
+    callback_function = lambda fnc: Time(default_timer(fnc))
+
+    obj = sample(
+        result,
+        ["preprocessing", "searching"],
+        callback=callback_function,
+        callback_calls=True,
+    )
+    assert isinstance(obj, Sampler)
+    assert callable(obj.callback)
+    assert obj.callback == callback_function
+    assert len(obj.variables) == 2
+    assert obj.variables[0] in fz.variables
+    assert obj.variables[1] in fz.variables
+    assert not obj.compute_kwargs
+    assert isinstance(obj.tunable, HighLevel)
+
+    assert obj.max_samples == 16
+    assert obj.n_samples == 16
+    assert obj.samples == tuple(
+        itertools.product([*preprocessing.keys()], [*searching.keys()]),
+    )
+    values = obj.sample_values()
+    for a in [x[1] for x in values]:
+        assert isinstance(a, Time)
+    assert [x[0] for x in values] == list(obj.samples)
+    assert obj.value == fz.compute()
+
+    obj_B = benchmark(fz.copy(reset=True))
+    assert isinstance(obj_B, Sampler)
+    assert obj_B.samples == obj.samples
+    assert obj_B.value == obj.value
+
+    obj_C = crosscheck(fz.copy(reset=True))
+    assert isinstance(obj_C, Sampler)
+    values2 = obj_C.sample_values()
+    for a in [x[1] for x in values2]:
+        assert isinstance(a, bool)
+    assert obj_C.samples == obj.samples

test/test_tuner.py

@@ -0,0 +1,86 @@
+from tuneit import *
+import numpy
+from tuneit.tools.tuner import Tuner
+from tuneit.tools.time import Time, default_timer
+from tuneit.tools.optuna import OptunaSampler
+from tuneit.tools.base import Sampler
+import pytest
+
+
+def test_tuner():
+    # simple example to use in tests
+    # building a graph with variables for sorting (preprocessing) and searching to be tuned:
+
+    @alternatives(
+        mergesort=lambda a: numpy.sort(a, kind="mergesort"),
+        heapsort=lambda a: numpy.sort(a, kind="heapsort"),
+        timsort=lambda a: numpy.array(sorted(a)),
+    )
+    def preprocessing(array):
+        res = numpy.sort(array)
+        return res
+
+    @alternatives(
+        indices=lambda a, b: [i for i, x in enumerate(a.tolist()) if x == b][0],
+        array_search=lambda a, b: numpy.where(a == b)[0][0],
+        binary_search=lambda a, b: numpy.searchsorted(a, b),
+    )
+    def searching(array, element):
+        l = array.tolist()
+        index = l.index(element)
+        return index
+
+    element = 65
+    result = searching(
+        preprocessing(numpy.random.randint(1000, size=(10000))), element
+    )  # input size: 10 000, type: integers
+
+    # test optimise function
+    obj_A = optimise(result, sampler="optuna")
+    assert isinstance(obj_A, Tuner)
+
+    # test tune function
+    obj_B = tune(result, callback=lambda fnc: Time(default_timer(fnc)))
+    assert isinstance(obj_B, Tuner)
+
+    # test Tuner class
+    obj_C = Tuner(
+        result, sampler="optuna", callback=lambda fnc: Time(default_timer(fnc))
+    )
+    assert isinstance(obj_C, Tuner)
+    assert bool(obj_C.tuner_kwargs)
+    assert callable(obj_B.tuner_kwargs.get("callback", None))
+    # test compute function in Tuner
+    assert obj_C.compute() == finalize(result).compute()
+    # test get_best_trial function in Tuner
+    res = obj_C.get_best_trial()
+    assert isinstance(res, dict)
+    assert "preprocessing" in {k.split("-")[0]: v for k, v in res.items()}
+    assert (
+        next(v for k, v in res.items() if k.startswith("preprocessing"))
+        in preprocessing.keys()
+    )
+    assert "searching" in {k.split("-")[0]: v for k, v in res.items()}
+    assert (
+        next(v for k, v in res.items() if k.startswith("searching")) in searching.keys()
+    )
+    # test get_sampler function in Tuner
+    assert obj_C.get_sampler() == OptunaSampler
+    assert (
+        Tuner(
+            result, sampler=None, callback=lambda fnc: Time(default_timer(fnc))
+        ).get_sampler()
+        == Sampler
+    )
+    with pytest.raises(ValueError):
+        Tuner(
+            result, sampler="hello", callback=lambda fnc: Time(default_timer(fnc))
+        ).get_sampler()
+    # test get_sampler_kwargs function in Tuner
+    kwargs = obj_C.get_sampler_kwargs()
+    assert kwargs["storage"] == "sqlite:///example.db"
+    assert kwargs["callback"] == obj_C.tuner_kwargs.get("callback", None)
+    kwargs = Tuner(
+        result, sampler=None, callback=lambda fnc: Time(default_timer(fnc))
+    ).get_sampler_kwargs()
+    assert not bool(kwargs)