# Path: loom/util.py
# def cp_ns(source, destin):
#     'like cp -ns, link destin to source if destin does not exist'
#     if not os.path.exists(destin):
#         assert os.path.exists(source), source
#         dirname = os.path.dirname(destin)
#         if dirname:
#             mkdir_p(dirname)
#         try:
#             os.symlink(source, destin)
#         except OSError as e:
#             if not os.path.exists(destin):
#                 raise e
# 
# Path: loom/util.py
# def LOG(message):
#     if VERBOSITY:
#         sys.stdout.write('{}\n'.format(message))
#         sys.stdout.flush()
# 
# Path: loom/util.py
# def parallel_map(fun, args):
#     if not isinstance(args, list):
#         args = list(args)
#     is_daemon = multiprocessing.current_process().daemon
#     if THREADS == 1 or len(args) < 2 or is_daemon:
#         print 'Running {} in this thread'.format(fun.__name__)
#         return map(fun, args)
#     else:
#         print 'Running {} in {:d} threads'.format(fun.__name__, THREADS)
#         pool = multiprocessing.Pool(THREADS)
#         fun_args = [(fun, arg) for arg in args]
#         return pool.map(print_trace, fun_args, chunksize=1)
# 
# Path: loom/util.py
# def pickle_dump(data, filename):
#     with open_compressed(filename, 'wb') as f:
#         pickle.dump(data, f)
# 
# Path: loom/util.py
# def pickle_load(filename):
#     with open_compressed(filename, 'rb') as f:
#         return pickle.load(f)
# 
# Path: loom/format.py
# def load_encoder(encoder):
#     model = encoder['model']
#     if 'symbols' in encoder:
#         encode = encoder['symbols'].__getitem__
#     elif model == 'bb':
#         encode = BOOLEAN_SYMBOLS.__getitem__
#     else:
#         encode = loom.schema.MODELS[model].Value
#     return encode
# 
# Path: loom/format.py
# def load_decoder(encoder):
#     model = encoder['model']
#     if 'symbols' in encoder:
#         decoder = {value: key for key, value in encoder['symbols'].iteritems()}
#         decode = decoder.__getitem__
#     elif model == 'bb':
#         decode = ('0', '1').__getitem__
#     else:
#         decode = str
#     return decode


import os
import re
import datetime
import dateutil.parser
import loom.util
import loom.documented
import parsable
from itertools import izip
from collections import Counter
from contextlib2 import ExitStack
from distributions.io.stream import json_dump
from distributions.io.stream import json_load
from loom.util import cp_ns
from loom.util import LOG
from loom.util import parallel_map
from loom.util import pickle_dump
from loom.util import pickle_load
from loom.format import load_encoder
from loom.format import load_decoder
            row = transform.forward_row(header, transformed_header, row)
            writer.writerow(row)


@loom.documented.transform(
    inputs=['ingest.schema', 'ingest.transforms', 'rows_csv'],
    outputs=['ingest.rows_csv'])
@parsable.command
def transform_rows(schema_in, transforms_in, rows_in, rows_out, id_field=None):
    transforms = pickle_load(transforms_in)
    if not transforms:
        cp_ns(rows_in, rows_out)
    else:
        transform = TransformSequence(transforms)
        transformed_header = sorted(json_load(schema_in).iterkeys())
        if id_field is not None:
            assert id_field not in transformed_header
            transformed_header = [id_field] + transformed_header
        tasks = []
        if os.path.isdir(rows_in):
            loom.util.mkdir_p(rows_out)
            for f in os.listdir(rows_in):
                tasks.append((
                    transform,
                    transformed_header,
                    os.path.join(rows_in, f),
                    os.path.join(rows_out, f),
                ))
        else:
            tasks.append((transform, transformed_header, rows_in, rows_out))

# Path: loom/util.py
# def cp_ns(source, destin):
#     'like cp -ns, link destin to source if destin does not exist'
#     if not os.path.exists(destin):
#         assert os.path.exists(source), source
#         dirname = os.path.dirname(destin)
#         if dirname:
#             mkdir_p(dirname)
#         try:
#             os.symlink(source, destin)
#         except OSError as e:
#             if not os.path.exists(destin):
#                 raise e
# 
# Path: loom/util.py
# def LOG(message):
#     if VERBOSITY:
#         sys.stdout.write('{}\n'.format(message))
#         sys.stdout.flush()
# 
# Path: loom/util.py
# def parallel_map(fun, args):
#     if not isinstance(args, list):
#         args = list(args)
#     is_daemon = multiprocessing.current_process().daemon
#     if THREADS == 1 or len(args) < 2 or is_daemon:
#         print 'Running {} in this thread'.format(fun.__name__)
#         return map(fun, args)
#     else:
#         print 'Running {} in {:d} threads'.format(fun.__name__, THREADS)
#         pool = multiprocessing.Pool(THREADS)
#         fun_args = [(fun, arg) for arg in args]
#         return pool.map(print_trace, fun_args, chunksize=1)
# 
# Path: loom/util.py
# def pickle_dump(data, filename):
#     with open_compressed(filename, 'wb') as f:
#         pickle.dump(data, f)
# 
# Path: loom/util.py
# def pickle_load(filename):
#     with open_compressed(filename, 'rb') as f:
#         return pickle.load(f)
# 
# Path: loom/format.py
# def load_encoder(encoder):
#     model = encoder['model']
#     if 'symbols' in encoder:
#         encode = encoder['symbols'].__getitem__
#     elif model == 'bb':
#         encode = BOOLEAN_SYMBOLS.__getitem__
#     else:
#         encode = loom.schema.MODELS[model].Value
#     return encode
# 
# Path: loom/format.py
# def load_decoder(encoder):
#     model = encoder['model']
#     if 'symbols' in encoder:
#         decoder = {value: key for key, value in encoder['symbols'].iteritems()}
#         decode = decoder.__getitem__
#     elif model == 'bb':
#         decode = ('0', '1').__getitem__
#     else:
#         decode = str
#     return decode


# Path: loom/transforms.py
import os
import re
import datetime
import dateutil.parser
import loom.util
import loom.documented
import parsable
from itertools import izip
from collections import Counter
from contextlib2 import ExitStack
from distributions.io.stream import json_dump
from distributions.io.stream import json_load
from loom.util import cp_ns
from loom.util import LOG
from loom.util import parallel_map
from loom.util import pickle_dump
from loom.util import pickle_load
from loom.format import load_encoder
from loom.format import load_decoder

            row = transform.forward_row(header, transformed_header, row)
            writer.writerow(row)


@loom.documented.transform(
    inputs=['ingest.schema', 'ingest.transforms', 'rows_csv'],
    outputs=['ingest.rows_csv'])
@parsable.command
def transform_rows(schema_in, transforms_in, rows_in, rows_out, id_field=None):
    transforms = pickle_load(transforms_in)
    if not transforms:
        cp_ns(rows_in, rows_out)
    else:
        transform = TransformSequence(transforms)
        transformed_header = sorted(json_load(schema_in).iterkeys())
        if id_field is not None:
            assert id_field not in transformed_header
            transformed_header = [id_field] + transformed_header
        tasks = []
        if os.path.isdir(rows_in):
            loom.util.mkdir_p(rows_out)
            for f in os.listdir(rows_in):
                tasks.append((
                    transform,
                    transformed_header,
                    os.path.join(rows_in, f),
                    os.path.join(rows_out, f),
                ))
        else:
            tasks.append((transform, transformed_header, rows_in, rows_out))

        parallel_map(_transform_rows, tasks)