File size: 2,730 Bytes
53dc0ac
dfe6290
5117e0a
53dc0ac
dfe6290
a565e7e
53dc0ac
 
2db4636
54b3256
5117e0a
a565e7e
5117e0a
a565e7e
 
 
5117e0a
 
 
 
a565e7e
 
dfe6290
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
cd5bd85
a565e7e
 
 
cd5bd85
a565e7e
 
 
 
 
 
aff284c
 
 
 
 
 
 
 
 
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
import os
import requests
import streamlit as st

from typing import List, Tuple


data_dir = os.path.join(os.path.dirname(__file__), '..', 'data')
img_dir = os.path.join(os.path.dirname(__file__), '..', 'img')
config_dir = os.path.join(os.path.dirname(__file__), '..', 'config')


def hf_api_token() -> str:
    """
    Utility single access point to look up the hugging face access token.
    """
    token = st.secrets['hf_token']
    if token is None:
        raise ValueError('No HF access token found in streamlit secrets')
    return token


HF_RUNNING = 'running'
HF_SCALEDTOZERO = 'scaleToZero'
HF_PAUSED = 'paused'


def hf_endpoint_status(username: str, endpoint_name: str, api_token: str = None) -> Tuple[bool, bool]:
    """
    Utility to check the status of a hugging face inference endpoint
    :return: tuple of booleans first is if it is available, second is if it fully shutdown.  False, True indicates it is not fully available or fully shutdown so is in process of starting up or shutting down
    """
    url = f'https://api.endpoints.huggingface.cloud/v2/endpoint/{username}/{endpoint_name}'
    if api_token is None:
        api_token = hf_api_token()
    response = requests.get(url, headers={"Authorization" : f"Bearer {api_token}"})
    if response.status_code != 200:
        raise ValueError(f"Likely config error. Received status code {response.status_code} - {response.text}")
    else:
        state = response.json()['status']['state']
        return state


def pause_hf_endpoint(username: str, endpoint_name: str, api_token: str = None) -> None:
    url = f'https://api.endpoints.huggingface.cloud/v2/endpoint/{username}/{endpoint_name}/pause'
    if api_token is None:
        api_token = hf_api_token()
    requests.post(url, headers={"Authorization": f"Bearer {api_token}"})


def resume_hf_endpoint(username: str, endpoint_name: str, api_token: str = None) -> None:
    url = f'https://api.endpoints.huggingface.cloud/v2/endpoint/{username}/{endpoint_name}/resume'
    if api_token is None:
        api_token = hf_api_token()
    requests.post(url, headers={"Authorization": f"Bearer {api_token}"})


def join_items_comma_and(items: List[str]) -> str:
    """
    Utility to convert a list of items to lowercase strings, comma separated and ending with and
    """
    items = [str(i).strip() for i in items]
    string_count = len(items)
    if string_count == 0:
        return ""
    if string_count == 1:
        return items[0]
    return f"{', '.join(items[:-1])} and {items[-1]}"


def escape_dollars(text: str) -> str:
    """
    Convenience function to escape dollar signs for prices in markdown
    """
    if text is None:
        return text
    return text.replace("$", "\\$")