Map-It-engine

fileRecieverServer.py

@@ -0,0 +1,127 @@
+import os
+from http.server import BaseHTTPRequestHandler
+from email.parser import BytesParser
+from email.policy import default
+import re
+
+
+UPLOAD_DIR = 'map'
+UPLOAD_MAP_NAME = 'custom_map.csv'
+
+
+def parse_content_disposition(header_value):
+    parts = header_value.split(';')
+    disposition = parts[0].strip().lower()
+    params = {}
+    for part in parts[1:]:
+        if '=' in part:
+            key, val = part.strip().split('=', 1)
+            params[key.lower()] = val.strip('"')
+    return disposition, params
+
+
+class FileRecieverServer(BaseHTTPRequestHandler):
+
+    def _set_cors_headers(self):
+        self.send_header('Access-Control-Allow-Origin', '*')  # Allow all origins
+        self.send_header('Access-Control-Allow-Methods', 'POST, OPTIONS')
+        self.send_header('Access-Control-Allow-Headers', 'Content-Type')
+
+    def do_OPTIONS(self):
+        self.send_response(204)  # No content
+        self._set_cors_headers()
+        self.end_headers()
+
+    def do_POST(self):
+        if self.path == '/activateRandomMap':
+            self.send_response(200)
+            self._set_cors_headers()
+            self.end_headers()
+
+            src = os.path.join(UPLOAD_DIR, "disabled_custom_map.csv")
+
+            if os.path.exists(src):
+                os.rename(src, "map/custom_map.csv")
+                self.wfile.write(b"Map activated")
+            else:
+                self.send_response(404)
+                self.end_headers()
+                self.wfile.write(b"custom_map.csv not found")
+            return
+
+        elif self.path == '/deactivateRandomMap':
+            active_map_path = os.path.join(UPLOAD_DIR, UPLOAD_MAP_NAME)
+            if os.path.exists(active_map_path):
+                os.rename(active_map_path, f"map/disabled_custom_map.csv")
+
+            self.send_response(200)
+            self._set_cors_headers()
+            self.end_headers()
+            self.wfile.write(b"Map deactivated")
+            return
+
+        elif self.path == '/upload':
+            content_type = self.headers.get('Content-Type')
+            if not content_type or not content_type.startswith('multipart/form-data'):
+                self.send_response(400)
+                self._set_cors_headers()
+                self.end_headers()
+                self.wfile.write(b"Invalid content type")
+                return
+
+            match = re.search('boundary=(.*)', content_type)
+            if not match:
+                self.send_response(400)
+                self._set_cors_headers()
+                self.end_headers()
+                self.wfile.write(b"Missing boundary")
+                return
+
+            content_length = int(self.headers.get('Content-Length', 0))
+            body = self.rfile.read(content_length)
+
+            full_message = (
+                f"Content-Type: {content_type}\r\n\r\n".encode() + body
+            )
+            msg = BytesParser(policy=default).parsebytes(full_message)
+
+            for part in msg.iter_parts():
+                disposition_header = part.get('Content-Disposition', '')
+                disposition, params = parse_content_disposition(disposition_header)
+
+                if disposition != 'form-data' or 'filename' not in params:
+                    continue
+
+                file_data = part.get_payload(decode=True)
+
+                os.makedirs(UPLOAD_DIR, exist_ok=True)
+                filepath = os.path.join(UPLOAD_DIR, UPLOAD_MAP_NAME)
+
+                with open(filepath, 'wb') as f:
+                    f.write(file_data)
+
+                self.send_response(200)
+                self._set_cors_headers()
+                self.end_headers()
+                self.wfile.write(b"File uploaded successfully.")
+                return
+
+            self.send_response(400)
+            self._set_cors_headers()
+            self.end_headers()
+            self.wfile.write(b"No valid file part found")
+
+        else:
+            self.send_response(404)
+            self._set_cors_headers()
+            self.end_headers()
+            self.wfile.write(b"Endpoint not found")
+
+
+    def do_GET(self):
+        self.send_response(405)
+        self._set_cors_headers()
+        self.end_headers()
+        self.wfile.write(b"Use POST for all operations")
+
+

main.py

@@ -1,7 +1,11 @@
 import argparse
 import asyncio
+from http.server import HTTPServer
 import logging
+import os
+import threading
 
+from fileRecieverServer import FileRecieverServer
 from rose.engine import server
 
 
@@ -55,5 +59,28 @@ def main():
     )
 
 
+def startServer():
+    server_address = ('', 8000)
+    httpd = HTTPServer(server_address, FileRecieverServer)
+    print("Serving on http://localhost:8000")
+    httpd.serve_forever()
+
+
+def deleteCustomMap():
+    if os.path.exists("map/custom_map.csv"):
+        os.remove("map/custom_map.csv")
+    if os.path.exists("map/disabled_custom_map.csv"):
+        os.remove("map/disabled_custom_map.csv")
+
 if __name__ == "__main__":
-    main()
+
+    threading.Thread(target=startServer,daemon=True).start()
+    try:
+        main()
+    finally:
+        try:
+            deleteCustomMap()    
+        except FileNotFoundError:
+            print("No custom map files to remove.")
+
+

rose/engine/csv_file_handler.py

@@ -0,0 +1,75 @@
+import csv
+from pathlib import Path
+from typing import Union
+
+class CsvFileHandler:
+    @staticmethod
+    def add_line(file_path: Union[Path, str], row: list[str]) -> bool:
+        """
+        Appends a list of strings as a new row to a CSV file.
+
+        Args:
+            file_path (Path | str): Path to the CSV file.
+            row (list[str]): List of strings representing a row.
+
+        Returns:
+            bool: True if the row was successfully written, False on failure.
+        """
+        try:
+            path = Path(file_path)
+
+            if not path.exists():
+                print(f"[ERROR] File '{path}' does not exist.")
+                return False
+
+            if path.suffix.lower() != ".csv":
+                print(f"[ERROR] File '{path}' is not a CSV file.")
+                return False
+
+            with path.open(mode='a', newline='', encoding='utf-8') as file:
+                writer = csv.writer(file)
+                writer.writerow(row)
+
+            return True
+
+        except OSError as os_err:
+            print(f"[ERROR] File system error: {os_err}")
+            return False
+        except Exception as e:
+            print(f"[ERROR] Unexpected error while writing to CSV: {e}")
+            return False
+
+    @staticmethod
+    def read_as_matrix(file_path: Union[Path, str]) -> list[list[str]]:
+        """
+        Reads a CSV file and returns its contents as a matrix (list of rows).
+
+        Args:
+            file_path (Path | str): Path to the CSV file.
+
+        Returns:
+            list[list[str]]: Matrix of strings representing the CSV content.
+        """
+        try:
+            path = Path(file_path)
+
+            if not path.exists():
+                print(f"[ERROR] File '{path}' does not exist.")
+                return []
+
+            if path.suffix.lower() != ".csv":
+                print(f"[ERROR] File '{path}' is not a CSV file.")
+                return []
+
+            with path.open(mode='r', newline='', encoding='utf-8') as file:
+                reader = csv.reader(file)
+                matrix = [row for row in reader]
+
+            return matrix
+
+        except OSError as os_err:
+            print(f"[ERROR] File system error: {os_err}")
+            return []
+        except Exception as e:
+            print(f"[ERROR] Unexpected error while reading from CSV: {e}")
+            return []

rose/engine/track.py

@@ -1,21 +1,31 @@
 import random
 
+import os
 from rose.engine import config
 from rose.common import obstacles
+from rose.engine import csv_file_handler
 
 
 class Track(object):
     def __init__(self, is_track_random=False):
         self._matrix = None
         self.is_track_random = is_track_random
         self.reset()
+        self.custom_index = 0
+        self.custom_map = csv_file_handler.CsvFileHandler.read_as_matrix("map/custom_map.csv")
 
-    # Game state interface
 
+
+
+        # Game state interface
     def update(self):
         """Go to the next game state"""
         self._matrix.pop()
-        self._matrix.insert(0, self._generate_row())
+        if os.path.exists("map/custom_map.csv") and self.custom_map != []:
+            self.custom_map = self.check_obstacle(self.custom_map)
+            self._matrix.insert(0, self.generate_custom_map(self.custom_map))
+        else:
+            self._matrix.insert(0, self._generate_row())
 
     def state(self):
         """Return read only serialize-able state for sending to client"""
@@ -80,3 +90,24 @@ def _generate_row(self):
                 row[cell + lane * config.cells_per_player] = obstacle
 
         return row
+
+    def check_obstacle(self, custom_map):
+        for row in range(len(custom_map)-1):
+            for col in range(len(custom_map[row])-1):
+                if custom_map[row][col] not in obstacles.ALL:
+                    print(custom_map[row][col])
+                    custom_map[row][col] = obstacles.get_random_obstacle()
+        return custom_map
+
+    def generate_custom_map(self,custom_map):
+        if self.custom_index >= len(custom_map):
+            self.custom_index = 0
+
+        row = custom_map[self.custom_index]
+        self.custom_index += 1
+
+        return [
+            getattr(obstacles, value.upper(), obstacles.NONE) if value else obstacles.NONE
+            for value in row
+        ]
+