test / app /main.py
Vsevolod's picture
init
c104e4a
import os
import csv
import time
import gradio as gr
from fastapi import FastAPI
from fastapi_utils.tasks import repeat_every
class Timer(object):
def __enter__(self):
self.start = time.perf_counter()
return self
def __exit__(self, typ, value, traceback):
self.elapsed = time.perf_counter() - self.start
file_changed = False
HF_TOKEN = os.getenv('HF_TOKEN')
root_dataset_path = "datasets"
huggingface_dataset_path = "Vsevolod/text2rec-users-feedback"
feedback_dataset = gr.HuggingFaceDatasetSaver(HF_TOKEN, huggingface_dataset_path)
local_dataset_path = f"{root_dataset_path}/{huggingface_dataset_path}"
data_path = f"{local_dataset_path}/data.csv"
print(os.getcwd())
print(local_dataset_path)
def foo(x):
global file_changed
now = time.time()
with open(data_path, mode="a", newline="", encoding="utf-8") as file:
writer = csv.writer(file)
writer.writerow(("Фильмы про путешествия во времени","Грань будущего","Релевантно", "mNpuTAujx-j1E6mm4PVOMQ", now, 228))
file_changed = True
return x
def create_interface():
with gr.Blocks() as demo:
query = gr.Textbox(label="Запрос")
search_btn = gr.Button(label="Поиск")
output = gr.Textbox(label="Ответ")
search_btn.click(foo, inputs=query, outputs=output)
feedback_dataset.setup([], root_dataset_path)
return demo
def get_line_count(data_path):
with open(data_path, "r", encoding="utf-8") as csvfile:
line_count = len([None for row in csv.reader(csvfile)]) - 1
return line_count
demo = create_interface()
app = FastAPI(docs_url=None, redoc_url=None)
@app.on_event("startup")
@repeat_every(seconds=20)
def repeat():
global file_changed
if not file_changed:
return
print("Updating remote repo")
with Timer() as timer:
feedback_dataset.repo.git_pull(lfs=True)
line_count = get_line_count(data_path)
feedback_dataset.repo.push_to_hub(commit_message="Flagged sample #{}".format(line_count))
print(f"Elapsed: {timer.elapsed:.3f}")
file_changed = False
app.mount("/", demo.app)
# from fastapi import FastAPI import gradio as gr app = FastAPI()
# @app.get("/")
# def read_main():
# return {"message": "This is your main app"}
# io = gr.Interface(lambda x: "Hello, " + x + "!", "textbox", "textbox")
# app = gr.mount_gradio_app(app, io, path="/gradio")