Sheep compatiability

.gitignore

@@ -144,5 +144,10 @@ cython_debug/
 # Non-generic gitignores
 ##################################################################
 
-usercode/*
-logs.txt
+usercode/round/*
+usercode/editable/*
+!usercode/editable/main.py
+!usercode/editable/minimal_example.py
+!usercode/editable/snippests.py
+logs.txt
+run.sh

README.md

@@ -22,6 +22,24 @@ Additionally there are a whole load of features which a new shepherd could suppo
 
 ## Getting started
 
+## Use the script
+
+Automatically install everything and set up the virtual env:
+
+```
+sh install.sh
+```
+
+The script is able to build python3.10 if you do not have it, alternatively you
+can follow the instructions here:
+ - Guide: https://realpython.com/installing-python/#how-to-build-python-from-source-code
+ - Offical python docs: https://docs.python.org/3/using/unix.html
+
+If building on a pi you will need to increase your swap space. 2GB has been
+tested to work on a pi 3A+.
+
+## Manual install
+
 Dependencies are managed using poetry which you can get
 [here](https://python-poetry.org/docs/master/#installing-with-the-official-installer)
 
@@ -39,16 +57,23 @@ You can enter this virtual environment by:
 poetry shell
 ```
 
-Start the server in development mode (hot reload)
+## Running shepherd
+
+Start the server in development mode (hot reload):
 ```
-dev = "poetry run uvicorn app:app --reload"
+poetry run uvicorn app:app --reload
 ```
 
-Run tests
+To deploy run (chosen port and ip are optional):
 ```
-pytest -x
+poetry run uvicorn app:app --host 0.0.0.0 --port 80
 ```
 
-## Documentation
+## Tests
 
-Documentation is available for this project here: TODO
+Tests are located in `test/`
+
+Run all tests and stop on the first fail with a verbose output:
+```
+pytest -xv
+```

app/__init__.py

@@ -1,6 +1,6 @@
 import pretty_errors
 
-from app.main import app
+from app.main import shepherd
 
 pretty_errors.configure(
     separator_character = '*',

app/config.py

@@ -1,29 +1,105 @@
 import os
 
-from pydantic import BaseSettings
+import logging
 from pathlib import Path
+import tempfile
 
 
-class Settings(BaseSettings):
+class Settings:
 
-    output_file_path: Path = Path("logs.txt").absolute()
-    usercode_path: Path = Path("usercode/").absolute()
-    usercode_entry_point = Path("main.py")
-    usercode_entry_path: Path = (usercode_path / usercode_entry_point).absolute()
-    sheep_files_path: Path = Path("static/editor").absolute()
+    log_file_path: Path = Path("logs.txt")
+    round_path: Path = Path("usercode/round").absolute()
+    round_entry_point = Path("main.py")
+    round_entry_path: Path = (round_path / round_entry_point).absolute()
+
+    editor_path: Path = Path("static/editor/").absolute()
+    docs_path: Path = Path("static/docs/").absolute()
 
     round_len: float = 180.0  # seconds
     reap_grace_time: float = 5.0  # seconds
-    arenaUSB_path: Path = Path("/media/ArenaUSB")
+    arena_usb_path: Path = Path("/media/ArenaUSB")
 
-    robot_path: Path = Path("/home/pi/robot").absolute()
+    robot_path: Path = Path("/home/pi/robot")
     robot_env: dict = dict(os.environ)
     on_brain: bool = False
 
-    upload_tmp_dir: str = "shepherd-user-code-"
+    robot_usb_path: Path = Path("/media/RobotUSB")
+    teamname_file: Path = Path("/home/pi/teamname.txt")
+    zone: bool = False
 
+    # tempfile.mktemp is deprecated, but there's no possibility of a race --
+    usr_fifo_path = tempfile.mktemp(prefix="shepherd-fifo-")
 
-config = Settings()
+    def __init__(self):
+        self._on_brain()
+        self._init_usercode_folder()
+        self._zone_from_USB()
+
+        # os.mkfifo raises if its path already exists.
+        os.mkfifo(self.usr_fifo_path)
+
+    def _on_brain(self):
+        """Detects if we are on a brain and alters the config accordingly
+        Looks to see if the robot_usb path exists, if it does then we are
+        probably on a configured brain rather than a dev PC
+        """
+        if self.robot_usb_path.exists():
+            logging.warning("Detected RobotUSB path assuming on brain")
+            self.log_file_path = self.robot_usb_path / self.log_file_path
+            config.robot_env["PYTHONPATH"] = config.robot_path
+            self._get_team_specifics()
+
+    def _init_usercode_folder(self):
+        """Ensure that the saved usercode has a main.py and blocks.json"""
+        self.usr_src_path = Path("usercode/editable").absolute()
+        if not self.usr_src_path.exists():
+            os.mkdir(self.usr_src_path)
+
+        self.usr_src_main_path = self.usr_src_path / Path('main.py')
+        with open(self.usr_src_main_path, 'w') as main_file:
+            main_file.write('# DO NOT DELETE\n')
+        self.blocks_path = self.usr_src_path / Path('blocks.json')
+
+        if self.round_path.exists() is False:
+            os.mkdir(self.round_path)
+
+    def _zone_from_USB(self):
+        """Set the zone based on the ARENAUSB, defaulting to zone 0"""
+        self.zone = "0"
+        for i in range(1, 4):
+            if (self.arena_usb_path / f"zone{i}.txt").exists():
+                self.zone = str(i)
+                return
 
-if config.on_brain is True:
-    config.robot_env["PYTHONPATH"] = config.robot_path
+    def _get_team_specifics(self):
+        """Find information set on each brain about the team
+
+        Only makes sense to run this if we are on a brain
+        Teamname is set per brain before shipping and allows unique graphics
+        for ID'ing teams in the arena.
+
+        Pick a start image in order of preference :
+            1) We have a team corner image on the USB
+            2) The team have uploaded their own image to the robot
+            3) We have a generic corner image on the USB
+            4) The game image
+        """
+        if self.teamname_file.exists():
+            teamname_jpg = self.teamname_file.read_text().replace('\n', '') + '.jpg'
+        else:
+            teamname_jpg = 'none'
+
+        start_img_path = self.arena_usb_path / teamname_jpg
+        if not start_img_path.exists():
+            start_img_path = Path('usercode/editable/team_logo.jpg')
+        if not start_img_path.exists():
+            start_img_path = self.arena_usb_path / 'Corner.jpg'
+        if not start_img_path.exists():
+            start_img_path = Path('/home/pi/game_logo.jpg')
+
+        if start_img_path.exists():
+            displayed_img_path = Path('shepherd/static/image.jpg')
+            displayed_img_path.write_bytes(start_img_path.read_bytes())
+
+
+config = Settings()

app/editor.py

@@ -0,0 +1,73 @@
+"""Functions used by endpoints for sheep"""
+from app.config import config
+import os
+import json
+import re
+from pathlib import Path
+
+
+def get_files() -> dict:
+    """This read the editable files from the file system and returns them as json
+    The blockly blocks are saved in a `blocks.json` and are converted into a
+    standardized dictionary.
+    """
+    project_paths = [f for f in os.listdir(config.usr_src_path)
+                     if os.path.isfile(os.path.join(config.usr_src_path, f))
+                     and (f.endswith('.py') or f.endswith(".xml") or f == "blocks.json")
+                     and f != 'main.py']
+
+    def read_project(project_path: Path) -> dict:
+        with open(config.usr_src_path / project_path, 'r') as project_file:
+            content = project_file.read()
+        return {
+            'filename': project_path,
+            'content': content
+        }
+
+    blocks = {}
+    if config.blocks_path.exists():
+        with open(config.blocks_path, 'r') as blocks_file:
+            try:
+                blocks = json.load(blocks_file)
+            except ValueError:
+                pass
+
+    if "requires" not in blocks:
+        blocks["requires"] = []
+    if "header" not in blocks:
+        blocks["header"] = ""
+    if "footer" not in blocks:
+        blocks["footer"] = ""
+    if "blocks" not in blocks:
+        blocks["blocks"] = []
+
+    return {
+        'main': config.usr_src_main_path.absolute(),
+        'blocks': blocks,
+        'projects': [read_project(p) for p in project_paths]
+    }
+
+
+def save_file(filename, body):
+    """Write a file with `filename` and `body` to the filesystem"""
+    dots = len(re.findall("\\.", filename))
+    if dots == 1:
+        with open(os.path.join(config.usr_src_path, filename), 'w') as f:
+            f.write(body.decode('utf-8'))
+    else:
+        pass
+        logging.warning("A file was attempted to be saved with too many dots: "
+                     f"{filename}")
+
+
+def delete_file(filename):
+    """Remove a file with `filename`"""
+    if filename == "blocks.json":
+        return ""
+    dots = len(re.findall("\\.", filename))
+    if dots == 1:
+        os.unlink(os.path.join(config.usr_src_path, filename))
+    else:
+        pass
+        logging.warning("A file was attempted to be saved with too many dots: "
+                     f"{filename}")

app/logging.py

@@ -0,0 +1,119 @@
+"""Configure the logging of shepherd so it is readable and pretty"""
+import logging
+import click
+from copy import copy
+import http
+
+logger = logging.getLogger("app")
+
+
+class CustomFormatter(logging.Formatter):
+    """A base formatter class containing the colours"""
+
+    grey = "\x1b[38;20m"
+    green = "\x1b[32;20m"
+    yellow = "\x1b[33;20m"
+    bold_red = "\x1b[31;20m"
+    pale_red = "\x1b[31;1m"
+    reset = "\x1b[0m"
+    message_format = "%(message)s"
+
+    coloured_levels = {
+        logging.DEBUG: grey + "DEBUG" + reset + ":    ",
+        logging.INFO: green + "INFO" + reset + ":     ",
+        logging.WARNING: yellow + "WARNING" + reset + ":  ",
+        logging.ERROR: pale_red + "ERROR" + reset + ":    ",
+        logging.CRITICAL: bold_red + "CRITICAL" + reset + ": ",
+    }
+
+
+class ShepherdFormatter(CustomFormatter):
+    """A formatter which prints the file and line number"""
+
+    def format(self, record):
+        location = f"{record.filename}:{record.lineno}"
+        message = record.msg % record.args  # Old style python formatting
+        return f"{self.coloured_levels[record.levelno]}{location:<21}| {message}"
+
+
+class UvicornFormatter(CustomFormatter):
+    """The same as the ShepherdFormatter but prints uvicorn instead of path"""
+
+    def format(self, record):
+        location = "Uvicorn"
+        message = record.msg % record.args  # Old style python formatting
+        return f"{self.coloured_levels[record.levelno]}{location:<21}| {message}"
+
+
+class UvicornAccessFormatter(CustomFormatter):
+    """All API calls are logged using this formatter
+    Based on the uvicorn.logging
+    """
+    status_code_colours = {
+        1: lambda code: click.style(str(code), fg="bright_white"),
+        2: lambda code: click.style(str(code), fg="green"),
+        3: lambda code: click.style(str(code), fg="yellow"),
+        4: lambda code: click.style(str(code), fg="red"),
+        5: lambda code: click.style(str(code), fg="bright_red"),
+    }
+
+    def get_status_code(self, status_code: int) -> str:
+        """Format the status code using"""
+        try:
+            status_phrase = http.HTTPStatus(status_code).phrase
+        except ValueError:
+            status_phrase = ""
+        status_and_phrase = "%s %s" % (status_code, status_phrase)
+
+        def default(code: int) -> str:
+            return status_and_phrase
+
+        func = self.status_code_colours.get(status_code // 100, default)
+        return func(status_and_phrase)
+
+    def format(self, record):
+        """Assemble a complete message with a coloured http code"""
+        recordcopy = copy(record)
+        (
+            client_addr,
+            method,
+            full_path,
+            http_version,
+            status_code,
+        ) = recordcopy.args
+        status_code = self.get_status_code(int(status_code))
+        request_line = "%s %s HTTP/%s" % (method, full_path, http_version)
+        request_line = click.style(request_line, bold=True)
+        location = "Uvicorn.access"
+        return (f"{self.coloured_levels[record.levelno]}{location:<20} | "
+                f"{client_addr:<16} {status_code:<24} {request_line} ")
+
+
+def configure_logger(name: str, new_fomater, level):
+    """Applies `new_formater` to a logger specified by `name`
+    Ensure that this logger does not interfer with the root logger
+    Sets the loggers level to `level`
+    """
+    logger = logging.getLogger(name)
+    logger.propagate = False
+    if len(logger.handlers) > 0:
+        logger.handlers[0].setFormatter(new_fomater())
+    logger.setLevel(level)
+
+
+def configure_logging(level=logging.DEBUG, third_party_level=logging.INFO):
+    """Apply the logging formatters to the logging handlers
+    Prevents uvicorn's handlers from double logging
+    Sets logging level to `level`
+    `uvicorn_level` allows the shepherd logs to come through by raising the uvicorn level
+    """
+    configure_logger("uvicorn", UvicornFormatter, third_party_level)
+    configure_logger("uvicorn.access", UvicornAccessFormatter, third_party_level)
+    configure_logger("multipart", logging.Formatter, third_party_level)
+
+    logger.setLevel(level)
+    ch = logging.StreamHandler()
+    ch.setLevel(level)
+    ch.setFormatter(ShepherdFormatter())
+    logger.addHandler(ch)
+    logger.info("Logging configured")