File size: 2,937 Bytes
97869d5 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 |
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()
|