Skip to content

FocoosHUB

ApiClient #

A simple HTTP client for making GET, POST, and DELETE requests.

This client is initialized with an API key and a host URL, and it automatically includes the API key in the headers of each request.

Attributes:

Name Type Description
api_key str

The API key for authorization.
host_url str

The base URL for the API.
default_headers dict

Default headers including authorization and user agent.
Source code in focoos/hub/api_client.py 
 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
234
235
236
237
238
239
240
class ApiClient:
    """
    A simple HTTP client for making GET, POST, and DELETE requests.

    This client is initialized with an API key and a host URL, and it
    automatically includes the API key in the headers of each request.

    Attributes:
        api_key (str): The API key for authorization.
        host_url (str): The base URL for the API.
        default_headers (dict): Default headers including authorization and user agent.
    """

    def __init__(
        self,
        api_key: Optional[str] = None,
        host_url: Optional[str] = None,
    ):
        """
        Initialize the ApiClient with an API key and host URL.

        Args:
            api_key (str): The API key for authorization.
            host_url (str): The base URL for the API.
        """
        # Use provided api_key if not None, otherwise use config
        self.api_key = api_key if api_key is not None else FOCOOS_CONFIG.focoos_api_key
        self.host_url = host_url or FOCOOS_CONFIG.default_host_url

        self.default_headers = {
            "X-API-Key": self.api_key,
            "user_agent": f"focoos/{get_focoos_version()}",
        }

    def _check_api_key(self):
        if not self.api_key or (isinstance(self.api_key, str) and self.api_key.strip() == ""):
            raise ValueError("API key is required")

    def external_get(self, path: str, params: Optional[dict] = None, stream: bool = False):
        """
        Perform a GET request to an external URL.

        Args:
            path (str): The URL path to request.
            params (Optional[dict], optional): Query parameters for the request. Defaults to None.
            stream (bool, optional): Whether to stream the response. Defaults to False.

        Returns:
            Response: The response object from the requests library.
        """
        if params is None:
            params = {}
        return requests.get(path, params=params, stream=stream)

    def get(
        self,
        path: str,
        params: Optional[dict] = None,
        extra_headers: Optional[dict] = None,
        stream: bool = False,
    ):
        """
        Perform a GET request to the specified path on the host URL.

        Args:
            path (str): The URL path to request.
            params (Optional[dict], optional): Query parameters for the request. Defaults to None.
            extra_headers (Optional[dict], optional): Additional headers to include in the request. Defaults to None.
            stream (bool, optional): Whether to stream the response. Defaults to False.

        Returns:
            Response: The response object from the requests library.
        """
        self._check_api_key()
        url = f"{self.host_url}/{path}"
        headers = self.default_headers.copy()
        if extra_headers:
            headers.update(extra_headers)
        return requests.get(url, headers=headers, params=params, stream=stream)

    def post(
        self,
        path: str,
        data: Optional[dict] = None,
        extra_headers: Optional[dict] = None,
        files=None,
    ):
        """
        Perform a POST request to the specified path on the host URL.

        Args:
            path (str): The URL path to request.
            data (Optional[dict], optional): The JSON data to send in the request body. Defaults to None.
            extra_headers (Optional[dict], optional): Additional headers to include in the request. Defaults to None.
            files (optional): Files to send in the request. Defaults to None.

        Returns:
            Response: The response object from the requests library.
        """
        self._check_api_key()
        url = f"{self.host_url}/{path}"
        headers = self.default_headers.copy()
        if extra_headers:
            headers.update(extra_headers)
        return requests.post(url, headers=headers, json=data, files=files)

    def patch(
        self,
        path: str,
        data: Optional[dict] = None,
        extra_headers: Optional[dict] = None,
    ):
        self._check_api_key()
        url = f"{self.host_url}/{path}"
        headers = self.default_headers.copy()
        if extra_headers:
            headers.update(extra_headers)
        return requests.patch(url, headers=headers, json=data)

    def external_post(
        self,
        path: str,
        data: Optional[dict] = None,
        extra_headers: Optional[dict] = None,
        files=None,
        stream: bool = False,
    ):
        """
        Perform a POST request to an external URL without using the default host URL.

        This method is used for making POST requests to external services, such as file upload
        endpoints that require direct access without the base host URL prefix.

        Args:
            path (str): The complete URL to send the request to.
            data (Optional[dict], optional): The JSON data to send in the request body. Defaults to None.
            extra_headers (Optional[dict], optional): Headers to include in the request. Defaults to None.
            files (optional): Files to send in the request. Defaults to None.

        Returns:
            Response: The response object from the requests library.
        """
        headers = {}
        if extra_headers:
            headers.update(extra_headers)
        return requests.post(path, headers=headers, json=data, files=files, stream=stream)

    def delete(self, path: str, extra_headers: Optional[dict] = None):
        """
        Perform a DELETE request to the specified path on the host URL.

        Args:
            path (str): The URL path to request.
            extra_headers (Optional[dict], optional): Additional headers to include in the request. Defaults to None.

        Returns:
            Response: The response object from the requests library.
        """
        self._check_api_key()
        url = f"{self.host_url}/{path}"
        headers = self.default_headers.copy()
        if extra_headers:
            headers.update(extra_headers)
        return requests.delete(url, headers=headers)

    def upload_file(self, path: str, file_path: str, file_size: int):
        """
        Upload a file to the specified path.

        Args:
            path (str): The API endpoint path to upload to
            file_path (str): Path to the file to upload
            file_size (int): Size of the file in bytes

        Returns:
            Response: The response from the upload request
        """
        self._check_api_key()
        return self.post(path, data={"path": file_path, "file_size_bytes": file_size})

    def download_ext_file(self, uri: str, file_dir: str, file_name: Optional[str] = None, skip_if_exists: bool = False):
        """
        Download a file from a URI to a local directory.

        Args:
            uri (str): The URI to download the file from
            file_dir (str): Local directory to save the downloaded file

        Returns:
            str: Path to the downloaded file

        Raises:
            ValueError: If the download fails or filename cannot be determined
        """
        if os.path.exists(file_dir) and not os.path.isdir(file_dir):
            raise ValueError(f"Path is not a directory: {file_dir}")
        if not os.path.exists(file_dir):
            logger.debug(f"📥 Creating directory: {file_dir}")
            os.makedirs(file_dir)
        parsed_url = urlparse(uri)
        file_name = file_name or os.path.basename(parsed_url.path)
        res = self.external_get(uri, stream=True)
        if res.status_code != 200:
            logger.error(f"Failed to download file {file_name}: {res.status_code} {res.text}")
            raise ValueError(f"Failed to download file {file_name}: {res.status_code} {res.text}")

        file_path = os.path.join(file_dir, file_name)
        if skip_if_exists and os.path.exists(file_path):
            logger.debug(f"📥 File already exists: {file_path}")
            return file_path
        total_size = int(res.headers.get("content-length", 0))
        with (
            open(file_path, "wb") as f,
            tqdm(
                desc=file_name,
                total=total_size,
                unit="B",
                unit_scale=True,
                unit_divisor=1024,
            ) as bar,
        ):
            for chunk in res.iter_content(chunk_size=8192):
                f.write(chunk)
                bar.update(len(chunk))
        logger.debug(f"📥 File downloaded: {file_path} Size: {total_size / (1024**2):.2f} MB")
        return file_path

__init__(api_key=None, host_url=None) #

Initialize the ApiClient with an API key and host URL.

Parameters:

Name Type Description Default
api_key str

The API key for authorization.

 None
host_url str

The base URL for the API.

 None
Source code in focoos/hub/api_client.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def __init__(
    self,
    api_key: Optional[str] = None,
    host_url: Optional[str] = None,
):
    """
    Initialize the ApiClient with an API key and host URL.

    Args:
        api_key (str): The API key for authorization.
        host_url (str): The base URL for the API.
    """
    # Use provided api_key if not None, otherwise use config
    self.api_key = api_key if api_key is not None else FOCOOS_CONFIG.focoos_api_key
    self.host_url = host_url or FOCOOS_CONFIG.default_host_url

    self.default_headers = {
        "X-API-Key": self.api_key,
        "user_agent": f"focoos/{get_focoos_version()}",
    }

delete(path, extra_headers=None) #

Perform a DELETE request to the specified path on the host URL.

Parameters:

Name Type Description Default
path str

The URL path to request.

 required
extra_headers Optional[dict]

Additional headers to include in the request. Defaults to None.

 None

Returns:

Name Type Description
Response

The response object from the requests library.
Source code in focoos/hub/api_client.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
def delete(self, path: str, extra_headers: Optional[dict] = None):
    """
    Perform a DELETE request to the specified path on the host URL.

    Args:
        path (str): The URL path to request.
        extra_headers (Optional[dict], optional): Additional headers to include in the request. Defaults to None.

    Returns:
        Response: The response object from the requests library.
    """
    self._check_api_key()
    url = f"{self.host_url}/{path}"
    headers = self.default_headers.copy()
    if extra_headers:
        headers.update(extra_headers)
    return requests.delete(url, headers=headers)

download_ext_file(uri, file_dir, file_name=None, skip_if_exists=False) #

Download a file from a URI to a local directory.

Parameters:

Name Type Description Default
uri str

The URI to download the file from

 required
file_dir str

Local directory to save the downloaded file

 required

Returns:

Name Type Description
str

Path to the downloaded file

Raises:

Type Description
ValueError

If the download fails or filename cannot be determined
Source code in focoos/hub/api_client.py 
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
234
235
236
237
238
239
240
def download_ext_file(self, uri: str, file_dir: str, file_name: Optional[str] = None, skip_if_exists: bool = False):
    """
    Download a file from a URI to a local directory.

    Args:
        uri (str): The URI to download the file from
        file_dir (str): Local directory to save the downloaded file

    Returns:
        str: Path to the downloaded file

    Raises:
        ValueError: If the download fails or filename cannot be determined
    """
    if os.path.exists(file_dir) and not os.path.isdir(file_dir):
        raise ValueError(f"Path is not a directory: {file_dir}")
    if not os.path.exists(file_dir):
        logger.debug(f"📥 Creating directory: {file_dir}")
        os.makedirs(file_dir)
    parsed_url = urlparse(uri)
    file_name = file_name or os.path.basename(parsed_url.path)
    res = self.external_get(uri, stream=True)
    if res.status_code != 200:
        logger.error(f"Failed to download file {file_name}: {res.status_code} {res.text}")
        raise ValueError(f"Failed to download file {file_name}: {res.status_code} {res.text}")

    file_path = os.path.join(file_dir, file_name)
    if skip_if_exists and os.path.exists(file_path):
        logger.debug(f"📥 File already exists: {file_path}")
        return file_path
    total_size = int(res.headers.get("content-length", 0))
    with (
        open(file_path, "wb") as f,
        tqdm(
            desc=file_name,
            total=total_size,
            unit="B",
            unit_scale=True,
            unit_divisor=1024,
        ) as bar,
    ):
        for chunk in res.iter_content(chunk_size=8192):
            f.write(chunk)
            bar.update(len(chunk))
    logger.debug(f"📥 File downloaded: {file_path} Size: {total_size / (1024**2):.2f} MB")
    return file_path

external_get(path, params=None, stream=False) #

Perform a GET request to an external URL.

Parameters:

Name Type Description Default
path str

The URL path to request.

 required
params Optional[dict]

Query parameters for the request. Defaults to None.

 None
stream bool

Whether to stream the response. Defaults to False.

 False

Returns:

Name Type Description
Response

The response object from the requests library.
Source code in focoos/hub/api_client.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def external_get(self, path: str, params: Optional[dict] = None, stream: bool = False):
    """
    Perform a GET request to an external URL.

    Args:
        path (str): The URL path to request.
        params (Optional[dict], optional): Query parameters for the request. Defaults to None.
        stream (bool, optional): Whether to stream the response. Defaults to False.

    Returns:
        Response: The response object from the requests library.
    """
    if params is None:
        params = {}
    return requests.get(path, params=params, stream=stream)

external_post(path, data=None, extra_headers=None, files=None, stream=False) #

Perform a POST request to an external URL without using the default host URL.

This method is used for making POST requests to external services, such as file upload endpoints that require direct access without the base host URL prefix.

Parameters:

Name Type Description Default
path str

The complete URL to send the request to.

 required
data Optional[dict]

The JSON data to send in the request body. Defaults to None.

 None
extra_headers Optional[dict]

Headers to include in the request. Defaults to None.

 None
files optional

Files to send in the request. Defaults to None.

 None

Returns:

Name Type Description
Response

The response object from the requests library.
Source code in focoos/hub/api_client.py 
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
def external_post(
    self,
    path: str,
    data: Optional[dict] = None,
    extra_headers: Optional[dict] = None,
    files=None,
    stream: bool = False,
):
    """
    Perform a POST request to an external URL without using the default host URL.

    This method is used for making POST requests to external services, such as file upload
    endpoints that require direct access without the base host URL prefix.

    Args:
        path (str): The complete URL to send the request to.
        data (Optional[dict], optional): The JSON data to send in the request body. Defaults to None.
        extra_headers (Optional[dict], optional): Headers to include in the request. Defaults to None.
        files (optional): Files to send in the request. Defaults to None.

    Returns:
        Response: The response object from the requests library.
    """
    headers = {}
    if extra_headers:
        headers.update(extra_headers)
    return requests.post(path, headers=headers, json=data, files=files, stream=stream)

get(path, params=None, extra_headers=None, stream=False) #

Perform a GET request to the specified path on the host URL.

Parameters:

Name Type Description Default
path str

The URL path to request.

 required
params Optional[dict]

Query parameters for the request. Defaults to None.

 None
extra_headers Optional[dict]

Additional headers to include in the request. Defaults to None.

 None
stream bool

Whether to stream the response. Defaults to False.

 False

Returns:

Name Type Description
Response

The response object from the requests library.
Source code in focoos/hub/api_client.py 
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
def get(
    self,
    path: str,
    params: Optional[dict] = None,
    extra_headers: Optional[dict] = None,
    stream: bool = False,
):
    """
    Perform a GET request to the specified path on the host URL.

    Args:
        path (str): The URL path to request.
        params (Optional[dict], optional): Query parameters for the request. Defaults to None.
        extra_headers (Optional[dict], optional): Additional headers to include in the request. Defaults to None.
        stream (bool, optional): Whether to stream the response. Defaults to False.

    Returns:
        Response: The response object from the requests library.
    """
    self._check_api_key()
    url = f"{self.host_url}/{path}"
    headers = self.default_headers.copy()
    if extra_headers:
        headers.update(extra_headers)
    return requests.get(url, headers=headers, params=params, stream=stream)

post(path, data=None, extra_headers=None, files=None) #

Perform a POST request to the specified path on the host URL.

Parameters:

Name Type Description Default
path str

The URL path to request.

 required
data Optional[dict]

The JSON data to send in the request body. Defaults to None.

 None
extra_headers Optional[dict]

Additional headers to include in the request. Defaults to None.

 None
files optional

Files to send in the request. Defaults to None.

 None

Returns:

Name Type Description
Response

The response object from the requests library.
Source code in focoos/hub/api_client.py 
 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
def post(
    self,
    path: str,
    data: Optional[dict] = None,
    extra_headers: Optional[dict] = None,
    files=None,
):
    """
    Perform a POST request to the specified path on the host URL.

    Args:
        path (str): The URL path to request.
        data (Optional[dict], optional): The JSON data to send in the request body. Defaults to None.
        extra_headers (Optional[dict], optional): Additional headers to include in the request. Defaults to None.
        files (optional): Files to send in the request. Defaults to None.

    Returns:
        Response: The response object from the requests library.
    """
    self._check_api_key()
    url = f"{self.host_url}/{path}"
    headers = self.default_headers.copy()
    if extra_headers:
        headers.update(extra_headers)
    return requests.post(url, headers=headers, json=data, files=files)

upload_file(path, file_path, file_size) #

Upload a file to the specified path.

Parameters:

Name Type Description Default
path str

The API endpoint path to upload to

 required
file_path str

Path to the file to upload

 required
file_size int

Size of the file in bytes

 required

Returns:

Name Type Description
Response

The response from the upload request
Source code in focoos/hub/api_client.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
def upload_file(self, path: str, file_path: str, file_size: int):
    """
    Upload a file to the specified path.

    Args:
        path (str): The API endpoint path to upload to
        file_path (str): Path to the file to upload
        file_size (int): Size of the file in bytes

    Returns:
        Response: The response from the upload request
    """
    self._check_api_key()
    return self.post(path, data={"path": file_path, "file_size_bytes": file_size})

Focoos Module

This module provides a Python interface for interacting with Focoos APIs, allowing users to manage machine learning models and datasets in the Focoos ecosystem. The module supports operations such as retrieving model metadata, downloading models, and listing shared datasets.

Classes:

Name Description
FocoosHUB

Main class to interface with Focoos APIs.

Raises:

Type Description
ValueError

Raised for invalid API responses or missing parameters.

FocoosHUB #

Main class to interface with Focoos APIs.

This class provides methods to interact with Focoos-hosted models and datasets. It supports functionalities such as listing models, retrieving model metadata, downloading models, and creating new models.

Attributes:

Name Type Description
api_key str

The API key for authentication.
api_client ApiClient

HTTP client for making API requests.
user_info User

Information about the currently authenticated user.
host_url str

Base URL for the Focoos API.
Source code in focoos/hub/focoos_hub.py 
 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
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
class FocoosHUB:
    """
    Main class to interface with Focoos APIs.

    This class provides methods to interact with Focoos-hosted models and datasets.
    It supports functionalities such as listing models, retrieving model metadata,
    downloading models, and creating new models.

    Attributes:
        api_key (str): The API key for authentication.
        api_client (ApiClient): HTTP client for making API requests.
        user_info (User): Information about the currently authenticated user.
        host_url (str): Base URL for the Focoos API.
    """

    def __init__(
        self,
        api_key: Optional[str] = None,
        host_url: Optional[str] = None,
    ):
        """
        Initializes the FocoosHUB client.

        This client provides authenticated access to the Focoos API, enabling various operations
        through the configured HTTP client. It retrieves user information upon initialization and
        logs the environment details.

        Args:
            api_key (Optional[str]): API key for authentication. Defaults to the `focoos_api_key`
                specified in the FOCOOS_CONFIG.
            host_url (Optional[str]): Base URL for the Focoos API. Defaults to the `default_host_url`
                specified in the FOCOOS_CONFIG.

        Raises:
            ValueError: If the API key is not provided, or if the host URL is not specified in the
                arguments or the configuration.

        Attributes:
            api_key (str): The API key used for authentication.
            api_client (ApiClient): An HTTP client instance configured with the API key and host URL.
            user_info (User): Information about the authenticated user retrieved from the API.
            host_url (str): The base URL used for API requests.

        Logs:
            - Error if the API key or host URL is missing.
            - Info about the authenticated user and environment upon successful initialization.

        Example:
            ```python
            from focoos import FocoosHUB

            focoos = FocoosHUB()
            ```
        """
        self.api_key = api_key or FOCOOS_CONFIG.focoos_api_key
        if not self.api_key:
            logger.error("API key is required 🤖")
            raise ValueError("API key is required 🤖")

        self.host_url = host_url or FOCOOS_CONFIG.default_host_url

        self.api_client = ApiClient(api_key=self.api_key, host_url=self.host_url)
        self.user_info = self.get_user_info()
        logger.info(f"Currently logged as: {self.user_info.email} environment: {self.host_url}")

    def get_user_info(self) -> User:
        """
        Retrieves information about the authenticated user.

        Returns:
            User: User object containing account information and usage quotas.

        Raises:
            ValueError: If the API request fails.

        Example:
            ```python
            from focoos import FocoosHUB

            focoos = FocoosHUB()
            user_info = focoos.get_user_info()

            # Access user info fields
            print(f"Email: {user_info.email}")
            print(f"Created at: {user_info.created_at}")
            print(f"Updated at: {user_info.updated_at}")
            print(f"Company: {user_info.company}")
            print(f"API key: {user_info.api_key.key}")

            # Access quotas
            quotas = user_info.quotas
            print(f"Total inferences: {quotas.total_inferences}")
            print(f"Max inferences: {quotas.max_inferences}")
            print(f"Used storage (GB): {quotas.used_storage_gb}")
            print(f"Max storage (GB): {quotas.max_storage_gb}")
            print(f"Active training jobs: {quotas.active_training_jobs}")
            print(f"Max active training jobs: {quotas.max_active_training_jobs}")
            print(f"Used MLG4DNXLarge training jobs hours: {quotas.used_mlg4dnxlarge_training_jobs_hours}")
            print(f"Max MLG4DNXLarge training jobs hours: {quotas.max_mlg4dnxlarge_training_jobs_hours}")
            ```
        """
        res = self.api_client.get("user/")
        if res.status_code != 200:
            logger.error(f"Failed to get user info: {res.status_code} {res.text}")
            raise ValueError(f"Failed to get user info: {res.status_code} {res.text}")
        return User.from_json(res.json())

    def get_model_info(self, model_ref: str) -> RemoteModelInfo:
        """
        Retrieves metadata for a specific model.

        Args:
            model_ref (str): Reference identifier for the model.

        Returns:
            RemoteModelInfo: Metadata of the specified model.

        Raises:
            ValueError: If the API request fails.

        Example:
            ```python
            from focoos import FocoosHUB

            focoos = FocoosHUB()
            model_info = focoos.get_model_info(model_ref="user-or-fai-model-ref")
            ```
        """
        res = self.api_client.get(f"models/{model_ref}")
        if res.status_code != 200:
            logger.error(f"Failed to get model info: {res.status_code} {res.text}")
            raise ValueError(f"Failed to get model info: {res.status_code} {res.text}")
        return RemoteModelInfo.from_json(res.json())

    def list_remote_models(self) -> list[ModelPreview]:
        """
        Lists all models owned by the user.

        Returns:
            list[ModelPreview]: List of model previews.

        Raises:
            ValueError: If the API request fails.

        Example:
            ```python
            from focoos import FocoosHUB

            focoos = FocoosHUB()
            models = focoos.list_remote_models()
            ```
        """
        res = self.api_client.get("models/")
        if res.status_code != 200:
            logger.error(f"Failed to list models: {res.status_code} {res.text}")
            raise ValueError(f"Failed to list models: {res.status_code} {res.text}")
        return [ModelPreview.from_json(r) for r in res.json()]

    def get_remote_model(self, model_ref: str) -> RemoteModel:
        """
        Retrieves a remote model instance for cloud-based inference.

        Args:
            model_ref (str): Reference identifier for the model.

        Returns:
            RemoteModel: The remote model instance configured for cloud-based inference.

        Example:
            ```python
            from focoos import FocoosHUB

            focoos = FocoosHUB()
            model = focoos.get_remote_model(model_ref="fai-model-ref")
            results = model.infer("image.jpg", threshold=0.5)  # inference is remote!
            ```
        """
        return RemoteModel(model_ref, self.api_client)

    def download_model_pth(self, model_ref: str, skip_if_exists: bool = True) -> str:
        """
        Downloads a model from the Focoos API.

        Args:
            model_ref (str): Reference identifier for the model.
            skip_if_exists (bool): If True, skips the download if the model file already exists.
                Defaults to True.

        Returns:
            str: Path to the downloaded model file.

        Raises:
            ValueError: If the API request fails or the download fails.
        """
        model_dir = os.path.join(MODELS_DIR, model_ref)
        model_pth_path = os.path.join(model_dir, ArtifactName.WEIGHTS)
        if os.path.exists(model_pth_path) and skip_if_exists:
            logger.info("📥 Model already downloaded")
            return model_pth_path
        if not os.path.exists(model_dir):
            os.makedirs(model_dir)
        ## download model metadata
        res = self.api_client.get(f"models/{model_ref}/download?format=pth")
        if res.status_code != 200:
            logger.error(f"Failed to retrieve download url for model: {res.status_code} {res.text}")
            raise ValueError(f"Failed to retrieve download url for model: {res.status_code} {res.text}")

        download_data = res.json()

        download_uri = download_data.get("download_uri")
        if download_uri is None:
            logger.error(f"Failed to retrieve download url for model: {res.status_code} {res.text}")
            raise ValueError(f"Failed to retrieve download url for model: {res.status_code} {res.text}")
        ## download model from Focoos Cloud
        logger.debug(f"Model URI: {download_uri}")
        logger.info("📥 Downloading model from Focoos Cloud.. ")
        try:
            model_pth_path = self.api_client.download_ext_file(download_uri, model_dir, skip_if_exists=skip_if_exists)
        except Exception as e:
            logger.error(f"Failed to download model: {e}")
            raise ValueError(f"Failed to download model: {e}")
        if model_pth_path is None:
            logger.error(f"Failed to download model: {res.status_code} {res.text}")
            raise ValueError(f"Failed to download model: {res.status_code} {res.text}")

        return model_pth_path

    def list_remote_datasets(self, include_shared: bool = False) -> list[DatasetPreview]:
        """
        Lists all datasets available to the user.

        This method retrieves all datasets owned by the user and optionally includes
        shared datasets as well.

        Args:
            include_shared (bool): If True, includes datasets shared with the user.
                Defaults to False.

        Returns:
            list[DatasetPreview]: A list of DatasetPreview objects representing the available datasets.

        Raises:
            ValueError: If the API request to list datasets fails.

        Example:
            ```python
            from focoos import FocoosHUB

            focoos = FocoosHUB()

            # List only user's datasets
            datasets = focoos.list_remote_datasets()

            # List user's datasets and shared datasets
            all_datasets = focoos.list_remote_datasets(include_shared=True)

            for dataset in all_datasets:
                print(f"Dataset: {dataset.name}, Task: {dataset.task}")
            ```
        """
        res = self.api_client.get("datasets/")
        if res.status_code != 200:
            logger.error(f"Failed to list datasets: {res.status_code} {res.text}")
            raise ValueError(f"Failed to list datasets: {res.status_code} {res.text}")
        datasets = [DatasetPreview.from_json(r) for r in res.json()]
        if include_shared:
            res = self.api_client.get("datasets/shared")
            if res.status_code != 200:
                logger.error(f"Failed to list datasets: {res.status_code} {res.text}")
                raise ValueError(f"Failed to list datasets: {res.status_code} {res.text}")
            datasets.extend([DatasetPreview.from_json(sh_dataset) for sh_dataset in res.json()])
        return datasets

    def get_remote_dataset(self, ref: str) -> RemoteDataset:
        """
        Retrieves a remote dataset by its reference ID.

        Args:
            ref (str): The reference ID of the dataset to retrieve.

        Returns:
            RemoteDataset: A RemoteDataset instance for the specified reference.

        Example:
            ```python
            from focoos import FocoosHUB

            focoos = FocoosHUB()
            dataset = focoos.get_remote_dataset(ref="my-dataset-ref")
            ```
        """
        return RemoteDataset(ref, self.api_client)

    def new_model(self, model_info: ModelInfo) -> RemoteModel:
        """
        Creates a new model in the Focoos platform.

        Args:
            name (str): Name of the new model.
            focoos_model (str): Reference to the base Focoos model.
            description (str): Description of the new model.

        Returns:
            Optional[RemoteModel]: The created model instance, or None if creation fails.

        Raises:
            ValueError: If the API request fails.

        Example:
            ```python
            from focoos import Focoos

            focoos = Focoos()
            model = focoos.new_model(name="my-model", focoos_model="fai-model-ref", description="my-model-description")
            ```
        """

        res = self.api_client.post(
            "models/local-model",
            data={
                "name": model_info.name,
                "focoos_model": model_info.focoos_model,
                "description": model_info.description,
                "config": model_info.config if model_info.config else {},
                "task": model_info.task,
                "classes": model_info.classes,
                "im_size": model_info.im_size,
                "train_args": asdict(model_info.train_args) if model_info.train_args else None,
                "focoos_version": model_info.focoos_version,
            },
        )
        if res.status_code in [200, 201]:
            return RemoteModel(res.json()["ref"], self.api_client)
        if res.status_code == 409:
            logger.warning(f"Model already exists: {model_info.name}")
            raise ValueError(f"Failed to create new model: {res.status_code} {res.text}")
        else:
            logger.warning(f"Failed to create new model: {res.status_code} {res.text}")
            raise ValueError(f"Failed to create new model: {res.status_code} {res.text}")

__init__(api_key=None, host_url=None) #

Initializes the FocoosHUB client.

This client provides authenticated access to the Focoos API, enabling various operations through the configured HTTP client. It retrieves user information upon initialization and logs the environment details.

Parameters:

Name Type Description Default
api_key Optional[str]

API key for authentication. Defaults to the focoos_api_key specified in the FOCOOS_CONFIG.

 None
host_url Optional[str]

Base URL for the Focoos API. Defaults to the default_host_url specified in the FOCOOS_CONFIG.

 None

Raises:

Type Description
ValueError

If the API key is not provided, or if the host URL is not specified in the arguments or the configuration.

Attributes:

Name Type Description
api_key str

The API key used for authentication.
api_client ApiClient

An HTTP client instance configured with the API key and host URL.
user_info User

Information about the authenticated user retrieved from the API.
host_url str

The base URL used for API requests.
Logs
  • Error if the API key or host URL is missing.
  • Info about the authenticated user and environment upon successful initialization.
Example 
1
2
3
from focoos import FocoosHUB

focoos = FocoosHUB()
Source code in focoos/hub/focoos_hub.py 
 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
def __init__(
    self,
    api_key: Optional[str] = None,
    host_url: Optional[str] = None,
):
    """
    Initializes the FocoosHUB client.

    This client provides authenticated access to the Focoos API, enabling various operations
    through the configured HTTP client. It retrieves user information upon initialization and
    logs the environment details.

    Args:
        api_key (Optional[str]): API key for authentication. Defaults to the `focoos_api_key`
            specified in the FOCOOS_CONFIG.
        host_url (Optional[str]): Base URL for the Focoos API. Defaults to the `default_host_url`
            specified in the FOCOOS_CONFIG.

    Raises:
        ValueError: If the API key is not provided, or if the host URL is not specified in the
            arguments or the configuration.

    Attributes:
        api_key (str): The API key used for authentication.
        api_client (ApiClient): An HTTP client instance configured with the API key and host URL.
        user_info (User): Information about the authenticated user retrieved from the API.
        host_url (str): The base URL used for API requests.

    Logs:
        - Error if the API key or host URL is missing.
        - Info about the authenticated user and environment upon successful initialization.

    Example:
        ```python
        from focoos import FocoosHUB

        focoos = FocoosHUB()
        ```
    """
    self.api_key = api_key or FOCOOS_CONFIG.focoos_api_key
    if not self.api_key:
        logger.error("API key is required 🤖")
        raise ValueError("API key is required 🤖")

    self.host_url = host_url or FOCOOS_CONFIG.default_host_url

    self.api_client = ApiClient(api_key=self.api_key, host_url=self.host_url)
    self.user_info = self.get_user_info()
    logger.info(f"Currently logged as: {self.user_info.email} environment: {self.host_url}")

download_model_pth(model_ref, skip_if_exists=True) #

Downloads a model from the Focoos API.

Parameters:

Name Type Description Default
model_ref str

Reference identifier for the model.

 required
skip_if_exists bool

If True, skips the download if the model file already exists. Defaults to True.

 True

Returns:

Name Type Description
str str

Path to the downloaded model file.

Raises:

Type Description
ValueError

If the API request fails or the download fails.
Source code in focoos/hub/focoos_hub.py 
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
def download_model_pth(self, model_ref: str, skip_if_exists: bool = True) -> str:
    """
    Downloads a model from the Focoos API.

    Args:
        model_ref (str): Reference identifier for the model.
        skip_if_exists (bool): If True, skips the download if the model file already exists.
            Defaults to True.

    Returns:
        str: Path to the downloaded model file.

    Raises:
        ValueError: If the API request fails or the download fails.
    """
    model_dir = os.path.join(MODELS_DIR, model_ref)
    model_pth_path = os.path.join(model_dir, ArtifactName.WEIGHTS)
    if os.path.exists(model_pth_path) and skip_if_exists:
        logger.info("📥 Model already downloaded")
        return model_pth_path
    if not os.path.exists(model_dir):
        os.makedirs(model_dir)
    ## download model metadata
    res = self.api_client.get(f"models/{model_ref}/download?format=pth")
    if res.status_code != 200:
        logger.error(f"Failed to retrieve download url for model: {res.status_code} {res.text}")
        raise ValueError(f"Failed to retrieve download url for model: {res.status_code} {res.text}")

    download_data = res.json()

    download_uri = download_data.get("download_uri")
    if download_uri is None:
        logger.error(f"Failed to retrieve download url for model: {res.status_code} {res.text}")
        raise ValueError(f"Failed to retrieve download url for model: {res.status_code} {res.text}")
    ## download model from Focoos Cloud
    logger.debug(f"Model URI: {download_uri}")
    logger.info("📥 Downloading model from Focoos Cloud.. ")
    try:
        model_pth_path = self.api_client.download_ext_file(download_uri, model_dir, skip_if_exists=skip_if_exists)
    except Exception as e:
        logger.error(f"Failed to download model: {e}")
        raise ValueError(f"Failed to download model: {e}")
    if model_pth_path is None:
        logger.error(f"Failed to download model: {res.status_code} {res.text}")
        raise ValueError(f"Failed to download model: {res.status_code} {res.text}")

    return model_pth_path

get_model_info(model_ref) #

Retrieves metadata for a specific model.

Parameters:

Name Type Description Default
model_ref str

Reference identifier for the model.

 required

Returns:

Name Type Description
RemoteModelInfo RemoteModelInfo

Metadata of the specified model.

Raises:

Type Description
ValueError

If the API request fails.
Example 
1
2
3
4
from focoos import FocoosHUB

focoos = FocoosHUB()
model_info = focoos.get_model_info(model_ref="user-or-fai-model-ref")
Source code in focoos/hub/focoos_hub.py 
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
def get_model_info(self, model_ref: str) -> RemoteModelInfo:
    """
    Retrieves metadata for a specific model.

    Args:
        model_ref (str): Reference identifier for the model.

    Returns:
        RemoteModelInfo: Metadata of the specified model.

    Raises:
        ValueError: If the API request fails.

    Example:
        ```python
        from focoos import FocoosHUB

        focoos = FocoosHUB()
        model_info = focoos.get_model_info(model_ref="user-or-fai-model-ref")
        ```
    """
    res = self.api_client.get(f"models/{model_ref}")
    if res.status_code != 200:
        logger.error(f"Failed to get model info: {res.status_code} {res.text}")
        raise ValueError(f"Failed to get model info: {res.status_code} {res.text}")
    return RemoteModelInfo.from_json(res.json())

get_remote_dataset(ref) #

Retrieves a remote dataset by its reference ID.

Parameters:

Name Type Description Default
ref str

The reference ID of the dataset to retrieve.

 required

Returns:

Name Type Description
RemoteDataset RemoteDataset

A RemoteDataset instance for the specified reference.
Example 
1
2
3
4
from focoos import FocoosHUB

focoos = FocoosHUB()
dataset = focoos.get_remote_dataset(ref="my-dataset-ref")
Source code in focoos/hub/focoos_hub.py 
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
def get_remote_dataset(self, ref: str) -> RemoteDataset:
    """
    Retrieves a remote dataset by its reference ID.

    Args:
        ref (str): The reference ID of the dataset to retrieve.

    Returns:
        RemoteDataset: A RemoteDataset instance for the specified reference.

    Example:
        ```python
        from focoos import FocoosHUB

        focoos = FocoosHUB()
        dataset = focoos.get_remote_dataset(ref="my-dataset-ref")
        ```
    """
    return RemoteDataset(ref, self.api_client)

get_remote_model(model_ref) #

Retrieves a remote model instance for cloud-based inference.

Parameters:

Name Type Description Default
model_ref str

Reference identifier for the model.

 required

Returns:

Name Type Description
RemoteModel RemoteModel

The remote model instance configured for cloud-based inference.
Example 
1
2
3
4
5
from focoos import FocoosHUB

focoos = FocoosHUB()
model = focoos.get_remote_model(model_ref="fai-model-ref")
results = model.infer("image.jpg", threshold=0.5)  # inference is remote!
Source code in focoos/hub/focoos_hub.py 
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
def get_remote_model(self, model_ref: str) -> RemoteModel:
    """
    Retrieves a remote model instance for cloud-based inference.

    Args:
        model_ref (str): Reference identifier for the model.

    Returns:
        RemoteModel: The remote model instance configured for cloud-based inference.

    Example:
        ```python
        from focoos import FocoosHUB

        focoos = FocoosHUB()
        model = focoos.get_remote_model(model_ref="fai-model-ref")
        results = model.infer("image.jpg", threshold=0.5)  # inference is remote!
        ```
    """
    return RemoteModel(model_ref, self.api_client)

get_user_info() #

Retrieves information about the authenticated user.

Returns:

Name Type Description
User User

User object containing account information and usage quotas.

Raises:

Type Description
ValueError

If the API request fails.
Example 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
from focoos import FocoosHUB

focoos = FocoosHUB()
user_info = focoos.get_user_info()

# Access user info fields
print(f"Email: {user_info.email}")
print(f"Created at: {user_info.created_at}")
print(f"Updated at: {user_info.updated_at}")
print(f"Company: {user_info.company}")
print(f"API key: {user_info.api_key.key}")

# Access quotas
quotas = user_info.quotas
print(f"Total inferences: {quotas.total_inferences}")
print(f"Max inferences: {quotas.max_inferences}")
print(f"Used storage (GB): {quotas.used_storage_gb}")
print(f"Max storage (GB): {quotas.max_storage_gb}")
print(f"Active training jobs: {quotas.active_training_jobs}")
print(f"Max active training jobs: {quotas.max_active_training_jobs}")
print(f"Used MLG4DNXLarge training jobs hours: {quotas.used_mlg4dnxlarge_training_jobs_hours}")
print(f"Max MLG4DNXLarge training jobs hours: {quotas.max_mlg4dnxlarge_training_jobs_hours}")
Source code in focoos/hub/focoos_hub.py 
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
def get_user_info(self) -> User:
    """
    Retrieves information about the authenticated user.

    Returns:
        User: User object containing account information and usage quotas.

    Raises:
        ValueError: If the API request fails.

    Example:
        ```python
        from focoos import FocoosHUB

        focoos = FocoosHUB()
        user_info = focoos.get_user_info()

        # Access user info fields
        print(f"Email: {user_info.email}")
        print(f"Created at: {user_info.created_at}")
        print(f"Updated at: {user_info.updated_at}")
        print(f"Company: {user_info.company}")
        print(f"API key: {user_info.api_key.key}")

        # Access quotas
        quotas = user_info.quotas
        print(f"Total inferences: {quotas.total_inferences}")
        print(f"Max inferences: {quotas.max_inferences}")
        print(f"Used storage (GB): {quotas.used_storage_gb}")
        print(f"Max storage (GB): {quotas.max_storage_gb}")
        print(f"Active training jobs: {quotas.active_training_jobs}")
        print(f"Max active training jobs: {quotas.max_active_training_jobs}")
        print(f"Used MLG4DNXLarge training jobs hours: {quotas.used_mlg4dnxlarge_training_jobs_hours}")
        print(f"Max MLG4DNXLarge training jobs hours: {quotas.max_mlg4dnxlarge_training_jobs_hours}")
        ```
    """
    res = self.api_client.get("user/")
    if res.status_code != 200:
        logger.error(f"Failed to get user info: {res.status_code} {res.text}")
        raise ValueError(f"Failed to get user info: {res.status_code} {res.text}")
    return User.from_json(res.json())

list_remote_datasets(include_shared=False) #

Lists all datasets available to the user.

This method retrieves all datasets owned by the user and optionally includes shared datasets as well.

Parameters:

Name Type Description Default
include_shared bool

If True, includes datasets shared with the user. Defaults to False.

 False

Returns:

Type Description
list[DatasetPreview]

list[DatasetPreview]: A list of DatasetPreview objects representing the available datasets.

Raises:

Type Description
ValueError

If the API request to list datasets fails.
Example 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
from focoos import FocoosHUB

focoos = FocoosHUB()

# List only user's datasets
datasets = focoos.list_remote_datasets()

# List user's datasets and shared datasets
all_datasets = focoos.list_remote_datasets(include_shared=True)

for dataset in all_datasets:
    print(f"Dataset: {dataset.name}, Task: {dataset.task}")
Source code in focoos/hub/focoos_hub.py 
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
def list_remote_datasets(self, include_shared: bool = False) -> list[DatasetPreview]:
    """
    Lists all datasets available to the user.

    This method retrieves all datasets owned by the user and optionally includes
    shared datasets as well.

    Args:
        include_shared (bool): If True, includes datasets shared with the user.
            Defaults to False.

    Returns:
        list[DatasetPreview]: A list of DatasetPreview objects representing the available datasets.

    Raises:
        ValueError: If the API request to list datasets fails.

    Example:
        ```python
        from focoos import FocoosHUB

        focoos = FocoosHUB()

        # List only user's datasets
        datasets = focoos.list_remote_datasets()

        # List user's datasets and shared datasets
        all_datasets = focoos.list_remote_datasets(include_shared=True)

        for dataset in all_datasets:
            print(f"Dataset: {dataset.name}, Task: {dataset.task}")
        ```
    """
    res = self.api_client.get("datasets/")
    if res.status_code != 200:
        logger.error(f"Failed to list datasets: {res.status_code} {res.text}")
        raise ValueError(f"Failed to list datasets: {res.status_code} {res.text}")
    datasets = [DatasetPreview.from_json(r) for r in res.json()]
    if include_shared:
        res = self.api_client.get("datasets/shared")
        if res.status_code != 200:
            logger.error(f"Failed to list datasets: {res.status_code} {res.text}")
            raise ValueError(f"Failed to list datasets: {res.status_code} {res.text}")
        datasets.extend([DatasetPreview.from_json(sh_dataset) for sh_dataset in res.json()])
    return datasets

list_remote_models() #

Lists all models owned by the user.

Returns:

Type Description
list[ModelPreview]

list[ModelPreview]: List of model previews.

Raises:

Type Description
ValueError

If the API request fails.
Example 
1
2
3
4
from focoos import FocoosHUB

focoos = FocoosHUB()
models = focoos.list_remote_models()
Source code in focoos/hub/focoos_hub.py 
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
def list_remote_models(self) -> list[ModelPreview]:
    """
    Lists all models owned by the user.

    Returns:
        list[ModelPreview]: List of model previews.

    Raises:
        ValueError: If the API request fails.

    Example:
        ```python
        from focoos import FocoosHUB

        focoos = FocoosHUB()
        models = focoos.list_remote_models()
        ```
    """
    res = self.api_client.get("models/")
    if res.status_code != 200:
        logger.error(f"Failed to list models: {res.status_code} {res.text}")
        raise ValueError(f"Failed to list models: {res.status_code} {res.text}")
    return [ModelPreview.from_json(r) for r in res.json()]

new_model(model_info) #

Creates a new model in the Focoos platform.

Parameters:

Name Type Description Default
name str

Name of the new model.

 required
focoos_model str

Reference to the base Focoos model.

 required
description str

Description of the new model.

 required

Returns:

Type Description
RemoteModel

Optional[RemoteModel]: The created model instance, or None if creation fails.

Raises:

Type Description
ValueError

If the API request fails.
Example 
1
2
3
4
from focoos import Focoos

focoos = Focoos()
model = focoos.new_model(name="my-model", focoos_model="fai-model-ref", description="my-model-description")
Source code in focoos/hub/focoos_hub.py 
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
def new_model(self, model_info: ModelInfo) -> RemoteModel:
    """
    Creates a new model in the Focoos platform.

    Args:
        name (str): Name of the new model.
        focoos_model (str): Reference to the base Focoos model.
        description (str): Description of the new model.

    Returns:
        Optional[RemoteModel]: The created model instance, or None if creation fails.

    Raises:
        ValueError: If the API request fails.

    Example:
        ```python
        from focoos import Focoos

        focoos = Focoos()
        model = focoos.new_model(name="my-model", focoos_model="fai-model-ref", description="my-model-description")
        ```
    """

    res = self.api_client.post(
        "models/local-model",
        data={
            "name": model_info.name,
            "focoos_model": model_info.focoos_model,
            "description": model_info.description,
            "config": model_info.config if model_info.config else {},
            "task": model_info.task,
            "classes": model_info.classes,
            "im_size": model_info.im_size,
            "train_args": asdict(model_info.train_args) if model_info.train_args else None,
            "focoos_version": model_info.focoos_version,
        },
    )
    if res.status_code in [200, 201]:
        return RemoteModel(res.json()["ref"], self.api_client)
    if res.status_code == 409:
        logger.warning(f"Model already exists: {model_info.name}")
        raise ValueError(f"Failed to create new model: {res.status_code} {res.text}")
    else:
        logger.warning(f"Failed to create new model: {res.status_code} {res.text}")
        raise ValueError(f"Failed to create new model: {res.status_code} {res.text}")

RemoteDataset #

A class to manage remote datasets through the Focoos API.

This class provides functionality to interact with datasets stored remotely, including uploading, downloading, and managing dataset data.

Parameters:

Name Type Description Default
ref str

The reference identifier for the dataset.

 required
api_client ApiClient

The API client instance for making requests.

 required

Attributes:

Name Type Description
ref str

The dataset reference identifier.
api_client ApiClient

The API client instance.
metadata DatasetPreview

The dataset metadata.
Source code in focoos/hub/remote_dataset.py 
 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
class RemoteDataset:
    """
    A class to manage remote datasets through the Focoos API.

    This class provides functionality to interact with datasets stored remotely,
    including uploading, downloading, and managing dataset data.

    Args:
        ref (str): The reference identifier for the dataset.
        api_client (ApiClient): The API client instance for making requests.

    Attributes:
        ref (str): The dataset reference identifier.
        api_client (ApiClient): The API client instance.
        metadata (DatasetPreview): The dataset metadata.
    """

    def __init__(self, ref: str, api_client: ApiClient):
        self.ref = ref
        self.api_client = api_client
        self.metadata: DatasetPreview = self.get_info()

    def get_info(self) -> DatasetPreview:
        """
        Retrieves the dataset information from the API.

        Returns:
            DatasetPreview: The dataset preview information.
        """
        res = self.api_client.get(f"datasets/{self.ref}")
        if res.status_code != 200:
            raise ValueError(f"Failed to get dataset info: {res.status_code} {res.text}")
        return DatasetPreview.from_json(res.json())

    def upload_data(self, path: str) -> Optional[DatasetSpec]:
        """
        Uploads dataset data from a local zip file to the remote storage.

        Args:
            path (str): Local path to the zip file containing dataset data.

        Returns:
            Optional[DatasetSpec]: The dataset specification after successful upload.

        Raises:
            FileNotFoundError: If the specified file does not exist.
            ValueError: If the file is not a zip file or upload fails.
        """
        if not path.endswith(".zip"):
            raise ValueError("Dataset must be .zip compressed")
        if not os.path.exists(path):
            raise FileNotFoundError(f"File not found: {path}")

        file_name = os.path.basename(path)
        file_size = os.path.getsize(path)
        file_size_mb = file_size / (1024 * 1024)
        logger.info(f"🔗 Requesting upload url for {file_name} of size {file_size_mb:.2f} MB")
        presigned_url = self.api_client.post(
            f"datasets/{self.ref}/generate-upload-url",
            data={"file_size_bytes": file_size, "file_name": file_name},
        )
        if presigned_url.status_code != 200:
            raise ValueError(f"Failed to generate upload url: {presigned_url.status_code} {presigned_url.text}")
        presigned_url = presigned_url.json()
        fields = {k: v for k, v in presigned_url["fields"].items()}
        logger.info(f"📤 Uploading file {file_name}..")

        # Use context manager to properly handle file closure
        with open(path, "rb") as file_obj:
            fields["file"] = (file_name, file_obj, "application/zip")

            res = self.api_client.external_post(
                presigned_url["url"],
                files=fields,
                data=presigned_url["fields"],
                stream=True,
            )

        logger.info("✅ Upload file done.")
        if res.status_code not in [200, 201, 204]:
            raise ValueError(f"Failed to upload dataset: {res.status_code} {res.text}")

        logger.info("🔗 Validating dataset..")
        complete_upload = self.api_client.post(
            f"datasets/{self.ref}/complete-upload",
        )
        if complete_upload.status_code not in [200, 201, 204]:
            raise ValueError(f"Failed to validate dataset: {complete_upload.status_code} {complete_upload.text}")
        self.metadata = self.get_info()
        logger.info(f"✅ Dataset validated! => {self.metadata.spec}")
        return self.metadata.spec

    @property
    def name(self):
        return self.metadata.name

    @property
    def task(self):
        return self.metadata.task

    @property
    def layout(self):
        return self.metadata.layout

    def download_data(self, path: str = DATASETS_DIR):
        """
        Downloads the dataset data to a local path.

        Args:
            path (str): Local path where the dataset should be downloaded.

        Returns:
            str: The path where the file was downloaded.

        Raises:
            ValueError: If the download fails.
        """
        res = self.api_client.get(f"datasets/{self.ref}/download")
        if res.status_code != 200:
            raise ValueError(f"Failed to download dataset data: {res.status_code} {res.text}")
        url = res.json()["download_uri"]

        path = self.api_client.download_ext_file(url, path, skip_if_exists=True)
        logger.info(f"✅ Dataset data downloaded to {path}")
        return path

    def __str__(self):
        return f"RemoteDataset(ref={self.ref}, name={self.name}, task={self.task}, layout={self.layout})"

download_data(path=DATASETS_DIR) #

Downloads the dataset data to a local path.

Parameters:

Name Type Description Default
path str

Local path where the dataset should be downloaded.

 DATASETS_DIR

Returns:

Name Type Description
str

The path where the file was downloaded.

Raises:

Type Description
ValueError

If the download fails.
Source code in focoos/hub/remote_dataset.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def download_data(self, path: str = DATASETS_DIR):
    """
    Downloads the dataset data to a local path.

    Args:
        path (str): Local path where the dataset should be downloaded.

    Returns:
        str: The path where the file was downloaded.

    Raises:
        ValueError: If the download fails.
    """
    res = self.api_client.get(f"datasets/{self.ref}/download")
    if res.status_code != 200:
        raise ValueError(f"Failed to download dataset data: {res.status_code} {res.text}")
    url = res.json()["download_uri"]

    path = self.api_client.download_ext_file(url, path, skip_if_exists=True)
    logger.info(f"✅ Dataset data downloaded to {path}")
    return path

get_info() #

Retrieves the dataset information from the API.

Returns:

Name Type Description
DatasetPreview DatasetPreview

The dataset preview information.
Source code in focoos/hub/remote_dataset.py 
33
34
35
36
37
38
39
40
41
42
43
def get_info(self) -> DatasetPreview:
    """
    Retrieves the dataset information from the API.

    Returns:
        DatasetPreview: The dataset preview information.
    """
    res = self.api_client.get(f"datasets/{self.ref}")
    if res.status_code != 200:
        raise ValueError(f"Failed to get dataset info: {res.status_code} {res.text}")
    return DatasetPreview.from_json(res.json())

upload_data(path) #

Uploads dataset data from a local zip file to the remote storage.

Parameters:

Name Type Description Default
path str

Local path to the zip file containing dataset data.

 required

Returns:

Type Description
Optional[DatasetSpec]

Optional[DatasetSpec]: The dataset specification after successful upload.

Raises:

Type Description
FileNotFoundError

If the specified file does not exist.
ValueError

If the file is not a zip file or upload fails.
Source code in focoos/hub/remote_dataset.py 
 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
def upload_data(self, path: str) -> Optional[DatasetSpec]:
    """
    Uploads dataset data from a local zip file to the remote storage.

    Args:
        path (str): Local path to the zip file containing dataset data.

    Returns:
        Optional[DatasetSpec]: The dataset specification after successful upload.

    Raises:
        FileNotFoundError: If the specified file does not exist.
        ValueError: If the file is not a zip file or upload fails.
    """
    if not path.endswith(".zip"):
        raise ValueError("Dataset must be .zip compressed")
    if not os.path.exists(path):
        raise FileNotFoundError(f"File not found: {path}")

    file_name = os.path.basename(path)
    file_size = os.path.getsize(path)
    file_size_mb = file_size / (1024 * 1024)
    logger.info(f"🔗 Requesting upload url for {file_name} of size {file_size_mb:.2f} MB")
    presigned_url = self.api_client.post(
        f"datasets/{self.ref}/generate-upload-url",
        data={"file_size_bytes": file_size, "file_name": file_name},
    )
    if presigned_url.status_code != 200:
        raise ValueError(f"Failed to generate upload url: {presigned_url.status_code} {presigned_url.text}")
    presigned_url = presigned_url.json()
    fields = {k: v for k, v in presigned_url["fields"].items()}
    logger.info(f"📤 Uploading file {file_name}..")

    # Use context manager to properly handle file closure
    with open(path, "rb") as file_obj:
        fields["file"] = (file_name, file_obj, "application/zip")

        res = self.api_client.external_post(
            presigned_url["url"],
            files=fields,
            data=presigned_url["fields"],
            stream=True,
        )

    logger.info("✅ Upload file done.")
    if res.status_code not in [200, 201, 204]:
        raise ValueError(f"Failed to upload dataset: {res.status_code} {res.text}")

    logger.info("🔗 Validating dataset..")
    complete_upload = self.api_client.post(
        f"datasets/{self.ref}/complete-upload",
    )
    if complete_upload.status_code not in [200, 201, 204]:
        raise ValueError(f"Failed to validate dataset: {complete_upload.status_code} {complete_upload.text}")
    self.metadata = self.get_info()
    logger.info(f"✅ Dataset validated! => {self.metadata.spec}")
    return self.metadata.spec

RemoteModel Module

This module provides a class to manage remote models in the Focoos ecosystem. It supports various functionalities including model training, deployment, inference, and monitoring.

Classes:

Name Description
RemoteModel

A class for interacting with remote models, managing their lifecycle, and performing inference.

Functions:

Name Description
__init__

Initializes the RemoteModel instance.
get_info

Retrieves model metadata.
train

Initiates model training.
train_info

Retrieves training status.
train_logs

Retrieves training logs.
metrics

Retrieves model metrics.

RemoteModel #

Represents a remote model in the Focoos platform.

Attributes:

Name Type Description
model_ref str

Reference ID for the model.
api_client ApiClient

Client for making HTTP requests.
model_info RemoteModelInfo

Model information of the model.
Source code in focoos/hub/remote_model.py 
 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
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
class RemoteModel:
    """
    Represents a remote model in the Focoos platform.

    Attributes:
        model_ref (str): Reference ID for the model.
        api_client (ApiClient): Client for making HTTP requests.
        model_info (RemoteModelInfo): Model information of the model.
    """

    def __init__(
        self,
        model_ref: str,
        api_client: ApiClient,
    ):
        """
        Initialize the RemoteModel instance.

        Args:
            model_ref (str): Reference ID for the model.
            api_client (ApiClient): HTTP client instance for communication.

        Raises:
            ValueError: If model metadata retrieval fails.
        """
        self.model_ref = model_ref
        self.api_client = api_client
        self.model_info: RemoteModelInfo = self.get_info()

        logger.info(
            f"[RemoteModel]: ref: {self.model_ref} name: {self.model_info.name} description: {self.model_info.description} status: {self.model_info.status}"
        )

    @property
    def ref(self) -> str:
        return self.model_ref

    def get_info(self) -> RemoteModelInfo:
        """
        Retrieve model metadata.

        Returns:
            ModelMetadata: Metadata of the model.

        Raises:
            ValueError: If the request fails.

        Example:
            ```python
            from focoos import Focoos, RemoteModel

            focoos = Focoos()
            model = focoos.get_remote_model(model_ref="<model_ref>")
            model_info = model.get_info()
            ```
        """
        res = self.api_client.get(f"models/{self.model_ref}")
        if res.status_code != 200:
            logger.error(f"Failed to get model info: {res.status_code} {res.text}")
            raise ValueError(f"Failed to get model info: {res.status_code} {res.text}")
        self.metadata = RemoteModelInfo(**res.json())
        return self.metadata

    def sync_local_training_job(
        self, local_training_info: HubSyncLocalTraining, dir: str, upload_artifacts: Optional[List[ArtifactName]] = None
    ) -> None:
        if not os.path.exists(os.path.join(dir, ArtifactName.INFO)):
            logger.warning(f"Model info not found in {dir}")
            raise ValueError(f"Model info not found in {dir}")
        metrics = parse_metrics(os.path.join(dir, ArtifactName.METRICS))
        local_training_info.metrics = metrics
        logger.debug(
            f"[Syncing Training] iter: {metrics.iterations} {self.metadata.name} status: {local_training_info.status} ref: {self.model_ref}"
        )

        ## Update metrics
        res = self.api_client.patch(
            f"models/{self.model_ref}/sync-local-training",
            data=asdict(local_training_info),
        )
        if res.status_code != 200:
            logger.error(f"Failed to sync local training: {res.status_code} {res.text}")
            raise ValueError(f"Failed to sync local training: {res.status_code} {res.text}")
        if upload_artifacts:
            for artifact in [ArtifactName.METRICS, ArtifactName.WEIGHTS]:
                file_path = os.path.join(dir, artifact.value)
                if os.path.isfile(file_path):
                    try:
                        self._upload_model_artifact(file_path)
                    except Exception:
                        logger.error(f"Failed to upload artifact: {artifact.value}")
                        pass

    def _upload_model_artifact(self, path: str) -> None:
        """
        Uploads an model artifact to the Focoos platform.
        """
        if not os.path.exists(path):
            raise ValueError(f"File not found: {path}")
        file_ext = os.path.splitext(path)[1]
        if file_ext not in [".pt", ".onnx", ".pth", ".json", ".txt"]:
            raise ValueError(f"Unsupported file extension: {file_ext}")
        file_name = os.path.basename(path)
        file_size = os.path.getsize(path)
        file_size_mb = file_size / (1024 * 1024)
        logger.debug(f"🔗 Requesting upload url for {file_name} of size {file_size_mb:.2f} MB")
        presigned_url = self.api_client.post(
            f"models/{self.model_ref}/generate-upload-url",
            data={"file_size_bytes": file_size, "file_name": file_name},
        )
        if presigned_url.status_code != 200:
            raise ValueError(f"Failed to generate upload url: {presigned_url.status_code} {presigned_url.text}")
        presigned_url = presigned_url.json()
        fields = {k: v for k, v in presigned_url["fields"].items()}
        logger.info(f"📤 Uploading file {file_name} to HUB, size: {file_size_mb:.2f} MB")
        fields["file"] = (file_name, open(path, "rb"))
        res = self.api_client.external_post(
            presigned_url["url"],
            files=fields,
            data=presigned_url["fields"],
        )
        if res.status_code not in [200, 201, 204]:
            raise ValueError(f"Failed to upload model artifact: {res.status_code} {res.text}")
        logger.info(f"✅ Model artifact {file_name} uploaded to HUB.")

    def train_info(self) -> Optional[TrainingInfo]:
        """
        Retrieve the current status of the model training.

        Sends a request to check the training status of the model referenced by `self.model_ref`.

        Returns:
            dict: A dictionary containing the training status information.

        Raises:
            ValueError: If the request to get training status fails.
        """
        res = self.api_client.get(f"models/{self.model_ref}/train/status")
        if res.status_code != 200:
            logger.error(f"Failed to get train info: {res.status_code} {res.text}")
            raise ValueError(f"Failed to get train info: {res.status_code} {res.text}")
        dct = {k: v for k, v in res.json().items() if k in TrainingInfo.__dataclass_fields__}
        return TrainingInfo(**dct)

    def train_logs(self) -> list[str]:
        """
        Retrieve the training logs for the model.

        This method sends a request to fetch the logs of the model's training process. If the request
        is successful (status code 200), it returns the logs as a list of strings. If the request fails,
        it logs a warning and returns an empty list.

        Returns:
            list[str]: A list of training logs as strings.

        Raises:
            None: Returns an empty list if the request fails.
        """
        res = self.api_client.get(f"models/{self.model_ref}/train/logs")
        if res.status_code != 200:
            logger.warning(f"Failed to get train logs: {res.status_code} {res.text}")
            return []
        return res.json()

    def metrics(self) -> Metrics:  # noqa: F821
        """
        Retrieve the metrics of the model.

        This method sends a request to fetch the metrics of the model identified by `model_ref`.
        If the request is successful (status code 200), it returns the metrics as a `Metrics` object.
        If the request fails, it logs a warning and returns an empty `Metrics` object.

        Returns:
            Metrics: An object containing the metrics of the model.

        Raises:
            None: Returns an empty `Metrics` object if the request fails.
        """
        res = self.api_client.get(f"models/{self.model_ref}/metrics")
        if res.status_code != 200:
            logger.warning(f"Failed to get metrics: {res.status_code} {res.text}")
            return Metrics()  # noqa: F821
        return Metrics(**{k: v for k, v in res.json().items() if k in Metrics.__dataclass_fields__})

    def __call__(
        self, image: Union[str, Path, np.ndarray, bytes, Image.Image], threshold: float = 0.5
    ) -> FocoosDetections:
        return self.infer(image, threshold)

    def infer(
        self,
        image: Union[str, Path, np.ndarray, bytes, Image.Image],
        threshold: float = 0.5,
    ) -> FocoosDetections:
        """
        Perform inference on the provided image using the remote model.

        This method sends an image to the remote model for inference and retrieves the detection results.

        Args:
            image (Union[str, Path, np.ndarray, bytes]): The image to infer on, which can be a file path,
                a string representing the path, a NumPy array, or raw bytes.
            threshold (float, optional): The confidence threshold for detections. Defaults to 0.5.
                Detections with confidence scores below this threshold will be discarded.

        Returns:
            FocoosDetections: The detection results including class IDs, confidence scores, bounding boxes,
                and segmentation masks (if applicable).

        Raises:
            FileNotFoundError: If the provided image file path is invalid.
            ValueError: If the inference request fails.

        Example:
            ```python
            from focoos import Focoos

            focoos = Focoos()
            model = focoos.get_remote_model("my-model")
            results = model.infer("image.jpg", threshold=0.5)

            # Print detection results
            for det in results.detections:
                print(f"Found {det.label} with confidence {det.conf:.2f}")
                print(f"Bounding box: {det.bbox}")
                if det.mask:
                    print("Instance segmentation mask included")
            ```
        """
        if isinstance(image, Image.Image):
            image = np.array(image)

        image_bytes = None
        if isinstance(image, str) or isinstance(image, Path):
            if not os.path.exists(image):
                logger.error(f"Image file not found: {image}")
                raise FileNotFoundError(f"Image file not found: {image}")
            image_bytes = open(image, "rb").read()
        elif isinstance(image, np.ndarray):
            _, buffer = cv2.imencode(".jpg", image)
            image_bytes = buffer.tobytes()
        else:
            image_bytes = image
        files = {"file": ("image", image_bytes, "image/jpeg")}
        t0 = time.time()
        res = self.api_client.post(
            f"models/{self.model_ref}/inference?confidence_threshold={threshold}",
            files=files,
        )
        t1 = time.time()
        if res.status_code == 200:
            detections = FocoosDetections(
                detections=[FocoosDet.from_json(d) for d in res.json().get("detections", [])],
                latency=res.json().get("latency", None),
            )
            logger.debug(
                f"Found {len(detections.detections)} detections. Inference Request time: {(t1 - t0) * 1000:.0f}ms"
            )

            return detections
        else:
            logger.error(f"Failed to infer: {res.status_code} {res.text}")
            raise ValueError(f"Failed to infer: {res.status_code} {res.text}")

    def notebook_monitor_train(self, interval: int = 30, plot_metrics: bool = False, max_runtime: int = 36000) -> None:
        """
        Monitor the training process in a Jupyter notebook and display metrics.

        Periodically checks the training status and displays metrics in a notebook cell.
        Clears previous output to maintain a clean view.

        Args:
            interval (int): Time between status checks in seconds. Must be 30-240. Default: 30
            plot_metrics (bool): Whether to plot metrics graphs. Default: False
            max_runtime (int): Maximum monitoring time in seconds. Default: 36000 (10 hours)

        Returns:
            None
        """
        from IPython.display import clear_output

        if not 30 <= interval <= 240:
            raise ValueError("Interval must be between 30 and 240 seconds")

        last_update = self.get_info().updated_at
        start_time = time.time()
        status_history = []

        while True:
            # Get current status
            model_info = self.get_info()
            status = model_info.status

            # Clear and display status
            clear_output(wait=True)
            status_msg = f"[Live Monitor {self.metadata.name}] {status.value}"
            status_history.append(status_msg)
            for msg in status_history:
                logger.info(msg)

            # Show metrics if training completed
            if status == ModelStatus.TRAINING_COMPLETED:
                metrics = self.metrics()
                if metrics.best_valid_metric:
                    logger.info(f"Best Checkpoint (iter: {metrics.best_valid_metric.get('iteration', 'N/A')}):")
                    for k, v in metrics.best_valid_metric.items():
                        logger.info(f"  {k}: {v}")
                    visualizer = MetricsVisualizer(metrics)
                    visualizer.log_metrics()
                    if plot_metrics:
                        visualizer.notebook_plot_training_metrics()

            # Update metrics during training
            if status == ModelStatus.TRAINING_RUNNING and model_info.updated_at > last_update:
                last_update = model_info.updated_at
                metrics = self.metrics()
                visualizer = MetricsVisualizer(metrics)
                visualizer.log_metrics()
                if plot_metrics:
                    visualizer.notebook_plot_training_metrics()

            # Check exit conditions
            if status not in [ModelStatus.CREATED, ModelStatus.TRAINING_RUNNING, ModelStatus.TRAINING_STARTING]:
                return

            if time.time() - start_time > max_runtime:
                logger.warning(f"Monitoring exceeded {max_runtime} seconds limit")
                return

            sleep(interval)

__init__(model_ref, api_client) #

Initialize the RemoteModel instance.

Parameters:

Name Type Description Default
model_ref str

Reference ID for the model.

 required
api_client ApiClient

HTTP client instance for communication.

 required

Raises:

Type Description
ValueError

If model metadata retrieval fails.
Source code in focoos/hub/remote_model.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def __init__(
    self,
    model_ref: str,
    api_client: ApiClient,
):
    """
    Initialize the RemoteModel instance.

    Args:
        model_ref (str): Reference ID for the model.
        api_client (ApiClient): HTTP client instance for communication.

    Raises:
        ValueError: If model metadata retrieval fails.
    """
    self.model_ref = model_ref
    self.api_client = api_client
    self.model_info: RemoteModelInfo = self.get_info()

    logger.info(
        f"[RemoteModel]: ref: {self.model_ref} name: {self.model_info.name} description: {self.model_info.description} status: {self.model_info.status}"
    )

get_info() #

Retrieve model metadata.

Returns:

Name Type Description
ModelMetadata RemoteModelInfo

Metadata of the model.

Raises:

Type Description
ValueError

If the request fails.
Example 
1
2
3
4
5
from focoos import Focoos, RemoteModel

focoos = Focoos()
model = focoos.get_remote_model(model_ref="<model_ref>")
model_info = model.get_info()
Source code in focoos/hub/remote_model.py 
 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 get_info(self) -> RemoteModelInfo:
    """
    Retrieve model metadata.

    Returns:
        ModelMetadata: Metadata of the model.

    Raises:
        ValueError: If the request fails.

    Example:
        ```python
        from focoos import Focoos, RemoteModel

        focoos = Focoos()
        model = focoos.get_remote_model(model_ref="<model_ref>")
        model_info = model.get_info()
        ```
    """
    res = self.api_client.get(f"models/{self.model_ref}")
    if res.status_code != 200:
        logger.error(f"Failed to get model info: {res.status_code} {res.text}")
        raise ValueError(f"Failed to get model info: {res.status_code} {res.text}")
    self.metadata = RemoteModelInfo(**res.json())
    return self.metadata

infer(image, threshold=0.5) #

Perform inference on the provided image using the remote model.

This method sends an image to the remote model for inference and retrieves the detection results.

Parameters:

Name Type Description Default
image Union[str, Path, ndarray, bytes]

The image to infer on, which can be a file path, a string representing the path, a NumPy array, or raw bytes.

 required
threshold float

The confidence threshold for detections. Defaults to 0.5. Detections with confidence scores below this threshold will be discarded.

 0.5

Returns:

Name Type Description
FocoosDetections FocoosDetections

The detection results including class IDs, confidence scores, bounding boxes, and segmentation masks (if applicable).

Raises:

Type Description
FileNotFoundError

If the provided image file path is invalid.
ValueError

If the inference request fails.
Example 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
from focoos import Focoos

focoos = Focoos()
model = focoos.get_remote_model("my-model")
results = model.infer("image.jpg", threshold=0.5)

# Print detection results
for det in results.detections:
    print(f"Found {det.label} with confidence {det.conf:.2f}")
    print(f"Bounding box: {det.bbox}")
    if det.mask:
        print("Instance segmentation mask included")
Source code in focoos/hub/remote_model.py 
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
def infer(
    self,
    image: Union[str, Path, np.ndarray, bytes, Image.Image],
    threshold: float = 0.5,
) -> FocoosDetections:
    """
    Perform inference on the provided image using the remote model.

    This method sends an image to the remote model for inference and retrieves the detection results.

    Args:
        image (Union[str, Path, np.ndarray, bytes]): The image to infer on, which can be a file path,
            a string representing the path, a NumPy array, or raw bytes.
        threshold (float, optional): The confidence threshold for detections. Defaults to 0.5.
            Detections with confidence scores below this threshold will be discarded.

    Returns:
        FocoosDetections: The detection results including class IDs, confidence scores, bounding boxes,
            and segmentation masks (if applicable).

    Raises:
        FileNotFoundError: If the provided image file path is invalid.
        ValueError: If the inference request fails.

    Example:
        ```python
        from focoos import Focoos

        focoos = Focoos()
        model = focoos.get_remote_model("my-model")
        results = model.infer("image.jpg", threshold=0.5)

        # Print detection results
        for det in results.detections:
            print(f"Found {det.label} with confidence {det.conf:.2f}")
            print(f"Bounding box: {det.bbox}")
            if det.mask:
                print("Instance segmentation mask included")
        ```
    """
    if isinstance(image, Image.Image):
        image = np.array(image)

    image_bytes = None
    if isinstance(image, str) or isinstance(image, Path):
        if not os.path.exists(image):
            logger.error(f"Image file not found: {image}")
            raise FileNotFoundError(f"Image file not found: {image}")
        image_bytes = open(image, "rb").read()
    elif isinstance(image, np.ndarray):
        _, buffer = cv2.imencode(".jpg", image)
        image_bytes = buffer.tobytes()
    else:
        image_bytes = image
    files = {"file": ("image", image_bytes, "image/jpeg")}
    t0 = time.time()
    res = self.api_client.post(
        f"models/{self.model_ref}/inference?confidence_threshold={threshold}",
        files=files,
    )
    t1 = time.time()
    if res.status_code == 200:
        detections = FocoosDetections(
            detections=[FocoosDet.from_json(d) for d in res.json().get("detections", [])],
            latency=res.json().get("latency", None),
        )
        logger.debug(
            f"Found {len(detections.detections)} detections. Inference Request time: {(t1 - t0) * 1000:.0f}ms"
        )

        return detections
    else:
        logger.error(f"Failed to infer: {res.status_code} {res.text}")
        raise ValueError(f"Failed to infer: {res.status_code} {res.text}")

metrics() #

Retrieve the metrics of the model.

This method sends a request to fetch the metrics of the model identified by model_ref. If the request is successful (status code 200), it returns the metrics as a Metrics object. If the request fails, it logs a warning and returns an empty Metrics object.

Returns:

Name Type Description
Metrics Metrics

An object containing the metrics of the model.

Raises:

Type Description
None

Returns an empty Metrics object if the request fails.
Source code in focoos/hub/remote_model.py 
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
def metrics(self) -> Metrics:  # noqa: F821
    """
    Retrieve the metrics of the model.

    This method sends a request to fetch the metrics of the model identified by `model_ref`.
    If the request is successful (status code 200), it returns the metrics as a `Metrics` object.
    If the request fails, it logs a warning and returns an empty `Metrics` object.

    Returns:
        Metrics: An object containing the metrics of the model.

    Raises:
        None: Returns an empty `Metrics` object if the request fails.
    """
    res = self.api_client.get(f"models/{self.model_ref}/metrics")
    if res.status_code != 200:
        logger.warning(f"Failed to get metrics: {res.status_code} {res.text}")
        return Metrics()  # noqa: F821
    return Metrics(**{k: v for k, v in res.json().items() if k in Metrics.__dataclass_fields__})

notebook_monitor_train(interval=30, plot_metrics=False, max_runtime=36000) #

Monitor the training process in a Jupyter notebook and display metrics.

Periodically checks the training status and displays metrics in a notebook cell. Clears previous output to maintain a clean view.

Parameters:

Name Type Description Default
interval int

Time between status checks in seconds. Must be 30-240. Default: 30

 30
plot_metrics bool

Whether to plot metrics graphs. Default: False

 False
max_runtime int

Maximum monitoring time in seconds. Default: 36000 (10 hours)

 36000

Returns:

Type Description
None

None
Source code in focoos/hub/remote_model.py 
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
def notebook_monitor_train(self, interval: int = 30, plot_metrics: bool = False, max_runtime: int = 36000) -> None:
    """
    Monitor the training process in a Jupyter notebook and display metrics.

    Periodically checks the training status and displays metrics in a notebook cell.
    Clears previous output to maintain a clean view.

    Args:
        interval (int): Time between status checks in seconds. Must be 30-240. Default: 30
        plot_metrics (bool): Whether to plot metrics graphs. Default: False
        max_runtime (int): Maximum monitoring time in seconds. Default: 36000 (10 hours)

    Returns:
        None
    """
    from IPython.display import clear_output

    if not 30 <= interval <= 240:
        raise ValueError("Interval must be between 30 and 240 seconds")

    last_update = self.get_info().updated_at
    start_time = time.time()
    status_history = []

    while True:
        # Get current status
        model_info = self.get_info()
        status = model_info.status

        # Clear and display status
        clear_output(wait=True)
        status_msg = f"[Live Monitor {self.metadata.name}] {status.value}"
        status_history.append(status_msg)
        for msg in status_history:
            logger.info(msg)

        # Show metrics if training completed
        if status == ModelStatus.TRAINING_COMPLETED:
            metrics = self.metrics()
            if metrics.best_valid_metric:
                logger.info(f"Best Checkpoint (iter: {metrics.best_valid_metric.get('iteration', 'N/A')}):")
                for k, v in metrics.best_valid_metric.items():
                    logger.info(f"  {k}: {v}")
                visualizer = MetricsVisualizer(metrics)
                visualizer.log_metrics()
                if plot_metrics:
                    visualizer.notebook_plot_training_metrics()

        # Update metrics during training
        if status == ModelStatus.TRAINING_RUNNING and model_info.updated_at > last_update:
            last_update = model_info.updated_at
            metrics = self.metrics()
            visualizer = MetricsVisualizer(metrics)
            visualizer.log_metrics()
            if plot_metrics:
                visualizer.notebook_plot_training_metrics()

        # Check exit conditions
        if status not in [ModelStatus.CREATED, ModelStatus.TRAINING_RUNNING, ModelStatus.TRAINING_STARTING]:
            return

        if time.time() - start_time > max_runtime:
            logger.warning(f"Monitoring exceeded {max_runtime} seconds limit")
            return

        sleep(interval)

train_info() #

Retrieve the current status of the model training.

Sends a request to check the training status of the model referenced by self.model_ref.

Returns:

Name Type Description
dict Optional[TrainingInfo]

A dictionary containing the training status information.

Raises:

Type Description
ValueError

If the request to get training status fails.
Source code in focoos/hub/remote_model.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
def train_info(self) -> Optional[TrainingInfo]:
    """
    Retrieve the current status of the model training.

    Sends a request to check the training status of the model referenced by `self.model_ref`.

    Returns:
        dict: A dictionary containing the training status information.

    Raises:
        ValueError: If the request to get training status fails.
    """
    res = self.api_client.get(f"models/{self.model_ref}/train/status")
    if res.status_code != 200:
        logger.error(f"Failed to get train info: {res.status_code} {res.text}")
        raise ValueError(f"Failed to get train info: {res.status_code} {res.text}")
    dct = {k: v for k, v in res.json().items() if k in TrainingInfo.__dataclass_fields__}
    return TrainingInfo(**dct)

train_logs() #

Retrieve the training logs for the model.

This method sends a request to fetch the logs of the model's training process. If the request is successful (status code 200), it returns the logs as a list of strings. If the request fails, it logs a warning and returns an empty list.

Returns:

Type Description
list[str]

list[str]: A list of training logs as strings.

Raises:

Type Description
None

Returns an empty list if the request fails.
Source code in focoos/hub/remote_model.py 
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
def train_logs(self) -> list[str]:
    """
    Retrieve the training logs for the model.

    This method sends a request to fetch the logs of the model's training process. If the request
    is successful (status code 200), it returns the logs as a list of strings. If the request fails,
    it logs a warning and returns an empty list.

    Returns:
        list[str]: A list of training logs as strings.

    Raises:
        None: Returns an empty list if the request fails.
    """
    res = self.api_client.get(f"models/{self.model_ref}/train/logs")
    if res.status_code != 200:
        logger.warning(f"Failed to get train logs: {res.status_code} {res.text}")
        return []
    return res.json()