docs: usage: Integration example

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

Author: John Andersen

dffml/cli.py

@@ -15,6 +15,7 @@
 
 from .log import LOGGER
 from .version import VERSION
+from .repo import Repo
 from .port import Port
 from .feature import Feature, Features, Data
 from .source.source import BaseSource, Sources, SubsetSources
@@ -135,15 +136,17 @@ class Merge(CMD):
     '''
 
     arg_dest = Arg('dest', help='Sources merge repos into',
-            type=BaseSource.load)
+            type=BaseSource.load_labeled)
     arg_src = Arg('src', help='Sources to pull repos from',
-            type=BaseSource.load)
+            type=BaseSource.load_labeled)
 
     async def run(self):
         async with self.src.withconfig(self.extra_config) as src, \
                 self.dest.withconfig(self.extra_config) as dest:
-            async with self.src() as sctx, dest() as dctx:
-                async for repo in sctx.repos():
+            async with src() as sctx, dest() as dctx:
+                async for src in sctx.repos():
+                    repo = Repo(src.src_url)
+                    repo.merge(src)
                     repo.merge(await dctx.repo(repo.src_url))
                     await dctx.update(repo)
 
@@ -313,27 +316,27 @@ async def run(self):
         async with self.sources as sources, self.features as features, \
                 self.model as model:
             async with sources() as sctx, model() as mctx:
-                return float(await mctx.accuracy(sources, features,
+                return float(await mctx.accuracy(sctx, features,
                     self.classifications))
 
 class PredictAll(EvaluateAll, MLCMD):
     '''Predicts for all sources'''
 
-    async def predict(self, model, sources, features, repos):
-        async with sources() as sctx, model() as mctx:
-            async for repo, classification, confidence in \
-                    mctx.predict(repos, features, self.classifications):
-                repo.predicted(classification, confidence)
-                yield repo
-                if self.update:
-                    await sctx.update(repo)
+    async def predict(self, mctx, sctx, features, repos):
+        async for repo, classification, confidence in \
+                mctx.predict(repos, features, self.classifications):
+            repo.predicted(classification, confidence)
+            yield repo
+            if self.update:
+                await sctx.update(repo)
 
     async def run(self):
         async with self.sources as sources, self.features as features, \
                 self.model as model:
-            async for repo in self.predict(model, sources, features,
-                    self.evaluate(sources, features)):
-                yield repo
+            async with sources() as sctx, model() as mctx:
+                async for repo in self.predict(mctx, sctx, features,
+                        sctx.repos()):
+                    yield repo
 
 class PredictRepo(PredictAll, EvaluateRepo):
     '''Predictions for individual repos'''

dffml/feature/feature.py

@@ -212,6 +212,23 @@ async def close(self):
         '''
         pass
 
+    @classmethod
+    def load(cls, loading=None):
+        if loading is not None and loading.startswith('def:'):
+            return cls.load_def(*loading.replace('def:', '').split(':'))
+        return super().load(loading)
+
+    @classmethod
+    def load_def(cls, name: str, dtype: str, length: str):
+        return DefFeature(name, cls.convert_dtype(dtype), int(length))
+
+    @classmethod
+    def convert_dtype(cls, dtype: str):
+        found = pydoc.locate(dtype)
+        if found is None:
+            raise TypeError('Failed to convert_dtype %r' % (dtype,))
+        return found
+
     async def __aenter__(self):
         await self.open()
         # TODO Context management
@@ -373,46 +390,6 @@ async def submit(self, src: str):
         return await super().start(partial(self.evaluate, src), src,
                 mktask=self.mktask)
 
-    @classmethod
-    def load(cls, *these: str):
-        '''
-        Loads all installed loading and returns them as a list. Sources to be
-        loaded should be registered to ENTRY_POINT via setuptools.
-        '''
-        these, loading_classes = cls.load_defs(*these)
-        for i in pkg_resources.iter_entry_points(Feature.ENTRY_POINT):
-            loaded = i.load()
-            if issubclass(loaded, Feature) and loaded.NAME in these:
-                loading_classes.append(loaded())
-        self = cls(*loading_classes)
-        for name in these:
-            if not name in self.names():
-                raise KeyError('%s was not found in (%s)' % \
-                        (repr(name), ', '.join(map(str, loading_classes))))
-        if not self.names():
-            raise KeyError('No features were loaded')
-        return self
-
-    @classmethod
-    def load_defs(cls, *args: str):
-        defs = []
-        no_def = [arg for arg in args if not arg.startswith('def:')]
-        for arg in args:
-            if arg.startswith('def:'):
-                defs.append(cls.load_def(*arg.replace('def:', '').split(':')))
-        return no_def, defs
-
-    @classmethod
-    def load_def(cls, name: str, dtype: str, length: str):
-        return DefFeature(name, cls.convert_dtype(dtype), int(length))
-
-    @classmethod
-    def convert_dtype(cls, dtype: str):
-        found = pydoc.locate(dtype)
-        if found is None:
-            raise TypeError('Failed to convert_dtype %r' % (dtype,))
-        return found
-
     async def __aenter__(self):
         self._stack = AsyncExitStack()
         await self._stack.__aenter__()

dffml/repo.py

@@ -7,6 +7,7 @@
 from datetime import datetime
 from typing import Optional, List, Dict, Any, AsyncIterator
 
+from .util.data import merge
 from .log import LOGGER
 
 LOGGER = LOGGER.getChild('repo')
@@ -77,7 +78,9 @@ def __init__(self, *,
 
     def dict(self):
         data = {key: getattr(self, key, []) for key in self.EXPORTED \
-                if len(getattr(self, key, []))}
+                if not isinstance(getattr(self, key, {}), dict) or \
+                (isinstance(getattr(self, key, {}), dict) and \
+                getattr(self, key, {}))}
         # Do not report if there has been no change since instantiation to
         # a default time value
         if self.last_updated != self.last_updated_default:
@@ -136,7 +139,7 @@ def __str__(self):
 
     def merge(self, repo: 'Repo'):
         data = self.data.dict()
-        data.update(repo.data.dict())
+        merge(data, repo.data.dict())
         self.data = self.REPO_DATA(**data)
         self.extra.update(repo.extra) # type: ignore
 

dffml/util/data.py

@@ -3,6 +3,13 @@
 '''
 from functools import wraps
 
+def merge(one, two):
+    for key, value in two.items():
+        if key in one and isinstance(value, dict):
+            merge(one[key], two[key])
+        else:
+            one[key] = two[key]
+
 def traverse_config_set(target, *args):
     '''
     >>> traverse_set({'level': {'one': 1}}, 'level', 'one', 42)

docs/concepts/data_flow.rst

@@ -0,0 +1,5 @@
+Data Flow
+=========
+
+An operation is run if there is new data of data types matching it's parameters
+in the data flow network. For

docs/concepts/index.rst

@@ -0,0 +1,9 @@
+Models
+======
+
+.. toctree::
+    :glob:
+    :maxdepth: 2
+    :caption: Contents:
+
+    base