base.py

from typing import Optional
import pydantic as dt
import json
from dgl import DGLError

class PartitionMeta(dt.BaseModel):
    """ Metadata that describes the partition assignment results.

    Regardless of the choice of partitioning algorithm, a metadata JSON file
    will be created in the output directory which includes the meta information
    of the partition algorithm.

    To generate a metadata JSON:

    >>> part_meta = PartitionMeta(version='1.0.0', num_parts=4, algo_name='random')
    >>> with open('metadata.json', 'w') as f:
    ...     json.dump(part_meta.dict(), f)

    To read a metadata JSON:

    >>> with open('metadata.json') as f:
    ...     part_meta = PartitionMeta(**(json.load(f)))

    """
    # version of metadata JSON.
    version: Optional[str] = '1.0.0'
    # number of partitions.
    num_parts: int
    # name of partition algorithm.
    algo_name: str

def dump_partition_meta(part_meta, meta_file):
    """ Dump partition metadata into json file.

    Parameters
    ----------
    part_meta : PartitionMeta
        The partition metadata.
    meta_file : str
        The target file to save data.
    """
    with open(meta_file, 'w') as f:
        json.dump(part_meta.dict(), f, sort_keys=True, indent=4)

def load_partition_meta(meta_file):
    """ Load partition metadata and do sanity check.

    Parameters
    ----------
    meta_file : str
        The path of the partition metadata file.

    Returns
    -------
    PartitionMeta
        The partition metadata.
    """
    with open(meta_file) as f:
        try:
            part_meta = PartitionMeta(**(json.load(f)))
        except dt.ValidationError as e:
            raise DGLError(
                f"Invalid partition metadata JSON. Error details: {e.json()}")
        if part_meta.version != '1.0.0':
            raise DGLError(
                f"Invalid version[{part_meta.version}]. Supported versions: '1.0.0'")
        if part_meta.num_parts <= 0:
            raise DGLError(
                f"num_parts[{part_meta.num_parts}] should be greater than 0.")
        if part_meta.algo_name not in ['random', 'metis']:
            raise DGLError(
                f"algo_name[{part_meta.num_parts}] is not supported.")
        return part_meta