File size: 3,075 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 |
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()
|