custom_client_server.py

"Client-server interface custom implementation for filter models."

import json
import concrete.numpy as cnp
from filters import Filter

from concrete.ml.deployment import FHEModelClient
from concrete.ml.version import __version__ as CML_VERSION


class CustomFHEClient(FHEModelClient):
    """Client interface to encrypt and decrypt FHE data associated to a Filter."""

    def load(self):
        """Load the parameters along with the FHE specs."""

        # Load the client
        self.client = cnp.Client.load(self.path_dir / "client.zip", self.key_dir)

        # Load the filter's parameters from the json file
        with (self.path_dir / "serialized_processing.json").open("r", encoding="utf-8") as f:
            serialized_processing = json.load(f)

        # Make sure the version in serialized_model is the same as CML_VERSION
        assert serialized_processing["cml_version"] == CML_VERSION, (
            f"The version of Concrete ML library ({CML_VERSION}) is different "
            f"from the one used to save the model ({serialized_processing['cml_version']}). "
            "Please update to the proper Concrete ML version.",
        )

        # Initialize the filter model using its filter name
        filter_name = serialized_processing["model_post_processing_params"]["filter_name"]
        self.model = Filter(filter_name)

        return self.model