Source code for jitx.events
"""
Event System
============
This module declares the JITX event system for triggering behavior at specific
points during design processing and instantiation.
"""
from collections import defaultdict
from collections.abc import Callable
from dataclasses import dataclass, field
import inspect
from typing import Any, Self, TypeVar
from .context import Context
from ._structural import PrePostInit
from ._instantiation import InstantiationStructure, instantiation
import sys
if sys.version_info > (3, 13):
T = TypeVar("T", default=None)
[docs]
class Event[T]:
"""Events can be used to trigger various behavior throughout your design,
such as verifying a particular condition after the design has been fully
instantiated.
Some events are predefined and will be triggered automatically by the
framework, and user defined events have to be triggered manually. Subclass
this class to create a user defined event.
"""
[docs]
@classmethod
def on(
cls, method: Callable[[Any, Self], T] | Callable[[Self], T] | Callable[[], T]
):
"""Register a method to be called when this event fires. The method
should accept an instance of the event as argument.
The method will be called with the context set to the same context that
was enabled when the method was registered.
>>> class MyComponent(Component):
@Design.Initialized.on
def on_initialized(self, init: Design.Initialized):
... # do something after initialization
"""
if instantiation.active():
ec = EventContext.require()
ec.events[cls].append((method, instantiation.current_frame))
return method
else:
def register(ob, _):
m = method.__get__(ob, type(ob))
ec = EventContext.require()
ec.events[cls].append((m, instantiation.current_frame))
return method
return PrePostInit(register, lambda _, ob: ob)
[docs]
def fire(self):
"""Fire this event.
Note that firing predefined events when they're not expected may cause
unpredictable results as internal infrastructure may rely on them.
>>> @dataclass
... class MyEvent(Event):
some_field: str
>>> class MyComponent(Component):
... @MyEvent.on
... def my_event_handler(self, myevent):
... print(f"Hello {myevent.some_field}!")
...
... @Design.Initialized.on
... def on_initialized(self):
... MyEvent("World").fire()
"""
ec = EventContext.require()
cls = self.__class__
if cls in ec.events:
for recv, frame in ec.events[cls]:
with instantiation.frame(frame):
if inspect.signature(recv).parameters:
recv(self)
else:
recv()
[docs]
@dataclass
class EventContext(Context):
"""Context for managing event handlers during design processing."""
events: dict[type[Event], list[tuple[Callable, InstantiationStructure.Frame]]] = (
field(default_factory=lambda: defaultdict(list))
)