app.py

from threading import Thread
import gradio as gr
import inspect
from gradio import routes
from typing import List, Type

import requests, os, re, asyncio, json
import math, random
import time
import datetime
import hashlib
from Blockchain import Blockchain

loop = asyncio.get_event_loop()

# init code
def get_types(cls_set: List[Type], component: str):
    docset = []
    types = []
    if component == "input":
        for cls in cls_set:
            doc = inspect.getdoc(cls)
            doc_lines = doc.split("\n")
            docset.append(doc_lines[1].split(":")[-1])
            types.append(doc_lines[1].split(")")[0].split("(")[-1])
    else:
        for cls in cls_set:
            doc = inspect.getdoc(cls)
            doc_lines = doc.split("\n")
            docset.append(doc_lines[-1].split(":")[-1])
            types.append(doc_lines[-1].split(")")[0].split("(")[-1])
    return docset, types
routes.get_types = get_types

# App code

account_list = dict()

account_list['id'] = "pass"

name_list = dict()
name_list['id'] = 'name'

p2p_list = dict()
p2p_list['id'] = '11111111'

gpu_add_list = []


def register(id, pw):
    if id in account_list:
        return "exist"
    else:
        account_list[id] = pw
        return "ok"

def login(id, pw):
    if id in account_list:
        if account_list[id] == pw:
            return "ok"
        else:
            return "password error"
    else:
        return "no id"

def add_name(id, name):
    name_list[id] = name
    return "ok"

def get_name(id):
    if id in name_list:
        return name_list[id]
    else:
        return "no id"

def get_id(name):
    reverse_dict= dict(map(reversed,name_list.items()))
    if name in reverse_dict:
        return reverse_dict[name]
    else:
        return "no name"

def add_p(id, p_id):
    p2p_list[id] = p_id
    return "ok"

def get_p(id):
    if id in p2p_list:
        return p2p_list[id]
    else:
        return "no id"

def get_id_from_p2p(i):
    reverse_dict= dict(map(reversed,p2p_list.items()))
    if i in reverse_dict:
        return reverse_dict[i]
    else:
        return "no id"

# Blockchain code


model_name = "petals-team/StableBeluga2"
def get_peers(model_name):
    data = requests.get("https://health.petals.dev/api/v1/state").json()
    out = []
    for d in data['model_reports']:
        if d['name'] == model_name:
            for r in d['server_rows']:
                out.append(r['peer_id'])
    return out


def get_used_gpus(model_name):
    data = requests.get("https://health.petals.dev/api/v1/state").json()
    out = []
    for d in data['model_reports']:
        if d['name'] == model_name:
            return round(d['num_blocks']/0.45, 2)
    return 0

blockchain = Blockchain()

def add_transaction(id, kind, data):
    if kind == "add" or kind == "inference" or kind == "out":
        blockchain.new_transaction(id, kind, data)
        
        if len(blockchain.current_transactions)>=10:
            proof(model_name)        
            new_block = blockchain.new_block()
        
        return "ok"
    else:
        return "fail"

def proof(model_name):
    peers = get_peers(model_name)
    for p in gpu_add_list:
        if not p in peers:
            add_transaction(get_id_from_p2p(peer), "out", 0)    

def get_coin(id):
    c = blockchain.get_user_balance(id)
    return c


def get_gpus():
    output = []
    for id, mem in blockchain.user_gpus.items():
        output.append({"name":get_name(id),"gpu":mem})
    return output
def used_gpu():
    output = 0
    for mem in get_gpus():
        output += mem * (5 + random.randrange(1,4)/10)
    return output

def get_data():
    output = []
    with open('Chain_data.json') as json_file:
        json_data = json.load(json_file)
    output.append({"gpus":get_gpus(), "total":{"total" : blockchain.get_total_gpu_mem(), "used":get_used_gpus(model_name)}, "chain":json_data+blockchain.chain})

    with open('Chain_data.json', 'w') as outfile:
        json.dump(output[0]['chain'], outfile)
    return output


def chat(id, npc, prompt):
    if get_coin(id) == 0:
        return "no coin"
    
    # model inference
    output = "AI 응답입니다."
    add_transaction(id, "inference", {"prompt":prompt, "output":output})

    if len(blockchain.current_transactions)>=10:
        proof(model_name)        
        new_block = blockchain.new_block()      
    return output


with gr.Blocks() as demo:

    rr = gr.Interface(
      fn=register,
      inputs=["text", "text"],
      outputs="text",
      description="register, 회원가입(성공시:ok, 중복시:exist 반환)\n /run/predict",
    )

    ll = gr.Interface(
      fn=login,
      inputs=["text", "text"],
      outputs="text",
      description="login, 로그인(성공시: ok, 실패시: password error, 아이디가 없으면: no id) \n /run/predict_",
    )

    ad = gr.Interface(
      fn=add_name,
      inputs=["text", "text"],
      outputs="text",
      description="add_name, id로 닉네임 추가. ok 반환.\n /run/predict_2",
    )
    
    nn = gr.Interface(
      fn=get_name,
      inputs=["text"],
      outputs="text",
      description="get_name, id로 닉네임 반환(없으면 no id)\n /run/predict_3",
    )
    
    adp = gr.Interface(
      fn=add_p,
      inputs=["text", "text"],
      outputs="text",
      description="add_p, id로 p2p id 추가. ok 반환. \n /run/predict_4",
    )

    addp = gr.Interface(
      fn=add_transaction,
      inputs=["text", "text", "text"],
      outputs="text",
      description="add_transaction  \n /run/predict_5",
    )

    gc = gr.Interface(
      fn=get_coin,
      inputs=["text"],
      outputs="text",
      description="get_coin, id로 잔여 코인(행동력) 반환. \n /run/predict_6",
    )

    gd = gr.Interface(
      fn=get_data,
      inputs=[],
      outputs="text",
      description="get_data, 시각화용 모든 데이터 반환. gpu기여 목록, total/used, chain \n /run/predict_7",
    )
    
    demo.queue(max_size=32).launch(enable_queue=True)