Skip to content

nos.hub

nos.hub.Hub

Registry for models.

Source code in nos/hub/__init__.py 
 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
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
class Hub:
    """Registry for models."""

    _instance: Optional["Hub"] = None
    """Singleton instance."""
    _registry: Dict[str, ModelSpec] = {}
    """Model specifications lookup for all models registered."""

    @classmethod
    def get(cls: "Hub") -> "Hub":
        """Get the singleton instance.

        Returns:
            Hub: Singleton instance.
        """
        if cls._instance is None:
            cls._instance = cls()
            # Register models / Populate the registry
            import nos.models  # noqa: F401, E402
        return cls._instance

    def __contains__(self, model_id: str) -> bool:
        return model_id in self.get()._registry

    @classmethod
    def list(cls, private: bool = False) -> List[str]:
        """List models in the registry.

        Args:
            private (bool): Whether to include private models.

        Returns:
            List[str]: List of model names.
        """
        if private:
            raise NotImplementedError("Private models not supported.")
        return [k for k in cls.get()._registry.keys()]

    @classmethod
    def load_spec(cls, model_id: str) -> ModelSpec:
        """Load model spec from the registry.

        Args:
            model_id (str): Model identifier (e.g. `openai/clip-vit-base-patch32`).

        Returns:
            ModelSpec: Model specification.
        """
        try:
            return cls.get()._registry[model_id]
        except KeyError:
            raise KeyError(f"Unavailable model (name={model_id}).")

    @classmethod
    def load(cls, model_id: str) -> Any:
        """Instantiate model from the registry.

        Args:
            model_id (str): Model identifier (e.g. `openai/clip-vit-base-patch32`).

        Returns:
            Any: Instantiated model.
        """
        spec: ModelSpec = cls.load_spec(model_id)
        # Note (spillai): Loading the default signature here is OK
        # since all the signatures have the same `func_or_cls`.
        sig: FunctionSignature = spec.default_signature
        return sig.func_or_cls(*sig.init_args, **sig.init_kwargs)

    @classmethod
    def register(
        cls,
        model_id: str,
        task: TaskType,
        func_or_cls: Callable,
        init_args: Tuple[Any] = (),
        init_kwargs: Dict[str, Any] = {},  # noqa: B006
        method: str = "__call__",
        inputs: Dict[str, Any] = {},  # noqa: B006
        outputs: Union[Any, Dict[str, Any], None] = None,  # noqa: B006
        resources: ModelResources = None,
        **kwargs,
    ) -> ModelSpec:
        """Model registry decorator.

        Args:
            model_id (str): Model identifier (e.g. `openai/clip-vit-base-patch32`).
            task (TaskType): Task type (e.g. `TaskType.OBJECT_DETECTION_2D`).
            func_or_cls (Type[Any]): Model function or class.
            **kwargs: Additional keyword arguments.
        Returns:
            ModelSpec: Model specification.
        """
        logger.debug(
            f"""Registering model [model={model_id}, task={task}, func_or_cls={func_or_cls}, """
            f"""inputs={inputs}, outputs={outputs}, """
            f"""init_args={init_args}, init_kwargs={init_kwargs}, method={method}]"""
        )

        # Create signature
        signature: Dict[str, FunctionSignature] = {
            method: FunctionSignature(
                func_or_cls,
                method=method,
                init_args=init_args,
                init_kwargs=init_kwargs,
                input_annotations=inputs,
                output_annotations=outputs,
            ),
        }

        # Get hub instance
        hub = cls.get()

        # Add metadata for the model
        catalog = ModelSpecMetadataCatalog.get()
        spec = ModelSpec(model_id, signature=signature)
        logger.debug(f"Created model spec [id={model_id}, spec={spec}]")

        # Add task metadata for the model
        if task is not None:
            metadata = ModelSpecMetadata(model_id, method, task)
            catalog._metadata_catalog[f"{model_id}/{method}"] = metadata

        # Add model resources
        if resources is not None:
            catalog._resources_catalog[f"{model_id}/{method}"] = resources

        # Register model id to model spec registry
        if model_id not in hub._registry:
            hub._registry[model_id] = spec
            logger.debug(f"Registered model to hub registry [id={model_id}, spec={spec}]")

        # Add another signature if the model is already registered
        else:
            _spec = hub._registry[model_id]
            if method not in _spec.signature:
                logger.debug(
                    f"Adding task signature [model={model_id}, task={task}, method={method}, sig={spec.signature}]"
                )
                _spec.signature[method] = spec.signature[method]
                catalog._metadata_catalog[f"{model_id}/{method}"] = spec.metadata(method)
            else:
                logger.debug(
                    f"Task signature already registered [model={model_id}, task={task}, method={method}, sig={spec.signature}]"
                )

        logger.debug(f"Registered model [id={model_id}, spec={spec}]")
        return spec

    @classmethod
    def register_spec(cls, spec: ModelSpec, task: TaskType = None, resources: ModelResources = None) -> ModelSpec:
        """Register model spec to the registry.

        Args:
            spec (ModelSpec): Model specification.
            task (TaskType): Task type (e.g. `TaskType.OBJECT_DETECTION_2D`).
            resources (ModelResources): Model resources.
        Returns:
            ModelSpec: Model specification.
        """
        logger.debug(f"Registering model spec [id={spec.id}, spec={spec}]")

        # Get hub instance
        hub = cls.get()

        # Add metadata for the model
        catalog = ModelSpecMetadataCatalog.get()

        # Add task metadata for the model
        if task is not None:
            metadata = ModelSpecMetadata(spec.id, spec.default_method, task)
            catalog._metadata_catalog[f"{spec.id}/{spec.default_method}"] = metadata

        # Add model resources
        if resources is not None:
            catalog._resources_catalog[f"{spec.id}/{spec.default_method}"] = resources

        # Register model id to model spec registry
        if spec.id not in hub._registry:
            hub._registry[spec.id] = spec
            logger.debug(f"Registered model spec [id={spec.id}, spec={spec}]")
        return spec

    @classmethod
    def register_from_yaml(cls, filename: str) -> List[Any]:
        """Register models from a catalog YAML.

        Args:
            filename (str): Path to the deployment config YAML file.
        Returns:
            List[ModelSpec]: List of model specifications.
        """

        @dataclass
        class _ModelImportConfig:
            """Model import configuration."""

            id: str
            """Model identifier."""
            runtime_env: str
            """Runtime environment."""
            model_path: str
            """Model path."""
            model_cls: Callable
            """Model class name."""
            default_method: str
            """Default model method name."""
            init_args: Tuple[Any, ...] = field(default_factory=tuple)
            """Arguments to initialize the model instance."""
            init_kwargs: Dict[str, Any] = field(default_factory=dict)
            """Keyword arguments to initialize the model instance."""
            deployment: ModelDeploymentSpec = field(default_factory=ModelDeploymentSpec)
            """Model deployment specification."""

            @model_validator(mode="before")
            @classmethod
            def _validate_model_cls_import(cls, values):
                """Validate the model."""
                import importlib
                import importlib.util
                import sys

                model_cls_name = values.kwargs.get("model_cls")
                model_path = values.kwargs.get("model_path")

                # Check if model_path is a valid path relative to the directory containing
                # the catalog.yaml file.
                model_path = Path(filename).parent / model_path
                if not model_path.exists():
                    logger.error(f"Invalid model path provided, model_path={model_path}.")
                    raise FileNotFoundError(f"Invalid model path provided, model_path={model_path}.")

                # Check if the model_cls is importable from the model_path
                try:
                    # Load `model_cls` from the `model_path`
                    logger.debug(f"Loading model class [model_cls={model_cls_name}, model_path={model_path}].")
                    sys.path.append(str(model_path.parent))
                    logger.debug(f"Added model path to sys.path [model_path={model_path.parent}].")
                    spec = importlib.util.spec_from_file_location(model_cls_name, model_path)
                    logger.debug(f"Loaded spec from file location [spec={spec}].")
                    module = importlib.util.module_from_spec(spec)
                    logger.debug(f"Loaded module from spec [module={module}].")
                    spec.loader.exec_module(module)
                    logger.debug(f"Executed module [module={module}].")
                    model_cls = getattr(module, model_cls_name)
                    logger.debug(f"Loaded model class [model_cls={model_cls}].")
                except Exception as e:
                    import traceback

                    tback_str = traceback.format_exc()
                    logger.error(
                        f"Failed to import model class, model_cls={model_cls_name}, model_path={model_path}, e={e}\n\n{tback_str}."
                    )
                    raise ValueError(
                        f"Invalid model class provided, model_cls={model_cls_name}, model_path={model_path}, e={e}\n\n{tback_str}."
                    )

                values.kwargs.update(model_cls=model_cls)
                return values

        # Check if the file exists and has a YAML extension
        path = Path(filename)
        logger.debug(f"Loading deployment configuration from {path}")
        if not path.exists():
            raise FileNotFoundError(f"YAML file {path.absolute()} does not exist")
        if not (path.name.endswith(".yaml") or path.name.endswith(".yml")):
            raise ValueError(f"YAML file {path.absolute()} must have a .yaml or .yml extension")

        # Load the YAML file
        with path.open("r") as f:
            data = yaml.safe_load(f)
        if "models" not in data:
            raise ValueError("Missing `models` specification in the YAML file")

        # Service the models
        services: List[ModelServiceSpec] = []
        for model_id, mconfig in data["models"].items():
            # Check if the model is already registered
            logger.debug(f"Checking if model is already registered [id={model_id}].")
            try:
                spec: ModelSpec = cls.load_spec(model_id)
                deployment: ModelDeploymentSpec = ModelDeploymentSpec(**mconfig.get("deployment", {}))
                logger.debug(f"Model already registered [id={model_id}, spec={spec}, deployment={deployment}]")
                services.append(ModelServiceSpec(model=spec, deployment=deployment))
                logger.debug(f"Registered service [id={model_id}, svc={services[-1]}]")
                continue
            except KeyError as e:
                logger.error(f"Failed to load model spec, model_id={model_id}, e={e}")

            # If the model_id is not previously registered, register it
            # Add the model id to the config
            mconfig.update({"id": model_id})

            # Generate the model spec from the config
            try:
                mconfig = _ModelImportConfig(**mconfig)
            except (ValidationError, PydanticUserError) as e:
                raise ValueError(f"Invalid model config provided, filename={filename}, e={e}")

            # Register the model as a custom model
            spec = ModelSpec.from_cls(
                mconfig.model_cls,
                method=mconfig.default_method,
                init_args=mconfig.init_args,
                init_kwargs=mconfig.init_kwargs,
                model_id=mconfig.id,
            )
            cls.register_spec(spec, task=TaskType.CUSTOM, resources=mconfig.deployment.resources)
            services.append(ModelServiceSpec(model=spec, deployment=mconfig.deployment))
            logger.debug(f"Registered service [id={model_id}, svc={services[-1]}]")
        return services

    @classmethod
    def register_from_catalog(cls):
        """Register models from the catalog.

        The current workflow for registering models is as follows:
         - Load all .yaml files from the environment variable `NOS_HUB_CATALOG_PATH`
         - Register models from each of the catalog files (via `register_from_yaml`)

            `register_from_catalog` -> `register_from_yaml` -> `register`

        Raises:
            TypeError: If `NOS_HUB_CATALOG_PATH` is not a string.
            FileNotFoundError: If a specified catalog file does not exist.
        """
        import os

        warn_msg = "register_from_catalog will be deprecated soon, use register_from_yaml instead."
        logger.warning(warn_msg)

        NOS_HUB_CATALOG_PATH = os.getenv("NOS_HUB_CATALOG_PATH", "")
        if not isinstance(NOS_HUB_CATALOG_PATH, str):
            raise TypeError(f"NOS_HUB_CATALOG_PATH must be a string, got {type(NOS_HUB_CATALOG_PATH)}")

        logger.debug("Loading hub models from catalog.")
        paths: List[Path] = [
            Path(filename)
            for filename in NOS_HUB_CATALOG_PATH.split(":")
            if filename.endswith(".yaml") or filename.endswith(".yml")
        ]
        logger.debug(f"Found {len(paths)} catalog files.")

        specs = []
        for path in paths:
            logger.debug(f"Loading catalog file {path}.")
            if not path.exists():
                raise FileNotFoundError(f"Catalog file {path} does not exist.")
            _specs = cls.register_from_yaml(str(path))
            specs.extend(_specs)
        logger.debug(f"Registered {len(specs)} models from catalog.")

get classmethod 

get() -> Hub

Get the singleton instance.

Returns:

  • Hub ( Hub ) –

    Singleton instance.

Source code in nos/hub/__init__.py 
37
38
39
40
41
42
43
44
45
46
47
48
@classmethod
def get(cls: "Hub") -> "Hub":
    """Get the singleton instance.

    Returns:
        Hub: Singleton instance.
    """
    if cls._instance is None:
        cls._instance = cls()
        # Register models / Populate the registry
        import nos.models  # noqa: F401, E402
    return cls._instance

list classmethod 

list(private: bool = False) -> List[str]

List models in the registry.

Parameters:

  • private (bool, default: False ) –

    Whether to include private models.

Returns:

  • List[str] –

    List[str]: List of model names.

Source code in nos/hub/__init__.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
@classmethod
def list(cls, private: bool = False) -> List[str]:
    """List models in the registry.

    Args:
        private (bool): Whether to include private models.

    Returns:
        List[str]: List of model names.
    """
    if private:
        raise NotImplementedError("Private models not supported.")
    return [k for k in cls.get()._registry.keys()]

load_spec classmethod 

load_spec(model_id: str) -> ModelSpec

Load model spec from the registry.

Parameters:

  • model_id (str) –

    Model identifier (e.g. openai/clip-vit-base-patch32).

Returns:

  • ModelSpec ( ModelSpec ) –

    Model specification.

Source code in nos/hub/__init__.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
@classmethod
def load_spec(cls, model_id: str) -> ModelSpec:
    """Load model spec from the registry.

    Args:
        model_id (str): Model identifier (e.g. `openai/clip-vit-base-patch32`).

    Returns:
        ModelSpec: Model specification.
    """
    try:
        return cls.get()._registry[model_id]
    except KeyError:
        raise KeyError(f"Unavailable model (name={model_id}).")

load classmethod 

load(model_id: str) -> Any

Instantiate model from the registry.

Parameters:

  • model_id (str) –

    Model identifier (e.g. openai/clip-vit-base-patch32).

Returns:

  • Any ( Any ) –

    Instantiated model.

Source code in nos/hub/__init__.py 
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
@classmethod
def load(cls, model_id: str) -> Any:
    """Instantiate model from the registry.

    Args:
        model_id (str): Model identifier (e.g. `openai/clip-vit-base-patch32`).

    Returns:
        Any: Instantiated model.
    """
    spec: ModelSpec = cls.load_spec(model_id)
    # Note (spillai): Loading the default signature here is OK
    # since all the signatures have the same `func_or_cls`.
    sig: FunctionSignature = spec.default_signature
    return sig.func_or_cls(*sig.init_args, **sig.init_kwargs)

register classmethod 

register(model_id: str, task: TaskType, func_or_cls: Callable, init_args: Tuple[Any] = (), init_kwargs: Dict[str, Any] = {}, method: str = '__call__', inputs: Dict[str, Any] = {}, outputs: Union[Any, Dict[str, Any], None] = None, resources: ModelResources = None, **kwargs) -> ModelSpec

Model registry decorator.

Parameters:

  • model_id (str) –

    Model identifier (e.g. openai/clip-vit-base-patch32).

  • task (TaskType) –

    Task type (e.g. TaskType.OBJECT_DETECTION_2D).

  • func_or_cls (Type[Any]) –

    Model function or class.

  • **kwargs –

    Additional keyword arguments.

Returns: ModelSpec: Model specification.

Source code in nos/hub/__init__.py 
 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
@classmethod
def register(
    cls,
    model_id: str,
    task: TaskType,
    func_or_cls: Callable,
    init_args: Tuple[Any] = (),
    init_kwargs: Dict[str, Any] = {},  # noqa: B006
    method: str = "__call__",
    inputs: Dict[str, Any] = {},  # noqa: B006
    outputs: Union[Any, Dict[str, Any], None] = None,  # noqa: B006
    resources: ModelResources = None,
    **kwargs,
) -> ModelSpec:
    """Model registry decorator.

    Args:
        model_id (str): Model identifier (e.g. `openai/clip-vit-base-patch32`).
        task (TaskType): Task type (e.g. `TaskType.OBJECT_DETECTION_2D`).
        func_or_cls (Type[Any]): Model function or class.
        **kwargs: Additional keyword arguments.
    Returns:
        ModelSpec: Model specification.
    """
    logger.debug(
        f"""Registering model [model={model_id}, task={task}, func_or_cls={func_or_cls}, """
        f"""inputs={inputs}, outputs={outputs}, """
        f"""init_args={init_args}, init_kwargs={init_kwargs}, method={method}]"""
    )

    # Create signature
    signature: Dict[str, FunctionSignature] = {
        method: FunctionSignature(
            func_or_cls,
            method=method,
            init_args=init_args,
            init_kwargs=init_kwargs,
            input_annotations=inputs,
            output_annotations=outputs,
        ),
    }

    # Get hub instance
    hub = cls.get()

    # Add metadata for the model
    catalog = ModelSpecMetadataCatalog.get()
    spec = ModelSpec(model_id, signature=signature)
    logger.debug(f"Created model spec [id={model_id}, spec={spec}]")

    # Add task metadata for the model
    if task is not None:
        metadata = ModelSpecMetadata(model_id, method, task)
        catalog._metadata_catalog[f"{model_id}/{method}"] = metadata

    # Add model resources
    if resources is not None:
        catalog._resources_catalog[f"{model_id}/{method}"] = resources

    # Register model id to model spec registry
    if model_id not in hub._registry:
        hub._registry[model_id] = spec
        logger.debug(f"Registered model to hub registry [id={model_id}, spec={spec}]")

    # Add another signature if the model is already registered
    else:
        _spec = hub._registry[model_id]
        if method not in _spec.signature:
            logger.debug(
                f"Adding task signature [model={model_id}, task={task}, method={method}, sig={spec.signature}]"
            )
            _spec.signature[method] = spec.signature[method]
            catalog._metadata_catalog[f"{model_id}/{method}"] = spec.metadata(method)
        else:
            logger.debug(
                f"Task signature already registered [model={model_id}, task={task}, method={method}, sig={spec.signature}]"
            )

    logger.debug(f"Registered model [id={model_id}, spec={spec}]")
    return spec

register_spec classmethod 

register_spec(spec: ModelSpec, task: TaskType = None, resources: ModelResources = None) -> ModelSpec

Register model spec to the registry.

Parameters:

  • spec (ModelSpec) –

    Model specification.

  • task (TaskType, default: None ) –

    Task type (e.g. TaskType.OBJECT_DETECTION_2D).

  • resources (ModelResources, default: None ) –

    Model resources.

Returns: ModelSpec: Model specification.

Source code in nos/hub/__init__.py 
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
@classmethod
def register_spec(cls, spec: ModelSpec, task: TaskType = None, resources: ModelResources = None) -> ModelSpec:
    """Register model spec to the registry.

    Args:
        spec (ModelSpec): Model specification.
        task (TaskType): Task type (e.g. `TaskType.OBJECT_DETECTION_2D`).
        resources (ModelResources): Model resources.
    Returns:
        ModelSpec: Model specification.
    """
    logger.debug(f"Registering model spec [id={spec.id}, spec={spec}]")

    # Get hub instance
    hub = cls.get()

    # Add metadata for the model
    catalog = ModelSpecMetadataCatalog.get()

    # Add task metadata for the model
    if task is not None:
        metadata = ModelSpecMetadata(spec.id, spec.default_method, task)
        catalog._metadata_catalog[f"{spec.id}/{spec.default_method}"] = metadata

    # Add model resources
    if resources is not None:
        catalog._resources_catalog[f"{spec.id}/{spec.default_method}"] = resources

    # Register model id to model spec registry
    if spec.id not in hub._registry:
        hub._registry[spec.id] = spec
        logger.debug(f"Registered model spec [id={spec.id}, spec={spec}]")
    return spec

register_from_yaml classmethod 

register_from_yaml(filename: str) -> List[Any]

Register models from a catalog YAML.

Parameters:

  • filename (str) –

    Path to the deployment config YAML file.

Returns: List[ModelSpec]: List of model specifications.

Source code in nos/hub/__init__.py 
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
@classmethod
def register_from_yaml(cls, filename: str) -> List[Any]:
    """Register models from a catalog YAML.

    Args:
        filename (str): Path to the deployment config YAML file.
    Returns:
        List[ModelSpec]: List of model specifications.
    """

    @dataclass
    class _ModelImportConfig:
        """Model import configuration."""

        id: str
        """Model identifier."""
        runtime_env: str
        """Runtime environment."""
        model_path: str
        """Model path."""
        model_cls: Callable
        """Model class name."""
        default_method: str
        """Default model method name."""
        init_args: Tuple[Any, ...] = field(default_factory=tuple)
        """Arguments to initialize the model instance."""
        init_kwargs: Dict[str, Any] = field(default_factory=dict)
        """Keyword arguments to initialize the model instance."""
        deployment: ModelDeploymentSpec = field(default_factory=ModelDeploymentSpec)
        """Model deployment specification."""

        @model_validator(mode="before")
        @classmethod
        def _validate_model_cls_import(cls, values):
            """Validate the model."""
            import importlib
            import importlib.util
            import sys

            model_cls_name = values.kwargs.get("model_cls")
            model_path = values.kwargs.get("model_path")

            # Check if model_path is a valid path relative to the directory containing
            # the catalog.yaml file.
            model_path = Path(filename).parent / model_path
            if not model_path.exists():
                logger.error(f"Invalid model path provided, model_path={model_path}.")
                raise FileNotFoundError(f"Invalid model path provided, model_path={model_path}.")

            # Check if the model_cls is importable from the model_path
            try:
                # Load `model_cls` from the `model_path`
                logger.debug(f"Loading model class [model_cls={model_cls_name}, model_path={model_path}].")
                sys.path.append(str(model_path.parent))
                logger.debug(f"Added model path to sys.path [model_path={model_path.parent}].")
                spec = importlib.util.spec_from_file_location(model_cls_name, model_path)
                logger.debug(f"Loaded spec from file location [spec={spec}].")
                module = importlib.util.module_from_spec(spec)
                logger.debug(f"Loaded module from spec [module={module}].")
                spec.loader.exec_module(module)
                logger.debug(f"Executed module [module={module}].")
                model_cls = getattr(module, model_cls_name)
                logger.debug(f"Loaded model class [model_cls={model_cls}].")
            except Exception as e:
                import traceback

                tback_str = traceback.format_exc()
                logger.error(
                    f"Failed to import model class, model_cls={model_cls_name}, model_path={model_path}, e={e}\n\n{tback_str}."
                )
                raise ValueError(
                    f"Invalid model class provided, model_cls={model_cls_name}, model_path={model_path}, e={e}\n\n{tback_str}."
                )

            values.kwargs.update(model_cls=model_cls)
            return values

    # Check if the file exists and has a YAML extension
    path = Path(filename)
    logger.debug(f"Loading deployment configuration from {path}")
    if not path.exists():
        raise FileNotFoundError(f"YAML file {path.absolute()} does not exist")
    if not (path.name.endswith(".yaml") or path.name.endswith(".yml")):
        raise ValueError(f"YAML file {path.absolute()} must have a .yaml or .yml extension")

    # Load the YAML file
    with path.open("r") as f:
        data = yaml.safe_load(f)
    if "models" not in data:
        raise ValueError("Missing `models` specification in the YAML file")

    # Service the models
    services: List[ModelServiceSpec] = []
    for model_id, mconfig in data["models"].items():
        # Check if the model is already registered
        logger.debug(f"Checking if model is already registered [id={model_id}].")
        try:
            spec: ModelSpec = cls.load_spec(model_id)
            deployment: ModelDeploymentSpec = ModelDeploymentSpec(**mconfig.get("deployment", {}))
            logger.debug(f"Model already registered [id={model_id}, spec={spec}, deployment={deployment}]")
            services.append(ModelServiceSpec(model=spec, deployment=deployment))
            logger.debug(f"Registered service [id={model_id}, svc={services[-1]}]")
            continue
        except KeyError as e:
            logger.error(f"Failed to load model spec, model_id={model_id}, e={e}")

        # If the model_id is not previously registered, register it
        # Add the model id to the config
        mconfig.update({"id": model_id})

        # Generate the model spec from the config
        try:
            mconfig = _ModelImportConfig(**mconfig)
        except (ValidationError, PydanticUserError) as e:
            raise ValueError(f"Invalid model config provided, filename={filename}, e={e}")

        # Register the model as a custom model
        spec = ModelSpec.from_cls(
            mconfig.model_cls,
            method=mconfig.default_method,
            init_args=mconfig.init_args,
            init_kwargs=mconfig.init_kwargs,
            model_id=mconfig.id,
        )
        cls.register_spec(spec, task=TaskType.CUSTOM, resources=mconfig.deployment.resources)
        services.append(ModelServiceSpec(model=spec, deployment=mconfig.deployment))
        logger.debug(f"Registered service [id={model_id}, svc={services[-1]}]")
    return services

register_from_catalog classmethod 

register_from_catalog()

Register models from the catalog.

The current workflow for registering models is as follows
  • Load all .yaml files from the environment variable NOS_HUB_CATALOG_PATH
  • Register models from each of the catalog files (via register_from_yaml)

register_from_catalog -> register_from_yaml -> register

Raises:

  • TypeError –

    If NOS_HUB_CATALOG_PATH is not a string.

  • FileNotFoundError –

    If a specified catalog file does not exist.

Source code in nos/hub/__init__.py 
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
@classmethod
def register_from_catalog(cls):
    """Register models from the catalog.

    The current workflow for registering models is as follows:
     - Load all .yaml files from the environment variable `NOS_HUB_CATALOG_PATH`
     - Register models from each of the catalog files (via `register_from_yaml`)

        `register_from_catalog` -> `register_from_yaml` -> `register`

    Raises:
        TypeError: If `NOS_HUB_CATALOG_PATH` is not a string.
        FileNotFoundError: If a specified catalog file does not exist.
    """
    import os

    warn_msg = "register_from_catalog will be deprecated soon, use register_from_yaml instead."
    logger.warning(warn_msg)

    NOS_HUB_CATALOG_PATH = os.getenv("NOS_HUB_CATALOG_PATH", "")
    if not isinstance(NOS_HUB_CATALOG_PATH, str):
        raise TypeError(f"NOS_HUB_CATALOG_PATH must be a string, got {type(NOS_HUB_CATALOG_PATH)}")

    logger.debug("Loading hub models from catalog.")
    paths: List[Path] = [
        Path(filename)
        for filename in NOS_HUB_CATALOG_PATH.split(":")
        if filename.endswith(".yaml") or filename.endswith(".yml")
    ]
    logger.debug(f"Found {len(paths)} catalog files.")

    specs = []
    for path in paths:
        logger.debug(f"Loading catalog file {path}.")
        if not path.exists():
            raise FileNotFoundError(f"Catalog file {path} does not exist.")
        _specs = cls.register_from_yaml(str(path))
        specs.extend(_specs)
    logger.debug(f"Registered {len(specs)} models from catalog.")

nos.hub.config.NosHubConfig dataclass

NOS Hub configuration.

Source code in nos/hub/config.py 
11
12
13
14
15
16
17
18
@dataclass(frozen=True)
class NosHubConfig:
    """NOS Hub configuration."""

    namespace: str
    """Namespace (repository, organization)."""
    name: str
    """Model name."""

namespace instance-attribute 

namespace: str

Namespace (repository, organization).

name instance-attribute 

name: str

Model name.

nos.hub.config.TorchHubConfig dataclass

PyTorch Hub configuration.

Source code in nos/hub/config.py 
21
22
23
24
25
26
27
28
29
30
@dataclass(frozen=True)
class TorchHubConfig:
    """PyTorch Hub configuration."""

    repo: str
    """Repository name (e.g. pytorch/vision)."""
    model_name: str
    """Model name (e.g. resnet18)."""
    checkpoint: str = None
    """Checkpoint name (e.g. resnet18-5c106cde.pth)."""

repo instance-attribute 

repo: str

Repository name (e.g. pytorch/vision).

model_name instance-attribute 

model_name: str

Model name (e.g. resnet18).

checkpoint class-attribute instance-attribute 

checkpoint: str = None

Checkpoint name (e.g. resnet18-5c106cde.pth).

nos.hub.config.HuggingFaceHubConfig dataclass

HuggingFace Hub configuration.

Source code in nos/hub/config.py 
33
34
35
36
37
38
39
40
@dataclass(frozen=True)
class HuggingFaceHubConfig:
    """HuggingFace Hub configuration."""

    model_name: str
    """Model name (e.g. bert-base-uncased)."""
    checkpoint: str = None
    """Checkpoint name (e.g. bert-base-uncased-pytorch_model.bin)."""

model_name instance-attribute 

model_name: str

Model name (e.g. bert-base-uncased).

checkpoint class-attribute instance-attribute 

checkpoint: str = None

Checkpoint name (e.g. bert-base-uncased-pytorch_model.bin).