Skip to content

Entity and methods

Model

Bases: Entity

A class representing a model.

Source code in digitalhub_ml/entities/models/entity.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
class Model(Entity):
    """
    A class representing a model.
    """

    ENTITY_TYPE = EntityTypes.MODELS.value

    def __init__(
        self,
        project: str,
        name: str,
        uuid: str,
        kind: str,
        metadata: Metadata,
        spec: ModelSpec,
        status: ModelStatus,
        user: str | None = None,
    ) -> None:
        """
        Constructor.

        Parameters
        ----------
        project : str
            Project name.
        name : str
            Name of the object.
        uuid : str
            Version of the object.
        kind : str
            Kind the object.
        metadata : Metadata
            Metadata of the object.
        spec : ModelSpec
            Specification of the object.
        status : ModelStatus
            Status of the object.
        user : str
            Owner of the object.
        """
        super().__init__()
        self.project = project
        self.name = name
        self.id = uuid
        self.kind = kind
        self.key = f"store://{project}/{self.ENTITY_TYPE}/{kind}/{name}:{uuid}"
        self.metadata = metadata
        self.spec = spec
        self.status = status
        self.user = user

        # Add attributes to be used in the to_dict method
        self._obj_attr.extend(["project", "name", "id", "key"])

    #############################
    #  Save / Refresh / Export
    #############################

    def save(self, update: bool = False) -> Model:
        """
        Save entity into backend.

        Parameters
        ----------
        update : bool
            Flag to indicate update.

        Returns
        -------
        Model
            Entity saved.
        """
        obj = self.to_dict()

        if not update:
            new_obj = create_entity_api_ctx(self.project, self.ENTITY_TYPE, obj)
            self._update_attributes(new_obj)
            return self

        self.metadata.updated = obj["metadata"]["updated"] = get_timestamp()
        new_obj = update_entity_api_ctx(self.project, self.ENTITY_TYPE, self.id, obj)
        self._update_attributes(new_obj)
        return self

    def refresh(self) -> Model:
        """
        Refresh object from backend.

        Returns
        -------
        Model
            Entity refreshed.
        """
        new_obj = read_entity_api_ctx(self.key)
        self._update_attributes(new_obj)
        return self

    def export(self, filename: str | None = None) -> None:
        """
        Export object as a YAML file.

        Parameters
        ----------
        filename : str
            Name of the export YAML file. If not specified, the default value is used.

        Returns
        -------
        None
        """
        obj = self.to_dict()
        if filename is None:
            filename = f"{self.kind}_{self.name}_{self.id}.yml"
        pth = self._context().root / filename
        pth.parent.mkdir(parents=True, exist_ok=True)
        write_yaml(pth, obj)

    #############################
    #  Context
    #############################

    def _context(self) -> Context:
        """
        Get context.

        Returns
        -------
        Context
            Context.
        """
        return get_context(self.project)

    #############################
    #  Model methods
    #############################

    def as_file(self) -> str:
        """
        Get model as file. In the case of a local store, the store returns the current
        path of the model. In the case of a remote store, the model is downloaded in
        a temporary directory.

        Returns
        -------
        str
            Path of the model.
        """
        path = self.spec.path
        store = get_store(path)
        if store.is_local():
            return path
        return store.download(path)

    def download(
        self,
        destination: str | None = None,
        overwrite: bool = False,
    ) -> str:
        """
        Download model from remote storage.

        Parameters
        ----------
        destination : str
            Destination path as filename.
        overwrite : bool
            Specify if overwrite an existing file. Default value is False.

        Returns
        -------
        str
            Path of the downloaded model.
        """

        # Check if target path is remote
        path = self.spec.path
        store = get_store(path)
        if store.is_local():
            raise RuntimeError("Local files cannot be downloaded. Use as_file().")

        # Check if download destination path is specified and rebuild it if necessary
        if destination is None:
            filename = Path(urlparse(path).path).name
            destination = f"{self.project}/{self.ENTITY_TYPE}/{self.name}/{self.id}/{filename}"

        # Check if destination path is local
        self._check_local(destination)

        # Check if destination path exists for overwrite
        check_overwrite(destination, overwrite)

        # Download dataitem and return path
        return store.download(path, destination)

    def upload(self, source: str | None = None) -> str:
        """
        Upload model to remote storage from given local path to
        spec path destination.

        Parameters
        ----------
        source : str
            Source path is the local path of the model.

        Returns
        -------
        str
            Path of the uploaded model.
        """
        # Check if target path is remote
        path = self.spec.path
        store = get_store(path)
        if store.is_local():
            raise RuntimeError("Only remote paths are supported for upload.")

        # Check if source path is provided and if it is local
        src = source if source is not None else self.spec.src_path
        self._check_local(src)

        # Get store and upload model and return remote path
        target = store.upload(src, path)
        file_info = store.get_file_info(target, src)

        if file_info is not None:
            self.refresh()
            self.status.add_file(file_info)
            self.save(update=True)

        return target

    #############################
    #  Private Helpers
    #############################

    @staticmethod
    def _check_local(path: str) -> None:
        """
        Check through URI scheme if given path is local or not.

        Parameters
        ----------
        path : str
            Path of some source.

        Returns
        -------
        None

        Raises
        ------
        EntityError
            If source path is not local.
        """
        if not check_local_path(path):
            raise EntityError("Only local paths are supported for source paths.")

    #############################
    #  Static interface methods
    #############################

    @staticmethod
    def _parse_dict(obj: dict, validate: bool = True) -> dict:
        """
        Get dictionary and parse it to a valid entity dictionary.

        Parameters
        ----------
        obj : dict
            Dictionary to parse.

        Returns
        -------
        dict
            A dictionary containing the attributes of the entity instance.
        """
        project = obj.get("project")
        name = build_name(obj.get("name"))
        kind = obj.get("kind")
        uuid = build_uuid(obj.get("id"))
        metadata = build_metadata(kind, **obj.get("metadata", {}))
        spec = build_spec(kind, validate=validate, **obj.get("spec", {}))
        status = build_status(kind, **obj.get("status", {}))
        user = obj.get("user")
        return {
            "project": project,
            "name": name,
            "uuid": uuid,
            "kind": kind,
            "metadata": metadata,
            "spec": spec,
            "status": status,
            "user": user,
        }

__init__(project, name, uuid, kind, metadata, spec, status, user=None)

Constructor.

Parameters:

Name Type Description Default
project str

Project name.

 required
name str

Name of the object.

 required
uuid str

Version of the object.

 required
kind str

Kind the object.

 required
metadata Metadata

Metadata of the object.

 required
spec ModelSpec

Specification of the object.

 required
status ModelStatus

Status of the object.

 required
user str

Owner of the object.

 None
Source code in digitalhub_ml/entities/models/entity.py 
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
def __init__(
    self,
    project: str,
    name: str,
    uuid: str,
    kind: str,
    metadata: Metadata,
    spec: ModelSpec,
    status: ModelStatus,
    user: str | None = None,
) -> None:
    """
    Constructor.

    Parameters
    ----------
    project : str
        Project name.
    name : str
        Name of the object.
    uuid : str
        Version of the object.
    kind : str
        Kind the object.
    metadata : Metadata
        Metadata of the object.
    spec : ModelSpec
        Specification of the object.
    status : ModelStatus
        Status of the object.
    user : str
        Owner of the object.
    """
    super().__init__()
    self.project = project
    self.name = name
    self.id = uuid
    self.kind = kind
    self.key = f"store://{project}/{self.ENTITY_TYPE}/{kind}/{name}:{uuid}"
    self.metadata = metadata
    self.spec = spec
    self.status = status
    self.user = user

    # Add attributes to be used in the to_dict method
    self._obj_attr.extend(["project", "name", "id", "key"])

as_file()

Get model as file. In the case of a local store, the store returns the current path of the model. In the case of a remote store, the model is downloaded in a temporary directory.

Returns:

Type Description
str

Path of the model.
Source code in digitalhub_ml/entities/models/entity.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def as_file(self) -> str:
    """
    Get model as file. In the case of a local store, the store returns the current
    path of the model. In the case of a remote store, the model is downloaded in
    a temporary directory.

    Returns
    -------
    str
        Path of the model.
    """
    path = self.spec.path
    store = get_store(path)
    if store.is_local():
        return path
    return store.download(path)

download(destination=None, overwrite=False)

Download model from remote storage.

Parameters:

Name Type Description Default
destination str

Destination path as filename.

 None
overwrite bool

Specify if overwrite an existing file. Default value is False.

 False

Returns:

Type Description
str

Path of the downloaded model.
Source code in digitalhub_ml/entities/models/entity.py 
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
def download(
    self,
    destination: str | None = None,
    overwrite: bool = False,
) -> str:
    """
    Download model from remote storage.

    Parameters
    ----------
    destination : str
        Destination path as filename.
    overwrite : bool
        Specify if overwrite an existing file. Default value is False.

    Returns
    -------
    str
        Path of the downloaded model.
    """

    # Check if target path is remote
    path = self.spec.path
    store = get_store(path)
    if store.is_local():
        raise RuntimeError("Local files cannot be downloaded. Use as_file().")

    # Check if download destination path is specified and rebuild it if necessary
    if destination is None:
        filename = Path(urlparse(path).path).name
        destination = f"{self.project}/{self.ENTITY_TYPE}/{self.name}/{self.id}/{filename}"

    # Check if destination path is local
    self._check_local(destination)

    # Check if destination path exists for overwrite
    check_overwrite(destination, overwrite)

    # Download dataitem and return path
    return store.download(path, destination)

export(filename=None)

Export object as a YAML file.

Parameters:

Name Type Description Default
filename str

Name of the export YAML file. If not specified, the default value is used.

 None

Returns:

Type Description
None
Source code in digitalhub_ml/entities/models/entity.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def export(self, filename: str | None = None) -> None:
    """
    Export object as a YAML file.

    Parameters
    ----------
    filename : str
        Name of the export YAML file. If not specified, the default value is used.

    Returns
    -------
    None
    """
    obj = self.to_dict()
    if filename is None:
        filename = f"{self.kind}_{self.name}_{self.id}.yml"
    pth = self._context().root / filename
    pth.parent.mkdir(parents=True, exist_ok=True)
    write_yaml(pth, obj)

refresh()

Refresh object from backend.

Returns:

Type Description
Model

Entity refreshed.
Source code in digitalhub_ml/entities/models/entity.py 
113
114
115
116
117
118
119
120
121
122
123
124
def refresh(self) -> Model:
    """
    Refresh object from backend.

    Returns
    -------
    Model
        Entity refreshed.
    """
    new_obj = read_entity_api_ctx(self.key)
    self._update_attributes(new_obj)
    return self

save(update=False)

Save entity into backend.

Parameters:

Name Type Description Default
update bool

Flag to indicate update.

 False

Returns:

Type Description
Model

Entity saved.
Source code in digitalhub_ml/entities/models/entity.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 save(self, update: bool = False) -> Model:
    """
    Save entity into backend.

    Parameters
    ----------
    update : bool
        Flag to indicate update.

    Returns
    -------
    Model
        Entity saved.
    """
    obj = self.to_dict()

    if not update:
        new_obj = create_entity_api_ctx(self.project, self.ENTITY_TYPE, obj)
        self._update_attributes(new_obj)
        return self

    self.metadata.updated = obj["metadata"]["updated"] = get_timestamp()
    new_obj = update_entity_api_ctx(self.project, self.ENTITY_TYPE, self.id, obj)
    self._update_attributes(new_obj)
    return self

upload(source=None)

Upload model to remote storage from given local path to spec path destination.

Parameters:

Name Type Description Default
source str

Source path is the local path of the model.

 None

Returns:

Type Description
str

Path of the uploaded model.
Source code in digitalhub_ml/entities/models/entity.py 
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
def upload(self, source: str | None = None) -> str:
    """
    Upload model to remote storage from given local path to
    spec path destination.

    Parameters
    ----------
    source : str
        Source path is the local path of the model.

    Returns
    -------
    str
        Path of the uploaded model.
    """
    # Check if target path is remote
    path = self.spec.path
    store = get_store(path)
    if store.is_local():
        raise RuntimeError("Only remote paths are supported for upload.")

    # Check if source path is provided and if it is local
    src = source if source is not None else self.spec.src_path
    self._check_local(src)

    # Get store and upload model and return remote path
    target = store.upload(src, path)
    file_info = store.get_file_info(target, src)

    if file_info is not None:
        self.refresh()
        self.status.add_file(file_info)
        self.save(update=True)

    return target

model_from_dict(obj)

Create Model instance from a dictionary.

Parameters:

Name Type Description Default
obj dict

Dictionary to create object from.

 required

Returns:

Type Description
Model

Model instance.
Source code in digitalhub_ml/entities/models/entity.py 
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
def model_from_dict(obj: dict) -> Model:
    """
    Create Model instance from a dictionary.

    Parameters
    ----------
    obj : dict
        Dictionary to create object from.

    Returns
    -------
    Model
        Model instance.
    """
    return Model.from_dict(obj)

model_from_parameters(project, name, kind, uuid=None, description=None, git_source=None, labels=None, embedded=True, path=None, framework=None, algorithm=None, **kwargs)

Create a new Model instance with the specified parameters.

Parameters:

Name Type Description Default
project str

Project name.

 required
name str

Object name.

 required
kind str

Kind the object.

 required
uuid str

ID of the object (UUID4).

 None
git_source str

Remote git source for object.

 None
labels list[str]

List of labels.

 None
description str

Description of the object (human readable).

 None
embedded bool

Flag to determine if object must be embedded in project.

 True
path str

Object path on local file system or remote storage. If not provided, it's generated.

 None
framework str

Model framework (e.g. 'pytorch').

 None
algorithm str

Model algorithm (e.g. 'resnet').

 None
**kwargs dict

Spec keyword arguments.

 {}

Returns:

Type Description
Model

An instance of the created model.
Source code in digitalhub_ml/entities/models/entity.py 
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
392
393
394
395
396
397
398
399
400
401
402
403
404
def model_from_parameters(
    project: str,
    name: str,
    kind: str,
    uuid: str | None = None,
    description: str | None = None,
    git_source: str | None = None,
    labels: list[str] | None = None,
    embedded: bool = True,
    path: str | None = None,
    framework: str | None = None,
    algorithm: str | None = None,
    **kwargs,
) -> Model:
    """
    Create a new Model instance with the specified parameters.

    Parameters
    ----------
    project : str
        Project name.
    name : str
        Object name.
    kind : str
        Kind the object.
    uuid : str
        ID of the object (UUID4).
    git_source : str
        Remote git source for object.
    labels : list[str]
        List of labels.
    description : str
        Description of the object (human readable).
    embedded : bool
        Flag to determine if object must be embedded in project.
    path : str
        Object path on local file system or remote storage.
        If not provided, it's generated.
    framework : str
        Model framework (e.g. 'pytorch').
    algorithm : str
        Model algorithm (e.g. 'resnet').
    **kwargs : dict
        Spec keyword arguments.

    Returns
    -------
    Model
        An instance of the created model.
    """
    if path is None:
        raise EntityError("Path must be provided.")
    name = build_name(name)
    uuid = build_uuid(uuid)
    metadata = build_metadata(
        kind,
        project=project,
        name=name,
        version=uuid,
        description=description,
        source=git_source,
        labels=labels,
        embedded=embedded,
    )
    spec = build_spec(
        kind,
        path=path,
        framework=framework,
        algorithm=algorithm,
        **kwargs,
    )
    status = build_status(kind)
    return Model(
        project=project,
        name=name,
        uuid=uuid,
        kind=kind,
        metadata=metadata,
        spec=spec,
        status=status,
    )