Method _cull_dependencies

dask/blockwise.py:657–711

Determine the necessary dependencies to produce `output_blocks`. This method does not require graph materialization.

(self, output_blocks)

        return hasattr(self, "_cached_dict")

    def _cull_dependencies(self, output_blocks):
        """Determine the necessary dependencies to produce `output_blocks`.

        This method does not require graph materialization.
        """

        # Check `concatenate` option
        concatenate = None
        if self.concatenate is True:
            from dask.array.core import concatenate_axes as concatenate

        # Generate coordinate map
        coord_maps, concat_axes, dummies = _get_coord_mapping(
            self.dims,
            self.output_indices,
            self.numblocks,
            self.indices,
            concatenate,
        )

        # Gather constant dependencies (for all output keys)
        const_deps = set()
        for arg, _ in self.indices:
            if isinstance(arg, TaskRef):
                const_deps.add(arg.key)

        # Get dependencies for each output block
        key_deps = {}
        for out_coords in output_blocks:
            deps = set()
            coords = out_coords + dummies
            for cmap, axes, (arg, ind) in zip(coord_maps, concat_axes, self.indices):
                if ind is not None and arg not in self.io_deps:
                    arg_coords = tuple(coords[c] for c in cmap)
                    if axes:
                        tups = _lol_product((arg,), arg_coords)
                        deps.update(flatten(tups))
                        if concatenate:
                            tups = (concatenate, tups, axes)
                    else:
                        tups = (arg,) + arg_coords
                        deps.add(tups)
            key_deps[(self.output,) + out_coords] = deps | const_deps

        # Add valid-key dependencies from io_deps
        for key, io_dep in self.io_deps.items():
            if io_dep.produces_keys:
                for out_coords in output_blocks:
                    key = (self.output,) + out_coords
                    valid_key_dep = io_dep[out_coords]
                    if isinstance(valid_key_dep, TaskRef):
                        valid_key_dep = valid_key_dep.key
                    key_deps[key] |= {valid_key_dep}

        return key_deps

    def _cull(self, output_blocks):
        return Blockwise(