add hdf5 functionality

dsms/knowledge/kitem.py

@@ -3,6 +3,7 @@
 from datetime import datetime
 from enum import Enum
 from typing import TYPE_CHECKING, Any, List, Optional, Union
+from urllib.parse import urljoin
 from uuid import UUID, uuid4
 
 from rdflib import Graph
@@ -33,6 +34,8 @@
     ExternalLink,
     ExternalLinksProperty,
     KProperty,
+    # HDF5Container,
+    Column,
     LinkedKItem,
     LinkedKItemsProperty,
     Summary,
@@ -122,6 +125,8 @@ class KItem(BaseModel):
         None, description="KType of the KItem", exclude=True
     )
 
+    hdf5: Optional[Column] = Field(None, description="")
+
     model_config = ConfigDict(
         extra="forbid",
         validate_assignment=True,
@@ -416,6 +421,8 @@ def context(cls) -> "Context":
         return Context
 
     @property
-    def url(cls) -> Optional[str]:
+    def url(cls) -> str:
         """URL of the KItem"""
-        # Build url with base url, ktype and slug
+        return urljoin(
+            cls.context.dsms.config.host_url, f"{cls.ktype_id}/{cls.slug}"
+        )

dsms/knowledge/properties/__init__.py

@@ -13,13 +13,16 @@
 from dsms.knowledge.properties.base import KProperty, KPropertyItem
 from dsms.knowledge.properties.contacts import ContactInfo, ContactsProperty
 from dsms.knowledge.properties.custom_properties import CustomProperties
+from dsms.knowledge.properties.hdf5 import Column, HDF5Container
 from dsms.knowledge.properties.summary import Summary
 from dsms.knowledge.properties.user_groups import UserGroup, UserGroupsProperty
 
 from dsms.knowledge.properties.attachments import (  # isort:skip
     Attachment,
     AttachmentsProperty,
 )
+
+
 from dsms.knowledge.properties.linked_kitems import (  # isort:skip
     LinkedKItem,
     LinkedKItemsProperty,
@@ -54,4 +57,6 @@
     "Summary",
     "KProperty",
     "KPropertyItem",
+    "HDF5Container",
+    "Column",
 ]

dsms/knowledge/properties/hdf5.py

@@ -0,0 +1,59 @@
+"""HDF5 Properties of a KItem"""
+from typing import TYPE_CHECKING
+
+from pydantic import Field
+
+from dsms.knowledge.properties.base import KProperty, KPropertyItem
+from dsms.knowledge.utils import _get_hdf5_column
+
+if TYPE_CHECKING:
+    from typing import Any, Callable, List
+
+
+class Column(KPropertyItem):
+    """Column of an HDF5 data frame"""
+
+    column_id: int = Field(..., description="Column ID in the data frame")
+
+    name: str = Field(
+        ..., description="Name of the column in the data series."
+    )
+
+    def get(self) -> "List[Any]":
+        """Download the data for the column in a time series"""
+        return _get_hdf5_column(self.id, self.name)
+
+
+class HDF5Container(KProperty):
+    """HDF5 container of a data frame related to a KItem"""
+
+    # OVERRIDE
+    @property
+    def k_property_item(cls) -> "Callable":
+        return Column
+
+    # OVERRIDE
+    def _add(self, item: Column) -> None:
+        """Side effect when an Column is added to the KProperty"""
+        raise NotImplementedError(
+            "Adding columns to an HDF5 container is not supported yet"
+        )
+
+    # OVERRIDE
+    def _update(self, item: Column) -> None:
+        """Side effect when an Column is updated at the KProperty"""
+        raise NotImplementedError(
+            "Updating columns to an HDF5 container is not supported yet"
+        )
+
+    # OVERRIDE
+    def _delete(self, item: Column) -> None:
+        """Side effect when deleting the Column of a KItem"""
+        raise NotImplementedError(
+            "Deleting columns to an HDF5 container is not supported yet"
+        )
+
+    # OVERRIDE
+    def _get(self, item: Column) -> Column:
+        """Side effect when getting the Column for a specfic kitem"""
+        return item

dsms/knowledge/utils.py

@@ -334,3 +334,30 @@ def _slug_is_available(ktype_id: Union[str, UUID], value: str) -> bool:
         f"api/knowledge/kitems/{ktype_id}/{value}", "head"
     )
     return response.status_code == 404
+
+
+def _get_hdf5_column(kitem_id: str, column_id: int) -> List[Any]:
+    """Download the column of a hdf5 container of a certain kitem"""
+    response = _perform_request(
+        f"api/knowledge/data_api/{kitem_id}/column-{column_id}", "get"
+    )
+    if not response.ok:
+        message = f"""Something went wrong fetch column id `{column_id}`
+        for kitem `{kitem_id}`: {response.text}"""
+        raise ValueError(message)
+    return response.json().get("array")
+
+
+def _inspect_hdf5(kitem_id: str) -> List[Dict[str, Any]]:
+    """Get column info for the hdf5 container of a certain kitem"""
+    response = _perform_request(f"api/knowledge/data_api/{kitem_id}", "get")
+    if not response.ok and response.status_code == 404:
+        message = (
+            f"""KItem with id `{kitem_id}` does not own any data frame."""
+        )
+        raise ValueError(message)
+    if not response.ok:
+        message = f"""Something went wrong fetching intospection
+        for kitem `{kitem_id}`: {response.text}"""
+        raise ValueError(message)
+    return response.json().get("array")