How to export an instance of ADWIN class in JSON

[LorenzoCutrupi]

Hi,
I'm working on a project where I have to export a working instance of Adwin so that I can later import it again. Is it possible to do so? I'm having troubles doing it, as in particular I think it's necessary to save the buckets to preserve the perfect portability of the instance.
I've read the documentation at https://riverml.xyz/0.18.0/faq/#how-do-i-save-and-load-models where it's suggested to use pickle. Unfortunately I am restricted to export and import in json format and can't save bytestream. Is there any solution?
Please help me :c
If you need some extra information ask

[MaxHalford]

Hey there. I'm curious, why are you limited to not using a byte stream?

[LorenzoCutrupi]

I'm working for a company that is very strict in the type of elements that can be saved in their db, and the only format allowed is .json

[MaxHalford]

In an ideal world, River class should be JSON-serializable. For instance, we could that by using Pydantic everywhere. Alas, we're not there yet.

What you could do in the meantime is write a custom JSON serializer/parser. This would be basically mean converting ADWIN to a dict, and also instantiate ADWIN from a dict. There's some good examples here. Do you see what I mean?

[LorenzoCutrupi]

Ye actually today I wrote manually some functions to make adwin serializable. In practice I modified 4 of the original classes from the library and added the methods from_dict and to_dict to export and import an instance correctly. If it can be useful I can write the code I wrote or contribute myself to the library (I never did a contribution yet but that sounds cool)

[MaxHalford]

I think it's cool to share it here. But I wouldn't add it to River yet. If we solve JSON serialization, we should do it all in one go, not just for a few classes.

[LorenzoCutrupi]

To accomplish my goal I modified the original library, in particular editing four classes:

I modified the class Base from river/base/base.py adding these two methods:

def to_dict(self):
     """
     Convert the Base instance to a dictionary.

     Returns:
         dict: A dictionary representation of the Base instance.
     """
     clone_instance = self.clone(include_attributes=True)
     params = clone_instance._get_params()
     return params

 @classmethod
 def from_dict(cls, data):
     """
     Create a Base instance from a dictionary.

     Args:
         data (dict): Dictionary containing the serialized Base data.

     Returns:
         Base: A new Base instance.
     """
     instance = cls()
     instance.mutate(data)
     return instance

Then I modified two classes within river/base/drift_detector.py :

class _BaseDriftDetector(Base):
    ...old methods...

    def to_dict(self):
        return {
            "_drift_detected": self._drift_detected,
        }

    @classmethod
    def from_dict(cls, data):
        instance = cls()
        instance.mutate(data)
        return instance

class DriftDetector(_BaseDriftDetector):
    ...old methods...

    def to_dict(self):
        base_dict = super().to_dict()
        # Add specific attributes for DriftDetector if needed
        return base_dict

    @classmethod
    def from_dict(cls, data):
        instance = cls()
        instance.mutate(data)
        return instance

Then in river/drift/adwin_c.pyx I added these methods in the AdaptiveWindowing class:

def to_dict(self):
        """
        Convert the AdaptiveWindowing instance to a dictionary.

        Returns:
            dict: A dictionary representation of the AdaptiveWindowing instance.
        """
        buckets_data = [bucket.to_dict() for bucket in self.bucket_deque]

        return {
            "delta": self.delta,
            "bucket_deque": buckets_data,
            "total": self.total,
            "variance": self.variance,
            "width": self.width,
            "n_buckets": self.n_buckets,
            "grace_period": self.grace_period,
            "tick": self.tick,
            "total_width": self.total_width,
            "n_detections": self.n_detections,
            "clock": self.clock,
            "max_n_buckets": self.max_n_buckets,
            "min_window_length": self.min_window_length,
            "max_buckets": self.max_buckets
        }

@classmethod
def from_dict(cls, data):
    """
    Create an AdaptiveWindowing instance from a dictionary.

    Args:
        data (dict): Dictionary containing the serialized AdaptiveWindowing data.

    Returns:
        AdaptiveWindowing: A new AdaptiveWindowing instance.
    """
    windowing = cls(
        delta=data["delta"],
        clock=data["clock"],
        max_buckets=data["max_buckets"],
        min_window_length=data["min_window_length"],
        grace_period=data["grace_period"]
    )
    windowing.bucket_deque = deque([Bucket.from_dict(bucket_data) for bucket_data in data["bucket_deque"]])
    windowing.total = data["total"]
    windowing.variance = data["variance"]
    windowing.width = data["width"]
    windowing.n_buckets = data["n_buckets"]
    windowing.tick = data["tick"]
    windowing.total_width = data["total_width"]
    windowing.n_detections = data["n_detections"]
    windowing.max_n_buckets = data["max_n_buckets"]

    return windowing

(keep in mind that the original code was in cython while I rewrote in in python and added these methods).
Finally I added other methods to the class ADWIN of river/drift/adwin.py:

def to_dict(self):
       """
       Convert the ADWIN instance to a dictionary.

       Returns:
           dict: A dictionary representation of the ADWIN instance.
       """
       return {
           "delta": self.delta,
           "clock": self.clock,
           "max_buckets": self.max_buckets,
           "min_window_length": self.min_window_length,
           "grace_period": self.grace_period,
           "_helper": self._helper.to_dict(),
           # Add other key-value pairs for the state of ADWIN
       }

@classmethod
def from_dict(cls, data):
   """
   Create an ADWIN instance from a dictionary.

   Args:
       data (dict): Dictionary containing the serialized ADWIN data.

   Returns:
       ADWIN: A new ADWIN instance.
   """
   adwin = cls(
       delta=data["delta"],
       clock=data["clock"],
       max_buckets=data["max_buckets"],
       min_window_length=data["min_window_length"],
       grace_period=data["grace_period"]
   )
   adwin._helper = AdaptiveWindowing.from_dict(data["_helper"])
   # Initialize other state of ADWIN using data dictionary
   return adwin