Add config class

arteria/config_schemas/schema_arteria_runfolder.py

@@ -0,0 +1,29 @@
+runfolder_schema = {
+    "type": "object",
+    "properties": {
+        "monitored_directories": {
+            "type": "array",
+            "items": {
+                "type": "string",
+                "minLength": 1,
+            },
+            "minItems": 1,
+        },
+        "completed_marker_grace_minutes": {
+            "type": "number",
+        },
+        "port": {
+            "type": "number",
+        },
+        "logger_config_file": {
+            "type": "string",
+            "minLength": 1,
+        }
+    },
+    "required": [
+        "monitored_directories",
+        "completed_marker_grace_minutes",
+        "port",
+        "logger_config_file",
+    ]
+}

arteria/models/config.py

@@ -0,0 +1,70 @@
+import copy
+
+from jsonschema import validate
+
+
+class Config:
+    """
+    Class to make the app's config easily available throughout the program.
+
+    A Config instance possesses a global config dictionary available to all
+    instances, as well as a local one that can be used to provide default
+    values, when they are not defined in the global dictionary.
+
+    Variables stored in the config are meant to be constant throughout the
+    program and updating them after initialization them should be avoided.
+    """
+
+    def __new__(cls, *args, **kwargs):
+        if hasattr(cls, "_instance"):
+            return cls._instance
+
+        config = super(Config, cls).__new__(cls)
+        config._global_config_dict = {}
+        return config
+
+    def __init__(self, local_config_dict=None):
+        config_dict = copy.deepcopy(local_config_dict) or {}
+        config_dict.update(self._global_config_dict)
+        self._config_dict = config_dict
+
+    @classmethod
+    def new(cls, global_config_dict, exist_ok=False, schema=None):
+        """
+        Initialize a new global config.
+
+        Raises AssertionError if a Config already exists, unles `exist_ok` is
+        set to True.
+
+        Validates the global_config_dict with jsonschema
+        """
+        assert not hasattr(cls, "_instance") or exist_ok, "Config has already been initialized"
+        if schema:
+            validate(instance=global_config_dict, schema=schema)
+
+        if not hasattr(cls, "_instance"):
+            cls._instance = super(Config, cls).__new__(cls)
+        cls._instance._global_config_dict = copy.deepcopy(global_config_dict)
+
+        return cls()
+
+    @classmethod
+    def clear(cls):
+        """
+        Clear existing config.
+        """
+        if hasattr(cls, "_instance"):
+            del cls._instance
+
+    def __getitem__(self, item):
+        return self._config_dict[item]
+
+    def get(self, item, default=None):
+        return self._config_dict.get(item, default)
+
+    def to_dict(self):
+        """
+        Returns config as dict
+        """
+        # Copy is returned to avoid accidentaly modifying the original config.
+        return copy.deepcopy(self._config_dict)

arteria/models/runfolder_utils.py

@@ -6,9 +6,14 @@
 
 from pathlib import Path
 from arteria.models.state import State
+from arteria.models.config import Config
 
 log = logging.getLogger(__name__)
 
+DEFAULT_CONFIG = {
+    "completed_marker_grace_minutes": 0,
+}
+
 
 def list_runfolders(monitored_directories, filter_key=lambda r: True):
     """
@@ -30,7 +35,11 @@ def list_runfolders(monitored_directories, filter_key=lambda r: True):
     return runfolders
 
 class Runfolder():
-    def __init__(self, path, grace_minutes=0):
+    """
+    A class to manipulate runfolders on disk
+    """
+    def __init__(self, path):
+        self.config = Config(DEFAULT_CONFIG)
         self.path = Path(path)
         assert self.path.is_dir()
         try:
@@ -43,14 +52,17 @@ def __init__(self, path, grace_minutes=0):
                 if path.exists()
             )
             self.run_parameters = xmltodict.parse(run_parameter_file.read_text())["RunParameters"]
-        except StopIteration as e:
-            raise AssertionError(f"File [Rr]unParameters.xml not found in runfolder {path}")
+        except StopIteration as exc:
+            raise AssertionError(f"File [Rr]unParameters.xml not found in runfolder {path}") from exc
 
         marker_file_name = Instrument(self.run_parameters).completed_marker_file
         marker_file = (self.path / marker_file_name)
         assert (
-                marker_file.exists()
-                and time.time() - os.path.getmtime(marker_file) > grace_minutes * 60
+            marker_file.exists()
+            and (
+                time.time() - os.path.getmtime(marker_file)
+                > self.config["completed_marker_grace_minutes"] * 60
+            )
         )
 
         (self.path / ".arteria").mkdir(exist_ok=True)
@@ -69,6 +81,14 @@ def state(self, new_state):
 
     @property
     def metadata(self):
+        """
+        Extract metadata from the runparameter file
+
+        Returns
+        -------
+            metadata: a dict containing up to two keys: "reagent_kit_barcode"
+            and "library_tube_barcode"
+        """
         if not self.run_parameters:
             log.warning(f"No metadata found for runfolder {self.path}")
 

arteria/services/arteria_runfolder.py

@@ -1,14 +1,39 @@
+import yaml
+import argparse
+import logging.config
+
 from aiohttp import web
 
+from arteria.models.config import Config
 from arteria.handlers.base import base_routes as routes
+from arteria.config_schemas.schema_arteria_runfolder import runfolder_schema
+
+
+def get_app(config_dict):
+    """
+    Creates an Application instance
+    Sets up logger from configuration file specified in the config
+    Registers the request handler
+    """
+    config = Config.new(config_dict, schema=runfolder_schema)
 
+    with open(config["logger_config_file"], "r", encoding="utf-8") as logger_config_file:
+        logger_config = yaml.safe_load(logger_config_file.read())
+    logging.config.dictConfig(logger_config)
 
-def get_app(config):
     app = web.Application()
     app.router.add_routes(routes)
+
     return app
 
 
 def main():
-    app = get_app({})
-    web.run_app(app)
+    parser = argparse.ArgumentParser(description="arteria-runfolder server")
+    parser.add_argument('--config_file')
+
+    args = parser.parse_args()
+    with open(args.config_file, "r", encoding="utf-8") as config_file:
+        config_dict = yaml.safe_load(config_file.read())
+
+    app = get_app(config_dict)
+    web.run_app(app, port=config.get("port"))