<jupyter_start><jupyter_text>IntroductionThis is just a very simple demonstation of how simple sciope core is, and this is how we build our workflows (inference and exploration). If we are not using any of predefined samplers in sciope, and one like to define their own samples it is required that the function have the arguments "n" (number of samples) and "chunk_size" (the number of samples in a batch/chunk). The function also need to create lazy objects in the form of dask delayed objects, where each chunk becomes a lazy object. See function "simple_sampler" below. Other than that we also need a simulator function and a function to compute summary stats. Obs. that these does not have to return lazy obejcts, that's being handled by sciope.core. The simulator function should take 1 sample as argument (obs, not a chunk) and the summary function should take the output from the simulator as argument.<jupyter_code>dim_params = 2

@dask.delayed
def normal(n, dim):
    return np.random.randn(n,dim)

#here is a simple sampler of a 2 dim unit normal
def simple_sampler(n, chunk_size=1):
    res = []
    for i in range(int(n/chunk_size)):
        res.append(normal(chunk_size, dim_params))
    return res

#here we don't modify the samples, for demonstration purposes later
def sim(x):
    return x + 0


#some summary stats, obs the reshape, the reason we want the shape (1,2) is to allow for cases 
#when ensembles (n_trajectories > 1 in gillespy2 ) are computed, when we then can compute e.g the mean 
def summ(x):
    return np.array([np.sum(x), np.min(x)]).reshape(1,2)<jupyter_output><empty_output><jupyter_text>Dask computational graph as lazy objectsOnce we have these three functions we can pass them to core.get_graph_chunked (there is also a core.get_graph_unchunked in case one would like to create one task per sample), this function will create the dask computational graph in the form of lazy objects. Here is some details about the arguments and output of the function:    Parameters    ----------    param_func : callable        the parameter sampling function, see sciope.designs, sciope.sampling        and sciope.utilities.priors    sim_func : callable        the simulator function which takes a parameter point as argument    summaries_func : callable, optional        the summaries statistics function which takes a simulation result,        by default None    dist_func : callable, optional        the distance function between , by default None    batch_size : int, optional        the number of points being sampled in each batch, by default 10    chunk_size : int, optional        decription, by default 2    ensemble_size : int, optional        [description], by default 1        Returns    -------    dict        with keys 'parameters', 'trajectories', 'summarystats' and 'distances'        values being dask delayed objects<jupyter_code>#number of samples
n = 10
#number of samples in a chunk (one task)
n_chunks = 2
test = core.get_graph_chunked(param_func=simple_sampler, 
                              sim_func=sim, summaries_func=summ, batch_size=n, chunk_size=n_chunks)
test<jupyter_output><empty_output><jupyter_text>Since the number of samples is 10 and the chunk size is 2 we will get 5 lazy objects (delayed dask objects) per function. The reason we "decoupled" functions and thus the lazy objects in this (why not have one lazy object for the whole workflow of one sample?) is that we want to keep the results from each function call (especially for exploration). This is just an easy way of doing it since we can compute each delayed object indvidually. There might other more efficient ways of doing this however, and as will be explained later (computations) this becomes a pains if one like to collect individual results as tasks are completed.    Visualize the graphIf you have graphviz installed on your machine (highly recommend to install it, but no worry if you don't have it, I have pre-rendered the graph in this notebook) you can visualize the computational graph by passing the last node (summarystats) to dask.visulaize(). Here, we see the delayed "normal" function (functions represented as circles) as the inital nodes and each edge to a box represent the results. The two last functions are both called "delay_func_chunk" which corresponds to the simulator and the summaries function. The reason they are called this, as mentioned earlier, because sciope.core wrap these functions to make the calls as delayed object and also handle the chunks from "simple_sampler". Future note: it would be great if the user could specify the names of the functions passed to core.get_graph_chunked and then rename the IDs of the dask tasks (delayed objects) accordingly, I don't know how to do this though. In general, we should be in more contol of the IDs.<jupyter_code>dask.visualize(test["summarystats"])<jupyter_output><empty_output><jupyter_text>Start the computationsThere several ways we can start the computations, either calling dask.compute() on the collections of delayed objects or using dask.persist() which will persist the results on cloud recourses if a cluster is being used. There is an third option when using a cloud cluster and that is to collect results to local RAM as they are completed, but for that we need the dask Futures objects. Sciope.stochmet (exploration) uses the dask.distributed.as_completed(). The problem with is that we lose track of the order of computations and thus also which samples are mapped to which summaries results. To overcomes this one can keep track of the indices and task ID's, but optimally all tasks (sampler, simulation and summaries) corresponding to the workflow of one sample should have a common substring in the task ID's.         Since we have not deployed a local dask cluster in this example, lets use dask.compute() on each node level (function) of the graph<jupyter_code>#compute the samples, kernel will be busy until all samples has been computed
samples, = dask.compute(test["parameters"])
#convert to numpy array
def ravel(lst):
    lst = np.array(lst)
    print(lst.shape)
    print("i.e, still chunks...lets ravel the chunks")
    lst = lst.reshape((n,dim_params))
    print("shape is now: ", lst.shape)
    return lst
samples = ravel(samples)
#then let compute the simulation function
trajectories, = dask.compute(test["trajectories"])
#convert to numpy array
trajectories = ravel(trajectories)
# lets plot the samples
plt.scatter(samples[:,0], samples[:,1], label='before')
plt.scatter(trajectories[:,0], trajectories[:,1], label='after')
plt.legend()<jupyter_output><empty_output><jupyter_text>Remember that we simply returned the samples without modifying them in the "sim" function? But why do we then see different samples in the plot? That's because of the delayed objects of "trajectories" are dependent on the delayed objects from "normal" so if we call compute on "trajectories" dask will compute the graph again from "normal" nodes and up the graph, i.e it's not deterministic in this case since we sample from a normal. The beauty with this for inference and exploration is that we only need to create the computational graph once and then simply call compute if we want new samples. But how do we get the results from each function call that are dependent? Like this:<jupyter_code>samples, trajectories = dask.compute(test["parameters"], test["trajectories"])
samples = ravel(samples)
trajectories = ravel(trajectories)
plt.scatter(samples[:,0], samples[:,1], label='before')
plt.scatter(trajectories[:,0], trajectories[:,1], label='after')
plt.legend()<jupyter_output><empty_output><jupyter_text>Now they are the same! In the same way, if we like to compute the whole workflow and keep the results from each function call/delayed object we do:<jupyter_code>samples, trajectories, summaries = dask.compute(test["parameters"], 
                                                test["trajectories"],
                                                test["summarystats"])
#convert to numpy array
summaries = ravel(summaries)
print("dim1: summation, dim2: minimum")
summaries<jupyter_output><empty_output><jupyter_text>Ok lets have a quick look at dask.persist(), for this we need to start a local cluster. You can click on the dashboard link.<jupyter_code>from dask.distributed import Client

c = Client()
c<jupyter_output><empty_output><jupyter_text>To actually observe how persit work we need to add some time delays to our functions. Lets just modify the "sim" function, this time we can modify the samples as well :). Since we change a function we need to rebuild the graph.<jupyter_code>from time import sleep

def sim(x):
    #sleep for 2 sec
    sleep(2)
    return x + 1

test = core.get_graph_chunked(param_func=simple_sampler, 
                              sim_func=sim, summaries_func=summ, batch_size=n, chunk_size=n_chunks)
test<jupyter_output><empty_output><jupyter_text>Ok, lets call persist and lets say we only care about the results from summaries for this case! First notice that the notebook cell completes immediately, then go to the dashboard.<jupyter_code>summaries, = dask.persist(test["summarystats"])<jupyter_output><empty_output><jupyter_text>You can see in the dashboard that each task takes 4 s = chunk size x 2 s.Now the summaries don't contain the results (it's persisted on the workers of the cluster), but if we want the results we call compute, and since the results has already been computed it will return the results to local RAM.<jupyter_code>#delayed objects
summaries
#get result
dask.compute(summaries)<jupyter_output><empty_output><jupyter_text>Ok, last thing to show (I promise!) is how as_computed works. Since most of us have more than 4 cores lets increase the number of points a bit.<jupyter_code>from dask.distributed import as_completed, futures_of

def sim(x):
    sec = np.random.randint(5)
    sleep(sec)
    return x + 1

n = 30

test = core.get_graph_chunked(param_func=simple_sampler, 
                              sim_func=sim, summaries_func=summ, batch_size=n, chunk_size=n_chunks)

summaries, = dask.persist(test["summarystats"])

futures = futures_of(summaries)

for f in as_completed(futures):
    print("summaries with ID now finished: {}".format(f.key))<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.06730862756541192, 'fraction_numerical': 0.004085205719288007, 'mean_word_length': 4.431590068674062, 'pattern_counts': {'":': 0, '<': 45, '<?xml version=': 0, '>': 46, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '11', 'dir': 'jupyter-structured-clean-dedup', 'id': '166983', 'license': 'Apache-2.0', 'n_tokens_mistral': 2838, 'n_tokens_neox': 2736, 'n_words': 1316, 'path': 'examples/core/intro_core.ipynb', 'repo_name': 'sciope/sciope', 'size': '10281'}