semvec/pypi-packages huggingface dataset:
cm-events 0.1.0
Minimal async pub/sub package with decorators and state machines
License: MIT
Requires Python: >=3.11

# cm-events

Simple async pub/sub package I made for an assignment of an Embedded Systems curricular unit.

## What it does
Just decouples pub/sub messaging from actual business logic like reading sensors and pushing notifications. 
Didn't like all the bloated code I had so I built this mini package to abstract away event handling 
using decorators and implementing just the required methods for an RPI controller needed for the assignment.

Basically, It lets you:

- Have components that publish events (like sensor readers)
- Have components that subscribe to events (like data processors)
- Auto-register components with decorators
- Basic state machine support for stateful components

## Basic usage

```python
from events import EventType, Event, Publisher, Subscriber, Broker, register

class YourEvents(EventType):
    SOME_EVENT = "some_event"

@register
class YourSensor(Publisher):
    
    async def startup(self):
       ... # start up your sensor
    
    async def run(self):
        while True:
            reading = sense()  # your sensor code
            await self.publish(Event(
                type=YourEvents.SOME_EVENT,
                source="sensor_1",
                payload={"value": reading}
            ))
            await asyncio.sleep(1)
        
    async def shutdown(self):
        ... # shutdown your sensor

@register  
class ReadingLogger(Subscriber):
    
    async def startup(self):
        self.subscribe_to(YourEvents.SOME_EVENT)
    
    async def handle_event(self, event):
        print(f"Reading: {event.payload['value']}")

# Run it
async def main():
    broker = Broker()
    await broker.start()
    
    try:
        while True:
            await asyncio.sleep(1)
    except KeyboardInterrupt:
        pass
    
    await broker.stop()

asyncio.run(main())
```

## Component types

- `Publisher` - broadcasts events via `publish()`
- `Subscriber` - receives subscribed events via `handle_event()`  
- `Transceiver` - does both

## Multiple instances

If you need multiple instances of the same component:

```python
from events import Publisher, register_multiple

@register_multiple([
    {"pin": 18, "name": "sensor1"},
    {"pin": 19, "name": "sensor2"}
])
class GPIOSensor(Publisher):
    def __init__(self, pin, name):
        self._pin = pin
        self._name = name

    # ...
```

## State machines

For components that need states:

```python
from events import StateMachine, Publisher, register, state, initial_state

@register
@initial_state("idle")
class StatefulComponent(StateMachine, Publisher):
    
    @state("idle", poll_interval=1.0)
    async def idle_state(self):
        if some_condition():
            return "working"
    
    @state("working", poll_interval=0.1) 
    async def working_state(self):
        do_work()
        if done():
            return "idle"
```

## Install

```bash
poetry add cm-events
```

Requires Python 3.11+ and pydantic.
readme_patcher 0.7.2
Generate README files from templates. Allow input from functions calls and cli output.
License: MIT
Requires Python: ~=3.8

.. image:: http://img.shields.io/pypi/v/readme-patcher.svg
    :target: https://pypi.org/project/readme-patcher
    :alt: This package on the Python Package Index

.. image:: https://github.com/Josef-Friedrich/readme_patcher/actions/workflows/tests.yml/badge.svg
    :target: https://github.com/Josef-Friedrich/readme_patcher/actions/workflows/tests.yml
    :alt: Tests

readme_patcher
==============

Generate README files from templates. Allow input from functions calls and cli
output.

.. code-block:: shell

    cd your-project
    vim README_template.rst
    poetry add --group dev readme-patcher
    poetry shell
    readme-patcher # README.rst

Global objects
--------------

py_project
^^^^^^^^^^

.. code-block:: jinja

    {{ py_project.repository }}

github
^^^^^^

https://docs.github.com/en/rest/repos/repos#get-a-repository

https://api.github.com/repos/Josef-Friedrich/readme_patcher

.. code-block:: jinja

    {{ github.name }}
    {{ github.full_name }}
    {{ github.description }}
    {{ github.owner.login }}

badge
^^^^^

.. code-block:: jinja

    {{ badge.pypi }}
    {{ badge.github_workflow('tests' 'Tests') }}
    {{ badge.readthedocs }}

Functions
---------

``cli`` (Command line interface): Combined output (stdout and stderr) of
command line interfaces (scripts / binaries)

.. code-block:: jinja

    {{ cli('awk --help') }}

It is recommended to use the ``cli`` function together with the literal filter.

.. code-block:: jinja

    {{ cli('awk --help') | literal }}

``func``: return values of Python functions

.. code-block:: jinja

    {{ func('os.getcwd') }}

read: read text files

.. code-block:: jinja

    {{ read('code/example.py') | code('python') }}

Filters
-------

code
^^^^

.. code-block:: jinja

    {{ 'print("example")' | code('python') }}

::

    .. code-block:: python

        print("example")

literal
^^^^^^^

.. code-block:: jinja

    {{ func('os.getcwd') | literal }}

::

    ::

        /home/repos/project

heading
^^^^^^^

.. code-block:: jinja

    {{ 'heading 1' | heading(1) }}

    {{ 'heading 2' | heading(2) }}

    {{ 'heading 3' | heading(3) }}

    {{ 'heading 4' | heading(4) }}

::

    heading 1
    =========

    heading 2
    ---------

    heading 3
    ^^^^^^^^^

    heading 4
    """""""""

Examples
--------

.. code-block:: jinja

    {% for command in [
                      'dns-ipv6-prefix.py',
                      'extract-pdftext.py',
                      'find-dupes-by-size.py',
                      'list-files.py',
                      'mac-to-eui64.py',
                      'pdf-compress.py',
                      'image-into-pdf.py'
                      ]
    %}

    ``{{ command }}``

    {{ cli('{} --help'.format(command)) | literal }}
    {% endfor %}

Configuration
-------------

.. code-block:: toml

    [[tool.readme_patcher.file]]
    src = "README_template.rst"
    dest = "README.rst"
    variables = { cwd = "func:os.getcwd", fortune = "cli:fortune --help" }