Standard Environment Plugin

The Standard Environment plugin (owa.env.std) is a core component of the Open World Agents framework. It provides essential functionalities related to time management and clock operations, which are fundamental for various time-based tasks and event scheduling within the system.

Features

• Time Functions: The plugin registers functions like std/time_ns that return the current time in nanoseconds.
• Tick Listener: It includes a std/tick listener that can be configured to execute callbacks at specified intervals.

Usage

The Standard Environment plugin is automatically available when you install owa-core. No manual activation needed!

# Components automatically available after installation
from owa.core.registry import CALLABLES, LISTENERS

You can access the registered functions and listeners via the global registries using the unified namespace/name pattern:

from owa.core.registry import CALLABLES, LISTENERS

# Get the current time in nanoseconds
current_time_ns = CALLABLES["std/time_ns"]()
print(f"Current time (ns): {current_time_ns}")

# Configure and start a tick listener
def on_tick():
    print(f"Tick at {CALLABLES['std/time_ns']()}")

tick_listener = LISTENERS["std/tick"]()
tick_listener.configure(callback=on_tick, interval=1)  # Tick every second
tick_listener.start()

# Run for a few seconds to see the tick listener in action
import time
time.sleep(5)

# Stop the tick listener
tick_listener.stop()
tick_listener.join()

Components

Time Functions

• std/time_ns: Returns the current time in nanoseconds. This function is registered in the CALLABLES registry.

Tick Listener

• std/tick: A listener that triggers a callback at specified intervals. This listener is registered in the LISTENERS registry and can be configured with an interval in seconds.

Example

Here is a complete example demonstrating how to use the Standard Environment plugin:

from owa.core.registry import CALLABLES, LISTENERS

# Components automatically available - no activation needed!

# Print the current time in nanoseconds
print(CALLABLES["std/time_ns"]())

# Define a callback function for the tick listener
def tick_callback():
    print(f"Tick at {CALLABLES['std/time_ns']()}")

# Configure and start the tick listener
tick_listener = LISTENERS["std/tick"]().configure(callback=tick_callback, interval=1)
tick_listener.start()

# Let the listener run for 5 seconds
import time
time.sleep(5)

# Stop the tick listener
tick_listener.stop()
tick_listener.join()

This example demonstrates how to use the automatically available components, retrieve the current time, and set up a tick listener that prints the current time every second.

The Standard Environment plugin is a fundamental part of the Open World Agents framework, providing essential time-based functionalities that can be leveraged by other modules and applications.

Auto-generated documentation

std plugin 0.1.0

Standard system components for OWA

Author: OWA Development Team

Callables

Usage: To use callable components, import CALLABLES from owa.core and access them by their component name:

from owa.core import CALLABLES

# Access a callable component (replace 'component_name' with actual name)
callable_func = CALLABLES["std/component_name"]
result = callable_func(your_arguments)

time_ns

time_ns() -> int

Return the current time in nanoseconds since the Unix epoch.

This function provides high-precision timing for OWA components, useful for performance measurement and precise scheduling.

Returns:

Name	Type	Description
int	int	Current time in nanoseconds since Unix epoch (January 1, 1970)

Examples:

Get current timestamp:

>>> current_time = time_ns()
>>> print(f"Current time: {current_time}")

Measure execution time:

>>> start = time_ns()
>>> # ... some operation ...
>>> duration = time_ns() - start
>>> print(f"Operation took {duration} nanoseconds")

Source code in projects/owa-core/owa/env/std/clock.py

def time_ns() -> int:
    """
    Return the current time in nanoseconds since the Unix epoch.

    This function provides high-precision timing for OWA components,
    useful for performance measurement and precise scheduling.

    Returns:
        int: Current time in nanoseconds since Unix epoch (January 1, 1970)

    Examples:
        Get current timestamp:

        >>> current_time = time_ns()
        >>> print(f"Current time: {current_time}")

        Measure execution time:

        >>> start = time_ns()
        >>> # ... some operation ...
        >>> duration = time_ns() - start
        >>> print(f"Operation took {duration} nanoseconds")
    """
    return time.time_ns()

Listeners

Usage: To use listener components, import LISTENERS from owa.core and call the configure() method with a callback function:

from owa.core import LISTENERS

# Configure a listener component (replace 'component_name' with actual name)
listener = LISTENERS["std/component_name"]
listener.configure(callback=my_callback, your_other_arguments)

# Use the listener in a context manager
with listener.session as active_listener:
    # The listener is now running and will call my_callback when events occur
    pass  # Your main code here

Note: The callback argument is required. The on_configure() method shown in the documentation is an internal method called by configure().

tick

Bases: Listener

A listener that triggers callbacks at regular intervals.

This listener provides precise timing for periodic tasks in OWA, supporting configurable intervals and automatic callback execution.

Examples:

Basic usage with 1-second interval:

>>> def on_tick():
...     print(f"Tick at {time_ns()}")
>>>
>>> listener = ClockTickListener()
>>> listener.configure(callback=on_tick, interval=1)
>>> listener.start()
>>> # ... listener runs in background ...
>>> listener.stop()
>>> listener.join()

Custom interval timing:

>>> listener = ClockTickListener()
>>> listener.configure(callback=my_callback, interval=0.5)  # 500ms
>>> listener.start()

on_configure

on_configure(*, interval: float = 1)

Configure the tick interval for the listener.

Parameters:

Name	Type	Description	Default
interval	float	Time between ticks in seconds. Defaults to 1.0.	1

Examples:

Configure for 100ms intervals:

>>> listener.configure(callback=my_func, interval=0.1)

Source code in projects/owa-core/owa/env/std/clock.py

def on_configure(self, *, interval: float = 1):
    """
    Configure the tick interval for the listener.

    Args:
        interval (float): Time between ticks in seconds. Defaults to 1.0.

    Examples:
        Configure for 100ms intervals:

        >>> listener.configure(callback=my_func, interval=0.1)
    """
    self.interval = interval * S_TO_NS

loop

loop(*, stop_event, callback)

Main loop that executes callbacks at configured intervals.

Parameters:

Name	Type	Description	Default
stop_event		Threading event to signal when to stop	required
callback		Function to call at each tick	required

Note

This method runs in a separate thread and maintains precise timing by accounting for callback execution time.

Source code in projects/owa-core/owa/env/std/clock.py

def loop(self, *, stop_event, callback):
    """
    Main loop that executes callbacks at configured intervals.

    Args:
        stop_event: Threading event to signal when to stop
        callback: Function to call at each tick

    Note:
        This method runs in a separate thread and maintains precise timing
        by accounting for callback execution time.
    """
    self._last_called = time.time()
    while not stop_event.is_set():
        callback()
        to_sleep = self.interval - (time.time() - self._last_called)
        if to_sleep > 0:
            stop_event.wait(to_sleep / S_TO_NS)