def setup(self):
        keys = self.keys

        while not keys.issubset(self.scheduler.tasks):
            yield gen.sleep(0.05)

        tasks = [self.scheduler.tasks[k] for k in keys]

        self.keys = None

        self.scheduler.add_plugin(self)  # subtle race condition here
        self.all_keys, errors = dependent_keys(tasks, complete=self.complete)
        if not self.complete:
            self.keys = self.all_keys.copy()
        else:
            self.keys, _ = dependent_keys(tasks, complete=False)
        self.all_keys.update(keys)
        self.keys |= errors & self.all_keys

        if not self.keys:
            self.stop(exception=None, key=None)

        # Group keys by func name
        self.keys = valmap(set, groupby(self.func, self.keys))
        self.all_keys = valmap(set, groupby(self.func, self.all_keys))
        for k in self.all_keys:
            if k not in self.keys:
                self.keys[k] = set()

        for k in errors:
            self.transition(k, None, 'erred', exception=True)
        logger.debug("Set up Progress keys")

 def function(scheduler, p):
     result = {'all': valmap(len, p.all_keys),
               'remaining': valmap(len, p.keys),
               'status': p.status}
     if p.status == 'error':
         result.update(p.extra)
     return result

    def _scatter(self, data, workers=None, broadcast=False):
        """ Scatter data to local data dictionary

        Rather than send data out to the cluster we keep data local.  However
        we do report to the scheduler that the local worker has the scattered
        data.  This allows other workers to come by and steal this data if
        desired.

        Keywords like ``broadcast=`` do not work, however operations like
        ``.replicate`` work fine after calling scatter, which can fill in for
        this functionality.
        """
        with log_errors():
            if not (workers is None and broadcast is False):
                raise NotImplementedError("Scatter from worker doesn't support workers or broadcast keywords")

            if isinstance(data, dict) and not all(isinstance(k, (bytes, str))
                                                   for k in data):
                d = yield self._scatter(keymap(tokey, data), workers, broadcast)
                raise gen.Return({k: d[tokey(k)] for k in data})

            if isinstance(data, (list, tuple, set, frozenset)):
                keys = []
                for x in data:
                    try:
                        keys.append(tokenize(x))
                    except:
                        keys.append(str(uuid.uuid1()))
                data2 = dict(zip(keys, data))
            elif isinstance(data, dict):
                keys = set(data)
                data2 = data
            else:
                raise TypeError("Don't know how to scatter %s" % type(data))

            nbytes = valmap(sizeof, data2)

            # self.worker.data.update(data2)  # thread safety matters
            self.worker.loop.add_callback(self.worker.data.update, data2)

            yield self.scheduler.update_data(
                    who_has={key: [self.worker.address] for key in data2},
                    nbytes=valmap(sizeof, data2),
                    client=self.id)

            if isinstance(data, dict):
                out = {k: Future(k, self) for k in data}
            elif isinstance(data, (tuple, list, set, frozenset)):
                out = type(data)([Future(k, self) for k in keys])
            else:
                raise TypeError(
                        "Input to scatter must be a list or dict")

            for key in keys:
                self.futures[key]['status'] = 'finished'
                self.futures[key]['event'].set()

            raise gen.Return(out)

    def setup(self, keys, complete):
        errors = Progress.setup(self, keys, complete)

        # Group keys by func name
        self.keys = valmap(set, groupby(self.func, self.keys))
        self.all_keys = valmap(set, groupby(self.func, self.all_keys))
        for k in self.all_keys:
            if k not in self.keys:
                self.keys[k] = set()

        logger.debug("Set up Progress keys")
        return errors

def expect_kinds(**named):
    """
    Preprocessing decorator that verifies inputs have expected dtype kinds.

    Usage
    -----
    >>> from numpy import int64, int32, float32
    >>> @expect_kinds(x='i')
    ... def foo(x):
    ...    return x
    ...
    >>> foo(int64(2))
    2
    >>> foo(int32(2))
    2
    >>> foo(float32(2))
    Traceback (most recent call last):
       ...n
    TypeError: foo() expected a numpy object of kind 'i' for argument 'x', but got 'f' instead.  # noqa
    """
    for name, kind in iteritems(named):
        if not isinstance(kind, (str, tuple)):
            raise TypeError(
                "expect_dtype_kinds() expected a string or tuple of strings"
                " for argument {name!r}, but got {kind} instead.".format(
                    name=name, kind=dtype,
                )
            )

    @preprocess(kinds=call(lambda x: x if isinstance(x, tuple) else (x,)))
    def _expect_kind(kinds):
        """
        Factory for kind-checking functions that work the @preprocess
        decorator.
        """
        def error_message(func, argname, value):
            # If the bad value has a dtype, but it's wrong, show the dtype
            # kind.  Otherwise just show the value.
            try:
                value_to_show = value.dtype.kind
            except AttributeError:
                value_to_show = value
            return (
                "{funcname}() expected a numpy object of kind {kinds} "
                "for argument {argname!r}, but got {value!r} instead."
            ).format(
                funcname=_qualified_name(func),
                kinds=' or '.join(map(repr, kinds)),
                argname=argname,
                value=value_to_show,
            )

        def _actual_preprocessor(func, argname, argvalue):
            if getattrs(argvalue, ('dtype', 'kind'), object()) not in kinds:
                raise TypeError(error_message(func, argname, argvalue))
            return argvalue

        return _actual_preprocessor

    return preprocess(**valmap(_expect_kind, named))

def merge_ownership_periods(mappings):
    """
    Given a dict of mappings where the values are lists of
    OwnershipPeriod objects, returns a dict with the same structure with
    new OwnershipPeriod objects adjusted so that the periods have no
    gaps.

    Orders the periods chronologically, and pushes forward the end date
    of each period to match the start date of the following period. The
    end date of the last period pushed forward to the max Timestamp.
    """
    return valmap(
        lambda v: tuple(
            OwnershipPeriod(
                a.start,
                b.start,
                a.sid,
                a.value,
            ) for a, b in sliding_window(
                2,
                concatv(
                    sorted(v),
                    # concat with a fake ownership object to make the last
                    # end date be max timestamp
                    [OwnershipPeriod(
                        pd.Timestamp.max.tz_localize('utc'),
                        None,
                        None,
                        None,
                    )],
                ),
            )
        ),
        mappings,
    )

    def __init__(self, constructors, name, *args, **kwargs):
        args = tuple(map(self._unwrap_name, args))
        kwargs = valmap(self._unwrap_name, kwargs)
        already_bound = {}
        for n, arg in enumerate(args):
            if arg in already_bound:
                raise TypeError(
                    'argument %r at position %d is already bound to the'
                    ' positional argument at index %d' % (
                        arg,
                        n,
                        already_bound[arg],
                    ),
                )
            already_bound[arg] = n

        for k, arg in kwargs.items():
            if arg in already_bound:
                loc = already_bound[arg]
                raise TypeError(
                    'argument %r at keyword %s is already bound to the %s' % (
                        arg,
                        k,
                        ('positional argument at index %d' % loc)
                        if isinstance(loc, int) else
                        ('keyword argument %r' % loc),
                    ),
                )

        super().__init__(constructors, name, *args, **kwargs)
        del constructors[name]
        self._constructors = constructors

    def test_retrieve_specific_type(self, type_, lookup_name, failure_type):
        equities = make_simple_equity_info(
            range(5), start_date=pd.Timestamp("2014-01-01"), end_date=pd.Timestamp("2015-01-01")
        )
        max_equity = equities.index.max()
        futures = make_commodity_future_info(first_sid=max_equity + 1, root_symbols=["CL"], years=[2014])
        equity_sids = [0, 1]
        future_sids = [max_equity + 1, max_equity + 2, max_equity + 3]
        if type_ == Equity:
            success_sids = equity_sids
            fail_sids = future_sids
        else:
            fail_sids = equity_sids
            success_sids = future_sids

        with tmp_asset_finder(equities=equities, futures=futures) as finder:
            # Run twice to exercise caching.
            lookup = getattr(finder, lookup_name)
            for _ in range(2):
                results = lookup(success_sids)
                self.assertIsInstance(results, dict)
                self.assertEqual(set(results.keys()), set(success_sids))
                self.assertEqual(valmap(int, results), dict(zip(success_sids, success_sids)))
                self.assertEqual({type_}, {type(asset) for asset in itervalues(results)})
                with self.assertRaises(failure_type):
                    lookup(fail_sids)
                with self.assertRaises(failure_type):
                    # Should fail if **any** of the assets are bad.
                    lookup([success_sids[0], fail_sids[0]])

def assemble_ast(tag:str, idsclasses: Mapping[str, str], attrs: Mapping[str, str], body: list):
  """
  Small helper function for the template_2_ast function that assembles the appropriate ast element
  given the tag name, a dictionary of ids/classes from the tag name, further attrs, and a list of children or the body.
  For most components, there won't be any children.
  
  :param tag:
  :param idsclasses:
  :param attrs:
  :param body:
  :return:
  """
  iscomponent = re.match(r'^[A-Z]', tag)
  attrs['id'] = (attrs.get('id', '') + ' ' + idsclasses.get('id', '')).strip()
  attrs['class'] = (attrs.get('class', '') + ' ' + idsclasses.get('class', '')).strip()
  # remove the empty attributes to avoid clutter and save bytes.
  attrs = dict(t.valfilter(lambda x: not (isinstance(x, str) and x.strip() == ''), attrs))
  # special handling for the "style" attribute, since that can be a dictionary
  attrs = t.valmap(lambda val:' '.join('{}: {};'.format(k,v) for k,v in val.items())
                   if isinstance(val, dict) else val,
                   attrs)
  
  if iscomponent:
    return {'name': tag, 'props': attrs, 'children': body}
  else:
    return {'tag': tag, 'attrs': attrs, 'body': body}

    def symbol_ownership_map(self):
        rows = sa.select(self.equity_symbol_mappings.c).execute().fetchall()

        mappings = {}
        for row in rows:
            mappings.setdefault((row.company_symbol, row.share_class_symbol), []).append(
                SymbolOwnership(
                    pd.Timestamp(row.start_date, unit="ns", tz="utc"),
                    pd.Timestamp(row.end_date, unit="ns", tz="utc"),
                    row.sid,
                    row.symbol,
                )
            )

        return valmap(
            lambda v: tuple(
                SymbolOwnership(a.start, b.start, a.sid, a.symbol)
                for a, b in sliding_window(
                    2,
                    concatv(
                        sorted(v),
                        # concat with a fake ownership object to make the last
                        # end date be max timestamp
                        [SymbolOwnership(pd.Timestamp.max.tz_localize("utc"), None, None, None)],
                    ),
                )
            ),
            mappings,
            factory=lambda: mappings,
        )

 def get_has_what(self, stream, keys=None):
     if keys:
         keys = [coerce_to_address(key) for key in keys]
     if keys is not None:
         return {k: list(self.has_what[k]) for k in keys}
     else:
         return valmap(list, self.has_what)

def expect_dtypes(**named):
    """
    Preprocessing decorator that verifies inputs have expected numpy dtypes.

    Usage
    -----
    >>> from numpy import dtype, arange, int8, float64
    >>> @expect_dtypes(x=dtype(int8))
    ... def foo(x, y):
    ...    return x, y
    ...
    >>> foo(arange(3, dtype=int8), 'foo')
    (array([0, 1, 2], dtype=int8), 'foo')
    >>> foo(arange(3, dtype=float64), 'foo')  # doctest: +NORMALIZE_WHITESPACE
    ...                                       # doctest: +ELLIPSIS
    Traceback (most recent call last):
       ...
    TypeError: ...foo() expected a value with dtype 'int8' for argument 'x',
    but got 'float64' instead.
    """
    for name, type_ in iteritems(named):
        if not isinstance(type_, (dtype, tuple)):
            raise TypeError(
                "expect_dtypes() expected a numpy dtype or tuple of dtypes"
                " for argument {name!r}, but got {dtype} instead.".format(
                    name=name, dtype=dtype,
                )
            )

    @preprocess(dtypes=call(lambda x: x if isinstance(x, tuple) else (x,)))
    def _expect_dtype(dtypes):
        """
        Factory for dtype-checking functions that work with the @preprocess
        decorator.
        """
        def error_message(func, argname, value):
            # If the bad value has a dtype, but it's wrong, show the dtype
            # name.  Otherwise just show the value.
            try:
                value_to_show = value.dtype.name
            except AttributeError:
                value_to_show = value
            return (
                "{funcname}() expected a value with dtype {dtype_str} "
                "for argument {argname!r}, but got {value!r} instead."
            ).format(
                funcname=_qualified_name(func),
                dtype_str=' or '.join(repr(d.name) for d in dtypes),
                argname=argname,
                value=value_to_show,
            )

        def _actual_preprocessor(func, argname, argvalue):
            if getattr(argvalue, 'dtype', object()) not in dtypes:
                raise TypeError(error_message(func, argname, argvalue))
            return argvalue

        return _actual_preprocessor

    return preprocess(**valmap(_expect_dtype, named))

def expect_element(*_pos, **named):
    """
    Preprocessing decorator that verifies inputs are elements of some
    expected collection.

    Usage
    -----
    >>> @expect_element(x=('a', 'b'))
    ... def foo(x):
    ...    return x.upper()
    ...
    >>> foo('a')
    'A'
    >>> foo('b')
    'B'
    >>> foo('c')
    Traceback (most recent call last):
       ...
    ValueError: foo() expected a value in ('a', 'b') for argument 'x', but got 'c' instead.  # noqa

    Notes
    -----
    This uses the `in` operator (__contains__) to make the containment check.
    This allows us to use any custom container as long as the object supports
    the container protocol.
    """
    if _pos:
        raise TypeError("expect_element() only takes keyword arguments.")

    return preprocess(**valmap(_expect_element, named))

 def check_subset(*ss_args, **ss_kwargs):
     """
     【装饰器】
     检查输入参数是否是某一集合的子集；检查失败raise CheckError
     :param ss_args: 参数集合tuple
     :param ss_kwargs: 参数集合dict
     :return: 
     """
     # 检查是否有不合规的tuple参数
     for ss in ss_args:
         if not isinstance(ss, (list, set, type(None))):
             raise TypeError(
                 "check_subset() expected a list or set or None of values"
                 ", but got {subset_} or tuple instead.".format(
                     subset_=str(type(ss)),
                 )
             )
     # 检查是否有不合规的dict参数
     for name, ss in six.iteritems(ss_kwargs):
         if not isinstance(ss, (list, set, type(None))):
             raise TypeError(
                 "check_subset() expected a list or set of values for "
                 "argument '{name_}', but got {subset_} or tuple instead.".format(
                     name_=name, subset_=str(type(ss)),
                 )
             )
     # 将subset_check函数作用在函数参数上
     return arg_process(*map(subset_check, list(ss_args)), **valmap(subset_check, ss_kwargs))

def md(template, *args, **kwargs):
    """Wraps string.format with naive markdown escaping"""
    def escape(s):
        for char in ('*', '#', '_', '~', '`', '>'):
            s = s.replace(char, '\\' + char)
        return s
    return template.format(*map(escape, args), **toolz.valmap(escape, kwargs))

def test_compression_binary(fmt):
    from dask.bytes.core import open_files
    files2 = valmap(compression.compress[fmt], files)
    with filetexts(files2, mode='b'):
        myfiles = open_files('.test.accounts.*', compression=fmt)
        data = compute(*[file.read() for file in myfiles])
        assert list(data) == [files[k] for k in sorted(files)]

def into(a, b, **kwargs):
    dialect = b.dialect.copy()
    del dialect['lineterminator']
    dates = [i for i, typ in enumerate(b.schema[0].types)
               if 'date' in str(typ)]
    schema = b.schema
    if '?' in str(schema):
        schema = dshape(str(schema).replace('?', ''))

    dtypes = valmap(to_numpy_dtype, schema[0].dict)

    datenames = [name for name in dtypes
                      if np.issubdtype(dtypes[name], np.datetime64)]

    dtypes = dict((k, v) for k, v in dtypes.items()
                         if not np.issubdtype(v, np.datetime64))

    if 'strict' in dialect:
        del dialect['strict']

    # Pass only keyword arguments appropriate for read_csv
    kws = keywords(pd.read_csv)
    options = toolz.merge(dialect, kwargs)
    options = toolz.keyfilter(lambda k: k in kws, options)

    if b.open == gzip.open:
        options['compression'] = 'gzip'

    return pd.read_csv(b.path,
                       skiprows=1 if b.header else 0,
                       dtype=dtypes,
                       parse_dates=datenames,
                       names=b.columns,
                       **options)

def test_multiprogress(s, a, b):
    sched, report = Queue(), Queue(); s.handle_queues(sched, report)
    s.update_graph(tasks=valmap(dumps_task, {'x-1': (inc, 1),
                                             'x-2': (inc, 'x-1'),
                                             'x-3': (inc, 'x-2'),
                                             'y-1': (dec, 'x-3'),
                                             'y-2': (dec, 'y-1')}),
                   keys=['y-2'],
                   dependencies={'x-2': ['x-1'], 'x-3': ['x-2'],
                                 'y-1': ['x-3'], 'y-2': ['y-1']})

    p = MultiProgress(['y-2'], scheduler=s, func=key_split)
    yield p.setup()

    assert p.keys == {'x': {'x-1', 'x-2', 'x-3'},
                      'y': {'y-1', 'y-2'}}

    while True:
        msg = yield report.get()
        if msg['op'] == 'key-in-memory' and msg['key'] == 'x-3':
            break

    assert p.keys == {'x': set(),
                      'y': {'y-1', 'y-2'}}

    while True:
        msg = yield report.get()
        if msg['op'] == 'key-in-memory' and msg['key'] == 'y-2':
            break

    assert p.keys == {'x': set(),
                      'y': set()}

    assert p.status == 'finished'

 def __init__(self,
              data=None,
              formats=None,
              authorization=None,
              allow_profiler=False,
              profiler_output=None,
              profile_by_default=False,
              allow_add=False):
     if isinstance(data, collections.Mapping):
         data = valmap(lambda v: v.data if isinstance(v, _Data) else v,
                       data)
     elif isinstance(data, _Data):
         data = data._resources()
     app = self.app = Flask('blaze.server.server')
     if data is None:
         data = {}
     app.register_blueprint(api,
                            data=data,
                            formats=formats if formats is not None else (json,),
                            authorization=authorization,
                            allow_profiler=allow_profiler,
                            profiler_output=profiler_output,
                            profile_by_default=profile_by_default,
                            allow_add=allow_add)
     self.data = data

def test_multi_progressbar_widget(s, a, b):
    s.update_graph(tasks=valmap(dumps_task, {'x-1': (inc, 1),
                                             'x-2': (inc, 'x-1'),
                                             'x-3': (inc, 'x-2'),
                                             'y-1': (dec, 'x-3'),
                                             'y-2': (dec, 'y-1'),
                                             'e': (throws, 'y-2'),
                                             'other': (inc, 123)}),
                   keys=['e'],
                   dependencies={'x-2': ['x-1'], 'x-3': ['x-2'],
                                 'y-1': ['x-3'], 'y-2': ['y-1'],
                                 'e': ['y-2']})

    p = MultiProgressWidget(['e'], scheduler=(s.ip, s.port))
    yield p.listen()

    assert p.bars['x'].value == 1.0
    assert p.bars['y'].value == 1.0
    assert p.bars['e'].value == 0.0
    assert '3 / 3' in p.bar_texts['x'].value
    assert '2 / 2' in p.bar_texts['y'].value
    assert '0 / 1' in p.bar_texts['e'].value

    assert p.bars['x'].bar_style == 'success'
    assert p.bars['y'].bar_style == 'success'
    # assert p.bars['e'].bar_style == 'danger'

    assert p.status == 'error'

    capacities = [int(re.search(r'\d+ / \d+', row.children[0].value)
                    .group().split(' / ')[1])
                  for row in p.bar_widgets.children]
    assert sorted(capacities, reverse=True) == capacities