Skip to content

API

API

This is the class to communicate with the API component

Source code in Agent/utils/api.py 
 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
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
class API:
    """
    This is the class to communicate with the API component
    """

    def __init__(
        self,
        domain: str = API_DOMAIN,
        token: str = "",
        uuid: str = "",
        task_name: str = "llm",
    ):
        """
        Init API class to communicate with the API
        Args:
            domain (str): The domain of the API
            token (str): The token to authenticate
            uuid (str): The UUID of the worker
            task_name (str): The task type of the worker
        """
        self.domain = domain
        self.token = token
        self.task_name = task_name
        self.uuid = uuid
        self.mac_address = getmac.get_mac_address()
        self.ip_address = self.get_local_ip()

    def verify_token(self) -> bool:
        try:
            url = f"{self.domain}/authenticate/api/token/verify/"
            r = requests.post(
                url,
                headers={"Authorization": f"Token {self.token}"},
                data={"token": self.token},
            )
            logger.info(f"POST {url} {r.status_code}")
            logger.info(r.json())
            if r.status_code != 200:
                return False
            return True
        except Exception as e:
            logger.error(f"Error verifying token: {e}")
            return False

    def get_available_models(self):
        """
        Get the available LLM models from the API
        Returns:

        """
        url = f"{self.domain}/llm/config"
        r = requests.get(url, headers={"Authorization": f"Token {self.token}"})
        logger.info(f"GET {url} {r.status_code}")
        return r.json()

    def get_task(self):
        """
        Get the task from the API
        Returns:

        """
        logger.debug(self.task_name)
        url = f"{self.domain}/queue_task/task/{self.task_name}/"
        r = requests.get(url, headers={"Authorization": f"Token {self.token}"})
        logger.info(f"GET {url} {r.status_code}")
        logger.info(r.text)
        if r.status_code != 200:
            return None
        return r.json()

    def post_task_result(
        self,
        task: Task,
    ):
        """
        Post the task result to the API
        Args:
            task[Task]: The task to post the result

        Returns:

        """
        url = f"{self.domain}/queue_task/{task.id}/update_result/"
        r = requests.post(
            url,
            data=task.json(),
            headers={
                "Authorization": f"Token {self.token}",
                "Content-Type": "application/json",
            },
        )
        logger.info(f"POST {url} {r.status_code}")
        logger.info(r.text)
        if r.status_code != 200:
            return None
        return r.json()

    def register_or_update_worker(self):
        """
        Register or update the  worker
        So we can know whether the worker is alive or not
        """
        try:
            url = f"{self.domain}/queue_task/worker/"
            r = requests.post(
                url,
                data={
                    "uuid": self.uuid,
                    "mac_address": self.mac_address,
                    "ip_address": self.ip_address,
                    "task_name": self.task_name,
                },
                headers={"Authorization": f"Token {self.token}"},
            )
            logger.info(f"POST {url} {r.status_code}")
            # logger.info(r.text)
            return r.json()
        except Exception as e:
            logger.error(f"Error registering worker: {e}")

    @staticmethod
    def get_local_ip() -> str:
        """
        Get the local IP address
        Returns:
            str: The local IP address

        """
        # Create a socket object
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        try:
            # doesn't matter if the address is reachable
            s.connect(("10.255.255.255", 1))
            ip = s.getsockname()[0]
        except Exception as e:
            logger.error(f"Error getting local IP: {e}")
            ip = "127.0.0.1"
        finally:
            s.close()
        return ip

    def get_storage_solution(self):
        """
        Get the storage solution from the API
        Returns:

        """
        url = f"{self.domain}/hardware/storage_solution/"
        r = requests.get(
            url, headers={"Authorization": f"Token {self.token}"}, timeout=30
        )
        logger.info(f"GET {url} {r.status_code}")
        if r.status_code != 200:
            return None
        data = r.json()
        logger.info(data)
        return data.get("storage_solution", "volume")

    def upload_file(
        self,
        source_file: str,
        dest_path: str,
    ):
        """
        Upload the file to the API
        """
        url = f"{self.domain}/hardware/upload_file/"
        files = {"file": open(source_file, "rb")}
        data = {
            "dest_path": dest_path,
        }
        r = requests.post(
            url,
            files=files,
            data=data,
            headers={"Authorization": f"Token {self.token}"},
            timeout=30,
        )
        logger.info(f"POST {url} {r.status_code}")
        if r.status_code != 200:
            return None
        return True

    def list_files(self, from_time=None):
        """
        List the files from the API
        """
        url = f"{self.domain}/hardware/list_files/"
        data = {
            "from_time": from_time,
        }
        r = requests.get(
            url,
            data=data,
            headers={"Authorization": f"Token {self.token}"},
        )
        logger.info(f"GET {url} {r.status_code}")
        if r.status_code != 200:
            return None
        return r.json()

    def download_file_link(self, file_id, file_type):
        """
        Get the download file link
        """
        url = f"{self.domain}/hardware/download_file_link/"
        data = {
            "file_id": file_id,
            "file_type": file_type,
        }

        r = requests.get(
            url,
            data=data,
            headers={"Authorization": f"Token {self.token}"},
        )
        logger.info(f"GET {url} {r.status_code}")
        if r.status_code != 200:
            return None
        return r.json()

__init__(domain=API_DOMAIN, token='', uuid='', task_name='llm')

Init API class to communicate with the API Args: domain (str): The domain of the API token (str): The token to authenticate uuid (str): The UUID of the worker task_name (str): The task type of the worker

Source code in Agent/utils/api.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
def __init__(
    self,
    domain: str = API_DOMAIN,
    token: str = "",
    uuid: str = "",
    task_name: str = "llm",
):
    """
    Init API class to communicate with the API
    Args:
        domain (str): The domain of the API
        token (str): The token to authenticate
        uuid (str): The UUID of the worker
        task_name (str): The task type of the worker
    """
    self.domain = domain
    self.token = token
    self.task_name = task_name
    self.uuid = uuid
    self.mac_address = getmac.get_mac_address()
    self.ip_address = self.get_local_ip()

Get the download file link

Source code in Agent/utils/api.py 
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
def download_file_link(self, file_id, file_type):
    """
    Get the download file link
    """
    url = f"{self.domain}/hardware/download_file_link/"
    data = {
        "file_id": file_id,
        "file_type": file_type,
    }

    r = requests.get(
        url,
        data=data,
        headers={"Authorization": f"Token {self.token}"},
    )
    logger.info(f"GET {url} {r.status_code}")
    if r.status_code != 200:
        return None
    return r.json()

get_available_models()

Get the available LLM models from the API Returns:

Source code in Agent/utils/api.py 
58
59
60
61
62
63
64
65
66
67
def get_available_models(self):
    """
    Get the available LLM models from the API
    Returns:

    """
    url = f"{self.domain}/llm/config"
    r = requests.get(url, headers={"Authorization": f"Token {self.token}"})
    logger.info(f"GET {url} {r.status_code}")
    return r.json()

get_local_ip() staticmethod

Get the local IP address Returns: str: The local IP address

Source code in Agent/utils/api.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
@staticmethod
def get_local_ip() -> str:
    """
    Get the local IP address
    Returns:
        str: The local IP address

    """
    # Create a socket object
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    try:
        # doesn't matter if the address is reachable
        s.connect(("10.255.255.255", 1))
        ip = s.getsockname()[0]
    except Exception as e:
        logger.error(f"Error getting local IP: {e}")
        ip = "127.0.0.1"
    finally:
        s.close()
    return ip

get_storage_solution()

Get the storage solution from the API Returns:

Source code in Agent/utils/api.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def get_storage_solution(self):
    """
    Get the storage solution from the API
    Returns:

    """
    url = f"{self.domain}/hardware/storage_solution/"
    r = requests.get(
        url, headers={"Authorization": f"Token {self.token}"}, timeout=30
    )
    logger.info(f"GET {url} {r.status_code}")
    if r.status_code != 200:
        return None
    data = r.json()
    logger.info(data)
    return data.get("storage_solution", "volume")

get_task()

Get the task from the API Returns:

Source code in Agent/utils/api.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def get_task(self):
    """
    Get the task from the API
    Returns:

    """
    logger.debug(self.task_name)
    url = f"{self.domain}/queue_task/task/{self.task_name}/"
    r = requests.get(url, headers={"Authorization": f"Token {self.token}"})
    logger.info(f"GET {url} {r.status_code}")
    logger.info(r.text)
    if r.status_code != 200:
        return None
    return r.json()

list_files(from_time=None)

List the files from the API

Source code in Agent/utils/api.py 
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
def list_files(self, from_time=None):
    """
    List the files from the API
    """
    url = f"{self.domain}/hardware/list_files/"
    data = {
        "from_time": from_time,
    }
    r = requests.get(
        url,
        data=data,
        headers={"Authorization": f"Token {self.token}"},
    )
    logger.info(f"GET {url} {r.status_code}")
    if r.status_code != 200:
        return None
    return r.json()

post_task_result(task)

Post the task result to the API Args: task[Task]: The task to post the result

Returns:

Source code in Agent/utils/api.py 
 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
def post_task_result(
    self,
    task: Task,
):
    """
    Post the task result to the API
    Args:
        task[Task]: The task to post the result

    Returns:

    """
    url = f"{self.domain}/queue_task/{task.id}/update_result/"
    r = requests.post(
        url,
        data=task.json(),
        headers={
            "Authorization": f"Token {self.token}",
            "Content-Type": "application/json",
        },
    )
    logger.info(f"POST {url} {r.status_code}")
    logger.info(r.text)
    if r.status_code != 200:
        return None
    return r.json()

register_or_update_worker()

Register or update the worker So we can know whether the worker is alive or not

Source code in Agent/utils/api.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def register_or_update_worker(self):
    """
    Register or update the  worker
    So we can know whether the worker is alive or not
    """
    try:
        url = f"{self.domain}/queue_task/worker/"
        r = requests.post(
            url,
            data={
                "uuid": self.uuid,
                "mac_address": self.mac_address,
                "ip_address": self.ip_address,
                "task_name": self.task_name,
            },
            headers={"Authorization": f"Token {self.token}"},
        )
        logger.info(f"POST {url} {r.status_code}")
        # logger.info(r.text)
        return r.json()
    except Exception as e:
        logger.error(f"Error registering worker: {e}")

upload_file(source_file, dest_path)

Upload the file to the API

Source code in Agent/utils/api.py 
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
def upload_file(
    self,
    source_file: str,
    dest_path: str,
):
    """
    Upload the file to the API
    """
    url = f"{self.domain}/hardware/upload_file/"
    files = {"file": open(source_file, "rb")}
    data = {
        "dest_path": dest_path,
    }
    r = requests.post(
        url,
        files=files,
        data=data,
        headers={"Authorization": f"Token {self.token}"},
        timeout=30,
    )
    logger.info(f"POST {url} {r.status_code}")
    if r.status_code != 200:
        return None
    return True