Hair Trigger

Simple, Subscribable, Custom Events
Explore the docs »

Report Bug · Request Feature

Table of Contents

1. About The Project
2. Getting Started
   • Installation
3. Usage
• Defining Events
• Assigning Instances
• Subscribing to Events
• Triggering Events
• Configuration
4. License
5. Contact
6. Acknowledgments

About The Project

Hair Trigger offers custom, subscribable events in the style of Luau events that allow for decoupled access between objects.

(back to top)

Getting Started

Hair Trigger is written in pure python, with no system dependencies, and should be OS-agnostic.

Installation

Hair Trigger can be installed from the PyPI using pip:

pip install hair_trigger

and can be imported for use with:

import hair_trigger

Hair Trigger has no dependencies beyond python itself.

(back to top)

Usage

Hair Trigger supplies no events by default, they must be custom made.

As an example, we'll make a simple event for detecting when an object is enabled.

Defining Events

from typing import Any
import hair_trigger

class OnEnable(hair_trigger.Event):
    """
    Called whenever the owner becomes enabled.

    :param this: The object being enabled.
    """

    def trigger(self, this: Any) -> None:
        return super().trigger(this)

Naming convention is suggested as On[Event name]. The trigger method must be defined, and must, at a minimum, call the super method. trigger's signature will also define the required signature of subscribing callbacks. It is recommended to put the docstring describing trigger's parameters in the class docstring, so that it is visible to users.

(back to top)

Assigning Instances

Now we'll need an object to have the event.

class Foo:
    def __init__(self, enabled: bool = False) -> None:
        self.OnEnable = OnEnable()
        self._enabled = enabled

When a Foo is created, a new instance of OnEnable is created for it, as well. It is recommended that the event attribute breaks normal snake_case style and uses PascalCase to make it clear that this is an event object rather than a method or a typical attribute.

(back to top)

Subscribing to Events

Let's say we want to print something when a Foo is enabled. Subscribing is done primarily by using the event instance as a decorator.

foo = Foo()

@foo.OnEnable
def do_the_thing(this: Foo) -> None:
    print(f"{this} has been enabled")


# For simple expressions, a lambda is also okay
# For this though, we do not use it as a decorator.

foo.OnEnable(lambda this: print(f"{this} has been enabled"))

Additionally, objects can subscribe to an event as well. It uses the event as a decorator, too, but requires an additional parameter, the subscribing object. Subscribers need an owner so they don't tie up garbage collection.

class FooListener:

    def __init__(self, foo: Foo) -> None:

        @foo.OnEnable(self)
        def _(self, this: Foo) -> None:
            # Note: `self` here will shadow the `self` of init. This is important!
            print(f"{self} noticed {this} is now enabled")

Alternatively, we can subscribe to a bound method directly, by using the event as a regular function. This doesn't require the subscribing object to be passed, it is extracted from the bound method.

class FooListener:

    def __init__(self, foo:Foo) -> None:

        foo.OnEnable(self.listen_in)
        
    def listen_in(self, this: Foo) -> None:
        print(f"{self} noticed {this} is now enabled")

Both versions have the same behavior, and if we have multiple of FooListener with the same Foo, the message will be printed once each.

Important notes:

• The callback must be subscribed in a method, not the class definition.
• The init method is a great candidate for callback subscription, but it can be done elsewhere if needed. Get creative!
• For new callbacks created inside the init/equivalent:
  • The callback does not need a name, "_" is fine.
  • The self inside the callback must shadow the self of the init. This allows the callback to use the object, but won't prevent garbage collection due to a closure.
  • Other than the self, the signature take all parameters as the trigger method of the event. Unused parameters can be caught with *args.

(back to top)

Triggering Events

Now that we have a listener, we'll need to actually to something to trigger the event. To do this, we'll simply need to call the trigger method of the event.

class Foo:
    # init definition as above

    @property
    def enabled(self) -> bool:
        return self._enabled
    
    @enabled.setter
    def enabled(self, enabled: bool) ->:
        self._enabled = enabled
        if enabled:
            self.OnEnable.trigger(self)

(back to top)

Configuration

By default, Hair Trigger will attempt to run callbacks immediately, in syncronous mode. If asynchronous behavior is needed, or events need to be run manually or in a particular order, this can be changed using the config function.

Synchronous vs Asynchronous

The default system will run callback synchronously, so any blocking that occurs will block the entire thread. If that's undesireable, you can also use:

• ThreadRunner: Uses the Python threading module to run callbacks in new threads, good for general purpose multithreading.
• AsyncioRunner: Uses Python's asyncio module, useful for when threading must be async-aware, such as in WASM deployments.

import hair_trigger
from hair_trigger.runner import AsyncioRunner, ThreadRunner

# Run standard threads
hair_trigger.config(runner=ThreadRunner())


# Run async-aware
hair_trigger.config(runner=AsyncioRunner())

Scheduling Modes

Without config, triggering an event instantly begins notifying the event's subscribers, and if those trigger additional events, they'll take over mid-call. Instead, you can use a deferred scheduler.

Included are:

• StackScheduler: New events are put onto a stack, so that the newest event resolve before olderone resolve.
• QueueScheduler: New events are put into a queue, so events resolve in the order they are triggered.

The deferred schedulers must be triggered manually, using hair_trigger.scheduler.pump_events().

import hair_trigger
import hair_trigger.scheduler
from hair_trigger.scheduler import QueueScheduler

hair_trigger.config(scheduler=QueueScheduler())

# Do things to trigger events

hair_trigger.scheduler.pump_events()

Custom Runners and Schedulers

Runners and schedulers are protocols, so custom one can be created to get specific behaviors.

For example:

class LoggingThreadRunner:

    def schedule(self, func: Callable[..., Any], *args, **kwds) -> None:
        print(f"Calling function {func}")
        threading.Thread(target=func, args=args, kwargs=kwds).start()


hair_trigger.config(LoggingThreadRunner())

This will log the function before starting the thread.

(back to top)

(back to top)

License

Distributed under the MIT License. See LICENSE.txt for more information.

(back to top)

Contact

Better Built Fool - betterbuiltfool@gmail.com

Bluesky - @betterbuiltfool.bsky.social

Project Link: https://github.com/BetterBuiltFool/hair_trigger

(back to top)