Improve code based on pylint and black suggestions

app.py

@@ -1,14 +1,23 @@
+"""
+Gradio web application
+"""
+
 import os
+import json
 import requests
 import gradio as gr
-from classification.classifier import Classifier
 from dotenv import load_dotenv, find_dotenv
-import json
+
+from classification.classifier import Classifier
+
+AWS_API = None
+
 
 # Initialize API URLs from env file or global settings
 def retrieve_api():
+    """Initialize API URLs from env file or global settings"""
 
-    env_path = find_dotenv('config_api.env')
+    env_path = find_dotenv("config_api.env")
     if env_path:
         load_dotenv(dotenv_path=env_path)
         print("config_api.env file loaded successfully.")
@@ -19,31 +28,35 @@ def retrieve_api():
     global AWS_API
     AWS_API = os.getenv("AWS_API", default="http://localhost:8000")
 
+
 def initialize_classifier():
-    global cls
+    """Initialize ML classifier"""
+
     cls = Classifier()
+    return cls
 
 
 def predict_class_local(sepl, sepw, petl, petw):
+    """ML prediction using direct source code - local"""
+
     data = list(map(float, [sepl, sepw, petl, petw]))
+    cls = initialize_classifier()
     results = cls.load_and_test(data)
     return results
 
 
 def predict_class_aws(sepl, sepw, petl, petw):
+    """ML prediction using AWS API endpoint"""
+
     if AWS_API == "http://localhost:8080":
-        API_endpoint = AWS_API + "/2015-03-31/functions/function/invocations"
+        api_endpoint = AWS_API + "/2015-03-31/functions/function/invocations"
     else:
-        API_endpoint = AWS_API + "/test/classify"
+        api_endpoint = AWS_API + "/test/classify"
 
     data = list(map(float, [sepl, sepw, petl, petw]))
-    json_object = {
-        "features": [
-            data
-        ]
-    }
+    json_object = {"features": [data]}
 
-    response = requests.post(API_endpoint, json=json_object)
+    response = requests.post(api_endpoint, json=json_object, timeout=60)
     if response.status_code == 200:
         # Process the response
         response_json = response.json()
@@ -54,11 +67,14 @@ def predict_class_aws(sepl, sepw, petl, petw):
     return results_dict
 
 
-def predict(sepl, sepw, petl, petw, type):
-    print("type: ", type)
-    if type == "Local":
+def predict(sepl, sepw, petl, petw, execution_type):
+    """ML prediction - local or via API endpoint"""
+
+    print("ML prediction type: ", execution_type)
+    results = None
+    if execution_type == "Local":
         results = predict_class_local(sepl, sepw, petl, petw)
-    elif type == "AWS API":
+    elif execution_type == "AWS API":
         results = predict_class_aws(sepl, sepw, petl, petw)
 
     prediction = results["predictions"][0]
@@ -69,30 +85,41 @@ def predict(sepl, sepw, petl, petw, type):
 
 # Define the Gradio interface
 def user_interface():
+    """Gradio application"""
+
+    description = """
+    Aims: Categorization of different species of iris flowers (Setosa, Versicolor, and Virginica) 
+    based on measurements of physical characteristics (sepals and petals).
+
+    Notes: This web application uses two types of machine learning predictions:
+       - local prediction (direct source code) 
+       - cloud prediction via an AWS API (i.e. use of ECR, Lambda function and API Gateway)
+    """
+
     with gr.Blocks() as demo:
         gr.Markdown("# IRIS classification task - use of AWS Lambda")
-        gr.Markdown(
-            """
-            Aims: Categorization of different species of iris flowers (Setosa, Versicolor, and Virginica) 
-            based on measurements of physical characteristics (sepals and petals).
-
-            Notes: This web application uses two types of predictions:
-               - local prediction (direct source code) 
-               - cloud prediction via an AWS API (i.e. use of ECR, Lambda function and API Gateway) to run the machine learning model.
-            
-            """
-        )
+        gr.Markdown(description)
 
         with gr.Row():
             with gr.Column():
                 with gr.Group():
-                    gr_sepl = gr.Slider(minimum=4.0, maximum=8.0, step=0.1, label="Sepal Length (in cm)")
-                    gr_sepw = gr.Slider(minimum=2.0, maximum=5.0, step=0.1, label="Sepal Width (in cm)")
-                    gr_petl = gr.Slider(minimum=1.0, maximum=7.0, step=0.1, label="Petal Length (in cm)")
-                    gr_petw = gr.Slider(minimum=0.1, maximum=2.8, step=0.1, label="Petal Width (in cm)")
+                    gr_sepl = gr.Slider(
+                        minimum=4.0, maximum=8.0, step=0.1, label="Sepal Length (in cm)"
+                    )
+                    gr_sepw = gr.Slider(
+                        minimum=2.0, maximum=5.0, step=0.1, label="Sepal Width (in cm)"
+                    )
+                    gr_petl = gr.Slider(
+                        minimum=1.0, maximum=7.0, step=0.1, label="Petal Length (in cm)"
+                    )
+                    gr_petw = gr.Slider(
+                        minimum=0.1, maximum=2.8, step=0.1, label="Petal Width (in cm)"
+                    )
             with gr.Column():
                 with gr.Row():
-                    gr_type = gr.Radio(["Local", "AWS API"], value="Local", label="Prediction type")
+                    gr_execution_type = gr.Radio(
+                        ["Local", "AWS API"], value="Local", label="Prediction type"
+                    )
                 with gr.Row():
                     gr_output = gr.Textbox(label="Prediction output")
 
@@ -100,12 +127,15 @@ def user_interface():
             submit_btn = gr.Button("Submit")
             clear_button = gr.ClearButton()
 
-        submit_btn.click(fn=predict, inputs=[gr_sepl, gr_sepw, gr_petl, gr_petw, gr_type], outputs=[gr_output])
+        submit_btn.click(
+            fn=predict,
+            inputs=[gr_sepl, gr_sepw, gr_petl, gr_petw, gr_execution_type],
+            outputs=[gr_output],
+        )
         clear_button.click(lambda: None, inputs=None, outputs=[gr_output], queue=False)
     demo.queue().launch(debug=True)
 
 
 if __name__ == "__main__":
     retrieve_api()
-    initialize_classifier()
     user_interface()

classification/classifier.py

@@ -1,40 +1,51 @@
-# from sklearn.ensemble import AdaBoostClassifier
+"""
+IRIS Classification - class definition
+"""
+
+import os
+import numpy as np
+import pandas as pd
+import joblib
+
 from sklearn.tree import DecisionTreeClassifier
 from sklearn.datasets import load_iris
 from sklearn.model_selection import train_test_split
-import joblib
-import pandas as pd
-import os
-import numpy as np
+
 
 class Classifier:
+    """Classifier class - ML training and testing"""
+
     def __init__(self):
         pass
 
     def train_and_save(self):
+        """ML training and saving"""
         print("\nIRIS model training...")
         iris = load_iris()
-        cart = DecisionTreeClassifier(max_depth = 3)
+        cart = DecisionTreeClassifier(max_depth=3)
 
-        X_train, X_test, y_train, y_test = train_test_split(iris.data, iris.target, test_size=0.1, random_state=42)
-        model = cart.fit(X_train, y_train)
+        x_train, x_test, y_train, y_test = train_test_split(
+            iris.data, iris.target, test_size=0.1, random_state=42
+        )
+        model = cart.fit(x_train, y_train)
 
-        print(f"Model score: {cart.score(X_train, y_train):.3f}")
-        print(f"Test Accuracy: {cart.score(X_test, y_test):.3f}")
+        print(f"Model score: {cart.score(x_train, y_train):.3f}")
+        print(f"Test Accuracy: {cart.score(x_test, y_test):.3f}")
 
         current_dir = os.path.dirname(os.path.abspath(__file__))
         parent_dir = os.path.dirname(current_dir)
         test_data_csv_path = os.path.join(parent_dir, "data", "test_data.csv")
 
-        pd.concat([pd.DataFrame(X_test), pd.DataFrame(y_test, columns=['4'])], axis=1).to_csv(test_data_csv_path,
-                                                                                              index=False)
+        pd.concat([pd.DataFrame(x_test), pd.DataFrame(y_test, columns=["4"])], axis=1).to_csv(
+            test_data_csv_path, index=False
+        )
 
         model_path = os.path.join(parent_dir, "models", "model.pkl")
         joblib.dump(model, model_path)
         print(f"Model saved to {model_path}")
 
-
     def load_and_test(self, data):
+        "ML loading and testing"
         print("\nIRIS model prediction...")
 
         current_dir = os.path.dirname(os.path.abspath(__file__))

inference_api.py

@@ -1,8 +1,11 @@
-import requests
-import io
+"""
+IRIS classification - command line inference via API
+"""
+
+import sys
 import json
 import argparse
-import sys
+import requests
 
 
 # Default examples
@@ -13,11 +16,13 @@ def arg_parser():
     """Parse arguments"""
 
     # Create an ArgumentParser object
-    parser = argparse.ArgumentParser(description='Object detection inference via API call')
+    parser = argparse.ArgumentParser(description="IRIS classification inference via API call")
     # Add arguments
-    parser.add_argument('-u', '--url', type=str, help='URL to the server (with endpoint location)', required=True)
-    parser.add_argument('-d', '--data', type=str, help='Input data', required=True)
-    parser.add_argument('-v', '--verbose', action='store_true', help='Increase output verbosity')
+    parser.add_argument(
+        "-u", "--url", type=str, help="URL to the server (with endpoint location)", required=True
+    )
+    parser.add_argument("-d", "--data", type=str, help="Input data", required=True)
+    parser.add_argument("-v", "--verbose", action="store_true", help="Increase output verbosity")
     return parser
 
 
@@ -27,19 +32,19 @@ def main(args=None):
     args = arg_parser().parse_args(args)
     # Use the arguments
     if args.verbose:
-        print(f'Input data: {args.data}')
-        print(f'Input data type: {type(args.data)}')
+        print(f"Input data: {args.data}")
+        print(f"Input data type: {type(args.data)}")
 
     # Send request to API
-    response = requests.post(args.url, json=json.loads(args.data))
+    response = requests.post(args.url, json=json.loads(args.data), timeout=60)
 
     if response.status_code == 200:
         # Process the response
         processed_data = json.loads(response.content)
-        print('processed_data', processed_data)
+        print("processed_data", processed_data)
     else:
         print(f"Error: {response.status_code}")
 
 
 if __name__ == "__main__":
-    sys.exit(main(sys.argv[1:]))
+    sys.exit(main(sys.argv[1:]))

inference_direct.py

@@ -1,5 +1,9 @@
-from classification.classifier import Classifier
+"""
+Direct inference with hard-coded data
+"""
+
 import json
+from classification.classifier import Classifier
 
 
 if __name__ == "__main__":
@@ -9,19 +13,16 @@ if __name__ == "__main__":
     cls.train_and_save()
 
     # Testing
-    data = {
-        "features": [[6.5, 3.0, 5.8, 2.2],[6.1, 2.8, 4.7, 1.2]]
-    }
+    data = {"features": [[6.5, 3.0, 5.8, 2.2], [6.1, 2.8, 4.7, 1.2]]}
     features = data["features"]
     results = cls.load_and_test(features)
     print("results:", results)
 
     # Response similar to REST API call
     response = {
-        'statusCode': 200,
-        'body': json.dumps({
-            'predictions': results["predictions"],
-            'probabilities': results["probabilities"]
-        })
+        "statusCode": 200,
+        "body": json.dumps(
+            {"predictions": results["predictions"], "probabilities": results["probabilities"]}
+        ),
     }
     print("Example REST API response: ", response)

lambda_function.py

@@ -1,30 +1,39 @@
-from classification.classifier import Classifier
+"""
+AWS Lambda function
+"""
+
 import json
+from classification.classifier import Classifier
 
 
 cls = Classifier()
 
+
 # Lambda handler (proxy integration option unchecked on AWS API Gateway)
 def lambda_handler(event, context):
+    """
+    Lambda handler (proxy integration option unchecked on AWS API Gateway)
+
+    Args:
+        event (dict): The event that triggered the Lambda function.
+        context (LambdaContext): Information about the execution environment.
+
+    Returns:
+        dict: The response to be returned from the Lambda function.
+    """
 
     try:
-        features = event.get('features', {})
+        features = event.get("features", {})
         if not features:
             raise ValueError("'features' key missing")
 
         response = cls.load_and_test(features)
         return {
-            'statusCode': 200,
-            'headers': {
-                'Content-Type': 'application/json'
-            },
-            'body': json.dumps({
-                'predictions': response["predictions"],
-                'probabilities': response["probabilities"]
-            })
+            "statusCode": 200,
+            "headers": {"Content-Type": "application/json"},
+            "body": json.dumps(
+                {"predictions": response["predictions"], "probabilities": response["probabilities"]}
+            ),
         }
     except Exception as e:
-        return {
-            'statusCode': 500,
-            'body': json.dumps({'error': str(e)})
-        }
+        return {"statusCode": 500, "body": json.dumps({"error": str(e)})}