<commit_before>import logging

from rx.internal import PriorityQueue, ArgumentOutOfRangeException

from .schedulerbase import SchedulerBase
from .scheduleditem import ScheduledItem
from .scheduleperiodicrecursive import SchedulePeriodicRecursive

log = logging.getLogger("Rx")


class VirtualTimeScheduler(SchedulerBase):
    """Virtual Scheduler. This scheduler should work with either
    datetime/timespan or ticks as int/int"""

    def __init__(self, initial_clock=0):
        """Creates a new virtual time scheduler with the specified initial
        clock value and absolute time comparer.

        Keyword arguments:
        initial_clock -- Initial value for the clock.
        comparer -- Comparer to determine causality of events based on absolute
            time.
        """
        self.clock = initial_clock

        self.is_enabled = False
        self.queue = PriorityQueue(1024)

        super(VirtualTimeScheduler, self).__init__()

    @property
    def now(self):
        """Gets the schedulers absolute time clock value as datetime offset."""

        return self.to_datetime(self.clock)

    def schedule(self, action, state=None):
        """Schedules an action to be executed."""

        return self.schedule_absolute(self.clock, action, state)

    def schedule_relative(self, duetime, action, state=None):
        """Schedules an action to be executed at duetime. Return the disposable
        object used to cancel the scheduled action (best effort)

        Keyword arguments:
        duetime -- Relative time after which to execute the action.
        action -- Action to be executed.
        state -- [Optional] State passed to the action to be executed."""

        runat = self.add(self.clock, self.to_relative(duetime))
        return self.schedule_absolute(duetime=runat, action=action, state=state)

    def schedule_absolute(self, duetime, action, state=None):
        """Schedules an action to be executed at duetime."""

        def run(scheduler, state1):
            return action(scheduler, state1)

        si = ScheduledItem(self, state, run, duetime)
        self.queue.enqueue(si)
        return si.disposable

    def schedule_periodic(self, period, action, state=None):
        scheduler = SchedulePeriodicRecursive(self, period, action, state)
        return scheduler.start()

    def start(self):
        """Starts the virtual time scheduler."""

        if self.is_enabled:
            return

        self.is_enabled = True
        while self.is_enabled:
            next = self.get_next()
            if not next:
                break<commit_msg>Run action not needed anymore<commit_after>import logging

from rx.internal import PriorityQueue, ArgumentOutOfRangeException

from .schedulerbase import SchedulerBase
from .scheduleditem import ScheduledItem
from .scheduleperiodicrecursive import SchedulePeriodicRecursive

log = logging.getLogger("Rx")


class VirtualTimeScheduler(SchedulerBase):
    """Virtual Scheduler. This scheduler should work with either
    datetime/timespan or ticks as int/int"""

    def __init__(self, initial_clock=0):
        """Creates a new virtual time scheduler with the specified initial
        clock value and absolute time comparer.

        Keyword arguments:
        initial_clock -- Initial value for the clock.
        comparer -- Comparer to determine causality of events based on absolute
            time.
        """
        self.clock = initial_clock

        self.is_enabled = False
        self.queue = PriorityQueue(1024)

        super(VirtualTimeScheduler, self).__init__()

    @property
    def now(self):
        """Gets the schedulers absolute time clock value as datetime offset."""

        return self.to_datetime(self.clock)

    def schedule(self, action, state=None):
        """Schedules an action to be executed."""

        return self.schedule_absolute(self.clock, action, state)

    def schedule_relative(self, duetime, action, state=None):
        """Schedules an action to be executed at duetime. Return the disposable
        object used to cancel the scheduled action (best effort)

        Keyword arguments:
        duetime -- Relative time after which to execute the action.
        action -- Action to be executed.
        state -- [Optional] State passed to the action to be executed."""

        runat = self.add(self.clock, self.to_relative(duetime))
        return self.schedule_absolute(duetime=runat, action=action, state=state)

    def schedule_absolute(self, duetime, action, state=None):
        """Schedules an action to be executed at duetime."""

        si = ScheduledItem(self, state, action, duetime)
        self.queue.enqueue(si)
        return si.disposable

    def schedule_periodic(self, period, action, state=None):
        scheduler = SchedulePeriodicRecursive(self, period, action, state)
        return scheduler.start()

    def start(self):
        """Starts the virtual time scheduler."""

        if self.is_enabled:
            return

        self.is_enabled = True
        while self.is_enabled:
            next = self.get_next()
            if not next:
                break
{'fraction_non_alphanumeric': 0.07913669064748201, 'fraction_numerical': 0.00233326852031888, 'mean_word_length': 2.765739385065886, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6598491', 'n_tokens_mistral': 1360, 'n_tokens_neox': 1290, 'n_words': 453}