Feature/ml streaming base (#115)

* basic definition of `Communicator` interface

* basic definition of `Compressor` interface

* basic definition of `Encryptor` interface

* `Compressor`, `Encryptor` and `Communicator` interfaces

* `PymiloClient` basic init

* refactor list deserialization

* add "\n" between functions and class declaration

* update interfaces, make Communicator interface stateful

* implement `RESTClientCommunicator`

* implement `RESTServerCommunicator`

* remove `DummyCommunicator`

* implement PyMilo REST Client

* implement PyMilo REST Server

* add `streaming` folder to pymilo package installations

* add script to run a test pymilo REST server

* implement a cohesive scenario1 as a test for ml streaming feature

* implement a cohesive scenario2 as a test for ml streaming feature

* add fastapi and uvicorn to requirments

* add requests to requirments

* update threshold to be able to have uvicorn in python 3.6

* add timeout to post & get calls

* `CHANGELOG.md` updated

* add `__init__.py` to `streaming` module

* split pymilo installation into 2 different modes, `basic` mode not having ml streaming features and `streaming` mode as a full version

* update docstring in `setup.py`

* add `__doc__` to attribute return + some refactorings applied

* rename `Communicator` interface to `ClientCommunicator`

* install the full version of pymilo in workflow

* remove trailing whitespaces + unused lines

* replace type check with type() with isinstance()

* remove whitespaces

* fix 500 error for /docs and /redoc

* make `ml streaming` testcases `deployable` next to other pytest testcases

* downgrade uvicorn + fastapi versions to fix python3.7 issue

* fix python3.7 and python3.6 issue in windows

* remove body modifier Middleware

* add `parser` instead of body modifier Middleware

* add session + Retry policy in `RESTClientCommunicator`

* replace `python` with `executable`

* remove unnecessary `pass` from @AbstractMethod

* rewrite requirements installation

* enhance the cohesiveness of the `execute_model` function output

* add dictionary mode to handle different installations

* replace return "Error" with raising exception

* update `toggle_mode` syntax

Author: AHReccese

.github/workflows/test.yml

@@ -26,7 +26,7 @@ jobs:
       - name: Installation
         run: |
           python -m pip install --upgrade pip
-          pip install .
+          pip install .["streaming"]
       - name: Test requirements Installation
         run: |
           python otherfiles/requirements-splitter.py

CHANGELOG.md

@@ -6,7 +6,19 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 ## [Unreleased]
 ### Added
+- `PymiloServer` class in `streaming.pymilo_server.py`
+- `PymiloClient` class in `streaming.pymilo_client.py`
+- `Communicator` interface in `streaming.interfaces.py`
+- `RESTClientCommunicator` class in `streaming.communicator.py`
+- `RESTServerCommunicator` class in `streaming.communicator.py`
+- `Compressor` interface in `streaming.interfaces.py`
+- `DummyCompressor` class in `streaming.compressor.py`
+- `Encryptor` interface in `streaming.interfaces.py`
+- `DummyEncryptor` class in `streaming.encryptor.py`
+- `ML Streaming` RESTful testcases
 ### Changed
+- `serialize` function in `GeneralDataStructureTransporter` Transporter refactored
+- `get_deserialized_list` function in `GeneralDataStructureTransporter` Transporter refactored
 - `Export` class call by reference bug fixed
 ## [0.9] - 2024-07-01
 ### Added

dev-requirements.txt

@@ -1,7 +1,9 @@
 numpy==2.0.1
 scikit-learn==1.5.1
 scipy>=0.19.1
-requests>=2.0.0
+uvicorn==0.14.0
+fastapi==0.68.0
+requests==2.0.0
 setuptools>=40.8.0
 vulture>=1.0
 bandit>=1.5.1

pymilo/streaming/__init__.py

@@ -0,0 +1,2 @@
+# -*- coding: utf-8 -*-
+"""PyMilo ML Streaming."""

pymilo/streaming/communicator.py

@@ -0,0 +1,101 @@
+import uvicorn
+import requests
+from fastapi import FastAPI, Request
+from pydantic import BaseModel
+from .interfaces import ClientCommunicator
+
+
+class RESTClientCommunicator(ClientCommunicator):
+
+    def __init__(self, server_url):
+        self._server_url = server_url
+        self.session = requests.Session()
+        retries = requests.adapters.Retry(
+            total=5,
+            backoff_factor=0.1,
+            status_forcelist=[500, 502, 503, 504]
+        )
+        self.session.mount('http://', requests.adapters.HTTPAdapter(max_retries=retries))
+        self.session.mount('https://', requests.adapters.HTTPAdapter(max_retries=retries))
+
+    def download(self, payload):
+        return self.session.get(url=self._server_url + "/download/", json=payload, timeout=5)
+
+    def upload(self, payload):
+        return self.session.post(url=self._server_url + "/upload/", json=payload, timeout=5)
+
+    def attribute_call(self, payload):
+        return self.session.post(url=self._server_url + "/attribute_call/", json=payload, timeout=5)
+
+
+
+class RESTServerCommunicator():
+
+    def __init__(
+            self,
+            ps,
+            host: str = "127.0.0.1",
+            port: int = 8000,
+            ):
+        self.app = FastAPI()
+        self.host = host
+        self.port = port
+        self._ps = ps
+        self.setup_routes()
+
+    def setup_routes(self):
+        class StandardPayload(BaseModel):
+            client_id: str
+            model_id: str
+        class DownloadPayload(StandardPayload):
+            pass
+        class UploadPayload(StandardPayload):
+            model: str
+        class AttributePayload(StandardPayload):
+            attribute: str
+            args: list
+            kwargs: dict
+
+        @self.app.get("/download/")
+        async def download(request: Request):
+            body = await request.json()
+            body = self.parse(body)
+            payload = DownloadPayload(**body)
+            message = "/download request from client: {} for model: {}".format(payload.client_id, payload.model_id)
+            return {
+                "message": message,
+                "payload": self._ps.export_model(),
+            }
+
+        @self.app.post("/upload/")
+        async def upload(request: Request):
+            body = await request.json()
+            body = self.parse(body)
+            payload = UploadPayload(**body)
+            message = "/upload request from client: {} for model: {}".format(payload.client_id, payload.model_id)
+            return {
+                "message": message,
+                "payload": self._ps.update_model(payload.model)
+            }
+
+        @self.app.post("/attribute_call/")
+        async def attribute_call(request: Request):
+            body = await request.json()
+            body = self.parse(body)
+            payload = AttributePayload(**body)
+            message = "/attribute_call request from client: {} for model: {}".format(payload.client_id, payload.model_id)
+            result = self._ps.execute_model(payload)
+            return {
+                "message": message,
+                "payload": result if result is not None else "The ML model has been updated in place."
+            }
+
+    def parse(self, body):
+        return self._ps._compressor.extract(
+            self._ps._encryptor.decrypt(
+                body
+            )
+        )
+
+    def run(self):
+        uvicorn.run(self.app, host=self.host, port=self.port)

pymilo/streaming/compressor.py

@@ -0,0 +1,12 @@
+from .interfaces import Compressor
+
+
+class DummyCompressor(Compressor):
+
+    @staticmethod
+    def compress(payload):
+        return payload
+
+    @staticmethod
+    def extract(payload):
+        return payload

pymilo/streaming/encryptor.py

@@ -0,0 +1,12 @@
+from .interfaces import Encryptor
+
+
+class DummyEncryptor(Encryptor):
+
+    @staticmethod
+    def encrypt(payload):
+        return payload
+
+    @staticmethod
+    def decrypt(payload):
+        return payload

pymilo/streaming/interfaces.py

@@ -0,0 +1,47 @@
+# -*- coding: utf-8 -*-
+"""PyMilo ML Streaming Interfaces."""
+from abc import ABC, abstractmethod
+
+
+class Compressor(ABC):
+
+    @abstractmethod
+    def compress(payload):
+        """
+        """
+
+    @abstractmethod
+    def extract(payload):
+        """
+        """
+
+
+class Encryptor(ABC):
+
+    @abstractmethod
+    def encrypt(payload):
+        """
+        """
+
+    @abstractmethod
+    def decrypt(payload):
+        """
+        """
+
+
+class ClientCommunicator(ABC):
+
+    @abstractmethod
+    def upload(self, payload):
+        """
+        """
+
+    @abstractmethod
+    def download(self, payload):
+        """
+        """
+
+    @abstractmethod
+    def attribute_call(self, payload):
+        """
+        """

pymilo/streaming/pymilo_client.py

@@ -0,0 +1,82 @@
+from .encryptor import DummyEncryptor
+from .compressor import DummyCompressor
+from ..pymilo_obj import Export, Import
+from .communicator import RESTClientCommunicator
+from ..transporters.general_data_structure_transporter import GeneralDataStructureTransporter
+class PymiloClient:
+
+    def __init__(
+            self,
+            model=None,
+            mode="LOCAL",
+            server="http://127.0.0.1",
+            port= 8000
+            ):
+        self._client_id = "0x_client_id"
+        self._model_id = "0x_model_id"
+        self._model = model
+        self._mode = mode
+        self._compressor = DummyCompressor()
+        self._encryptor = DummyEncryptor()
+        self._communicator = RESTClientCommunicator(
+            server_url="{}:{}".format(server, port)
+        )
+
+    def toggle_mode(self, mode="LOCAL"):
+        mode = mode.upper()
+        if mode not in ["LOCAL", "DELEGATE"]:
+            raise Exception("Invalid mode, the given mode should be either `LOCAL`[default] or `DELEGATE`.")
+        self._mode = mode
+
+    def download(self):
+        response = self._communicator.download({
+            "client_id": self._client_id,    
+            "model_id": self._model_id
+        })
+        if response.status_code != 200:
+            print("Remote model download failed.")
+        print("Remote model downloaded successfully.")
+        serialized_model = response.json()["payload"]
+        self._model = Import(file_adr=None, json_dump=serialized_model).to_model()
+        print("Local model updated successfully.")
+
+    def upload(self):
+        response = self._communicator.upload({
+            "client_id": self._client_id,    
+            "model_id": self._model_id,
+            "model": Export(self._model).to_json(),
+        })
+        if response.status_code == 200:
+            print("Local model uploaded successfully.")
+        else:
+            print("Local model upload failed.")
+
+    def __getattr__(self, attribute):
+        if self._mode == "LOCAL":
+            if attribute in dir(self._model):
+                return getattr(self._model, attribute)
+            else:
+                raise AttributeError("This attribute doesn't exist either in PymiloClient or the inner ML model.")
+        elif self._mode == "DELEGATE":
+            gdst = GeneralDataStructureTransporter()
+            def relayer(*args, **kwargs):
+                print(f"Method '{attribute}' called with args: {args} and kwargs: {kwargs}")
+                payload = {
+                    "client_id": self._client_id,                   
+                    "model_id": self._model_id,
+                    'attribute': attribute,
+                    'args': args,
+                    'kwargs': kwargs,
+                    }
+                payload["args"] = gdst.serialize(payload, "args", None)
+                payload["kwargs"] = gdst.serialize(payload, "kwargs", None)
+                result = self._communicator.attribute_call(
+                    self._encryptor.encrypt(
+                        self._compressor.compress(
+                            payload
+                        )
+                    )
+                ).json()
+                return gdst.deserialize(result, "payload", None)
+            relayer.__doc__ = getattr(self._model.__class__, attribute).__doc__
+            return relayer

pymilo/streaming/pymilo_server.py

@@ -0,0 +1,39 @@
+from ..pymilo_obj import Export, Import
+from .compressor import DummyCompressor
+from .encryptor import DummyEncryptor
+from .communicator import RESTServerCommunicator
+from ..transporters.general_data_structure_transporter import GeneralDataStructureTransporter
+
+
+class PymiloServer:
+
+    def __init__(self):
+        self._model = None
+        self._compressor = DummyCompressor()
+        self._encryptor = DummyEncryptor()
+        self._communicator = RESTServerCommunicator(ps=self)
+        self._communicator.run()
+
+    def export_model(self):
+        return Export(self._model).to_json()
+
+    def update_model(self, serialized_model):
+        self._model = Import(file_adr=None, json_dump=serialized_model).to_model()
+
+    def execute_model(self, request):
+        gdst = GeneralDataStructureTransporter()
+        attribute = request.attribute
+        retrieved_attribute = getattr(self._model, attribute, None)
+        if retrieved_attribute is None:
+            raise Exception("The requested attribute doesn't exist in this model.")
+        arguments = {
+            'args': request.args,
+            'kwargs': request.kwargs
+        }
+        args = gdst.deserialize(arguments, 'args', None)
+        kwargs = gdst.deserialize(arguments, 'kwargs', None)
+        output = retrieved_attribute(*args, **kwargs)
+        if isinstance(output, type(self._model)):
+            self._model = output
+            return None
+        return gdst.serialize({'output': output}, 'output', None)