mirror
/
micropython-lib
kopia lustrzana https://github.com/micropython/micropython-lib
1
0
Forkuj 0
Kod Zgłoszenia Projekty Wydania Wiki Aktywność

python-ecosys/pymitter: Added legacy branch from the original project. 

Signed-off-by: Oliver Maye <maye@ihp-microelectronics.com>
pull/961/head
Oliver Maye 2025-01-08 11:24:46 +01:00
rodzic e4cf09527b
commit 686fbffece
10 zmienionych plików z 749 dodań i 0 usunięć
Pokaż statystyki Ściągnij plik aktualizacji Ściągnij plik porównania Expand all files Collapse all files

27
python-ecosys/pymitter/LICENSE 100644
Wyświetl plik

@ -0,0 +1,27 @@
Copyright (c) 2014-2021, Marcel Rieger
All rights reserved.
Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are met:
* Redistributions of source code must retain the above copyright notice, this
list of conditions and the following disclaimer.
* Redistributions in binary form must reproduce the above copyright notice,
this list of conditions and the following disclaimer in the documentation
and/or other materials provided with the distribution.
* Neither the name of the copyright holder nor the names of its contributors
may be used to endorse or promote products derived from this software without
specific prior written permission.
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

2
python-ecosys/pymitter/MANIFEST.in 100644
Wyświetl plik

@ -0,0 +1,2 @@
include pymitter.py setup.py requirements.txt README.md LICENSE .flake8
global-exclude *.py[cod] __pycache__

183
python-ecosys/pymitter/README.md 100644
Wyświetl plik

@ -0,0 +1,183 @@
# pymitter
This is a fork of the [original pymitter project](https://pypi.org/project/pymitter/) by Marcel Rieger.
Sources are from the legacy/py2 branch which is a frozen v0.3.2 of that project.
At this state, the implementation is compatible to Python >= v2.7 including
MicroPython with a language level v3.4.
Later versions of that project make use of type hints, which were introduced
in Python 3.5. Type hints are currently not supported by MicroPython.
## Features
- Namespaces with wildcards
- Times to listen (TTL)
- Usage via decorators or callbacks
- Lightweight implementation, good performance
## Installation
*pymitter* is a registered [MicroPython module](https://github.com/olimaye/micropython-lib),
so the installation with *mip* is quite easy:
```console
mpremote mip install pymitter
```
## Examples
### Basic usage
```python
from pymitter import EventEmitter
ee = EventEmitter()
# decorator usage
@ee.on("myevent")
def handler1(arg):
print("handler1 called with", arg)
# callback usage
def handler2(arg):
print("handler2 called with", arg)
ee.on("myotherevent", handler2)
# emit
ee.emit("myevent", "foo")
# -> "handler1 called with foo"
ee.emit("myotherevent", "bar")
# -> "handler2 called with bar"
```
### TTL (times to listen)
```python
from pymitter import EventEmitter
ee = EventEmitter()
@ee.once("myevent")
def handler1():
print("handler1 called")
@ee.on("myevent", ttl=10)
def handler2():
print("handler2 called")
ee.emit("myevent")
# -> "handler1 called"
# -> "handler2 called"
ee.emit("myevent")
# -> "handler2 called"
```
### Wildcards
```python
from pymitter import EventEmitter
ee = EventEmitter(wildcard=True)
@ee.on("myevent.foo")
def handler1():
print("handler1 called")
@ee.on("myevent.bar")
def handler2():
print("handler2 called")
@ee.on("myevent.*")
def hander3():
print("handler3 called")
ee.emit("myevent.foo")
# -> "handler1 called"
# -> "handler3 called"
ee.emit("myevent.bar")
# -> "handler2 called"
# -> "handler3 called"
ee.emit("myevent.*")
# -> "handler1 called"
# -> "handler2 called"
# -> "handler3 called"
```
## API
### ``EventEmitter(wildcard=False, delimiter=".", new_listener=False, max_listeners=-1)``
EventEmitter constructor. **Note**: always use *kwargs* for configuration. When *wildcard* is
*True*, wildcards are used as shown in [this example](#wildcards). *delimiter* is used to seperate
namespaces within events. If *new_listener* is *True*, the *"new_listener"* event is emitted every
time a new listener is registered. Functions listening to this event are passed
``(func, event=None)``. *max_listeners* defines the maximum number of listeners per event. Negative
values mean infinity.
- #### ``on(event, func=None, ttl=-1)``
Registers a function to an event. When *func* is *None*, decorator usage is assumed. *ttl*
defines the times to listen. Negative values mean infinity. Returns the function.
- #### ``once(event, func=None)``
Registers a function to an event with ``ttl = 1``. When *func* is *None*, decorator usage is
assumed. Returns the function.
- #### ``on_any(func=None)``
Registers a function that is called every time an event is emitted. When *func* is *None*,
decorator usage is assumed. Returns the function.
- #### ``off(event, func=None)``
Removes a function that is registered to an event. When *func* is *None*, decorator usage is
assumed. Returns the function.
- #### ``off_any(func=None)``
Removes a function that was registered via ``on_any()``. When *func* is *None*, decorator usage
is assumed. Returns the function.
- #### ``off_all()``
Removes all functions of all events.
- #### ``listeners(event)``
Returns all functions that are registered to an event. Wildcards are not applied.
- #### ``listeners_any()``
Returns all functions that were registered using ``on_any()``.
- #### ``listeners_all()``
Returns all registered functions.
- #### ``emit(event, *args, **kwargs)``
Emits an event. All functions of events that match *event* are invoked with *args* and *kwargs*
in the exact order of their registeration. Wildcards might be applied.
## Development
- Source hosted at [GitHub](https://github.com/riga/pymitter)
- Python module hostet at [PyPI](https://pypi.python.org/pypi/pymitter)
- Report issues, questions, feature requests on [GitHub Issues](https://github.com/riga/pymitter/issues)

53
python-ecosys/pymitter/examples.py 100644
Wyświetl plik

@ -0,0 +1,53 @@
# coding: utf-8
# python imports
import os
import sys
from pymitter import EventEmitter
# create an EventEmitter instance
ee = EventEmitter(wildcard=True, new_listener=True, max_listeners=-1)
@ee.on("new_listener")
def on_new(func, event=None):
print("added listener", event, func)
@ee.on("foo")
def handler_foo1(arg):
print("foo handler 1 called with", arg)
@ee.on("foo")
def handler_foo2(arg):
print("foo handler 2 called with", arg)
@ee.on("foo.*", ttl=1)
def handler_fooall(arg):
print("foo.* handler called with", arg)
@ee.on("foo.bar")
def handler_foobar(arg):
print("foo.bar handler called with", arg)
@ee.on_any()
def handler_any(*args, **kwargs):
print("called every time")
print("emit foo")
ee.emit("foo", "test")
print(10 * "-")
print("emit foo.bar")
ee.emit("foo.bar", "test")
print(10 * "-")
print("emit foo.*")
ee.emit("foo.*", "test")
print(10 * "-")

7
python-ecosys/pymitter/manifest.py 100644
Wyświetl plik

@ -0,0 +1,7 @@
metadata(
description="Event subscription and publishing tools.",
version="0.3.2",
pypi="pymitter",
)
module("pymitter.py")

284
python-ecosys/pymitter/pymitter.py 100644
Wyświetl plik

@ -0,0 +1,284 @@
# coding: utf-8
"""
Python port of the extended Node.js EventEmitter 2 approach providing namespaces, wildcards and TTL.
"""
__author__ = "Marcel Rieger"
__author_email__ = "github.riga@icloud.com"
__copyright__ = "Copyright 2014-2022, Marcel Rieger"
__credits__ = ["Marcel Rieger"]
__contact__ = "https://github.com/riga/pymitter"
__license__ = "BSD-3-Clause"
__status__ = "Development"
__version__ = "0.3.2"
__all__ = ["EventEmitter", "Listener"]
import time
class EventEmitter(object):
"""
The EventEmitter class, ported from Node.js EventEmitter 2.
When *wildcard* is *True*, wildcards in event names are taken into account. When *new_listener*
is *True*, a ``"new_listener"`` event is emitted every time a new listener is registered with
arguments ``(func, event=None)``. *max_listeners* configures the maximum number of event
listeners. A negative numbers means that this number is unlimited. Event names have namespace
support with each namspace being separated by a *delimiter* which defaults to ``"."``.
"""
CB_KEY = "__callbacks"
WC_CHAR = "*"
def __init__(self, wildcard=False, new_listener=False, max_listeners=-1, delimiter="."):
super(EventEmitter, self).__init__()
self.wildcard = wildcard
self.delimiter = delimiter
self.new_listener = new_listener
self.max_listeners = max_listeners
self._tree = self._new_branch()
@classmethod
def _new_branch(cls):
"""
Returns a new branch. Essentially, a branch is just a dictionary with a special item
*CB_KEY* that holds registered functions. All other items are used to build a tree
structure.
"""
return {cls.CB_KEY: []}
def _find_branch(self, event):
"""
Returns a branch of the tree structure that matches *event*. Wildcards are not applied.
"""
parts = event.split(self.delimiter)
if self.CB_KEY in parts:
return None
branch = self._tree
for p in parts:
if p not in branch:
return None
branch = branch[p]
return branch
@classmethod
def _remove_listener(cls, branch, func):
"""
Removes a listener given by its function *func* from a *branch*.
"""
listeners = branch[cls.CB_KEY]
indexes = [i for i, l in enumerate(listeners) if l.func == func]
for i in indexes[::-1]:
listeners.pop(i)
def on(self, event, func=None, ttl=-1):
"""
Registers a function to an event. *ttl* defines the times to listen. Negative values mean
infinity. When *func* is *None*, decorator usage is assumed. Returns the function.
"""
def on(func):
if not callable(func):
return func
parts = event.split(self.delimiter)
if self.CB_KEY in parts:
return func
branch = self._tree
for p in parts:
branch = branch.setdefault(p, self._new_branch())
listeners = branch[self.CB_KEY]
if 0 <= self.max_listeners <= len(listeners):
return func
listener = Listener(func, event, ttl)
listeners.append(listener)
if self.new_listener:
self.emit("new_listener", func, event)
return func
return on(func) if func else on
def once(self, event, func=None):
"""
Registers a function to an event that is called once. When *func* is *None*, decorator usage
is assumed. Returns the function.
"""
return self.on(event, func=func, ttl=1)
def on_any(self, func=None, ttl=-1):
"""
Registers a function that is called every time an event is emitted. *ttl* defines the times
to listen. Negative values mean infinity. When *func* is *None*, decorator usage is assumed.
Returns the function.
"""
def on_any(func):
if not callable(func):
return func
listeners = self._tree[self.CB_KEY]
if 0 <= self.max_listeners <= len(listeners):
return func
listener = Listener(func, None, ttl)
listeners.append(listener)
if self.new_listener:
self.emit("new_listener", func)
return func
return on_any(func) if func else on_any
def off(self, event, func=None):
"""
Removes a function that is registered to an event. When *func* is *None*, decorator usage is
assumed. Returns the function.
"""
def off(func):
branch = self._find_branch(event)
if branch is None:
return func
self._remove_listener(branch, func)
return func
return off(func) if func else off
def off_any(self, func=None):
"""
Removes a function that was registered via :py:meth:`on_any`. When *func* is *None*,
decorator usage is assumed. Returns the function.
"""
def off_any(func):
self._remove_listener(self._tree, func)
return func
return off_any(func) if func else off_any
def off_all(self):
"""
Removes all registered functions.
"""
self._tree = self._new_branch()
def listeners(self, event):
"""
Returns all functions that are registered to an event. Wildcards are not applied.
"""
branch = self._find_branch(event)
if branch is None:
return []
return [listener.func for listener in branch[self.CB_KEY]]
def listeners_any(self):
"""
Returns all functions that were registered using :py:meth:`on_any`.
"""
return [listener.func for listener in self._tree[self.CB_KEY]]
def listeners_all(self):
"""
Returns all registered functions.
"""
listeners = list(self._tree[self.CB_KEY])
branches = list(self._tree.values())
for b in branches:
if not isinstance(b, dict):
continue
branches.extend(b.values())
listeners.extend(b[self.CB_KEY])
return [listener.func for listener in listeners]
def emit(self, event, *args, **kwargs):
"""
Emits an *event*. All functions of events that match *event* are invoked with *args* and
*kwargs* in the exact order of their registration. Wildcards might be applied.
"""
parts = event.split(self.delimiter)
if self.CB_KEY in parts:
return
listeners = list(self._tree[self.CB_KEY])
branches = [self._tree]
for p in parts:
_branches = []
for branch in branches:
for k, b in branch.items():
if k == self.CB_KEY:
continue
if k == p:
_branches.append(b)
elif self.wildcard and self.WC_CHAR in (p, k):
_branches.append(b)
branches = _branches
for b in branches:
listeners.extend(b[self.CB_KEY])
# sort listeners by registration time
listeners = sorted(listeners, key=lambda listener: listener.time)
# call listeners in the order of their registration time
for listener in sorted(listeners, key=lambda listener: listener.time):
listener(*args, **kwargs)
# remove listeners whose ttl value is 0
for listener in listeners:
if listener.ttl == 0:
self.off(listener.event, func=listener.func)
class Listener(object):
"""
A simple event listener class that wraps a function *func* for a specific *event* and that keeps
track of the times to listen left.
"""
def __init__(self, func, event, ttl):
super(Listener, self).__init__()
self.func = func
self.event = event
self.ttl = ttl
# store the registration time
self.time = time.time()
def __call__(self, *args, **kwargs):
"""
Invokes the wrapped function when ttl is non-zero, decreases the ttl value when positive and
returns whether it reached zero or not.
"""
if self.ttl != 0:
self.func(*args, **kwargs)
if self.ttl > 0:
self.ttl -= 1
return self.ttl == 0

0
python-ecosys/pymitter/requirements.txt 100644
Wyświetl plik

5
python-ecosys/pymitter/requirements_test.txt 100644
Wyświetl plik

@ -0,0 +1,5 @@
flake8<4;python_version<="2.7"
flake8>=4.0.1;python_version>="3.0"
flake8-commas>=2
flake8-quotes>=3,<3.3;python_version<="2.7"
flake8-quotes>=3;python_version>="3.0"

62
python-ecosys/pymitter/setup.py 100644
Wyświetl plik

@ -0,0 +1,62 @@
# coding: utf-8
import os
from setuptools import setup
import pymitter
this_dir = os.path.dirname(os.path.abspath(__file__))
keywords = [
"event",
"emitter",
"eventemitter",
"wildcard",
"node",
"nodejs",
]
classifiers = [
"Programming Language :: Python",
"Programming Language :: Python :: 2",
"Programming Language :: Python :: 2.7",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.4",
"Programming Language :: Python :: 3.5",
"Programming Language :: Python :: 3.6",
"Programming Language :: Python :: 3.7",
"Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9",
"Development Status :: 5 - Production/Stable",
"Operating System :: OS Independent",
"License :: OSI Approved :: BSD License",
"Intended Audience :: Developers",
]
# read the readme file
with open(os.path.join(this_dir, "README.md"), "r") as f:
long_description = f.read()
# load installation requirements
with open(os.path.join(this_dir, "requirements.txt"), "r") as f:
install_requires = [line.strip() for line in f.readlines() if line.strip()]
setup(
name=pymitter.__name__,
version=pymitter.__version__,
author=pymitter.__author__,
author_email=pymitter.__author_email__,
description=pymitter.__doc__.strip().split("\n")[0].strip(),
license=pymitter.__license__,
url=pymitter.__contact__,
keywords=keywords,
classifiers=classifiers,
long_description=long_description,
long_description_content_type="text/markdown",
install_requires=install_requires,
python_requires=">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*, !=3.4.*, <4",
zip_safe=False,
py_modules=[pymitter.__name__],
)

126
python-ecosys/pymitter/tests.py 100644
Wyświetl plik

@ -0,0 +1,126 @@
# coding: utf-8
import unittest
from pymitter import EventEmitter
class AllTestCase(unittest.TestCase):
def __init__(self, *args, **kwargs):
super(AllTestCase, self).__init__(*args, **kwargs)
self.ee1 = EventEmitter()
self.ee2 = EventEmitter(wildcard=True)
self.ee3 = EventEmitter(wildcard=True, delimiter=":")
self.ee4 = EventEmitter(new_listener=True)
self.ee5 = EventEmitter(max_listeners=1)
def test_1_callback_usage(self):
stack = []
def handler(arg):
stack.append("1_callback_usage_" + arg)
self.ee1.on("1_callback_usage", handler)
self.ee1.emit("1_callback_usage", "foo")
self.assertTrue(stack[-1] == "1_callback_usage_foo")
def test_1_decorator_usage(self):
stack = []
@self.ee1.on("1_decorator_usage")
def handler(arg):
stack.append("1_decorator_usage_" + arg)
self.ee1.emit("1_decorator_usage", "bar")
self.assertTrue(stack[-1] == "1_decorator_usage_bar")
def test_1_ttl_on(self):
stack = []
@self.ee1.on("1_ttl_on", ttl=1)
def handler(arg):
stack.append("1_ttl_on_" + arg)
self.ee1.emit("1_ttl_on", "foo")
self.assertTrue(stack[-1] == "1_ttl_on_foo")
self.ee1.emit("1_ttl_on", "bar")
self.assertTrue(stack[-1] == "1_ttl_on_foo")
def test_1_ttl_once(self):
stack = []
@self.ee1.once("1_ttl_once")
def handler(arg):
stack.append("1_ttl_once_" + arg)
self.ee1.emit("1_ttl_once", "foo")
self.assertTrue(stack[-1] == "1_ttl_once_foo")
self.ee1.emit("1_ttl_once", "bar")
self.assertTrue(stack[-1] == "1_ttl_once_foo")
def test_2_on_all(self):
stack = []
@self.ee2.on("2_on_all.*")
def handler():
stack.append("2_on_all")
self.ee2.emit("2_on_all.foo")
self.assertTrue(stack[-1] == "2_on_all")
def test_2_emit_all(self):
stack = []
@self.ee2.on("2_emit_all.foo")
def handler():
stack.append("2_emit_all.foo")
self.ee2.emit("2_emit_all.*")
self.assertTrue(stack[-1] == "2_emit_all.foo")
def test_3_delimiter(self):
stack = []
@self.ee3.on("3_delimiter:*")
def handler():
stack.append("3_delimiter")
self.ee3.emit("3_delimiter:foo")
self.assertTrue(stack[-1] == "3_delimiter")
def test_4_new(self):
stack = []
@self.ee4.on("new_listener")
def handler(func, event=None):
stack.append((func, event))
def newhandler():
pass
self.ee4.on("4_new", newhandler)
self.assertTrue(stack[-1] == (newhandler, "4_new"))
def test_5_max(self):
stack = []
@self.ee5.on("5_max")
def handler1():
stack.append("5_max_1")
@self.ee5.on("5_max")
def handler2():
stack.append("5_max_2")
self.ee5.emit("5_max")
self.assertTrue(stack[-1] == "5_max_1")
if __name__ == "__main__":
unittest.main()