Stateless drivers & components

MANIFEST.in

@@ -1,3 +1,2 @@
 include versioneer.py
 include src/tomato/_version.py
-include src/tomato/data/*

src/tomato/daemon/__init__.py

@@ -6,19 +6,17 @@
 
 """
 
-import logging
 import argparse
-from pathlib import Path
-from threading import Thread
-import toml
+import logging
 import time
-import zmq
-
-from tomato.models import Reply, Daemon
 import tomato.daemon.cmd as cmd
-import tomato.daemon.job
 import tomato.daemon.driver
 import tomato.daemon.io as io
+import tomato.daemon.job
+import zmq
+from pathlib import Path
+from threading import Thread
+from tomato.models import Daemon, Reply
 
 logger = logging.getLogger(__name__)
 
@@ -30,7 +28,7 @@ def setup_logging(daemon: Daemon):
     """
     logdir = Path(daemon.settings["logdir"])
     logdir.mkdir(parents=True, exist_ok=True)
-    logfile = logdir / f"daemon_{daemon.port}.log"
+    logfile = logdir / f"tomato_daemon_{daemon.port}.log"
     logging.basicConfig(
         level=daemon.verbosity,
         format="%(asctime)s - %(levelname)8s - %(name)-30s - %(message)s",
@@ -53,12 +51,13 @@ def tomato_daemon():
     parser.add_argument("--appdir", "-A", type=str, default=str(Path.cwd()))
 
     args = parser.parse_args()
-    settings = toml.load(Path(args.appdir) / "settings.toml")
 
-    daemon = Daemon(**vars(args), status="bootstrap", settings=settings)
+    daemon = Daemon(**vars(args), status="bootstrap")
     setup_logging(daemon)
     logger.info("logging set up with verbosity %s", daemon.verbosity)
 
+    # TODO: setup should not be a thing really.
+    cmd.setup(msg={}, daemon=daemon)
     logger.debug("attempting to restore daemon state")
     io.load(daemon)
 
@@ -71,10 +70,10 @@ def tomato_daemon():
 
     logger.debug("entering main loop")
     jmgr = Thread(target=tomato.daemon.job.manager, args=(daemon.port,), daemon=True)
-    jmgr.do_run = True
+    setattr(jmgr, "do_run", True)
     jmgr.start()
     dmgr = Thread(target=tomato.daemon.driver.manager, args=(daemon.port,), daemon=True)
-    dmgr.do_run = True
+    setattr(dmgr, "do_run", True)
     dmgr.start()
     t0 = time.process_time()
     while True:
@@ -93,9 +92,9 @@ def tomato_daemon():
             rep.send_pyobj(ret)
         if daemon.status == "stop":
             for mgr, label in [(jmgr, "job"), (dmgr, "driver")]:
-                if mgr is not None and mgr.do_run:
+                if mgr is not None and getattr(mgr, "do_run"):
                     logger.debug("stopping %s manager thread", label)
-                    mgr.do_run = False
+                    setattr(mgr, "do_run", False)
             if jmgr is not None:
                 jmgr.join(1e-3)
                 if not jmgr.is_alive():

src/tomato/daemon/cmd.py

@@ -12,21 +12,22 @@
 
 """
 
+import logging
+import tomato.daemon.io as io
+import tomato.daemon.jobdb as jobdb
+import tomato.utils
+from pathlib import Path
+from pydantic import BaseModel
 from tomato.models import (
+    Component,
     Daemon,
-    Driver,
     Device,
-    Reply,
-    Pipeline,
     Job,
-    Component,
+    Pipeline,
+    Reply,
+    SpawnData,
 )
-from pydantic import BaseModel
 from typing import Any
-import logging
-
-import tomato.daemon.io as io
-import tomato.daemon.jobdb as jobdb
 
 logger = logging.getLogger(__name__)
 
@@ -44,7 +45,7 @@ def merge_pipelines(
             if pip.jobid is not None:
                 ret[pname] = pip
         else:
-            if pip.devs == new[pname].devs:
+            if pip.components == new[pname].components:
                 ret[pname] = pip
             elif pip.jobid is None:
                 ret[pname] = new[pname]
@@ -74,27 +75,68 @@ def stop(msg: dict, daemon: Daemon) -> Reply:
 
 def setup(msg: dict, daemon: Daemon) -> Reply:
     logger = logging.getLogger(f"{__name__}.setup")
-    logger.debug("%s", msg)
+
+    # TODO: Rework this!
+    devicefile = tomato.utils.load_device_file(
+        Path(daemon.settings["devices"]["config"]),
+        logger,
+    )
+    logger.debug(f"{devicefile=}")
+    devs = {dev["name"]: Device(**dev) for dev in devicefile["devices"]}
+    logger.debug(f"{devs=}")
+    pips, cmps = tomato.utils.get_pipelines(
+        devs,
+        devicefile["pipelines"],
+        logger,
+    )
+    logger.debug(f"{pips=}")
+    logger.debug(f"{cmps=}")
+
     if daemon.status == "bootstrap":
-        for key in ["drvs", "devs", "pips", "cmps"]:
-            setattr(daemon, key, msg[key])
+        for key, val in [
+            # ("drvs", drvs),
+            # ("devs", devs),
+            ("pips", pips),
+            ("cmps", cmps),
+        ]:
+            setattr(daemon, key, val)
+
+        daemon.drivers = {
+            key: SpawnData(name=key) for key in daemon.devicefile.drivers.keys()
+        }
         logger.info("setup successful with pipelines: '%s'", daemon.pips.keys())
         daemon.status = "running"
     else:
+        try:
+            nd = Daemon(
+                status=daemon.status,
+                port=daemon.port,
+                appdir=daemon.appdir,
+                verbosity=daemon.verbosity,
+            )
+        except Exception as e:
+            logger.critical("Error", exc_info=e)
+            return Reply(
+                success=False,
+                msg="could not parse updated settings",
+            )
+        logger.debug(f"{nd=}")
+        ndf = nd.devicefile
         # First, check that we're not touching anything associated with a running job
         check_components = set()
         check_devices = set()
         check_drivers = set()
         for dpip in daemon.pips.values():
+            logger.debug(f"{dpip=}")
             if dpip.jobid is None:
                 continue
-            if dpip.name not in msg["pips"]:
+            if dpip.name not in pips:
                 return Reply(
                     success=False,
                     msg="reload would delete a running pipeline",
                     data=dpip,
                 )
-            pip = msg["pips"][dpip.name]
+            pip = pips[dpip.name]
             if pip.components != dpip.components:
                 return Reply(
                     success=False,
@@ -105,13 +147,13 @@ def setup(msg: dict, daemon: Daemon) -> Reply:
 
         for cname in check_components:
             dcomp = daemon.cmps[cname]
-            if cname not in msg["cmps"]:
+            if cname not in cmps:
                 return Reply(
                     success=False,
                     msg="reload would delete a component of a running pipeline",
                     data=dcomp,
                 )
-            comp = msg["cmps"][cname]
+            comp = cmps[cname]
             if (
                 dcomp.name != comp.name
                 or dcomp.driver != comp.driver
@@ -128,54 +170,37 @@ def setup(msg: dict, daemon: Daemon) -> Reply:
             check_devices.add(dcomp.device)
             check_drivers.add(dcomp.driver)
 
-        for dname in check_devices:
-            ddev = daemon.devs[dname]
-            if dname not in msg["devs"]:
-                return Reply(
-                    success=False,
-                    msg="reload would delete a device of a component in a running pipeline",
-                    data=ddev,
-                )
-            dev = msg["devs"][dname]
-            if (
-                ddev.name != dev.name
-                or ddev.driver != dev.driver
-                or ddev.address != dev.address
-                or ddev.pollrate != dev.pollrate
-                or any(ch not in dev.channels for ch in ddev.channels)
-            ):
-                return Reply(
-                    success=False,
-                    msg="reload would modify a device of a component in a running pipeline",
-                    data=ddev,
-                )
-
         for dname in check_drivers:
-            ddrv = daemon.drvs[dname]
-            if dname not in msg["drvs"]:
+            if dname not in ndf.drivers:
                 return Reply(
                     success=False,
                     msg="reload would delete a driver of a device in a running pipeline",
-                    data=ddev,
+                    data=daemon.drivers[dname],
                 )
-            drv = msg["drvs"][dname]
-            if ddrv.name != drv.name or ddrv.settings != drv.settings:
+
+            if daemon.devicefile.drivers[dname].settings != ndf.drivers[dname].settings:
                 return Reply(
                     success=False,
                     msg="reload would modify a driver of a device in a running pipeline",
-                    data=ddrv,
+                    data=daemon.devicefile.drivers[dname].settings,
                 )
 
-        _api_reload(msg["drvs"], daemon.drvs, "driver", ["settings"])
-
-        _api_reload(msg["pips"], daemon.pips, "pipeline", ["components"])
+        logger.critical("goint into api reload")
 
+        _api_reload(pips, daemon.pips, "pipeline", ["components"])
         attrlist = ["driver", "device", "address", "channel", "role"]
-        _api_reload(msg["cmps"], daemon.cmps, "component", attrlist)
-
-        _api_reload(msg["devs"], daemon.devs, "device", ["channels", "pollrate"])
-
+        _api_reload(cmps, daemon.cmps, "component", attrlist)
+        # Add new drivers, they will be spawned by driver.manager
+        for dname in ndf.drivers.keys():
+            if dname not in daemon.drivers:
+                logger.info("adding new driver '%s'", dname)
+                daemon.drivers[dname] = SpawnData(name=dname)
+
+        # We want to trigger re-parse of config files on daemon
+        daemon.settings = nd.settings
+        daemon.devicefile = ndf
         logger.info("reload successful with pipelines: '%s'", daemon.pips.keys())
+
     return Reply(success=True, data=daemon)
 
 
@@ -201,7 +226,7 @@ def pipeline(msg: dict, daemon: Daemon) -> Reply:
     logger.debug("%s", msg)
     pip = msg["params"]
     if pip["name"] is None:
-        logger.error()
+        logger.error("no pipeline name supplied")
         return Reply(success=False, msg="no pipeline name supplied", data=msg)
     if pip["name"] not in daemon.pips:
         dest = Pipeline(**pip)
@@ -243,15 +268,6 @@ def get_jobs(msg: dict, daemon: Daemon) -> Reply:
     return Reply(success=True, msg=f"found {len(jobs)} jobs", data=jobs)
 
 
-def driver(msg: dict, daemon: Daemon) -> Reply:
-    return _api(
-        otype="driver",
-        msg=msg,
-        ddict=daemon.drvs,
-        Cls=Driver,
-    )
-
-
 def device(msg: dict, daemon: Daemon) -> Reply:
     return _api(
         otype="device",
@@ -293,3 +309,22 @@ def _api(otype: str, msg: dict, ddict: dict[str, Any], Cls: BaseModel) -> Reply:
             msg=f"{otype} {obj['name']!r} updated",
             data=ddict[obj["name"]],
         )
+
+
+def reload(msg: dict, daemon: Daemon, **kwargs: dict) -> Reply:
+    # daemon.settings = toml.load(Path(daemon.appdir) / "settings.toml")
+    return Reply(success=True, msg="daemon settings reloaded", data=daemon.settings)
+
+
+def driver_set(msg: dict, daemon: Daemon) -> Reply:
+    params = msg.pop("params")
+    name = params.pop("name")
+    for k, v in params.items():
+        setattr(daemon.drivers[name], k, v)
+    return Reply(success=True, msg=f"updated driver {name}", data=daemon.drivers[name])
+
+
+def driver_del(msg: dict, daemon: Daemon) -> Reply:
+    name = msg.pop("params").pop("name")
+    del daemon.drivers[name]
+    return Reply(success=True, msg=f"removed driver {name}")