Spaces:
Runtime error
Runtime error
File size: 2,216 Bytes
763a996 6e17b85 763a996 e0c2374 763a996 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 |
"""Prep gradio API."""
# pylint: diable=invalid-name
from typing import List, Optional, Union
import os
from pathlib import Path
import multiprocessing as mp
import numpy as np
import psutil
import ray
import gradio as gr
# import joblib
import more_itertools as mit
import numpy as np
import psutil
import ray
from about_time import about_time
from logzero import logger
from radio_embed import radio_embed
num_cpus_m = mp.cpu_count()
num_cpus = psutil.cpu_count(logical=False)
filename = "fangfang-en.txt"
lines = Path(filename).read_text("utf8").splitlines()
lst = [_.strip() for _ in lines if _.strip()]
args_m = ["\n".join(elm) for elm in mit.divide(num_cpus_m, lst)]
args = ["\n".join(elm) for elm in mit.divide(num_cpus, lst)]
os.environ["TOKENIZERS_PARALLELISM"] = "false"
if not ray.is_initialized():
ray.init(num_cpus=num_cpus)
@ray.remote
def ray_embed(text):
"""Embed text to d-512."""
return radio_embed(text)
def test_pool(func, args_, num_procs=None):
"""Test mp.Pool."""
if num_procs is None:
num_procs = num_cpus_m
elif num_procs < 1:
num_procs = num_cpus_m
with mp.Pool(num_procs) as pool:
ret = pool.map(func, args_)
return ret
def fn(type_, num_cpus):
if type_ in ["mp_pool"]:
with about_time() as dur:
_ = test_pool(radio_embed, args, num_procs=num_cpus)
return dur.duration_human
with gr.Blocks() as blocks:
with gr.Row():
type_ = gr.Radio(
label="type",
choices=[
"mp_pool",
"joblib(loky)",
"joblib(mp)",
"ray",
],
value="mp_pool",
interactive=False,
)
num_cpus_ = gr.Slider(
label="num_procs",
value=num_cpus_m,
minimum=1,
maximum=2 * num_cpus_m,
step=1,
)
btn = gr.Button("Go")
out = gr.Textbox(
label="time elapsed",
value="",
)
btn.click(
fn=fn,
inputs=[type_, num_cpus_],
outputs=out,
api_name="time",
)
# gr.Markdown(f"{description}")
blocks.launch(enable_queue=True)
|