Serialization of KItems/KTypes

dsms/knowledge/kitem.py

@@ -74,6 +74,12 @@
 logger.addHandler(handler)
 logger.propagate = False
 
+class Format(Enum):
+    """Data formats"""
+
+    JSON = "json"
+    YAML = "yaml"
+    HDF5 = "hdf5"
 
 class KItem(BaseModel):
     """
@@ -655,3 +661,18 @@ def is_a(self, to_be_compared: KType) -> bool:
     def refresh(self) -> None:
         """Refresh the KItem"""
         _refresh_kitem(self)
+
+    def export(self, format: Format) -> Any:
+        """Export kitems to different formats"""
+
+        if format == Format.HDF5:
+            from dsms.knowledge.kitem_wrapper import to_hdf5
+            return to_hdf5(self)
+
+        elif format == Format.JSON:
+            # need to implement
+            return
+
+        elif format == Format.YAML:
+            # need to implement
+            return

dsms/knowledge/kitem_wrapper.py

@@ -0,0 +1,165 @@
+import io
+import h5py
+import numpy as np
+
+def to_hdf5(kItem) -> io.BytesIO:
+    """Export KItem to HDF5"""
+
+    data_bytes = io.BytesIO()
+    # with tempfile.NamedTemporaryFile(delete=False) as temp_file:
+    with h5py.File(data_bytes, 'w') as hdf:
+
+        # Store top-level attributes
+        keys = ['name', 'id', 'ktype_id', 'in_backend', 'slug', 'avatar_exists', 'created_at', 'updated_at', 'rdf_exists', 'context_id', 'access_url']
+        for key in keys:
+            value = getattr(kItem, key)
+            create_dataset(key, value, hdf)
+
+        # Store the summary
+        summary = getattr(kItem, 'summary')
+        if summary is not None:
+            if summary.text is not None:
+                hdf.create_dataset('summary', data = summary.text)
+
+        # Store dataframe
+        dataframe = getattr(kItem, 'dataframe')
+        if dataframe is not None:
+            value = dataframe.to_df().to_json()
+            hdf.create_dataset('dataframe', data = value)
+
+        # Store the avatar in binary
+        avatar = getattr(kItem, 'avatar')
+        if avatar is not None:
+            # Get the image
+            image = avatar.download()
+
+            # Create a BytesIO object and save the image to it
+            image_bytes = io.BytesIO()
+            image.save(image_bytes, format='PNG')
+
+            # Get the bytes value
+            value = image_bytes.getvalue()
+            img_arr = np.frombuffer(value, dtype=np.uint8)
+            hdf.create_dataset('avatar', data=img_arr, dtype=img_arr.dtype)
+
+        # Store the subgraph after serialization
+        subgraph = getattr(kItem, 'subgraph')
+        if subgraph is not None:
+            value = subgraph.serialize()
+            hdf.create_dataset('subgraph', data = value)  
+
+        # Store annotations
+        annotations_group = hdf.create_group('annotations')
+        for i, annotation in enumerate(getattr(kItem, 'annotations')):
+            annotation_group = annotations_group.create_group(f'annotation_{i}')
+            for key, value in annotation:
+                create_dataset(key, value, annotation_group)
+
+        # Store attachments
+        attachments_group = hdf.create_group('attachments')
+        for i, attachment in enumerate(getattr(kItem,'attachments')):
+            attachment_group = attachments_group.create_group(f'attachment_{i}')
+            for key, value in attachment:
+                if key == 'content':
+                    value = attachment.download().encode()
+                    binary_data = np.frombuffer(value, dtype='uint8')
+                    attachment_group.create_dataset(key, data=binary_data, dtype=binary_data.dtype)
+                else:
+                    create_dataset(key, value, attachment_group)
+
+        # Store linked_kitems
+        linked_kitems_group = hdf.create_group('linked_kitems')
+        for i, linked_kitem in enumerate(getattr(kItem,'linked_kitems')):
+            linked_kitem_group = linked_kitems_group.create_group(f'linked_kitem_{i}')
+            for key in ['id', 'name', 'slug', 'ktype_id']:
+                value = getattr(linked_kitem, key)
+                create_dataset(key, value, linked_kitem_group)
+
+        # Store affiliations
+        affiliations_group = hdf.create_group('affiliations')
+        for i, affiliation in enumerate(getattr(kItem,'affiliations')):
+            affiliation_group = affiliations_group.create_group(f'affiliation_{i}')
+            for key, value in affiliation:
+                create_dataset(key, value, affiliation_group)
+
+        # Store authors
+        authors_group = hdf.create_group('authors')
+        for i, author in enumerate(getattr(kItem,'authors')):
+            author_group = authors_group.create_group(f'author_{i}')
+            for key, value in author:
+                create_dataset(key, value, author_group)
+
+        # Store contacts
+        contacts_group = hdf.create_group('contacts')
+        for i, contact in enumerate(getattr(kItem,'contacts')):
+            contact_group = contacts_group.create_group(f'contact_{i}')
+            for key, value in contact:
+                create_dataset(key, value, contact_group)
+
+        # Store external links
+        external_links_group = hdf.create_group('external_links')
+        for i, external_link in enumerate(getattr(kItem,'external_links')):
+            external_link_group = external_links_group.create_group(f'external_link_{i}')
+            for key, value in external_link:
+                create_dataset(key, value, external_link_group)
+
+        # Store kitem_apps
+        kitem_apps_group = hdf.create_group('kitem_apps')
+        for i, app in enumerate(getattr(kItem,'kitem_apps')):
+            app_group = kitem_apps_group.create_group(f'app_{i}')
+            for key, value in app:
+                if key == 'additional_properties':
+                    for prop_key, prop_value in value:
+                        app_group.create_dataset(f'additional_properties/{prop_key}', data=prop_value)
+                else:
+                    create_dataset(key, value, app_group)
+
+        # Store user groups
+        user_groups_group = hdf.create_group('user_groups')
+        for i, user_group in enumerate(getattr(kItem,'user_groups')):
+            user_group_group = user_groups_group.create_group(f'user_group_{i}')
+            for key, value in user_group:
+                create_dataset(key, value, user_group_group)
+
+        # Store custom_properties
+        from dsms.knowledge.webform import KItemCustomPropertiesModel
+        custom_properties_group = hdf.create_group('custom_properties')
+        for item in kItem:
+            if 'custom_properties' in item:
+                break
+        for custom_property in item:
+
+            if isinstance(custom_property, KItemCustomPropertiesModel):
+                sections_group = custom_properties_group.create_group('sections')
+                for i, section in enumerate(custom_property):
+                    section_group = sections_group.create_group(f'section_{i}')
+                    section_group.create_dataset('id', data=section.id)
+                    section_group.create_dataset('name', data=section.name)
+                    entries_group = section_group.create_group('entries')
+
+                    for j, entry in enumerate(section):
+                        entry_group = entries_group.create_group(f'entry_{j}')
+                        entry_keys = ['measurement_unit', 'relation_mapping']
+                        for entry_key, entry_value in entry:
+                            if entry_key == 'kitem':
+                                continue
+                            elif entry_key in entry_keys and entry_value is not None:
+                                    group = entry_group.create_group(entry_key)
+                                    for key_, value_ in entry_value:
+                                        if key_ == 'kitem':
+                                            continue
+                                        create_dataset(key_, value_, group)
+                            else:
+                                create_dataset(key, value, entry_group)
+
+    return data_bytes
+
+
+def create_dataset(key, value, group):
+    """Create dataset depending on the type of the data"""
+
+    basic_types = (int, float, str, bool, list, tuple, dict, set)
+    if isinstance(value, basic_types):
+        group.create_dataset(key, data=value)
+    else:
+        group.create_dataset(key, data=str(value))

dsms/knowledge/ktype.py

@@ -4,6 +4,7 @@
 from datetime import datetime
 from typing import TYPE_CHECKING, Any, Optional, Union
 from uuid import UUID
+from enum import Enum
 
 from pydantic import BaseModel, Field, model_serializer
 
@@ -19,6 +20,12 @@
 logger.addHandler(handler)
 logger.propagate = False
 
+class Format(Enum):
+    """Data formats"""
+
+    JSON = "json"
+    YAML = "yaml"
+    HDF5 = "hdf5"
 
 class KType(BaseModel):
     """Knowledge type of the knowledge item."""
@@ -137,3 +144,19 @@ def serialize(self):
             )
             for key, value in self.__dict__.items()
         }
+
+    def export(self, format: Format) -> Any:
+        """Export ktypes to different formats"""
+
+        if format == Format.HDF5:
+            from dsms.knowledge.ktype_wrapper import to_hdf5
+            return to_hdf5(self)
+
+
+        elif format == Format.JSON:
+            # need to implement
+            return
+
+        elif format == Format.YAML:
+            # need to implement
+            return

dsms/knowledge/ktype_wrapper.py

@@ -0,0 +1,67 @@
+import io
+import h5py
+import numpy as np
+
+def to_hdf5(ktype) -> io.BytesIO:
+
+    data_bytes = io.BytesIO()
+    with h5py.File(data_bytes, 'w') as hdf:
+
+         # Store top-level attributes
+        keys = ['id', 'name', 'created_at', 'updated_at']
+        for key in keys:
+            value = getattr(ktype, key)
+            create_dataset(key, value, hdf)
+
+        # Store the Webform
+        webform = getattr(ktype, 'webform')
+        webform_group = hdf.create_group('webform')
+        if webform is not None:
+            sections_group = webform_group.create_group('sections')
+            section_keys = ['id', 'name', 'hidden']
+            input_keys = ['measurement_unit', 'relation_mapping', 'relation_mapping_extra', 'range_options']
+            for webform_key, webform_value in webform:
+                if webform_key == 'kitem':
+                    continue
+                elif webform_key == 'sections':
+                    for i, section in enumerate(webform_value):
+                        section_group = sections_group.create_group(f'section_{i}')
+                        for section_key in section_keys:
+                            section_value = getattr(section, section_key)
+                            create_dataset(section_key, section_value, section_group)
+
+                        inputs_group = section_group.create_group('inputs')
+
+                        for j, input in enumerate(section.inputs):
+                            input_group = inputs_group.create_group(f'input_{j}')
+                            for input_key, input_value in input:
+                                if input_key == 'kitem':
+                                    continue
+                                elif input_key == 'select_options':
+                                    select_options_group = input_group.create_group('select_options')
+                                    for k, select_option in enumerate(input_value):
+                                        select_option_group = select_options_group.create_group(f'option_{k}')
+                                        for option_key, option_value in select_option:                                            
+                                            create_dataset(option_key, option_value, select_option_group)
+                                elif input_key in input_keys and input_value is not None:
+                                    group = input_group.create_group(input_key)
+                                    for key_, value_ in input_value:
+                                        if key_ == 'kitem':
+                                            continue
+                                        create_dataset(key_, value_, group)
+                                else:
+                                    create_dataset(input_key, input_value, input_group)
+
+                else:
+                    create_dataset(webform_key, webform_value, webform_group)
+
+    return data_bytes
+
+def create_dataset(key, value, group):
+    """Create dataset depending on the type of the data"""
+
+    basic_types = (int, float, str, bool, list, tuple, dict, set)
+    if isinstance(value, basic_types):
+        group.create_dataset(key, data=value)
+    else:
+        group.create_dataset(key, data=str(value))