Skip to content

FocoosHUB

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 
 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
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
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) -> Optional[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")
            ```
        """
        if model_info.task == Task.KEYPOINT:
            logger.warning(
                "Unfortunatelly keypoint models are not supported in the hub yet. Use them only locally temporarily."
            )
            return None
        if model_info.model_family not in SUPPORTED_MODEL_FAMILIES:
            logger.warning(
                f"Unfortunatelly model family {model_info.model_family} is not supported in the hub yet. Use one of {SUPPORTED_MODEL_FAMILIES}."
            )
            return None

        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 
 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
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 
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
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 
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
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 
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
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 
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
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 
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
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 
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
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 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
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
Optional[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 
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
378
379
380
381
382
383
384
385
386
387
388
389
390
391
def new_model(self, model_info: ModelInfo) -> Optional[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")
        ```
    """
    if model_info.task == Task.KEYPOINT:
        logger.warning(
            "Unfortunatelly keypoint models are not supported in the hub yet. Use them only locally temporarily."
        )
        return None
    if model_info.model_family not in SUPPORTED_MODEL_FAMILIES:
        logger.warning(
            f"Unfortunatelly model family {model_info.model_family} is not supported in the hub yet. Use one of {SUPPORTED_MODEL_FAMILIES}."
        )
        return None

    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 
 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 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 upload_artifacts:
                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
                else:
                    logger.warning(f"Artifact {artifact.value} not found in {dir}")

    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,
        annotate: bool = False,
    ) -> FocoosDetections:
        """
        Run inference on an image using the remote model and return detection results.

        This method uploads an image to the remote model for inference. The image can be provided as a file path,
        a string path, a NumPy array, raw bytes, or a PIL Image. The method returns detection results, including
        class IDs, confidence scores, bounding boxes, and segmentation masks if available. Optionally, the results
        can be returned with the image annotated with detections.

        Args:
            image (Union[str, Path, np.ndarray, bytes, Image.Image]): The image to run inference on.
            threshold (float, optional): Minimum confidence threshold for detections. Detections below this
                threshold are filtered out. Default is 0.5.
            annotate (bool, optional): If True, returns the image with detections drawn on it. Default is False.

        Returns:
            FocoosDetections: Detection results, including a list of detections and optional annotated image.

        Raises:
            FileNotFoundError: If the image file path does not exist or cannot be loaded.
            ValueError: If the inference request to the remote model fails.

        Example:
            ```python
            from focoos import Focoos
            from PIL import Image

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

            for det in results.detections:
                print(f"Label: {det.label}, Confidence: {det.conf:.2f}, BBox: {det.bbox}")
                if det.mask is not None:
                    print("Segmentation mask available")
            if results.image is not None:
                Image.fromarray(results.image)
            ```
        """

        image = image_loader(image)
        _, buffer = cv2.imencode(".jpg", image)
        image_bytes = buffer.tobytes()
        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=InferLatency(**res.json().get("latency", {})),
            )
            detections.infer_print()
            print(f"Request time: {(t1 - t0) * 1000:.0f}ms")
            if annotate:
                detections.image = annotate_frame(image, detections, self.model_info.task, self.model_info.classes)
            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 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
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 
 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
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, annotate=False) #

Run inference on an image using the remote model and return detection results.

This method uploads an image to the remote model for inference. The image can be provided as a file path, a string path, a NumPy array, raw bytes, or a PIL Image. The method returns detection results, including class IDs, confidence scores, bounding boxes, and segmentation masks if available. Optionally, the results can be returned with the image annotated with detections.

Parameters:

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

The image to run inference on.

 required
threshold float

Minimum confidence threshold for detections. Detections below this threshold are filtered out. Default is 0.5.

 0.5
annotate bool

If True, returns the image with detections drawn on it. Default is False.

 False

Returns:

Name Type Description
FocoosDetections FocoosDetections

Detection results, including a list of detections and optional annotated image.

Raises:

Type Description
FileNotFoundError

If the image file path does not exist or cannot be loaded.
ValueError

If the inference request to the remote model fails.
Example 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
from focoos import Focoos
from PIL import Image

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

for det in results.detections:
    print(f"Label: {det.label}, Confidence: {det.conf:.2f}, BBox: {det.bbox}")
    if det.mask is not None:
        print("Segmentation mask available")
if results.image is not None:
    Image.fromarray(results.image)
Source code in focoos/hub/remote_model.py 
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
def infer(
    self,
    image: Union[str, Path, np.ndarray, bytes, Image.Image],
    threshold: float = 0.5,
    annotate: bool = False,
) -> FocoosDetections:
    """
    Run inference on an image using the remote model and return detection results.

    This method uploads an image to the remote model for inference. The image can be provided as a file path,
    a string path, a NumPy array, raw bytes, or a PIL Image. The method returns detection results, including
    class IDs, confidence scores, bounding boxes, and segmentation masks if available. Optionally, the results
    can be returned with the image annotated with detections.

    Args:
        image (Union[str, Path, np.ndarray, bytes, Image.Image]): The image to run inference on.
        threshold (float, optional): Minimum confidence threshold for detections. Detections below this
            threshold are filtered out. Default is 0.5.
        annotate (bool, optional): If True, returns the image with detections drawn on it. Default is False.

    Returns:
        FocoosDetections: Detection results, including a list of detections and optional annotated image.

    Raises:
        FileNotFoundError: If the image file path does not exist or cannot be loaded.
        ValueError: If the inference request to the remote model fails.

    Example:
        ```python
        from focoos import Focoos
        from PIL import Image

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

        for det in results.detections:
            print(f"Label: {det.label}, Confidence: {det.conf:.2f}, BBox: {det.bbox}")
            if det.mask is not None:
                print("Segmentation mask available")
        if results.image is not None:
            Image.fromarray(results.image)
        ```
    """

    image = image_loader(image)
    _, buffer = cv2.imencode(".jpg", image)
    image_bytes = buffer.tobytes()
    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=InferLatency(**res.json().get("latency", {})),
        )
        detections.infer_print()
        print(f"Request time: {(t1 - t0) * 1000:.0f}ms")
        if annotate:
            detections.image = annotate_frame(image, detections, self.model_info.task, self.model_info.classes)
        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 
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
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 
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
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 
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
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 
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
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()