AMIの独自のtimeモジュール(wip)

[Geson-anko]

概要

pythonのtimeモジュールをラップし、時間加速機能を実装した独自のtimeモジュールへ置換する。
from ami import timeで扱うことができる。

背景と目的

現在は生のtimeモジュールを使用し、sleep, time, perf_counter, monotonicなどを使用している。
しかし、AMIには時間加速機能が存在するため、AMIシステム全体のtime scaleを一貫して変更できる機能が欲しい。

提案内容

既存のtimeモジュールをラップし、置換することによってそれを実装する。

次の項目を実装する。

• time_scale設定機能
• sleep
• time
• perf_counter
• monotonic

そしてtimeモジュールを使用している全ての箇所において、ami timeモジュールへ置換する。

マルチスレッドで使用されるため、グローバル変数はスレッドセーフにする必要がある。

スケッチ

from threading import RLock
import time as _original_time
import copy

class _LockedGlobalVariables:
    __anthor_time = _original_time.time()
    __scaled_anchor_time = _original_time.time() 
    __anchor_perf_counter = _original_time.perf_counter()
    __scaled_anchor_perf_counter = _original_time.perf_counter()
    __anchor_monotonic = _original_time.monotonic()
    __scaled_anchor_monotonic = _original_time.monotonic()
    __time_scale = 1.0
    __lock = RLock()
    
    @classmethod
    @property
    def anchor_time(cls) -> float:
        with cls.__lock:
            return copy.copy(self.__anchor_time)
    # ... 
    # 以下 anchor_perf_counter, anchor_monotonic, scaled_*, time_scaleについても同様のgetter実装。

    def __init__(self) -> None:
        raise Error("Instantiation is not supported!")
    
    @classmethod
    def _update_anchor_values(cls) -> None:
        with cls.__lock:
            cls. __initial_time = _original_time.time()
            # perf_counter, monotonicに関しても同様
      
    @classmethod
    def _update_scaled_anchor_values(cls) -> None:
        ...
    
    @classmethod
    def set_time_scale(cls, time_scale: float) -> None:
        assert time_scale > 0, "Time scale must be > 0"
        with cls.__lock:
            cls._update_scaled_anchor_values()
            cls.__time_scale = time_scale
            cls. _update_initial_values()

def sleep(secs: float) -> None:
    _original_time.sleep(secs / _LockedGlobalVariables.time_scale)

def time() -> float:
    delta = _original_time.time() - _LockedGlobalVariables.anchor_time
    return _LockedGlobalVariables.scaled_anchor_time + delta * _LockedGlobalVariables.time_scale

def perf_counter() -> float:
    ...

def monotonic() -> float:
    ...

タスク

• ...

参考

[Geson-anko]

Lock デコレータを使うと便利かも

https://gist.github.com/ganesh-k13/f8a9c09192841570033c69ffa4665e8d

[Geson-anko]

Claudeに実装してもらったもの

"""
AMI Time Module

This module provides a custom implementation of time-related functions with a time acceleration feature.
It wraps the standard Python time module and allows for consistent time scaling across the AMI system.

Key features:
- Time acceleration: Adjust the speed of time passage in the system.
- Thread-safe: All operations are protected by locks for use in multi-threaded environments.
- Compatible API: Provides familiar time functions like sleep, time, perf_counter, and monotonic.

Usage:
    from ami import time

    # Get current time
    current_time = time.time()

    # Sleep for 1 second (affected by time scale)
    time.sleep(1)

    # Set time scale (e.g., 2x speed)
    time.set_time_scale(2.0)

    # Get current time scale
    current_scale = time.get_time_scale()

Note: This module is designed for use within the AMI system and may not be suitable for general-purpose time management.
"""

import time as _original_time
from threading import RLock
import copy

class _LockedGlobalVariables:
    _anchor_time = _original_time.time()
    _scaled_anchor_time = _original_time.time()
    _anchor_perf_counter = _original_time.perf_counter()
    _scaled_anchor_perf_counter = _original_time.perf_counter()
    _anchor_monotonic = _original_time.monotonic()
    _scaled_anchor_monotonic = _original_time.monotonic()
    _time_scale = 1.0
    _lock = RLock()

    def __init__(self) -> None:
        raise RuntimeError("Instantiation is not supported!")

    @classmethod
    @property
    def anchor_time(cls) -> float:
        with cls._lock:
            return copy.copy(cls._anchor_time)

    @classmethod
    @property
    def scaled_anchor_time(cls) -> float:
        with cls._lock:
            return copy.copy(cls._scaled_anchor_time)

    @classmethod
    @property
    def anchor_perf_counter(cls) -> float:
        with cls._lock:
            return copy.copy(cls._anchor_perf_counter)

    @classmethod
    @property
    def scaled_anchor_perf_counter(cls) -> float:
        with cls._lock:
            return copy.copy(cls._scaled_anchor_perf_counter)

    @classmethod
    @property
    def anchor_monotonic(cls) -> float:
        with cls._lock:
            return copy.copy(cls._anchor_monotonic)

    @classmethod
    @property
    def scaled_anchor_monotonic(cls) -> float:
        with cls._lock:
            return copy.copy(cls._scaled_anchor_monotonic)

    @classmethod
    @property
    def time_scale(cls) -> float:
        with cls._lock:
            return copy.copy(cls._time_scale)

    @classmethod
    def _update_anchor_values(cls) -> None:
        with cls._lock:
            cls._anchor_time = _original_time.time()
            cls._anchor_perf_counter = _original_time.perf_counter()
            cls._anchor_monotonic = _original_time.monotonic()

    @classmethod
    def _update_scaled_anchor_values(cls) -> None:
        with cls._lock:
            cls._scaled_anchor_time = time()
            cls._scaled_anchor_perf_counter = perf_counter()
            cls._scaled_anchor_monotonic = monotonic()

    @classmethod
    def set_time_scale(cls, time_scale: float) -> None:
        """
        Set the time scale for the AMI system.

        Args:
            time_scale (float): The new time scale. Must be greater than 0.

        Raises:
            AssertionError: If time_scale is not greater than 0.
        """
        assert time_scale > 0, "Time scale must be > 0"
        with cls._lock:
            cls._update_scaled_anchor_values()
            cls._update_anchor_values()
            cls._time_scale = time_scale
    
    @classmethod
    def get_time_scale(cls)
        """
        Get the current time scale of the AMI system.

        Returns:
            float: The current time scale.
        """
        return cls.__time_scale

def sleep(secs: float) -> None:
    """
    Suspend execution for the given number of seconds.

    This function is affected by the current time scale.

    Args:
        secs (float): The number of seconds to sleep.
    """
    _original_time.sleep(secs / _LockedGlobalVariables.time_scale)

def time() -> float:
    """
    Return the current time in seconds since the epoch.

    This function is affected by the current time scale.

    Returns:
        float: The current time in seconds since the epoch.
    """
    delta = _original_time.time() - _LockedGlobalVariables.anchor_time
    return _LockedGlobalVariables.scaled_anchor_time + delta * _LockedGlobalVariables.time_scale

def perf_counter() -> float:
    """
    Return the value (in fractional seconds) of a performance counter.

    This function is affected by the current time scale.

    Returns:
        float: The current value of the performance counter.
    """
    delta = _original_time.perf_counter() - _LockedGlobalVariables.anchor_perf_counter
    return _LockedGlobalVariables.scaled_anchor_perf_counter + delta * _LockedGlobalVariables.time_scale

def monotonic() -> float:
    """
    Return the value (in fractional seconds) of a monotonic time counter.

    This function is affected by the current time scale.

    Returns:
        float: The current value of the monotonic time counter.
    """
    delta = _original_time.monotonic() - _LockedGlobalVariables.anchor_monotonic
    return _LockedGlobalVariables.scaled_anchor_monotonic + delta * _LockedGlobalVariables.time_scale

# Expose the set_time_scale and get_time_scale functions
set_time_scale = _LockedGlobalVariables.set_time_scale
get_time_scale = _LockedGlobalVariables.time_scale.get_time_scale

[Geson-anko]

pause, resumeのイベントが発生したら その時間 ami.timeも止まっていてほしいよね。

[Geson-anko]

https://gist.github.com/Geson-anko/fd0af8b49b2cd9e22a95c08581a11c97