def __init__(self, data, processors=None, dialect=None, format='csv',
                 transform=True, encoding=None, decode_strategy='replace',
                 options=None, fail_fast=False, row_limit=20000,
                 report_limit=1000, report_stream=None, header_index=0,
                 break_on_invalid_processor=True, post_task=None,
                 report_type='basic'):

        if data is None:
            _msg = '`data` must be a filepath, url or stream.'
            raise exceptions.PipelineBuildError(_msg)

        self.openfiles = []
        self.data_source = data
        self.processors = processors or helpers.DEFAULT_PIPELINE
        self.dialect = self.get_dialect(dialect)
        self.format = format
        self.encoding = encoding
        self.decode_strategy = decode_strategy
        self.options = options or {}
        self.transform = transform
        self.fail_fast = fail_fast
        self.row_limit = self.get_row_limit(row_limit)
        self.report_limit = self.get_report_limit(report_limit)
        self.report_stream = report_stream
        self.header_index = header_index
        self.break_on_invalid_processor = break_on_invalid_processor

        helpers.validate_handler(post_task)

        if report_type == 'grouped':
            self.report_post_task = helpers.grouped_report
        else:
            self.report_post_task = helpers.basic_report

        self.post_task = post_task

        if self.report_stream:
            report_stream_tests = [isinstance(self.report_stream, io.TextIOBase),
                                   self.report_stream.writable(),
                                   self.report_stream.seekable()]

            if not all(report_stream_tests):
                _msg = '`report_stream` must be a seekable and writable text stream.'
                raise exceptions.PipelineBuildError(_msg)

            report_backend = 'client'
        else:
            report_backend = 'yaml'

        report_options = {
            'schema': helpers.report_schema,
            'backend': report_backend,
            'client_stream': self.report_stream,
            'limit': self.report_limit,
            'post_task': self.report_post_task
        }

        self.report = tellme.Report('Pipeline', **report_options)

        self.pipeline = self.get_pipeline()

        try:
            self.data = datatable.DataTable(self.data_source, format=self.format,
                                            encoding=encoding,
                                            decode_strategy=decode_strategy,
                                            header_index=self.header_index)
            self.openfiles.extend(self.data.openfiles)
            
        except datatable.DataTable.RAISES:
            self.data = self.data_source
dataset/ETHPy150Open frictionlessdata/goodtables/goodtables/pipeline/pipeline.py/Pipeline.__init__/original