<commit_before>"""
Asynchronous sequence execution
"""

# Twisted Imports
from twisted.internet import reactor, defer, task
from twisted.python import log
import twisted.internet.error

# System Imports
from exceptions import IndexError
import re

# Sibling Imports
from error import Error, NotRunning, AlreadyRunning, NotPaused, Stopped
import util

# Package Imports
from ..util import now, Event as util_Event
from ..constants import State
from ..data import Variable, Constant

__all__ = [
	"Step", "Sequence", "Parallel", "IfStep", "SetStep", "CancelStep",
	"LogStep", "WhileStep", "WaitStep", "WaitUntilStep", "CallStep", 
	"Error", "NotRunning", "AlreadyRunning", "NotPaused", "Stopped"
]

def _counter ():
	i = 1
	while i > 0:
		yield i
		i += 1

_counter = _counter()


class Step (util.BaseStep):

	type = "step"
	duration = 0

	@property
	def state (self):
		return self._state

	@state.setter
	def state (self, value):
		self._state = value
		self.event(item = self, state = value)

	def __init__ (self, expr = None):
		self.id = _counter.next()
		self.complete = defer.Deferred()
		self.dependents = util.Dependents()

		self.event = util_Event()
		self.log = util_Event()

		if expr is not None and not isinstance(expr, Variable):
			expr = Constant(expr)

		self._expr = expr

		self.state = State.READY

	def _run (self):
		self.dependents.event += self.event
		self.dependents.log += self.log

		self.dependents.run()

		return self.complete

	def _reset (self):
		self.complete = defer.Deferred()
		return self.dependents.reset()

	def _pause (self):
		return self.dependents.pause()

	def _resume (self):
		return self.dependents.resume()

	def _cancel (self, abort = False):
		if abort:
			return self._error(Stopped())
		else:
			return self._complete()

	def _complete (self, result = None):
		self.state = State.COMPLETE
		return self.__cb(self.complete.callback, self.dependents.cancel, result)

	def _complete_cb (self, result = None):
		self._complete(result)
		return result

	def _error (self, failure):
		self.state = State.ERROR
		return self.__cb(self.complete.errback, self.dependents.abort, failure)

	def _error_cb (self, failure):
		self._error(failure)
		return failure

	def __cb (self, cb_fn, dep_fn, value):
		def cb (result):
			self.dependents.event -= self.event
			self.dependents.log -= self.log

			try:
				cb_fn(value)
			except defer.AlreadyCalledError:
				pass

		dep_fn() \
			.addErrback(log.err) \
			.addBoth(cb) \
			.addErrback(log.err)

		return self.complete

	def serialize (self):
		serialized = {
			"id": self.id,
			"type": self.__class__.__name__.lower(),
			"state": self.state.value,
			"duration": self.duration
		}

		if self._expr is not None:
			try:
				serialized["expr"] = self._expr.serialize()
			except AttributeError:
				serialized["expr"] = str(self._expr)

		return serialized


class _StepWithChild (Step):

	def __init__ (self, expr = None, step = None):
		Step.__init__(self, expr)

		if isinstance(step, Step):
			self._step = step
		elif isinstance(step, list):
			self._step = Sequence(step)
		else:
			raise Error("Argument must be an instance of Step")

		step.event += self.event
		step.log += self.log

	def _cancel (self, abort = False):
		d = self.dependents.cancel(abort)

		try:
			d2 = self._step.cancel(abort)
			return defer.gatherResults([d, d2])
		except NotRunning:
			return d

	def _reset (self):
		return defer.gatherResults([
			defer.maybeDeferred(Step._reset, self),
			self._step.reset()
		])

	def _pause (self):
		return defer.gatherResults([<commit_msg>Accept lists in place of sequences.<commit_after>"""
Asynchronous sequence execution
"""

# Twisted Imports
from twisted.internet import reactor, defer, task
from twisted.python import log
import twisted.internet.error

# System Imports
from exceptions import IndexError
import re

# Sibling Imports
from error import Error, NotRunning, AlreadyRunning, NotPaused, Stopped
import util

# Package Imports
from ..util import now, Event as util_Event
from ..constants import State
from ..data import Variable, Constant

__all__ = [
	"Step", "Sequence", "Parallel", "IfStep", "SetStep", "CancelStep",
	"LogStep", "WhileStep", "WaitStep", "WaitUntilStep", "CallStep", 
	"Error", "NotRunning", "AlreadyRunning", "NotPaused", "Stopped"
]

def _counter ():
	i = 1
	while i > 0:
		yield i
		i += 1

_counter = _counter()


class Step (util.BaseStep):

	type = "step"
	duration = 0

	@property
	def state (self):
		return self._state

	@state.setter
	def state (self, value):
		self._state = value
		self.event(item = self, state = value)

	def __init__ (self, expr = None):
		self.id = _counter.next()
		self.complete = defer.Deferred()
		self.dependents = util.Dependents()

		self.event = util_Event()
		self.log = util_Event()

		if expr is not None and not isinstance(expr, Variable):
			expr = Constant(expr)

		self._expr = expr

		self.state = State.READY

	def _run (self):
		self.dependents.event += self.event
		self.dependents.log += self.log

		self.dependents.run()

		return self.complete

	def _reset (self):
		self.complete = defer.Deferred()
		return self.dependents.reset()

	def _pause (self):
		return self.dependents.pause()

	def _resume (self):
		return self.dependents.resume()

	def _cancel (self, abort = False):
		if abort:
			return self._error(Stopped())
		else:
			return self._complete()

	def _complete (self, result = None):
		self.state = State.COMPLETE
		return self.__cb(self.complete.callback, self.dependents.cancel, result)

	def _complete_cb (self, result = None):
		self._complete(result)
		return result

	def _error (self, failure):
		self.state = State.ERROR
		return self.__cb(self.complete.errback, self.dependents.abort, failure)

	def _error_cb (self, failure):
		self._error(failure)
		return failure

	def __cb (self, cb_fn, dep_fn, value):
		def cb (result):
			self.dependents.event -= self.event
			self.dependents.log -= self.log

			try:
				cb_fn(value)
			except defer.AlreadyCalledError:
				pass

		dep_fn() \
			.addErrback(log.err) \
			.addBoth(cb) \
			.addErrback(log.err)

		return self.complete

	def serialize (self):
		serialized = {
			"id": self.id,
			"type": self.__class__.__name__.lower(),
			"state": self.state.value,
			"duration": self.duration
		}

		if self._expr is not None:
			try:
				serialized["expr"] = self._expr.serialize()
			except AttributeError:
				serialized["expr"] = str(self._expr)

		return serialized


class _StepWithChild (Step):

	def __init__ (self, expr = None, step = None):
		Step.__init__(self, expr)

		if isinstance(step, list):
			step = Sequence(step)
		elif not isinstance(step, Step):
			raise Error("Argument must be an instance of Step")

		step.event += self.event
		step.log += self.log

		self._step = step

	def _cancel (self, abort = False):
		d = self.dependents.cancel(abort)

		try:
			d2 = self._step.cancel(abort)
			return defer.gatherResults([d, d2])
		except NotRunning:
			return d

	def _reset (self):
		return defer.gatherResults([
			defer.maybeDeferred(Step._reset, self),
			self._step.reset()
		])

	def _pause (self):
		return defer.gatherResults([
{'fraction_non_alphanumeric': 0.12482488091902494, 'fraction_numerical': 0.0016811431773606051, 'mean_word_length': 3.7340848806366047, 'pattern_counts': {'":': 8, '<': 3, '<?xml version=': 0, '>': 5, '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': '7193376', 'n_tokens_mistral': 2668, 'n_tokens_neox': 2358, 'n_words': 664}