From 4cb11583e84b2b6513edd738af3af56a3db83ff0 Mon Sep 17 00:00:00 2001 From: Pratyush-exe Date: Sat, 15 Apr 2023 15:00:54 +0530 Subject: [PATCH 1/5] FIX:fixed streamer.py --- main.py | 5 ++-- src/streamer.py | 70 +++++++++++++++++++++++-------------------------- 2 files changed, 35 insertions(+), 40 deletions(-) diff --git a/main.py b/main.py index c11e72a..3b97fca 100644 --- a/main.py +++ b/main.py @@ -21,10 +21,9 @@ python_exception_handler ) -import src.utils as utils +import src.utils as utils from src.streamer import RewardsStreamer - # TODO: # ----- # Add a response model for returning all the configuration in structured format @@ -238,7 +237,7 @@ def get_all_sessions(): @app.get('/api/v1/start_training/{session_id}') -async def start_training(session_id : str): +def start_training(session_id : str): """/start_training is the endpoint to start training the agent These are the sets of events that will happen during this session while triggering this endpoint diff --git a/src/streamer.py b/src/streamer.py index f93ea76..2bfcce1 100644 --- a/src/streamer.py +++ b/src/streamer.py @@ -93,43 +93,39 @@ def _convert_str_func_to_exec(self, str_function: str, function_name: str): new_func = globals_dict[function_name] return new_func - def stream_episode(self, yield_response: bool = False): - for episode in range(1, self.num_episodes + 1): - self.game.initialize() - self.game.FPS = self.car_speed - total_score, record = 0, 0 - done = False + def stream_episode(self, yield_response: bool = False): + self.game.FPS = self.car_speed + total_score, record = 0, 0 + done = False + while True: + if self.num_episodes == self.agent.n_games: + return { + "data": "False" + } + + _, done, score, pixel_data = self.agent.train_step(self.game) + self.game.timeTicking() - - while not done: - _, done, score, pixel_data = self.agent.train_step(self.game) - self.game.timeTicking() - - self.agent.n_games += 1 - self.agent.train_long_memory() - - if score > record: - self.agent.model.save( - self.checkpoint_folder_path, - 'model.pth', - device = self.device + if done: + self.game.initialize() + self.agent.n_games += 1 + self.agent.train_long_memory() + + if score > record: + self.agent.model.save( + self.checkpoint_folder_path, + 'model.pth', + device = self.device ) - total_score += score - - # stream all the metrics to a file - utils.add_inside_session( - self.session_id, - config_name="metrics", - rewrite=True, multi_config=True, - episode_number = episode, - episode_score = score, - mean_score = total_score / self.agent.n_games - ) - - response = { - 'episode_number' : episode, - 'episode_score' : score, - 'episode_mean_score' : total_score / self.agent.n_games - } + total_score += score + + utils.add_inside_session( + self.session_id, + config_name="metrics", + rewrite=True, multi_config=True, + episode_number = self.agent.n_games, + episode_score = score, + mean_score = total_score / self.agent.n_games + ) - yield json.dumps(response) + "\n" \ No newline at end of file + yield "sdafasfsdafsadfasdfdsfsdfdsf" \ No newline at end of file From 63372b6cc3e4123983cd4f285d79d2a3e6ab661f Mon Sep 17 00:00:00 2001 From: Pratyush-exe Date: Mon, 17 Apr 2023 09:09:33 +0530 Subject: [PATCH 2/5] FEAT:Added streaming functionality --- .gitignore | 3 +- flaskApp.py | 123 ++++++++++++++++++++++++++++++ main.py | 41 +--------- src/streamer.py | 196 ++++++++++++++++++++++++++++++++---------------- 4 files changed, 258 insertions(+), 105 deletions(-) create mode 100644 flaskApp.py diff --git a/.gitignore b/.gitignore index 222d4b8..412a5d2 100644 --- a/.gitignore +++ b/.gitignore @@ -89,4 +89,5 @@ target/ .mypy_cache/ rewards-api/src/__pycache__ -rewards-api/__pycache__ \ No newline at end of file +rewards-api/__pycache__ +tmp.py \ No newline at end of file diff --git a/flaskApp.py b/flaskApp.py new file mode 100644 index 0000000..a528cef --- /dev/null +++ b/flaskApp.py @@ -0,0 +1,123 @@ +import os +import cv2 +import pygame +import threading +import flask_cors +import src.utils as utils +import matplotlib.pyplot as plt +from src.config import CONFIG +from flask import Flask, Response, request, stream_with_context +from rewards import QTrainer, LinearQNet, CarGame + +app = Flask(__name__) +flask_cors.CORS(app) +lock = threading.Lock() + +def _convert_str_func_to_exec(str_function: str, function_name: str): + globals_dict = {} + exec(str_function, globals_dict) + new_func = globals_dict[function_name] + return new_func + +def generate(session_id): + r = utils.get_session_files(session_id) + print(session_id) + print("check" ,session_id) + enable_wandb = False + device = "cpu" + run_pygame_loop = False + record = 0 + done = False + + # environment parameters + env_name = r["env_params"]["environment_name"] + env_world = int(r["env_params"]["environment_world"]) + mode = r["env_params"]["mode"] + car_speed = r["env_params"]["car_speed"] + + # agent parameters + layer_config = eval(r["agent_params"]["model_configuration"]) + if type(layer_config) == str: + layer_config = eval(layer_config) + + lr = r["agent_params"]["learning_rate"] + loss = r["agent_params"]["loss_fn"] + optimizer = r["agent_params"]["optimizer"] + gamma = r["agent_params"]["gamma"] + epsilon = r["agent_params"]['epsilon'] + num_episodes = r["agent_params"]["num_episodes"] + + reward_function = r["training_params"]["reward_function"] + + global lock + + checkpoint_folder_path = os.path.join( + utils.get_home_path(), + CONFIG["REWARDS_PARENT_CONFIG_DIR"], + f"{session_id}/{CONFIG['REWARDS_CONFIG_MODEL_FOLDER_NAME']}/" + ) + + model = LinearQNet(layer_config) + + agent = QTrainer( + lr = lr, + gamma = gamma, + epsilon = epsilon, + model = model, + loss = loss, + optimizer = optimizer, + checkpoint_folder_path = checkpoint_folder_path, + model_name = "model.pth" + ) + + game = CarGame( + track_num=env_world, + mode = mode, + reward_function=_convert_str_func_to_exec( + reward_function, + function_name="reward_function" + ), + display_type="surface", + screen_size=(800, 700) + ) + game.FPS = car_speed + + record = 0 + done = False + + while True: + if agent.n_games == num_episodes: + return { + "status": 204, + "frame": "" + } + reward, done, score, pix = agent.train_step(game) + game.timeTicking() + + if done: + game.initialize() + agent.n_games += 1 + agent.train_long_memory() + if score > record: + record = score + agent.model.save( + checkpoint_folder_path, + 'model.pth', + device = "cpu" + ) + print('Game', agent.n_games, 'Score', score, 'Record:', record) + im_png = cv2.imencode(".png", pix)[1] + yield (b'--frame\r\n' b'Content-Type: image/jpeg\r\n\r\n' + bytearray(im_png) + b'\r\n') + +@app.route('/stream', methods = ['GET']) +def stream(): + session_id = request.args.get('id') + print(session_id) + return Response(generate(session_id), mimetype = "multipart/x-mixed-replace; boundary=frame") + +if __name__ == '__main__': + host = "127.0.0.1" + port = 8005 + debug = False + options = None + app.run(host, port, debug, options) \ No newline at end of file diff --git a/main.py b/main.py index 3b97fca..19c4cd1 100644 --- a/main.py +++ b/main.py @@ -3,9 +3,9 @@ import pygame from fastapi.logger import logger from fastapi import FastAPI, Request -from fastapi.responses import StreamingResponse from fastapi.middleware.cors import CORSMiddleware from fastapi.exceptions import RequestValidationError +from dotenv import set_key, load_dotenv # configs and import from other modules @@ -230,45 +230,6 @@ def get_all_sessions(): """ return utils.get_all_sessions_info() -# make streamer as the generator -# make this endpoint as the client -# so it will be back and forth connections between the client and the server - - - -@app.get('/api/v1/start_training/{session_id}') -def start_training(session_id : str): - """/start_training is the endpoint to start training the agent - These are the sets of events that will happen during this session while triggering this endpoint - - - Validation of all the parameters (TODO) - - Loading the model and the agent - - Start loading the game and streaming the results - - Args: - session_id (str): The session. - Using this session id we can train any of the experiment - """ - rewards_response = utils.get_session_files(session_id) - streamer = RewardsStreamer(session_id = session_id, response = rewards_response) - return StreamingResponse(streamer.stream_episode()) - - - -@app.get('/api/v1/stop_training/') -def stop_training(): - # NOTE: (TODO) - # Stop training does not work for now. - # One main reason is to make it into a different threading. This might introduce - # more bugs and problems. One can stop training by just clicking the cross button. - # Howevar it will automatically close once episode gets finished - - pygame.quit() - return { - "status" : 200, - "message" : "Stopped training successfully" - } - @app.post("/api/v1/validate_exp") def validate_latest_expriment(): diff --git a/src/streamer.py b/src/streamer.py index 2bfcce1..9fe26a5 100644 --- a/src/streamer.py +++ b/src/streamer.py @@ -1,16 +1,14 @@ import os -import cv2 -import ast -import sys -import time -import json +import cv2 import pygame -import asyncio -import multiprocessing as mp +import base64 +import numpy as np from typing import Dict, Any +import multiprocessing as mp import src.utils as utils from src.config import CONFIG +from dotenv import load_dotenv, set_key from rewards import workflow, QTrainer, LinearQNet, CarGame @@ -23,9 +21,13 @@ def __init__(self, session_id, response: Dict[str, Any]) -> None: RewardsStreamer is the class which is responsible for streaming, saving metrics and also streaming frames over the network. """ + print("called") self.session_id = session_id self.enable_wandb = False self.device = "cpu" + self.run_pygame_loop = False + self.record = 0 + self.done = False # environment parameters self.env_name = response["env_params"]["environment_name"] @@ -34,27 +36,46 @@ def __init__(self, session_id, response: Dict[str, Any]) -> None: self.car_speed = response["env_params"]["car_speed"] # agent parameters - self.layer_config = ast.literal_eval(response["agent_params"]["model_configuration"]) + self.layer_config = eval(response["agent_params"]["model_configuration"]) + if type(self.layer_config) == str: + self.layer_config = eval(self.layer_config) + self.lr = response["agent_params"]["learning_rate"] self.loss = response["agent_params"]["loss_fn"] self.optimizer = response["agent_params"]["optimizer"] self.gamma = response["agent_params"]["gamma"] self.epsilon = response["agent_params"]['epsilon'] self.num_episodes = response["agent_params"]["num_episodes"] + + self.reward_function = response["training_params"]["reward_function"] + + def _convert_str_func_to_exec(self, str_function: str, function_name: str): + """ + Converts a string like function skeleton to a Callable + + Args: + str_function (str): The string function skeleton + function_name (str): The name of the function + + Returns: + Callable: Actual callable function of type <'function'> + """ + globals_dict = {} + exec(str_function, globals_dict) + new_func = globals_dict[function_name] + return new_func + + def stream_episode(self, yield_response: bool = False): + global lock + self.checkpoint_folder_path = os.path.join( utils.get_home_path(), CONFIG["REWARDS_PARENT_CONFIG_DIR"], - f"{session_id}/{CONFIG['REWARDS_CONFIG_MODEL_FOLDER_NAME']}/" + f"{self.session_id}/{CONFIG['REWARDS_CONFIG_MODEL_FOLDER_NAME']}/" ) - - # reward function - self.reward_function = response["training_params"]["reward_function"] - - # make the model self.model = LinearQNet(self.layer_config) - # make the agent self.agent = QTrainer( lr = self.lr, gamma = self.gamma, @@ -66,7 +87,6 @@ def __init__(self, session_id, response: Dict[str, Any]) -> None: model_name = "model.pth" ) - # build the screen and the game self.game = CarGame( track_num=self.env_world, mode = self.mode, @@ -76,56 +96,104 @@ def __init__(self, session_id, response: Dict[str, Any]) -> None: ), display_type="surface", screen_size=(800, 700) if self.mode == "training" else (1000, 700) - ) - - def _convert_str_func_to_exec(self, str_function: str, function_name: str): - """Converts a string like function skeleton to a Callable - - Args: - str_function (str): The string function skeleton - function_name (str): The name of the function - - Returns: - Callable: Actual callable function of type <'function'> - """ - globals_dict = {} - exec(str_function, globals_dict) - new_func = globals_dict[function_name] - return new_func - - def stream_episode(self, yield_response: bool = False): + ) self.game.FPS = self.car_speed - total_score, record = 0, 0 - done = False + + self.record = 0 + self.done = False + while True: - if self.num_episodes == self.agent.n_games: - return { - "data": "False" - } - - _, done, score, pixel_data = self.agent.train_step(self.game) + print("in while") + reward, self.done, self.score, pix = self.agent.train_step(self.game) self.game.timeTicking() - - if done: - self.game.initialize() - self.agent.n_games += 1 - self.agent.train_long_memory() - - if score > record: - self.agent.model.save( - self.checkpoint_folder_path, - 'model.pth', - device = self.device - ) - total_score += score + + if self.done: + self.game.initialize() + self.agent.n_games += 1 + self.agent.train_long_memory() + if self.score > self.record: + self.record = self.score + # self.agent.model.save() + print('Game', self.agent.n_games, 'Score', self.score, 'Record:', self.record) + im_png = cv2.imencode(".png", pix)[1] + # yield b'--frame\r\n' b'Content-Type: image/jpeg\r\n\r\n' + bytearray(im_png) + b'\r\n' + yield { + "status": 200, + "frame": b'--frame\r\n' b'Content-Type: image/jpeg\r\n\r\n' + bytearray(im_png) + b'\r\n' + } + + # while True: + # if self.num_episodes == self.agent.n_games: + # yield "done showing" + # else: + # _, self.done, self.score, self.pixel_data = self.agent.train_step(self.game) + # self.game.timeTicking() - utils.add_inside_session( - self.session_id, - config_name="metrics", - rewrite=True, multi_config=True, - episode_number = self.agent.n_games, - episode_score = score, - mean_score = total_score / self.agent.n_games - ) + # if self.done: + # self.game.initialize() + # self.agent.n_games += 1 + # self.agent.train_long_memory() + + # if self.score > self.record: + # self.agent.model.save( + # self.checkpoint_folder_path, + # 'model.pth', + # device = self.device + # ) + # self.total_score += self.score + + # utils.add_inside_session( + # self.session_id, + # config_name="metrics", + # rewrite=True, multi_config=True, + # episode_number = self.agent.n_games, + # episode_score = self.score, + # mean_score = self.total_score / self.agent.n_games + # ) + + # im_png = cv2.imencode(".png", self.pixel_data)[1] + # img_b64 = base64.b64encode(im_png).decode('utf-8') + # yield "data:image/png;base64," + img_b64 + # global lock + # plot_scores = [] + # plot_mean_scores = [] + # total_score = 0 + # record = 0 + # linear_net = LinearQNet([[5, 9], [9, 3]]) + # agent = QTrainer( + # model=linear_net, + # lr=0.001, gamma=0.9, + # epsilon=0.2, loss='mse', + # optimizer="adam", + # checkpoint_folder_path=self.checkpoint_folder_path, + # model_name = "model.pth") + # game = CarGame( + # track_num=self.env_world, + # reward_function=self._convert_str_func_to_exec( + # self.reward_function, + # function_name="reward_function" + # ), + # mode = self.mode, + # display_type="surface", + # screen_size=(800, 700) if self.mode == "training" else (1000, 700) + # ) + # while True: + # # pygame.display.update() + # _, done, score, pix = agent.train_step(game) + # game.timeTicking() + + # if done: + # game.initialize() + # agent.n_games += 1 + # agent.train_long_memory() + # if score > record: + # record = score + # agent.model.save() + # print('Game', agent.n_games, 'Score', score, 'Record:', record) + # im_png = cv2.imencode(".png", pix)[1] + # img_b64 = base64.b64encode(im_png).decode('utf-8') + # yield "data:image/png;base64," + img_b64 + - yield "sdafasfsdafsadfasdfdsfsdfdsf" \ No newline at end of file + def stream(self): + yield "sending images" \ No newline at end of file From f2a6c2e56ce842b9e34d4ad7805a7997b96c8df8 Mon Sep 17 00:00:00 2001 From: Pratyush-exe Date: Tue, 18 Apr 2023 09:34:46 +0530 Subject: [PATCH 3/5] FEAT:Added streaming and pivoted to flask --- .gitignore | 5 +- Dockerfile | 3 +- README.md | 20 ++--- flaskApp.py | 123 -------------------------- main.py | 199 +++++++++++++++++++++++++++++++++---------- requirements.txt | 4 +- src/exceptions.py | 67 --------------- src/socket_server.py | 16 ---- src/streamer.py | 199 ------------------------------------------- src/utils.py | 15 ++-- 10 files changed, 179 insertions(+), 472 deletions(-) delete mode 100644 flaskApp.py delete mode 100644 src/exceptions.py delete mode 100644 src/socket_server.py delete mode 100644 src/streamer.py diff --git a/.gitignore b/.gitignore index 412a5d2..02b17a9 100644 --- a/.gitignore +++ b/.gitignore @@ -90,4 +90,7 @@ target/ rewards-api/src/__pycache__ rewards-api/__pycache__ -tmp.py \ No newline at end of file +tmp.py +flaskApp.py +old_code +saved_models \ No newline at end of file diff --git a/Dockerfile b/Dockerfile index d240184..8923de6 100644 --- a/Dockerfile +++ b/Dockerfile @@ -19,5 +19,4 @@ RUN mkdir rewards-api COPY ./rewards-api rewards-api WORKDIR "/rewards-api" EXPOSE 8900 -ENTRYPOINT [ "uvicorn", "main:app"] -CMD [ "--reload" ] \ No newline at end of file +CMD ["python", "main.py"] \ No newline at end of file diff --git a/README.md b/README.md index 945e8b2..ab3807c 100644 --- a/README.md +++ b/README.md @@ -1,22 +1,23 @@ -### **Rewards API** +### **Rewards API**
-**rewards API** is a `REST` API where creating experiments, integrating environments and building agents are just some sets of CRUD operations and API calls. Our API internally uses [**rewards-sdk**](https://github.com/rewards-ai/rewards-SDK). +**rewards API** is a `REST` API where creating experiments, integrating environments and building agents are just some sets of CRUD operations and API calls. Our API internally uses [**rewards-sdk**](https://github.com/rewards-ai/rewards-SDK).
-#### **How to run the project** +#### **How to run the project** First clone the project by running: + ```bash git clone https://github.com/rewards-ai/rewards-api.git ``` -`rewards-api` runs on `fastapi`. So install fastapi by running: +`rewards-api` runs on `flask`. So install fastapi by running: ```bash -pip install fastapi +pip install flask ``` Now for installing additional dependencies just run: @@ -28,15 +29,14 @@ pip install -r requirements.txt Now go to the `rewards-api` directory and run: ``` -uvicorn main:app --reload +python main.py ``` -This will open the the link `http://127.0.0.1:8000` and then go to `http://127.0.0.1:8000/docs/`. There you will find all the endpoints with it's instructions and curl commands after running each. - +This will open the the link `http://127.0.0.1:8000` and then go to `http://127.0.0.1:8000/docs/`. There you will find all the endpoints with it's instructions and curl commands after running each. TODO: - [ ] Logging functionality -- [ ] Support for custom exceptions +- [ ] Support for custom exceptions - [X] Supporting streaming the game screen -- [X] Integration with the frontend \ No newline at end of file +- [X] Integration with the frontend diff --git a/flaskApp.py b/flaskApp.py deleted file mode 100644 index a528cef..0000000 --- a/flaskApp.py +++ /dev/null @@ -1,123 +0,0 @@ -import os -import cv2 -import pygame -import threading -import flask_cors -import src.utils as utils -import matplotlib.pyplot as plt -from src.config import CONFIG -from flask import Flask, Response, request, stream_with_context -from rewards import QTrainer, LinearQNet, CarGame - -app = Flask(__name__) -flask_cors.CORS(app) -lock = threading.Lock() - -def _convert_str_func_to_exec(str_function: str, function_name: str): - globals_dict = {} - exec(str_function, globals_dict) - new_func = globals_dict[function_name] - return new_func - -def generate(session_id): - r = utils.get_session_files(session_id) - print(session_id) - print("check" ,session_id) - enable_wandb = False - device = "cpu" - run_pygame_loop = False - record = 0 - done = False - - # environment parameters - env_name = r["env_params"]["environment_name"] - env_world = int(r["env_params"]["environment_world"]) - mode = r["env_params"]["mode"] - car_speed = r["env_params"]["car_speed"] - - # agent parameters - layer_config = eval(r["agent_params"]["model_configuration"]) - if type(layer_config) == str: - layer_config = eval(layer_config) - - lr = r["agent_params"]["learning_rate"] - loss = r["agent_params"]["loss_fn"] - optimizer = r["agent_params"]["optimizer"] - gamma = r["agent_params"]["gamma"] - epsilon = r["agent_params"]['epsilon'] - num_episodes = r["agent_params"]["num_episodes"] - - reward_function = r["training_params"]["reward_function"] - - global lock - - checkpoint_folder_path = os.path.join( - utils.get_home_path(), - CONFIG["REWARDS_PARENT_CONFIG_DIR"], - f"{session_id}/{CONFIG['REWARDS_CONFIG_MODEL_FOLDER_NAME']}/" - ) - - model = LinearQNet(layer_config) - - agent = QTrainer( - lr = lr, - gamma = gamma, - epsilon = epsilon, - model = model, - loss = loss, - optimizer = optimizer, - checkpoint_folder_path = checkpoint_folder_path, - model_name = "model.pth" - ) - - game = CarGame( - track_num=env_world, - mode = mode, - reward_function=_convert_str_func_to_exec( - reward_function, - function_name="reward_function" - ), - display_type="surface", - screen_size=(800, 700) - ) - game.FPS = car_speed - - record = 0 - done = False - - while True: - if agent.n_games == num_episodes: - return { - "status": 204, - "frame": "" - } - reward, done, score, pix = agent.train_step(game) - game.timeTicking() - - if done: - game.initialize() - agent.n_games += 1 - agent.train_long_memory() - if score > record: - record = score - agent.model.save( - checkpoint_folder_path, - 'model.pth', - device = "cpu" - ) - print('Game', agent.n_games, 'Score', score, 'Record:', record) - im_png = cv2.imencode(".png", pix)[1] - yield (b'--frame\r\n' b'Content-Type: image/jpeg\r\n\r\n' + bytearray(im_png) + b'\r\n') - -@app.route('/stream', methods = ['GET']) -def stream(): - session_id = request.args.get('id') - print(session_id) - return Response(generate(session_id), mimetype = "multipart/x-mixed-replace; boundary=frame") - -if __name__ == '__main__': - host = "127.0.0.1" - port = 8005 - debug = False - options = None - app.run(host, port, debug, options) \ No newline at end of file diff --git a/main.py b/main.py index 19c4cd1..d4d2963 100644 --- a/main.py +++ b/main.py @@ -1,60 +1,43 @@ -import os -import json -import pygame -from fastapi.logger import logger -from fastapi import FastAPI, Request -from fastapi.middleware.cors import CORSMiddleware -from fastapi.exceptions import RequestValidationError -from dotenv import set_key, load_dotenv - -# configs and import from other modules +from rewards import QTrainer, LinearQNet, CarGame +from werkzeug.exceptions import HTTPException +from flask import Flask, Response, request +from flask import Flask, jsonify, Request +from werkzeug.exceptions import NotFound +import matplotlib.pyplot as plt +from src.config import CONFIG +from flask_cors import CORS +import src.utils as utils +import numpy as np +import cv2 +import os -from src.config import CONFIG from src.schemas import ( AgentConfiguration, TrainingConfigurations, EnvironmentConfigurations, RewardFunction ) -from src.exceptions import ( - validation_exception_handler, - python_exception_handler -) - import src.utils as utils -from src.streamer import RewardsStreamer -# TODO: -# ----- -# Add a response model for returning all the configuration in structured format -# Also add a error response model +app = Flask(__name__) +CORS(app) +@app.errorhandler(HTTPException) +def handle_exception(e): + response = e.get_response() + response.data = jsonify({ + "code": e.code, + "name": e.name, + "description": e.description + }) + response.content_type = "application/json" + return response -app = FastAPI( - title="RewardsAI API for interacting with rewards-platform", - version="1.0.0", - description=""" - rewards-api is the easy to use API for interacting with agents and environments. - It enables users to easily create experiments and manage each of the experiments - by changing different types of parameters and reward function and also pushing the - model to other location while competing. - """ -) - -app.add_middleware( - CORSMiddleware, allow_origins=["*"], - allow_credentials=True, allow_methods=["*"], - allow_headers=["*"], -) -app.add_exception_handler(RequestValidationError, validation_exception_handler) -app.add_exception_handler(Exception, python_exception_handler) - -@app.on_event('startup') -async def startup_event(): +@app.before_first_request +def startup_event(): utils.create_folder_struct() - logger.info("Folder Created") - logger.info("Starting up") - + app.logger.info("Folder Created") + app.logger.info("Starting up") @app.post('/api/v1/create_session/{session_id}') def create_new_session(session_id : str): @@ -271,3 +254,129 @@ def get_all_envs(): @app.post("/api/v1/get_all_tracks") def get_all_tracks(): return utils.get_all_tracks("car-racer") + +def enableStreaming(): + global stop_streaming + stop_streaming = False + +def convert_str_func_to_exec(str_function: str, function_name: str): + globals_dict = {} + exec(str_function, globals_dict) + new_func = globals_dict[function_name] + return new_func + +def generate(session_id): + r = utils.get_session_files(session_id) + print(session_id) + print("check" ,session_id) + record = 0 + done = False + enableStreaming() + + # environment parameters + env_name = r["env_params"]["environment_name"] + env_world = int(r["env_params"]["environment_world"]) + mode = r["env_params"]["mode"] + car_speed = r["env_params"]["car_speed"] + + # agent parameters + layer_config = eval(r["agent_params"]["model_configuration"]) + if type(layer_config) == str: + layer_config = eval(layer_config) + + lr = r["agent_params"]["learning_rate"] + loss = r["agent_params"]["loss_fn"] + optimizer = r["agent_params"]["optimizer"] + gamma = r["agent_params"]["gamma"] + epsilon = r["agent_params"]['epsilon'] + num_episodes = r["agent_params"]["num_episodes"] + + reward_function = r["training_params"]["reward_function"] + + global lock + + checkpoint_folder_path = os.path.join( + utils.get_home_path(), + CONFIG["REWARDS_PARENT_CONFIG_DIR"], + f"{session_id}/{CONFIG['REWARDS_CONFIG_MODEL_FOLDER_NAME']}/" + ) + + model = LinearQNet(layer_config) + + agent = QTrainer( + lr = lr, + gamma = gamma, + epsilon = epsilon, + model = model, + loss = loss, + optimizer = optimizer, + checkpoint_folder_path = checkpoint_folder_path, + model_name = "model.pth" + ) + + game = CarGame( + track_num=env_world, + mode = mode, + reward_function=convert_str_func_to_exec( + reward_function, + function_name="reward_function" + ), + display_type="surface", + screen_size=(800, 700) + ) + game.FPS = car_speed + + record = 0 + plot_scores = [] + plot_mean_scores = [] + total_score = 0 + done = False + + while True: + global stop_streaming + if stop_streaming or agent.n_games == num_episodes: + return {"status": 204} + reward, done, score, pix = agent.train_step(game) + game.timeTicking() + + if done: + game.initialize() + agent.n_games += 1 + agent.train_long_memory() + if score > record: + record = score + agent.model.save( + checkpoint_folder_path, + 'model.pth', + device = "cpu" + ) + print('Game', agent.n_games, 'Score', score, 'Record:', record) + plot_scores.append(score) + total_score += score + mean_score = total_score / agent.n_games + plot_mean_scores.append(mean_score) + utils.update_graphing_file(session_id, {"plot_scores": plot_scores, "plot_mean_scores": plot_mean_scores}) + img = np.fliplr(pix) + img = np.rot90(img) + img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB) + img = cv2.imencode(".png", img)[1] + yield (b'--frame\r\n' b'Content-Type: image/jpeg\r\n\r\n' + bytearray(img) + b'\r\n') + +@app.route('/stream', methods = ['GET']) +def stream(): + session_id = request.args.get('id') + print(session_id) + return Response(generate(session_id), mimetype = "multipart/x-mixed-replace; boundary=frame") + +@app.route('/stop') +def stop(): + global stop_streaming + stop_streaming = True + return {"status": 204} + +if __name__ == '__main__': + host = "127.0.0.1" + port = 8005 + debug = True + options = None + app.run(host, port, debug, options) diff --git a/requirements.txt b/requirements.txt index 5c0ba61..5b4a9b4 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,8 +2,8 @@ torch pandas numpy scikit-learn -fastapi==0.88.0 +Flask==2.2.3 +Flask-Cors==3.0.10 rewards -wandb matplotlib pygame \ No newline at end of file diff --git a/src/exceptions.py b/src/exceptions.py deleted file mode 100644 index b1ed5ff..0000000 --- a/src/exceptions.py +++ /dev/null @@ -1,67 +0,0 @@ -import json -import traceback - -from fastapi.logger import logger -from fastapi import Request, status -from fastapi.responses import JSONResponse -from fastapi.exceptions import RequestValidationError - -from src.config import CONFIG - -# TODO: Also put a custom error handler message and -# also a logging function that can log all the exceptions - - -def get_error_response(request, exc) -> dict: - """Generic error handling function - - Args: - request (Request): Incoming request - exc (_type_): execution - - Returns: - dict: Error response - """ - error_ressonse = {"error": True, "message": str(exc)} - - if CONFIG["DEBUG"]: - error_ressonse["traceback"] = "".join( - traceback.format_exception(type(exc), value=exc, tb=exc.__traceback__) - ) - return error_ressonse - - -async def validation_exception_handler(request: Request, exc: RequestValidationError): - """ - Handling error in validating requests - """ - return JSONResponse( - status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, content=get_error_response(request, exc) - ) - - -async def python_exception_handler(request: Request, exc: Exception): - """ - Handling any internal error - """ - - # Log requester infomation - logger.error( - "Request info:\n" - + json.dumps( - { - "host": request.client.host, - "method": request.method, - "url": str(request.url), - "headers": str(request.headers), - "path_params": str(request.path_params), - "query_params": str(request.query_params), - "cookies": str(request.cookies), - }, - indent=4, - ) - ) - - return JSONResponse( - status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content=get_error_response(request, exc) - ) diff --git a/src/socket_server.py b/src/socket_server.py deleted file mode 100644 index af711d8..0000000 --- a/src/socket_server.py +++ /dev/null @@ -1,16 +0,0 @@ -import asyncio -import websockets - -import src.utils as utils -from src.config import CONFIG -from src.streamer import RewardsStreamer - -class WebsocketStreamingServer: - def __init__(self, host, port, secret): - self._secret = secret - - async def _start(self, websocket, endpoint): - print(f"=> Connected to {endpoint}") - secret = await websocket.recv() - if secret == self._secret : - ... \ No newline at end of file diff --git a/src/streamer.py b/src/streamer.py deleted file mode 100644 index 9fe26a5..0000000 --- a/src/streamer.py +++ /dev/null @@ -1,199 +0,0 @@ -import os -import cv2 -import pygame -import base64 -import numpy as np -from typing import Dict, Any -import multiprocessing as mp - -import src.utils as utils -from src.config import CONFIG -from dotenv import load_dotenv, set_key - -from rewards import workflow, QTrainer, LinearQNet, CarGame - -from fastapi.responses import JSONResponse -from fastapi.encoders import jsonable_encoder - -class RewardsStreamer: - def __init__(self, session_id, response: Dict[str, Any]) -> None: - """ - RewardsStreamer is the class which is responsible for streaming, saving - metrics and also streaming frames over the network. - """ - print("called") - self.session_id = session_id - self.enable_wandb = False - self.device = "cpu" - self.run_pygame_loop = False - self.record = 0 - self.done = False - - # environment parameters - self.env_name = response["env_params"]["environment_name"] - self.env_world = int(response["env_params"]["environment_world"]) - self.mode = response["env_params"]["mode"] - self.car_speed = response["env_params"]["car_speed"] - - # agent parameters - self.layer_config = eval(response["agent_params"]["model_configuration"]) - if type(self.layer_config) == str: - self.layer_config = eval(self.layer_config) - - self.lr = response["agent_params"]["learning_rate"] - self.loss = response["agent_params"]["loss_fn"] - self.optimizer = response["agent_params"]["optimizer"] - self.gamma = response["agent_params"]["gamma"] - self.epsilon = response["agent_params"]['epsilon'] - self.num_episodes = response["agent_params"]["num_episodes"] - - self.reward_function = response["training_params"]["reward_function"] - - def _convert_str_func_to_exec(self, str_function: str, function_name: str): - """ - Converts a string like function skeleton to a Callable - - Args: - str_function (str): The string function skeleton - function_name (str): The name of the function - - Returns: - Callable: Actual callable function of type <'function'> - """ - globals_dict = {} - exec(str_function, globals_dict) - new_func = globals_dict[function_name] - return new_func - - def stream_episode(self, yield_response: bool = False): - global lock - - self.checkpoint_folder_path = os.path.join( - utils.get_home_path(), - CONFIG["REWARDS_PARENT_CONFIG_DIR"], - f"{self.session_id}/{CONFIG['REWARDS_CONFIG_MODEL_FOLDER_NAME']}/" - ) - - self.model = LinearQNet(self.layer_config) - - self.agent = QTrainer( - lr = self.lr, - gamma = self.gamma, - epsilon = self.epsilon, - model = self.model, - loss = self.loss, - optimizer = self.optimizer, - checkpoint_folder_path = self.checkpoint_folder_path, - model_name = "model.pth" - ) - - self.game = CarGame( - track_num=self.env_world, - mode = self.mode, - reward_function=self._convert_str_func_to_exec( - self.reward_function, - function_name="reward_function" - ), - display_type="surface", - screen_size=(800, 700) if self.mode == "training" else (1000, 700) - ) - self.game.FPS = self.car_speed - - self.record = 0 - self.done = False - - while True: - print("in while") - reward, self.done, self.score, pix = self.agent.train_step(self.game) - self.game.timeTicking() - - if self.done: - self.game.initialize() - self.agent.n_games += 1 - self.agent.train_long_memory() - if self.score > self.record: - self.record = self.score - # self.agent.model.save() - print('Game', self.agent.n_games, 'Score', self.score, 'Record:', self.record) - im_png = cv2.imencode(".png", pix)[1] - # yield b'--frame\r\n' b'Content-Type: image/jpeg\r\n\r\n' + bytearray(im_png) + b'\r\n' - yield { - "status": 200, - "frame": b'--frame\r\n' b'Content-Type: image/jpeg\r\n\r\n' + bytearray(im_png) + b'\r\n' - } - - # while True: - # if self.num_episodes == self.agent.n_games: - # yield "done showing" - # else: - # _, self.done, self.score, self.pixel_data = self.agent.train_step(self.game) - # self.game.timeTicking() - - # if self.done: - # self.game.initialize() - # self.agent.n_games += 1 - # self.agent.train_long_memory() - - # if self.score > self.record: - # self.agent.model.save( - # self.checkpoint_folder_path, - # 'model.pth', - # device = self.device - # ) - # self.total_score += self.score - - # utils.add_inside_session( - # self.session_id, - # config_name="metrics", - # rewrite=True, multi_config=True, - # episode_number = self.agent.n_games, - # episode_score = self.score, - # mean_score = self.total_score / self.agent.n_games - # ) - - # im_png = cv2.imencode(".png", self.pixel_data)[1] - # img_b64 = base64.b64encode(im_png).decode('utf-8') - # yield "data:image/png;base64," + img_b64 - # global lock - # plot_scores = [] - # plot_mean_scores = [] - # total_score = 0 - # record = 0 - # linear_net = LinearQNet([[5, 9], [9, 3]]) - # agent = QTrainer( - # model=linear_net, - # lr=0.001, gamma=0.9, - # epsilon=0.2, loss='mse', - # optimizer="adam", - # checkpoint_folder_path=self.checkpoint_folder_path, - # model_name = "model.pth") - # game = CarGame( - # track_num=self.env_world, - # reward_function=self._convert_str_func_to_exec( - # self.reward_function, - # function_name="reward_function" - # ), - # mode = self.mode, - # display_type="surface", - # screen_size=(800, 700) if self.mode == "training" else (1000, 700) - # ) - # while True: - # # pygame.display.update() - # _, done, score, pix = agent.train_step(game) - # game.timeTicking() - - # if done: - # game.initialize() - # agent.n_games += 1 - # agent.train_long_memory() - # if score > record: - # record = score - # agent.model.save() - # print('Game', agent.n_games, 'Score', score, 'Record:', record) - # im_png = cv2.imencode(".png", pix)[1] - # img_b64 = base64.b64encode(im_png).decode('utf-8') - # yield "data:image/png;base64," + img_b64 - - - def stream(self): - yield "sending images" \ No newline at end of file diff --git a/src/utils.py b/src/utils.py index c10e96a..f485d71 100644 --- a/src/utils.py +++ b/src/utils.py @@ -1,17 +1,18 @@ -import os -import json -import shutil -from typing import Optional, Union, Any, Dict - -# rewards package +from typing import Optional, Union, Any, Dict from src.config import CONFIG -from rewards import workflow, Agent +import shutil +import json +import os def get_home_path(): # get the home directory using os return os.path.expanduser("~") +def update_graphing_file(session_id: str, data: dict, dir: Optional[str] = None, name: Optional[str] = None) -> None: + f = open("D:/Prototypes/rewards.ai/training-platform\src/assets/temp.json", "w") + f.write(json.dumps(data)) + def create_folder_struct(dir: Optional[str] = None, name: Optional[str] = None) -> None: """Creates a root folder to store all the session configuration From 6a6ad313b78c24eefbf1fff6ec803ebe5a552621 Mon Sep 17 00:00:00 2001 From: Pratyush-exe Date: Tue, 18 Apr 2023 10:42:20 +0530 Subject: [PATCH 4/5] FIX:fixed minor bugs, removed schema --- main.py | 163 +++++++++++++++++++++++++++----------------------------- 1 file changed, 80 insertions(+), 83 deletions(-) diff --git a/main.py b/main.py index d4d2963..b2e03fb 100644 --- a/main.py +++ b/main.py @@ -1,37 +1,20 @@ from rewards import QTrainer, LinearQNet, CarGame from werkzeug.exceptions import HTTPException -from flask import Flask, Response, request -from flask import Flask, jsonify, Request +from flask import Flask, Response, request, Request from werkzeug.exceptions import NotFound import matplotlib.pyplot as plt from src.config import CONFIG from flask_cors import CORS import src.utils as utils +from flasgger import Swagger import numpy as np +import json import cv2 import os -from src.schemas import ( - AgentConfiguration, - TrainingConfigurations, - EnvironmentConfigurations, - RewardFunction -) -import src.utils as utils - app = Flask(__name__) CORS(app) - -@app.errorhandler(HTTPException) -def handle_exception(e): - response = e.get_response() - response.data = jsonify({ - "code": e.code, - "name": e.name, - "description": e.description - }) - response.content_type = "application/json" - return response +Swagger(app) @app.before_first_request def startup_event(): @@ -39,19 +22,20 @@ def startup_event(): app.logger.info("Folder Created") app.logger.info("Starting up") -@app.post('/api/v1/create_session/{session_id}') -def create_new_session(session_id : str): - """ - This create a new session in the rewards-platform where - user can now initialize with different environment, agent configuration. +@app.post('/api/v1/create_session') +def create_new_session(): + ''' + Create a new session in the rewards-platform where user can now initialize with different environment, agent configuration. - Args: - - - `session_id (str)`:The session ID is a unique string with a format of __