Upload 6 files

ai-analytics/endpoint.py

@@ -0,0 +1,100 @@
+import asyncio
+import json
+import os
+import re
+import sys
+from datetime import datetime
+from typing import cast
+import threading
+
+import tornado.web
+import tornado.websocket
+from tornado import ioloop
+
+sys.path.append(os.path.dirname(os.path.realpath(__file__)))
+from prophet_model import ProphetModel
+
+ISO_PATTERN = r'^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$'
+DATASET = os.getenv("DATASET", "data/traffic_data.csv")
+
+class MainHandler(tornado.web.RequestHandler):
+    """
+    Handler to render the index page
+    """
+    def get(self):
+        self.render("index.html")
+
+class PredictionHandler(tornado.web.RequestHandler):
+    """
+    Handler to process the request for predicting traffic data
+    """
+    def initialize(self):
+        """
+        Initializes and trains a model
+        """
+        self.model = ProphetModel()
+        self.model.train_model(DATASET)
+
+    def post(self):
+        """
+        POST method to process the request for traffic data generation
+        """
+        try:
+            print(self.request.body)
+            run_parameters = json.loads(self.request.body)
+
+            if run_parameters is None:
+                self.write("Invalid parameters: 400")
+                self.set_status(400)
+                return
+
+            predict_time = cast(str, run_parameters.get("predict_time"))
+
+            if re.match(ISO_PATTERN, predict_time):
+                #convert string of format 2024-10-01 08:00:00 to datetime object
+                predict_time = datetime.strptime(predict_time, "%Y-%m-%d %H:%M:%S")
+            else:
+                self.write("Please provide date in the format YYYY-MM-DD HH:mm:ss")
+                self.set_status(400)
+                return
+
+            dataset = self.model.predict(predict_time)
+            self.write(json.dumps(dataset))
+
+        except json.JSONDecodeError:
+            self.write("Invalid JSON payload: 400")
+            self.set_status(400)
+            return
+        except Exception as error_message:
+            self.set_status(500)
+            self.write(f"Error in data generation: {str(error_message)}")
+            return
+
+def make_app():
+    """
+    Create the tornado application
+    """
+    return tornado.web.Application([
+        (r"/", MainHandler),
+        (r"/predict", PredictionHandler),
+    ])
+
+def start():
+    """
+    Start the server
+    """
+    def run_server():
+        asyncio.set_event_loop(asyncio.new_event_loop())
+        try:
+            app = make_app()
+            app.listen(8889)
+            print("Server is running on port 8889")
+            ioloop.IOLoop.current().start()
+        except Exception as error_message:
+            print("An error occurred while starting the server:", error_message)
+
+    server_thread = threading.Thread(target=run_server)
+    server_thread.start()
+
+if __name__ == "__main__":
+    start()

ai-analytics/prophet_model.py

@@ -0,0 +1,78 @@
+from datetime import datetime
+import pickle
+import pandas as pd
+from prophet import Prophet
+
+class ProphetModel:
+    def __init__(self):
+        self.model = None
+
+    def train_model(self, csv_file):
+        """
+        Train the Prophet model using the data from the given CSV file.
+        
+        :param csv_file: Path to the CSV file containing the dataset.
+        """
+        # Load the dataset
+        df = pd.read_csv(csv_file)
+
+        # Prepare the data
+        df.rename(columns={'time': 'ds', 'volume': 'y'}, inplace=True)
+
+        # Initialize and fit the Prophet model
+        self.model = Prophet()
+        self.model.fit(df)
+
+    def predict(self, datetime):
+        """
+        Query the forecast for a given date.
+        
+        :param date: The date for which to query the forecast (format: 'YYYY-MM-DD').
+        :return: The forecasted value for the given date.
+        """
+        if self.model is None:
+            raise Exception("Model has not been trained. Call train_model() first.")
+
+        # Create a DataFrame with the given date
+        future = pd.DataFrame({'ds': [datetime]})
+
+        # Make predictions
+        forecast = self.model.predict(future)
+
+        # Return the forecasted value to json format
+        return forecast[['ds', 'yhat', 'yhat_lower', 'yhat_upper']].to_json(orient='records')
+
+    def save_model(self, model_file):
+        """
+        Save the trained model to a file.
+        
+        :param model_file: Path to the file where the model will be saved.
+        """
+        #check if trained model exists
+        if self.model is None:
+            raise Exception("Model has not been trained. Call train_model() first.")
+
+        with open(model_file, 'wb') as fout:
+            pickle.dump(self.model, fout)
+
+    def load_model(self, model_file):
+        """
+        Load a trained model from a file.
+        
+        :param model_file: Path to the file containing the saved model.
+        """
+        with open(model_file, 'rb') as fin:
+            self.model = pickle.load(fin)
+        return self
+
+# prophet_model = ProphetModel()
+# prophet_model.train_model('data/traffic_data.csv')
+# print(type(prophet_model))
+# predict_time = datetime.strptime('2023-10-01 09:00:00', "%Y-%m-%d %H:%M:%S")
+# forecast = prophet_model.predict(predict_time)
+# prophet_model.save_model('saved_models/forecast_model.pkl')
+# print(forecast)
+# prophet_model = prophet_model.load_model('saved-models/forecast_model.pkl')
+# print(type(prophet_model))
+# forecast = prophet_model.predict(predict_time)
+# print(forecast)

data-preprocessing/endpoint.py

@@ -0,0 +1,98 @@
+import asyncio
+import json
+import os
+import re
+import sys
+from datetime import datetime
+from typing import cast
+import threading
+
+import tornado.web
+import tornado.websocket
+from tornado import ioloop
+
+sys.path.append(os.path.dirname(os.path.realpath(__file__)))
+from generate_traffic_data import generate_data
+
+subscriptions = {}
+PATTERN = r'^\d{4}-\d{2}-\d{2}'
+ISO_PATTERN = r'^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$'
+
+class MainHandler(tornado.web.RequestHandler):
+    """
+    Handler to render the index page
+    """
+    def get(self):
+        self.render("index.html")
+
+class DataGeneratorHandler(tornado.web.RequestHandler):
+    """
+    Handler to process the request for generating traffic data
+    """
+    def post(self):
+        """
+        POST method to process the request for traffic data generation
+        """
+        try:
+            print(self.request.body)
+            run_parameters = json.loads(self.request.body)
+
+            if run_parameters is None:
+                self.write("Invalid parameters: 400")
+                self.set_status(400)
+                return
+
+            start_time = cast(str, run_parameters.get("start_time"))
+            end_time = cast(str, run_parameters.get("end_time"))
+            window_size = cast(str, run_parameters.get("window_size"))
+
+            if re.match(ISO_PATTERN, start_time) and re.match(ISO_PATTERN, end_time):
+                #convert string of format 2024-10-01 08:00:00 to datetime object
+                start_time = datetime.strptime(start_time, "%Y-%m-%d %H:%M:%S")
+                end_time = datetime.strptime(end_time, "%Y-%m-%d %H:%M:%S")
+                window_size = int(window_size)
+            else:
+                self.write("Please provide date in the format " + ISO_PATTERN)
+                self.set_status(400)
+                return
+
+            dataset = generate_data(start_time, end_time, window_size)
+            self.write(json.dumps(dataset))
+
+        except json.JSONDecodeError:
+            self.write("Invalid JSON payload: 400")
+            self.set_status(400)
+            return
+        except Exception as error_message:
+            self.set_status(500)
+            self.write(f"Error in data generation: {str(error_message)}")
+            return
+
+def make_app():
+    """
+    Create the tornado application
+    """
+    return tornado.web.Application([
+        (r"/", MainHandler),
+        (r"/traffic", DataGeneratorHandler),
+    ])
+
+def start():
+    """
+    Start the server
+    """
+    def run_server():
+        asyncio.set_event_loop(asyncio.new_event_loop())
+        try:
+            app = make_app()
+            app.listen(8889)
+            print("Server is running on port 8889")
+            ioloop.IOLoop.current().start()
+        except Exception as error_message:
+            print("An error occurred while starting the server:", error_message)
+
+    server_thread = threading.Thread(target=run_server)
+    server_thread.start()
+
+if __name__ == "__main__":
+    start()

data-preprocessing/generate_traffic_data.py

@@ -0,0 +1,32 @@
+import random
+from datetime import datetime, timedelta
+
+def generate_data(start_time: datetime, end_time: datetime, window_size: int):
+    """
+    Generate random traffic volume data for a given time range and window size.
+    """
+    traffic_data = []
+    current_time = start_time
+
+    while current_time < end_time:
+        # Generate random traffic volume data for the current window
+        traffic_volume = random.randint(0, 100)
+        traffic_data.append({
+            'time': current_time.strftime("%Y-%m-%d %H:%M:%S"),
+            'volume': traffic_volume
+        })
+        current_time += timedelta(minutes=window_size)
+
+    return traffic_data
+
+if __name__ == "__main__":
+    start_time = datetime(2023, 10, 1, 8, 0, 0)  # October 1, 2023, 08:00:00
+    end_time = datetime(2023, 10, 1, 10, 0, 0)  # October 1, 2023, 10:00:00
+    window_size = 15  # 15-minute window
+
+    traffic_data = generate_data(start_time, end_time, window_size)
+
+    with open('data/traffic_data.csv', 'w') as f:
+        f.write("time,volume\n")
+        for data in traffic_data:
+            f.write(f"{data['time']},{data['volume']}\n")