Refactorings (#124)

* autopep8.sh + minor refactorings

* add pydantic to `dev` & `streaming` requirements

* minor refactorings

* refactor hardcoded strings to `param.py`

* add docstring to `streaming.param.py`

* increase retry to 10 times

* add `attribute_type` endpoint to handle just field attribute requests

* handle case of field attribute request (non-callable attributes retrieval request) in PyMiloClient

* handle case of field attribute request (non-callable attributes retrieval request) in PyMiloServer

* add field attribute retrieval to check to scenario2

* `autopep8.sh` applied

Author: AHReccese

dev-requirements.txt

@@ -4,6 +4,7 @@ scipy>=0.19.1
 uvicorn==0.30.6
 fastapi==0.112.1
 requests==2.32.3
+pydantic>=1.5.0
 setuptools>=40.8.0
 vulture>=1.0
 bandit>=1.5.1

pymilo/streaming/communicator.py

@@ -2,8 +2,8 @@
 """PyMilo RESTFull Communication Mediums."""
 import uvicorn
 import requests
-from fastapi import FastAPI, Request
 from pydantic import BaseModel
+from fastapi import FastAPI, Request
 from .interfaces import ClientCommunicator
 
 
@@ -21,7 +21,7 @@ def __init__(self, server_url):
         self._server_url = server_url
         self.session = requests.Session()
         retries = requests.adapters.Retry(
-            total=5,
+            total=10,
             backoff_factor=0.1,
             status_forcelist=[500, 502, 503, 504]
         )
@@ -34,30 +34,45 @@ def download(self, payload):
 
         :param payload: download request payload
         :type payload: dict
-        :return: response of pymilo server
+        :return: string serialized model
         """
-        return self.session.get(url=self._server_url + "/download/", json=payload, timeout=5)
+        response = self.session.get(url=self._server_url + "/download/", json=payload, timeout=5)
+        if response.status_code != 200:
+            return None
+        return response.json()["payload"]
 
     def upload(self, payload):
         """
         Upload the local ML model to the remote server.
 
         :param payload: upload request payload
         :type payload: dict
-        :return: response of pymilo server
+        :return: True if upload was successful, False otherwise
         """
-        return self.session.post(url=self._server_url + "/upload/", json=payload, timeout=5)
+        response = self.session.post(url=self._server_url + "/upload/", json=payload, timeout=5)
+        return response.status_code == 200
 
     def attribute_call(self, payload):
         """
         Delegate the requested attribute call to the remote server.
 
         :param payload: attribute call request payload
         :type payload: dict
-        :return: response of pymilo server
+        :return: json-encoded response of pymilo server
         """
-        return self.session.post(url=self._server_url + "/attribute_call/", json=payload, timeout=5)
+        response = self.session.post(url=self._server_url + "/attribute_call/", json=payload, timeout=5)
+        return response.json()
 
+    def attribute_type(self, payload):
+        """
+        Identify the attribute type of the requested attribute.
+
+        :param payload: attribute type request payload
+        :type payload: dict
+        :return: response of pymilo server
+        """
+        response = self.session.post(url=self._server_url + "/attribute_type/", json=payload, timeout=5)
+        return response.json()
 
 
 class RESTServerCommunicator():
@@ -68,7 +83,7 @@ def __init__(
             ps,
             host: str = "127.0.0.1",
             port: int = 8000,
-            ):
+    ):
         """
         Initialize the Pymilo RESTServerCommunicator instance.
 
@@ -91,15 +106,21 @@ 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):
+
+        class AttributeCallPayload(StandardPayload):
             attribute: str
             args: list
             kwargs: dict
 
+        class AttributeTypePayload(StandardPayload):
+            attribute: str
+
         @self.app.get("/download/")
         async def download(request: Request):
             body = await request.json()
@@ -126,14 +147,29 @@ async def upload(request: Request):
         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)
+            payload = AttributeCallPayload(**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."
             }
 
+        @self.app.post("/attribute_type/")
+        async def attribute_type(request: Request):
+            body = await request.json()
+            body = self.parse(body)
+            payload = AttributeTypePayload(**body)
+            message = "/attribute_type request from client: {} for model: {}".format(
+                payload.client_id, payload.model_id)
+            is_callable, field_value = self._ps.is_callable_attribute(payload)
+            return {
+                "message": message,
+                "attribute type": "method" if is_callable else "field",
+                "attribute value": "" if is_callable else field_value,
+            }
+
     def parse(self, body):
         """
         Parse the compressed encrypted body of the request.

pymilo/streaming/param.py

@@ -0,0 +1,10 @@
+# -*- coding: utf-8 -*-
+"""Streaming Parameters and constants."""
+PYMILO_CLIENT_INVALID_MODE = "Invalid mode, the given mode should be either `LOCAL`[default] or `DELEGATE`."
+PYMILO_CLIENT_MODEL_SYNCHED = "PyMiloClient synched the local ML model with the remote one successfully."
+PYMILO_CLIENT_LOCAL_MODEL_UPLOADED = "PyMiloClient uploaded the local model successfully."
+PYMILO_CLIENT_LOCAL_MODEL_UPLOAD_FAILED = "PyMiloClient failed to upload the local model."
+PYMILO_CLIENT_INVALID_ATTRIBUTE = "This attribute doesn't exist in either PymiloClient or the inner ML model."
+PYMILO_CLIENT_FAILED_TO_DOWNLOAD_REMOTE_MODEL = "PyMiloClient failed to download the remote ML model."
+
+PYMILO_SERVER_NON_EXISTENT_ATTRIBUTE = "The requested attribute doesn't exist in this model."

pymilo/streaming/pymilo_client.py

@@ -4,6 +4,9 @@
 from .encryptor import DummyEncryptor
 from .compressor import DummyCompressor
 from ..pymilo_obj import Export, Import
+from .param import PYMILO_CLIENT_INVALID_MODE, PYMILO_CLIENT_MODEL_SYNCHED, \
+    PYMILO_CLIENT_LOCAL_MODEL_UPLOADED, PYMILO_CLIENT_LOCAL_MODEL_UPLOAD_FAILED, \
+    PYMILO_CLIENT_INVALID_ATTRIBUTE, PYMILO_CLIENT_FAILED_TO_DOWNLOAD_REMOTE_MODEL
 from .communicator import RESTClientCommunicator
 from ..transporters.general_data_structure_transporter import GeneralDataStructureTransporter
 
@@ -22,20 +25,17 @@ def __init__(
             self,
             model=None,
             mode=Mode.LOCAL,
-            server="http://127.0.0.1",
-            port= 8000
-            ):
+            server_url="http://127.0.0.1:8000",
+    ):
         """
         Initialize the Pymilo PymiloClient instance.
 
         :param model: the ML model PyMiloClient wrapped around
         :type model: Any
         :param mode: the mode in which PymiloClient should work, either LOCAL mode or DELEGATE
         :type mode: str (LOCAL|DELEGATE)
-        :param server: the url to which PyMilo Server listens
-        :type server: str
-        :param port: the port to which PyMilo Server listens
-        :type port: int
+        :param server_url: the url to which PyMilo Server listens
+        :type server_url: str
         :return: an instance of the Pymilo PymiloClient class
         """
         self._client_id = "0x_client_id"
@@ -44,10 +44,7 @@ def __init__(
         self._mode = mode
         self._compressor = DummyCompressor()
         self._encryptor = DummyEncryptor()
-        self._communicator = RESTClientCommunicator(
-            server_url="{}:{}".format(server, port)
-        )
-
+        self._communicator = RESTClientCommunicator(server_url)
 
     def toggle_mode(self, mode=Mode.LOCAL):
         """
@@ -56,41 +53,41 @@ def toggle_mode(self, mode=Mode.LOCAL):
         :return: None
         """
         if mode not in Mode.__members__.values():
-            raise Exception("Invalid mode, the given mode should be either `LOCAL`[default] or `DELEGATE`.")
-        self._mode = mode
+            raise Exception(PYMILO_CLIENT_INVALID_MODE)
+        if mode != self._mode:
+            self._mode = mode
 
     def download(self):
         """
         Request for the remote ML model to download.
 
         :return: None
         """
-        response = self._communicator.download({
-            "client_id": self._client_id,    
+        serialized_model = 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"]
+        if serialized_model is None:
+            print(PYMILO_CLIENT_FAILED_TO_DOWNLOAD_REMOTE_MODEL)
+            return
         self._model = Import(file_adr=None, json_dump=serialized_model).to_model()
-        print("Local model updated successfully.")
+        print(PYMILO_CLIENT_MODEL_SYNCHED)
 
     def upload(self):
         """
         Upload the local ML model to the remote server.
 
         :return: None
         """
-        response = self._communicator.upload({
-            "client_id": self._client_id,    
+        succeed = 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.")
+        if succeed:
+            print(PYMILO_CLIENT_LOCAL_MODEL_UPLOADED)
         else:
-            print("Local model upload failed.")
+            print(PYMILO_CLIENT_LOCAL_MODEL_UPLOAD_FAILED)
 
     def __getattr__(self, attribute):
         """
@@ -105,18 +102,31 @@ def __getattr__(self, attribute):
             if attribute in dir(self._model):
                 return getattr(self._model, attribute)
             else:
-                raise AttributeError("This attribute doesn't exist in either PymiloClient or the inner ML model.")
+                raise AttributeError(PYMILO_CLIENT_INVALID_ATTRIBUTE)
         elif self._mode == Mode.DELEGATE:
             gdst = GeneralDataStructureTransporter()
+            response = self._communicator.attribute_type(
+                self._encryptor.encrypt(
+                    self._compressor.compress(
+                        {
+                            "client_id": self._client_id,
+                            "model_id": self._model_id,
+                            "attribute": attribute,
+                        }
+                    )
+                )
+            )
+            if response["attribute type"] == "field":
+                return gdst.deserialize(response, "attribute value", None)
+
             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(
@@ -125,7 +135,6 @@ def relayer(*args, **kwargs):
                             payload
                         )
                     )
-                ).json()
+                )
                 return gdst.deserialize(result, "payload", None)
-            relayer.__doc__ = getattr(self._model.__class__, attribute).__doc__
             return relayer

pymilo/streaming/pymilo_server.py

@@ -1,9 +1,10 @@
 # -*- coding: utf-8 -*-
 """PyMiloServer for RESTFull protocol."""
 from ..pymilo_obj import Export, Import
-from .compressor import DummyCompressor
 from .encryptor import DummyEncryptor
+from .compressor import DummyCompressor
 from .communicator import RESTServerCommunicator
+from .param import PYMILO_SERVER_NON_EXISTENT_ATTRIBUTE
 from ..transporters.general_data_structure_transporter import GeneralDataStructureTransporter
 
 
@@ -13,7 +14,7 @@ class PymiloServer:
     def __init__(self, port=8000):
         """
         Initialize the Pymilo PymiloServer instance.
-        
+
         :param port: the port to which PyMiloServer listens
         :type port: int
         :return: an instance of the PymiloServer class
@@ -53,7 +54,7 @@ def execute_model(self, request):
         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.")
+            raise Exception(PYMILO_SERVER_NON_EXISTENT_ATTRIBUTE)
         arguments = {
             'args': request.args,
             'kwargs': request.kwargs
@@ -65,3 +66,18 @@ def execute_model(self, request):
             self._model = output
             return None
         return gdst.serialize({'output': output}, 'output', None)
+
+    def is_callable_attribute(self, request):
+        """
+        Check whether the requested attribute is callable or not.
+
+        :param request: request obj containing requested attribute to check it's type
+        :type request: obj
+        :return: True if it is callable False otherwise
+        """
+        attribute = request.attribute
+        retrieved_attribute = getattr(self._model, attribute, None)
+        if callable(retrieved_attribute):
+            return True, None
+        else:
+            return False, GeneralDataStructureTransporter().serialize({'output': retrieved_attribute}, 'output', None)

pymilo/transporters/randomstate_transporter.py

@@ -4,6 +4,7 @@
 from .transporter import AbstractTransporter
 from ..utils.util import check_str_in_iterable
 
+
 class RandomStateTransporter(AbstractTransporter):
     """Customized PyMilo Transporter developed to handle RandomState field."""
 
@@ -31,8 +32,8 @@ def serialize(self, data, key, model_type):
                     inner_random_state.get_state()[2],
                     inner_random_state.get_state()[3],
                     inner_random_state.get_state()[4],
-                    ),
-                }
+                ),
+            }
         return data[key]
 
     def deserialize(self, data, key, model_type):

streaming-requirements.txt

@@ -1,3 +1,4 @@
 uvicorn>=0.14.0
 fastapi>=0.68.0
-requests>=2.0.0
+requests>=2.0.0
+pydantic>=1.5.0

tests/test_ml_streaming/scenarios/scenario2.py

@@ -25,6 +25,7 @@ def scenario2():
     # 3.
     client.toggle_mode(Mode.DELEGATE)
     client.fit(x_train, y_train)
+    remote_field = client.coef_
 
     # 4.
     result = client.predict(x_test)
@@ -35,6 +36,7 @@ def scenario2():
 
     # 6.
     client.toggle_mode(mode=Mode.LOCAL)
+    local_field = client.coef_
     result = client.predict(x_test)
     mse_local = mean_squared_error(y_test, result)
-    return np.abs(mse_server-mse_local)
+    return np.abs(mse_server-mse_local) + np.abs(np.sum(local_field-remote_field))