Simple Serialization for Nested Parameterized Objects

serialize_param.py

import param as pm
import pandas as pd

def serialize_param(obj: pm.Parameterized, skip_dataframes=True, exclude_params=None):
    """
    Recursively serialize a Parameterized object into a nested dictionary.
    
    Parameters:
    - obj: pm.Parameterized
        The object to serialize.
    - skip_dataframes: bool
        Whether to skip DataFrame parameters (default: True).
    - exclude_params: list
        A list of parameter names to exclude from serialization (default: None).
    
    Returns:
    - dict: A nested dictionary representation of the Parameterized object.
    """
    if exclude_params is None:
        exclude_params = []

    serialized_data = {}

    for param_name, param_value in obj.param.values().items():
        param_obj = obj.param[param_name]
        
        # Skip excluded parameters
        if param_name in exclude_params:
            continue
        
        # Optionally skip DataFrame parameters
        if skip_dataframes and isinstance(param_value, pd.DataFrame):
            continue
        
        if isinstance(param_value, pm.Parameterized):
            # Recursively serialize nested Parameterized objects
            serialized_data[param_name] = serialize_param(param_value, skip_dataframes, exclude_params)
        
        elif isinstance(param_obj, pm.List) and isinstance(param_value, list):
            # Recursively serialize lists of Parameterized objects
            serialized_data[param_name] = [
                serialize_param(item, skip_dataframes, exclude_params) if isinstance(item, pm.Parameterized) else item
                for item in param_value
            ]
        
        else:
            # For basic types, just store the value directly
            serialized_data[param_name] = param_value

    return serialized_data


def deserialize_param(cls, data):
    """
    Recursively instantiate Parameterized classes from nested data.
    """
    if isinstance(data, list):
        return [instantiate_from_data(cls, item) for item in data]
    elif isinstance(data, dict):
        init_data = {}
        for key, value in data.items():
            param_obj = cls.param[key]
            # Check if the parameter is a ClassSelector or List and process accordingly
            if isinstance(param_obj, pm.ClassSelector):
                init_data[key] = instantiate_from_data(param_obj.class_, value)
            elif isinstance(param_obj, pm.List) and isinstance(value, list):
                # For List parameters, instantiate each item in the list
                init_data[key] = [deserialize_param(param_obj.item_type, v) for v in value]
            else:
                init_data[key] = value
        return cls(**init_data)
    else:
        return data


def serialize_param_flat(obj: pm.Parameterized, prefix='', skip_dataframes=True, exclude_params=None):
    flat_dict = {}
    exclude_params = exclude_params or []

    for param_name, param_value in obj.param.values().items():
        if param_name in exclude_params:
            continue

        new_prefix = f"{prefix}{param_name}."

        # Skip DataFrame parameters if required
        if skip_dataframes and isinstance(param_value, pd.DataFrame):
            continue

        # If the parameter value is a Parameterized object
        if isinstance(param_value, pm.Parameterized):
            flat_dict.update(serialize_param_flat(param_value, prefix=new_prefix, skip_dataframes=skip_dataframes, exclude_params=exclude_params))
        
        # If the parameter value is a list of Parameterized objects
        elif isinstance(param_value, list) and all(isinstance(item, pm.Parameterized) for item in param_value):
            for i, item in enumerate(param_value):
                item_prefix = f"{new_prefix}{i}."
                flat_dict.update(serialize_param_flat(item, prefix=item_prefix, skip_dataframes=skip_dataframes, exclude_params=exclude_params))
        
        # Otherwise, just add the value to the flat dict
        else:
            flat_dict[f"{prefix}{param_name}"] = param_value

    return flat_dict


# Example usage of serialization and deserialization
class Item(pm.Parameterized):
    name = pm.String(default="Item")
    value = pm.Number(default=0)

class Container(pm.Parameterized):
    items = pm.List(item_type=Item)

    def add_item(self, name, value):
        self.items.append(Item(name=name, value=value))


# Example instance creation
container = Container()
container.add_item("Item 1", 100)
container.add_item("Item 2", 200)

# Serialize the Container
data = serialize_param(container)
print("Serialized Nested Data:")
print(data)

# Verify the restored container
print("\nRestored Nested Data:")
print(serialize_param(deserialize_param(Container, data)))

# Flat Serialize the Container
flat_data = serialize_param_flat(container)
print("\nSerialized Flat Data:")
print(flat_data)

outputs = """
Serialized Nested Data:
{'items': [{'name': 'Item 1', 'value': 100}, {'name': 'Item 2', 'value': 200}], 'name': 'Container00050'}

Restored Nested Data:
{'items': [{'name': 'Item 1', 'value': 100}, {'name': 'Item 2', 'value': 200}], 'name': 'Container00050'}

Serialized Flat Data:
{'items.0.name': 'Item 1', 'items.0.value': 100, 'items.1.name': 'Item 2', 'items.1.value': 200, 'name': 'Container00050'}
"""