Standard Environment

Core utilities and timing functions for OWA agents.

Installation

pip install owa-core  # Included automatically

Components

Component	Type	Description
std/time_ns	Callable	Get current time in nanoseconds
std/tick	Listener	Periodic callback execution

Usage Examples

Time FunctionsPeriodic TasksManual Control

from owa.core import CALLABLES

# Get current time
current_time = CALLABLES["std/time_ns"]()
print(f"Current time: {current_time}")

from owa.core import LISTENERS
import time

def on_tick():
    print(f"Tick: {CALLABLES['std/time_ns']()}")

# Using context manager (recommended)
tick = LISTENERS["std/tick"]().configure(callback=on_tick, interval=1)
with tick.session:
    time.sleep(3)  # Prints every second for 3 seconds

# Manual start/stop control
tick = LISTENERS["std/tick"]().configure(callback=on_tick, interval=1)
tick.start()
time.sleep(3)
tick.stop()
tick.join()

API Reference

std plugin 0.5.3

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)