run model predict

Signed-off-by: John Andersen <[email protected]>

Author: John Andersen

.ci/run.sh

@@ -55,7 +55,7 @@ function run_plugin() {
     # Create the docs
     cd "${SRC_ROOT}"
     "${PYTHON}" -m pip install -U -e "${SRC_ROOT}[dev]"
-    "${PYTHON}" -m dffml service dev install
+    "${PYTHON}" -m dffml service dev install -user
     ./scripts/docs.sh
 
     # Log skipped tests to file

CHANGELOG.md

@@ -27,6 +27,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   performance issue.
 - MySQL packaging issue.
 - Develop service running one off operations correctly json-loads dict types.
+- Operations with configs can be run via the development service
 ### Removed
 - CLI command `operations` removed in favor of `dataflow run`
 - Duplicate dataflow diagram code from development service

dffml/base.py

@@ -29,6 +29,9 @@ def get_args(t):
 from .log import LOGGER
 
 
+ARGP = ArgumentParser()
+
+
 class ParseExpandAction(argparse.Action):
     def __call__(self, parser, namespace, values, option_string=None):
         if not isinstance(values, list):
@@ -106,11 +109,92 @@ def __str__(self):
         return repr(self)
 
 
+def mkarg(field):
+    arg = Arg(type=field.type)
+    # HACK For detecting dataclasses._MISSING_TYPE
+    if "dataclasses._MISSING_TYPE" not in repr(field.default):
+        arg["default"] = field.default
+    if field.type == bool:
+        arg["action"] = "store_true"
+    elif inspect.isclass(field.type):
+        if issubclass(field.type, list):
+            arg["nargs"] = "+"
+            if not hasattr(field.type, "SINGLETON"):
+                raise AttributeError(
+                    f"{field.type.__qualname__} missing attribute SINGLETON"
+                )
+            arg["action"] = list_action(field.type)
+            arg["type"] = field.type.SINGLETON
+        if hasattr(arg["type"], "load"):
+            # TODO (python3.8) Use Protocol
+            arg["type"] = arg["type"].load
+    elif get_origin(field.type) is list:
+        arg["type"] = get_args(field.type)[0]
+        arg["nargs"] = "+"
+    if "help" in field.metadata:
+        arg["help"] = field.metadata["help"]
+    return arg
+
+
+def convert_value(arg, value):
+    if value is None:
+        # Return default if not found and available
+        if "default" in arg:
+            return arg["default"]
+        raise MissingConfig
+
+    # TODO This is a oversimplification of argparse's nargs
+    if not "nargs" in arg:
+        value = value[0]
+    if "type" in arg:
+        # TODO This is a oversimplification of argparse's nargs
+        if "nargs" in arg:
+            value = list(map(arg["type"], value))
+        else:
+            value = arg["type"](value)
+    if "action" in arg:
+        if isinstance(arg["action"], str):
+            # HACK This accesses _pop_action_class from ArgumentParser
+            # which is prefaced with an underscore indicating it not an API
+            # we can rely on
+            arg["action"] = ARGP._pop_action_class(arg)
+        namespace = ConfigurableParsingNamespace()
+        action = arg["action"](dest="dest", option_strings="")
+        action(None, namespace, value)
+        value = namespace.dest
+    return value
+
+
+def is_config_dict(value):
+    return bool(
+        "arg" in value
+        and "config" in value
+        and isinstance(value["config"], dict)
+    )
+
+
+def _fromdict(cls, **kwargs):
+    for field in dataclasses.fields(cls):
+        if field.name in kwargs:
+            value = kwargs[field.name]
+            config = {}
+            if is_config_dict(value):
+                value, config = value["arg"], value["config"]
+            value = convert_value(mkarg(field), value)
+            if inspect.isclass(value) and issubclass(value, BaseConfigurable):
+                value = value.withconfig(
+                    {field.name: {"arg": None, "config": config}}
+                )
+            kwargs[field.name] = value
+    return cls(**kwargs)
+
+
 def config(cls):
     """
     Decorator to create a dataclass
     """
     datacls = dataclasses.dataclass(eq=True, frozen=True)(cls)
+    datacls._fromdict = classmethod(_fromdict)
     datacls._replace = lambda self, *args, **kwargs: dataclasses.replace(
         self, *args, **kwargs
     )
@@ -130,8 +214,6 @@ class BaseConfigurable(abc.ABC):
     only parameter to the __init__ of a BaseDataFlowFacilitatorObject.
     """
 
-    __argp = ArgumentParser()
-
     def __init__(self, config: BaseConfig) -> None:
         """
         BaseConfigurable takes only one argument to __init__,
@@ -225,41 +307,20 @@ def config_get(cls, config, above, *path) -> BaseConfig:
             with contextlib.suppress(KeyError):
                 value = traverse_config_get(config, *no_label_above)
 
-        if value is None:
-            # Return default if not found and available
-            if "default" in arg:
-                return arg["default"]
-            raise MissingConfig(
-                "%s missing %r from %s"
-                % (
-                    cls.__qualname__,
-                    label_above[-1],
-                    ".".join(label_above[:-1]),
-                )
+        try:
+            return convert_value(arg, value)
+        except MissingConfig as error:
+            error.args = (
+                (
+                    "%s missing %r from %s"
+                    % (
+                        cls.__qualname__,
+                        label_above[-1],
+                        ".".join(label_above[:-1]),
+                    )
+                ),
             )
-
-        if value is None and "default" in arg:
-            return arg["default"]
-        # TODO This is a oversimplification of argparse's nargs
-        if not "nargs" in arg:
-            value = value[0]
-        if "type" in arg:
-            # TODO This is a oversimplification of argparse's nargs
-            if "nargs" in arg:
-                value = list(map(arg["type"], value))
-            else:
-                value = arg["type"](value)
-        if "action" in arg:
-            if isinstance(arg["action"], str):
-                # HACK This accesses _pop_action_class from ArgumentParser
-                # which is prefaced with an underscore indicating it not an API
-                # we can rely on
-                arg["action"] = cls.__argp._pop_action_class(arg)
-            namespace = ConfigurableParsingNamespace()
-            action = arg["action"](dest="dest", option_strings="")
-            action(None, namespace, value)
-            value = namespace.dest
-        return value
+            raise
 
     @classmethod
     def args(cls, args, *above) -> Dict[str, Arg]:
@@ -271,30 +332,7 @@ def args(cls, args, *above) -> Dict[str, Arg]:
                 f"{cls.__qualname__} requires CONFIG property or implementation of args() classmethod"
             )
         for field in dataclasses.fields(cls.CONFIG):
-            arg = Arg(type=field.type)
-            # HACK For detecting dataclasses._MISSING_TYPE
-            if "dataclasses._MISSING_TYPE" not in repr(field.default):
-                arg["default"] = field.default
-            if field.type == bool:
-                arg["action"] = "store_true"
-            elif inspect.isclass(field.type):
-                if issubclass(field.type, list):
-                    arg["nargs"] = "+"
-                    if not hasattr(field.type, "SINGLETON"):
-                        raise AttributeError(
-                            f"{field.type.__qualname__} missing attribute SINGLETON"
-                        )
-                    arg["action"] = list_action(field.type)
-                    arg["type"] = field.type.SINGLETON
-                if hasattr(arg["type"], "load"):
-                    # TODO (python3.8) Use Protocol
-                    arg["type"] = arg["type"].load
-            elif get_origin(field.type) is list:
-                arg["type"] = get_args(field.type)[0]
-                arg["nargs"] = "+"
-            if "help" in field.metadata:
-                arg["help"] = field.metadata["help"]
-            cls.config_set(args, above, field.name, arg)
+            cls.config_set(args, above, field.name, mkarg(field))
         return args
 
     @classmethod

dffml/df/base.py

@@ -51,10 +51,14 @@ class BaseDataFlowObject(BaseDataFlowFacilitatorObject):
 
     @classmethod
     def args(cls, args, *above) -> Dict[str, Arg]:
+        if hasattr(cls, "CONFIG"):
+            return super(BaseDataFlowObject, cls).args(args, *above)
         return args
 
     @classmethod
     def config(cls, config, *above) -> BaseConfig:
+        if hasattr(cls, "CONFIG"):
+            return super(BaseDataFlowObject, cls).config(config, *above)
         return BaseConfig()
 
 
@@ -221,6 +225,9 @@ def __init__(self, config):
                         config = config_cls(**config)
                 super().__init__(config)
 
+        if config_cls is not None:
+            Implementation.CONFIG = config_cls
+
         if inspect.isclass(func) and issubclass(
             func, OperationImplementationContext
         ):
@@ -247,9 +254,8 @@ async def run(
                     if uses_self:
                         # We can't pass self to functions running in threads
                         # Its not thread safe!
-                        return await (
-                            func.__get__(self, self.__class__)(**inputs)
-                        )
+                        bound = func.__get__(self, self.__class__)
+                        return await bound(**inputs)
                     elif inspect.iscoroutinefunction(func):
                         return await func(**inputs)
                     else:

dffml/df/memory.py

@@ -1159,6 +1159,10 @@ async def initialize_dataflow(self, dataflow: DataFlow) -> None:
                             operation.name,
                         )
                         opimp_config = BaseConfig()
+                    if isinstance(opimp_config, dict) and hasattr(
+                        getattr(opimp, "CONFIG", False), "_fromdict"
+                    ):
+                        opimp_config = opimp.CONFIG._fromdict(**opimp_config)
                     await self.nctx.instantiate(
                         operation, opimp_config, opimp=opimp
                     )

dffml/operation/model.py

@@ -0,0 +1,38 @@
+from typing import Dict, Any
+
+from ..repo import Repo
+from ..base import config
+from ..model import Model
+from ..df.types import Definition
+from ..df.base import op
+
+
+@config
+class ModelPredictConfig:
+    model: Model
+    msg: str
+
+
+@op(
+    inputs={
+        "features": Definition(
+            name="repo_features", primitive="Dict[str, Any]"
+        )
+    },
+    outputs={
+        "prediction": Definition(
+            name="model_predictions", primitive="Dict[str, Any]"
+        )
+    },
+    config_cls=ModelPredictConfig,
+    imp_enter={"model": (lambda self: self.config.model)},
+    ctx_enter={"mctx": (lambda self: self.parent.model())},
+)
+async def model_predict(self, features: Dict[str, Any]) -> Dict[str, Any]:
+    async def repos():
+        yield Repo("", data={"features": features})
+
+    async for repo in self.mctx.predict(repos()):
+        return {
+            "prediction": {self.config.model.config.predict: repo.prediction()}
+        }

dffml/service/dev.py

@@ -209,10 +209,26 @@ async def run_op(self, name, opimp):
                 error.args = (f"{opimp.op.inputs}: {error.args[0]}",)
                 raise error
 
+        config = {}
+        extra_config = self.extra_config
+
+        for i in range(0, 2):
+            if "config" in extra_config and len(extra_config["config"]):
+                extra_config = extra_config["config"]
+
+        # TODO(p0) This only goes one level deep. This won't work for
+        # configs that are multi-leveled
+        if extra_config:
+            config = extra_config
+
+        dataflow = DataFlow.auto(GetSingle, opimp)
+        if config:
+            dataflow.configs[opimp.op.name] = config
+
         # Run the operation in the memory orchestrator
         async with MemoryOrchestrator.withconfig({}) as orchestrator:
             # Orchestrate the running of these operations
-            async with orchestrator(DataFlow.auto(GetSingle, opimp)) as octx:
+            async with orchestrator(dataflow) as octx:
                 async for ctx, results in octx.run(
                     [
                         Input(
@@ -290,7 +306,7 @@ class Install(CMD):
     """
 
     arg_user = Arg(
-        "-user", "Preform user install", default=True, action="store_false"
+        "-user", "Preform user install", default=False, action="store_true"
     )
 
     async def run(self):
@@ -307,12 +323,7 @@ async def run(self):
             )
         )
         self.logger.info("Installing %r in development mode", packages)
-        cmd = [
-            sys.executable,
-            "-m",
-            "pip",
-            "install",
-        ]
+        cmd = [sys.executable, "-m", "pip", "install"]
         if self.user:
             # --user sometimes fails
             local_path = Path("~", ".local").expanduser().absolute()