<commit_before># Copyright (c) 2014, <NAME>
# All rights reserved.

"""lazy dict-based i/o processing pipelines"""


import sys, types, functools


#--- setup logging

import logging
#(logging.NullHandler was introduced in 2.7, and this code is 2.6 compatible)
class NullHandler(logging.Handler):
    def emit(self, record):
        pass
logging.getLogger('dio').addHandler(NullHandler())


#--- decorators for the basic roles

def processor(f):
	"""Turn a processing function into a pipline participator.

	This serves two purposes:

	1) This assigns out and err, which all processors must expose as keyword
	   arguments, to the global default instances if no other value has been
	   given.  This accomplishes later binding than if they were ordinary
	   defaults in the original function definitions.

	2) Most processors are coroutines (extended generators) that get sent
	   input.  If f is a coroutine, this primes the coroutine so that it runs
	   until its first yield and is ready to get sent real input; this returns
	   the coroutine (generator object) for further use.

	   This is irrelevant for sources -- plain functions that don't get sent
	   any input.  In that case, calling the decorated function calls the
	   original function, which fully executes it, and there is nothing left to
	   use (this returns None).
	"""
	@functools.wraps(f)  #(for __doc__)
	def primed_f(*args,**kwargs):
		#set out and err to default values if not specified; default_out/err
		#won't be available for the standard sink processors, since they are
		#used to set default_out/err
		if 'default_out' in globals() and 'default_err' in globals():
			for k, v in (('out',default_out), ('err',default_err)):
				if k not in kwargs or kwargs[k] is None:
					kwargs[k] = v

		f2 = f(*args,**kwargs)

		#coroutine priming
		if isinstance(f2, types.GeneratorType):
			f2.send(None)
			return f2

	return primed_f

def restart_on_error(f):
	"""Decorate a processing function so that it's restarted upon any errors.

	Ideally, it would *continue* on error, but you can't resume a generator
	after it's raised an exception.  Therefore this restarts a new instance of
	the generator.

	This @restart_on_error decorator should be applied *before* (i.e. on a
	lower line) than the @processor decorator.
	"""
	def restarter(*args, **kwargs):
		while True:
			try:
				f2 = f(*args, **kwargs)
				f2.next()
				while True:
					d = yield
					f2.send(d)
			except Exception, e:
				#(GeneratorExit is not an Exception, just BaseException)
				kwargs['err'].send(e2d(e))
	return restarter


#--- exception handling

def e2d(e):
	"""Convert an Exception to a LazyDict.

	#FIXME this needs a lot of work -- it's really just a placeholder for now.
	"""
	return {'message': e.message}


#--- sources

@processor
def source(iterable, out=None, err=None):
	"""Turn any iterable into a source to start a processing pipeline."""
	for d in iterable:
		out.send(d)


#--- serialization

def pre_serialize(d):
	if type(d) != dict:
		d['__class__'] = '.'.join((d.__class__.__module__, d.__class__.__name__))
	return d

def post_deserialize(d):
	#LBYL since optimizing for built-in dicts
	if d.has_key('__class__'):
		m, c = d['__class__'].rsplit('.',1)  #assuming all subclasses are in modules
		m = __import__(m)
		d = getattr(m, c)(d)
	return d


#--- standard i/o

#though these have the form of standard processors, out and err should be
#file-like objects, not other processors.

#repr/eval (file-like)
import ast
@processor
def repr_in(inn=None, out=None, err=None):
	for line in inn:
		out.send(post_deserialize(ast.literal_eval(line)))<commit_msg>allow a package hierarchy in dynamically imported classes<commit_after># Copyright (c) 2014, <NAME>
# All rights reserved.

"""lazy dict-based i/o processing pipelines"""


import sys, types, functools


#--- setup logging

import logging
#(logging.NullHandler was introduced in 2.7, and this code is 2.6 compatible)
class NullHandler(logging.Handler):
    def emit(self, record):
        pass
logging.getLogger('dio').addHandler(NullHandler())


#--- decorators for the basic roles

def processor(f):
	"""Turn a processing function into a pipline participator.

	This serves two purposes:

	1) This assigns out and err, which all processors must expose as keyword
	   arguments, to the global default instances if no other value has been
	   given.  This accomplishes later binding than if they were ordinary
	   defaults in the original function definitions.

	2) Most processors are coroutines (extended generators) that get sent
	   input.  If f is a coroutine, this primes the coroutine so that it runs
	   until its first yield and is ready to get sent real input; this returns
	   the coroutine (generator object) for further use.

	   This is irrelevant for sources -- plain functions that don't get sent
	   any input.  In that case, calling the decorated function calls the
	   original function, which fully executes it, and there is nothing left to
	   use (this returns None).
	"""
	@functools.wraps(f)  #(for __doc__)
	def primed_f(*args,**kwargs):
		#set out and err to default values if not specified; default_out/err
		#won't be available for the standard sink processors, since they are
		#used to set default_out/err
		if 'default_out' in globals() and 'default_err' in globals():
			for k, v in (('out',default_out), ('err',default_err)):
				if k not in kwargs or kwargs[k] is None:
					kwargs[k] = v

		f2 = f(*args,**kwargs)

		#coroutine priming
		if isinstance(f2, types.GeneratorType):
			f2.send(None)
			return f2

	return primed_f

def restart_on_error(f):
	"""Decorate a processing function so that it's restarted upon any errors.

	Ideally, it would *continue* on error, but you can't resume a generator
	after it's raised an exception.  Therefore this restarts a new instance of
	the generator.

	This @restart_on_error decorator should be applied *before* (i.e. on a
	lower line) than the @processor decorator.
	"""
	def restarter(*args, **kwargs):
		while True:
			try:
				f2 = f(*args, **kwargs)
				f2.next()
				while True:
					d = yield
					f2.send(d)
			except Exception, e:
				#(GeneratorExit is not an Exception, just BaseException)
				kwargs['err'].send(e2d(e))
	return restarter


#--- exception handling

def e2d(e):
	"""Convert an Exception to a LazyDict.

	#FIXME this needs a lot of work -- it's really just a placeholder for now.
	"""
	return {'message': e.message}


#--- sources

@processor
def source(iterable, out=None, err=None):
	"""Turn any iterable into a source to start a processing pipeline."""
	for d in iterable:
		out.send(d)


#--- serialization

def pre_serialize(d):
	if type(d) != dict:
		d['__class__'] = '.'.join((d.__class__.__module__, d.__class__.__name__))
	return d

def post_deserialize(d):
	#LBYL since optimizing for built-in dicts
	if d.has_key('__class__'):
		m, c = d['__class__'].rsplit('.',1)  #assuming all subclasses are in modules
		m = __import__(m, fromlist=[None])  #any non-empty fromlist allows importing from a package hierarchy
		d = getattr(m, c)(d)
	return d


#--- standard i/o

#though these have the form of standard processors, out and err should be
#file-like objects, not other processors.

#repr/eval (file-like)
import ast
@processor
def repr_in(inn=None, out=None, err=None):
	for line in inn:
		out.send(post_deserialize(ast.literal_eval(line)))
{'fraction_non_alphanumeric': 0.1010840108401084, 'fraction_numerical': 0.005420054200542005, 'mean_word_length': 3.855921052631579, 'pattern_counts': {'":': 0, '<': 5, '<?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': '4284135', 'n_tokens_mistral': 2463, 'n_tokens_neox': 2252, 'n_words': 922}