Files
scgenerator/examples/playground_threading.py

34 lines
794 B
Python

from tqdm import tqdm
import scgenerator as sc
import numpy as np
import time
s = "The file to read. File-like objects must support the".split()
size = 5
def do_stuff(name: str, stuff: int, pbar=tqdm):
speed = np.random.rand()
for i in pbar(range(size)):
time.sleep(speed)
return np.arange(size * 4).reshape(size, 4) * len(name) + stuff
def main():
shape = (len(s), size, 4)
out = np.zeros(shape)
out_control = np.zeros(shape)
args = [(el, i) for i, el in enumerate(s)]
for i, result in sc.threading.apply_with_progress_iter(do_stuff, args, 2):
print(i, result)
out[i] = result
for i, arg in enumerate(args):
out_control[i] = do_stuff(*arg)
assert np.all(out == out_control)
if __name__ == "__main__":
main()