init

.gitignore

@@ -0,0 +1,5 @@
+/.idea
+/**/__pycache__/
+/dist
+/.tox
+/src/whenact.egg-info

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2022 Morvan
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

pyproject.toml

@@ -0,0 +1,3 @@
+[build-system]
+requires = ["setuptools>=42", "wheel"]
+build-backend = "setuptools.build_meta"

requirements.txt

@@ -0,0 +1 @@
+

requirements_dev.txt

@@ -0,0 +1,3 @@
+tox==3.24.3
+build
+twine

setup.cfg

@@ -0,0 +1,25 @@
+[metadata]
+name = whenact
+version = 0.0.1
+author = MorvanZhou
+author_email = morvanzhou@hotmail.com
+description = whenact decision pipeline
+long_description = file: README.md
+long_description_content_type = text/markdown
+url = https://github.com/MorvanZhou/whenact
+project_urls =
+    Bug Tracker = https://github.com/MorvanZhou/whenact/issues
+classifiers =
+    Programming Language :: Python :: 3
+    License :: OSI Approved :: MIT License
+    Operating System :: OS Independent
+
+[options]
+package_dir =
+    = src
+packages = find:
+python_requires = >=3.6
+install_requires =
+
+[options.packages.find]
+where = src

src/whenact/__init__.py

@@ -0,0 +1,4 @@
+from whenact.block.when_block import when
+from whenact.block.act_block import act
+from whenact.pipeline import Pipeline, create_pipeline, Policy
+from whenact.context import PipelineContext, BaseContext

src/whenact/block/__init__.py

@@ -0,0 +1,2 @@
+from whenact.block.act_block import act
+from whenact.block.when_block import when

src/whenact/block/act_block.py

@@ -0,0 +1,16 @@
+from functools import wraps
+
+from whenact.context import PipelineContext
+
+
+def act(func):
+    func._type = "act"
+
+    @wraps(func)
+    def wrapper(context: PipelineContext, *args, **kwargs):
+        res = func(context, *args, **kwargs)
+        context._set_pipeline_output(res)
+
+        return
+
+    return wrapper

src/whenact/block/when_block.py

@@ -0,0 +1,16 @@
+from functools import wraps
+
+from whenact.context import PipelineContext
+
+
+def when(func):
+    func._type = "when"
+
+    @wraps(func)
+    def wrapper(context: PipelineContext, *args, **kwargs):
+        res = func(context, *args, **kwargs)
+        if not isinstance(res, bool):
+            raise TypeError(f"{func.__name__} should return a boolean")
+        return res
+
+    return wrapper

src/whenact/context.py

@@ -0,0 +1,41 @@
+
+class BaseContext:
+    def __init__(self):
+        self._cache_data = {}
+
+    def __getitem__(self, item):
+        return self._cache_data[item]
+
+    def __setitem__(self, key, value):
+        self._cache_data[key] = value
+        self.__setattr__(key, value)
+
+    def __delitem__(self, key):
+        if key in self._cache_data:
+            del self._cache_data[key]
+            self.__delattr__(key)
+
+
+class PipelineContext:
+    def __init__(self, base_ctx: BaseContext):
+        self._ctx = base_ctx
+        self._pipeline_output = None
+
+    def _set_pipeline_output(self, data):
+        self._pipeline_output = data
+
+    def __getattr__(self, item):
+        try:
+            return self._ctx.__getattribute__(item)
+        except AttributeError:
+            return self.__getattribute__(item)
+
+    def __getitem__(self, item):
+        return self._ctx.__getitem__(item)
+
+    def __setitem__(self, key, value):
+        self._ctx.__setitem__(key, value)
+
+    @property
+    def last_output(self):
+        return self._pipeline_output

src/whenact/pipeline.py

@@ -0,0 +1,116 @@
+import typing
+from collections import OrderedDict
+from dataclasses import dataclass, field
+
+from whenact.context import PipelineContext, BaseContext
+
+_POLICY_NAME_COUNT = 0
+
+
+def _get_policy_name():
+    global _POLICY_NAME_COUNT
+    name = f"p{_POLICY_NAME_COUNT}"
+    _POLICY_NAME_COUNT += 1
+    return name
+
+
+def _reset_policy_name():
+    global _POLICY_NAME_COUNT
+    _POLICY_NAME_COUNT = 0
+
+
+@dataclass
+class Policy:
+    when: typing.Sequence[typing.Callable]
+    action: typing.Sequence[typing.Callable]
+    name: str = field(default_factory=_get_policy_name)
+
+    def __post_init__(self):
+        if self.name is None:
+            self.name = _get_policy_name()
+
+
+class Pipeline:
+    def __init__(self, pipe_list: typing.Optional[typing.Sequence[Policy]] = None):
+        self.data: typing.OrderedDict[str, Policy] = OrderedDict()
+        if pipe_list is not None:
+            for p in pipe_list:
+                self.add_policy(p)
+
+    def add(
+            self,
+            when: typing.Union[typing.Callable, typing.Sequence[typing.Callable]],
+            action: typing.Union[typing.Callable, typing.Sequence[typing.Callable]],
+            name: typing.Optional[str] = None
+    ):
+        if not isinstance(when, (tuple, list)):
+            when = [when]
+        if not isinstance(action, (tuple, list)):
+            action = [action]
+        self.add_policy(Policy(when=when, action=action, name=name))
+
+    def add_policy(self, policy: Policy):
+        if policy.name in self.data:
+            raise ValueError(f"{policy.name=} is exist in pipeline, please use new one")
+        self.data[policy.name] = policy
+
+    def remove_policy(self, name: str):
+        del self.data[name]
+
+    def run(self, context: BaseContext):
+        output = None
+        for res in self.iter_run(context=context):
+            output = res
+        return output
+
+    def iter_run(self, context: BaseContext):
+        p_ctx = PipelineContext(base_ctx=context)
+        for policy in self.data.values():
+            keep = True
+            for w in policy.when:
+                keep = w(p_ctx)
+                yield keep
+                if not keep:
+                    break
+            if keep:
+                for a in policy.action:
+                    a(p_ctx)
+                    yield p_ctx.last_output
+
+    def __getitem__(self, item) -> Policy:
+        if isinstance(item, int):
+            return list(self.data.values())[item]
+        return self.data[item]
+
+    def __str__(self):
+        return "\n".join(self.view_pipe())
+
+    def view_pipe(self):
+        res = []
+        for name, p in self.data.items():
+            res.append(f"{name}: "
+                       f"[{' > '.join([a.__name__ for a in p.when])}] "
+                       f"> [{' > '.join([a.__name__ for a in p.action])}]")
+        return res
+
+
+def create_pipeline(config: typing.Sequence[typing.Sequence[typing.Callable]]) -> Pipeline:
+    p = Pipeline()
+    for policy in config:
+        when = []
+        action = []
+        to_action = False
+        for wa in policy:
+            try:
+                t = wa._type
+            except AttributeError as e:
+                raise ValueError(f"when or act function must be decorated: {e}")
+            if t == "when" and not to_action:
+                when.append(wa)
+            elif t == "act":
+                to_action = True
+                action.append(wa)
+            else:
+                raise TypeError("[when] function must be set before [act] function")
+        p.add_policy(Policy(when=when, action=action))
+    return p