-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathtools.py
144 lines (120 loc) · 5.12 KB
/
tools.py
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
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
import random
import gradio as gr
import globales
from huggingface_hub import HfApi
import bridges
import sulkuPypi
def theme_selector():
temas_posibles = [
gr.themes.Base(),
gr.themes.Default(),
gr.themes.Glass(),
gr.themes.Monochrome(),
gr.themes.Soft()
]
tema = random.choice(temas_posibles)
return tema
def eligeAPI(opcion):
funciones = {
"eligeQuotaOCosto": eligeQuotaOCosto,
"eligeAOB": eligeAOB,
"eligeGratisOCosto": eligeGratisOCosto
}
if opcion in funciones:
print("Opción en Funciones")
funcion_elegida = funciones[opcion]
api, tipo_api = funcion_elegida()
else:
print("Opción no válida")
return api, tipo_api
#Los tipos de elección son diferentes porque tienen diferentes reglas de negocio.
def eligeGratisOCosto():
#Se eligirá en los casos en los que sin costo funciona bien como Astroblend pero por si se quiere mejorar hacia Costo.
#Por ahora funcionará exactamente igual que eligeAoB, en el futuro se basará en reglas de membresía.
apis = [globales.api_a, globales.api_b]
api_elegida = random.choice(apis)
print("Print api elegida: ", api_elegida)
api, tipo_api = api_elegida
return api, tipo_api
def eligeAOB():
#Se eligirá cuando se tenga un control sobre la cantidad en queu y se redirija hacia una segunda fuente alternativa.
# Lista con las opciones
apis = [globales.api_a, globales.api_b]
api_elegida = random.choice(apis)
print("Print api elegida: ", api_elegida)
api, tipo_api = api_elegida
return api, tipo_api
def eligeQuotaOCosto():
#Se eligirá en los casos en los que se use Zero, para extender las posibilidades de Quota y después usar Costo.
diferencia = sulkuPypi.getQuota() - globales.process_cost
if diferencia >= 0:
#Entonces puedes usar Zero.
api, tipo_api = globales.api_zero
#Además Si el resultado puede usar la Zero "por última vez", debe de ir prendiendo la otra.
#if diferencia es menor que el costo de un sig. del proceso, ve iniciando ya la otra API.
if diferencia < globales.process_cost:
print("Preventivamente iremos prendiendo la otra.")
initAPI(globales.api_cost)
else:
api, tipo_api = globales.api_cost
print("La API elegida es: ", api)
return api, tipo_api
def initAPI(api):
global result_from_initAPI
try:
repo_id = api
llave = HfApi(token=bridges.hug)
runtime = api.get_space_runtime(repo_id=repo_id)
print("Stage: ", runtime.stage)
#"RUNNING_BUILDING", "APP_STARTING", "SLEEPING", "RUNNING", "PAUSED", "RUNTIME_ERROR"
if runtime.stage == "SLEEPING" or runtime.stage == "PAUSED":
llave.restart_space(repo_id=repo_id)
print("Hardware: ", runtime.hardware)
result_from_initAPI = runtime.stage
except Exception as e:
#Creo que ya no debería de llegar aquí.
print("No api, encendiendo: ", e)
result_from_initAPI = str(e)
return result_from_initAPI
def titulizaExcepDeAPI(e):
print("El e recibido por tituliza es: ", e)
#Resume una excepción a un título manejable.
if "RUNTIME_ERROR" in str(e):
resultado = "RUNTIME_ERROR" #api mal construida tiene error.
elif "PAUSED" in str(e):
resultado = "PAUSED"
elif "The read operation timed out" in str(e): #IMPORTANTE, ESTO TAMBIÉN SUCEDE CUANDO LA DESPIERTAS Y ES INSTANTÁNEO.
resultado = "STARTING"
elif "GPU quota" in str(e):
resultado = recortadorQuota(str(e)) #Cuando se trata de quota regresa el resultado completo convertido a string.
elif "handshake operation timed out" in str(e):
resultado = "HANDSHAKE_ERROR"
elif "File None does not exist on local filesystem and is not a valid URL." in str(e):
resultado = "NO_FILE"
#A partir de aquí son casos propios de cada aplicación.
#image-blend
elif "no-source-face" in str(e):
resultado = "NO_FACE"
#splashmix
elif "Unable to detect a face" in str(e):
resultado = "NO_FACE"
else:
resultado = "GENERAL"
return resultado
def recortadorQuota(texto_quota):
# Encontrar el índice de inicio (después de "exception:")
indice_inicio = texto_quota.find("exception:") + len("exception:")
# Encontrar el índice de final (antes de "<a")
indice_final = texto_quota.find("<a")
if indice_final == -1: #Significa que no encontró el texto "<a" entonces buscará Sign-Up.
indice_final = texto_quota.find("Sign-up")
#Extraer la subcadena
subcadena = texto_quota[indice_inicio:indice_final]
#Y si el objetivo es nunca desplegar el texto Hugging Face, éste es el plan de escape final.
if "Hugging" in subcadena:
nuevo_mensaje = "Your quota is exceeded, try again in few hours please."
return nuevo_mensaje
else:
print("El recorte quedó: ")
print(subcadena)
return subcadena