o Nrf@sxddlmZddlZddlZddlZddlmZddlmZddl m Z ddl m Z ddl ZddlmZddlmZdd lmZmZmZmZdd lmZdd lmZdd lmZmZmZm Z Gd ddZ!GdddeZ"Gddde"Z#Gddde"Z$GdddeZ%d'ddZ&ddZ'ddZ(GdddeZ)Gdd d e)Z*Gd!d"d"eZ+Gd#d$d$eZ,Gd%d&d&eZ-dS)( annotationsN) defaultdict)Callable)product)Any)map)tokenize) Blockwise BlockwiseDepBlockwiseDepDictblockwise_token)flatten)Layer)apply cached_cumsumconcreteinsertc@s eZdZdZddZddZdS)CallableLazyImportzFunction Wrapper for Lazy Importing. This Class should only be used when materializing a graph on a distributed scheduler. cCs ||_dSN) function_path)selfrrT/var/www/html/software/conda/envs/catlas/lib/python3.10/site-packages/dask/layers.py__init__" zCallableLazyImport.__init__cOs ddlm}||j|i|S)Nr) import_term)Zdistributed.utilsrr)rargskwargsrrrr__call__%s zCallableLazyImport.__call__N)__name__ __module__ __qualname____doc__rrrrrrrs rc@sBeZdZUdZded<ded<dZded<dd d Zdd d ZdS)ArrayBlockwiseDepzg Blockwise dep for array-likes, which only needs chunking information to compute its data. tuple[tuple[int, ...], ...]chunkstuple[int, ...] numblocksFboolproduces_taskscCs$||_tdd|D|_d|_dS)Ncss|]}t|VqdSr)len).0chunkrrr >sz-ArrayBlockwiseDep.__init__..F)r&tupler(r*rr&rrrr<s zArrayBlockwiseDep.__init__idxcCstd)Nz%Subclasses must implement __getitem__)NotImplementedErrorrr1rrr __getitem__AzArrayBlockwiseDep.__getitem__Nr&r%r1r')r r!r"r#__annotations__r*rr4rrrrr$2s   r$c@seZdZdZdddZdS)ArrayChunkShapeDepz(Produce chunk shapes given a chunk indexr1r'cCstddt||jDS)Ncss|] \}}||VqdSrr)r,ir-rrrr.Iz1ArrayChunkShapeDep.__getitem__..)r/zipr&r3rrrr4HszArrayChunkShapeDep.__getitem__Nr7)r r!r"r#r4rrrrr9Esr9cs6eZdZUdZded<d fdd Zd d d ZZS) ArraySliceDepz>Produce slice(s) into the full-sized array given a chunk indexr%startsr&cs$t|tdd|D|_dS)Ncss|] }t|ddVqdS)T)Z initial_zeroN)r)r,crrrr.Sr;z)ArraySliceDep.__init__..)superrr/r>r0 __class__rrrQs zArraySliceDep.__init__r1r/cCs,tddt||jD}tdd|DS)Ncss(|]\}}||||dfVqdSNr)r,r:startrrrr.Vs&z,ArraySliceDep.__getitem__..css"|] }tg|dRVqdSrslicer,srrrr.Ws )r/r<r>)rr1locrrrr4UszArraySliceDep.__getitem__r6)r1r/)r r!r"r#r8rr4 __classcell__rrrArr=Ls r=csneZdZdZfddZddZeddZdd Zd d Z d d Z ddZ ddZ ddZ dddZZS)ArrayOverlapLayera\Simple HighLevelGraph array overlap layer. Lazily computed High-level graph layer for a array overlap operations. Parameters ---------- name : str Name of new output overlap array. array : Dask array axes: Mapping Axes dictionary indicating overlap in each dimension, e.g. ``{'0': 1, '1': 1}`` cs2t||_||_||_||_||_d|_dSr)r@rnameaxesr&r(token _cached_keys)rrMrNr&r(rOrArrris  zArrayOverlapLayer.__init__cCsd|jdS)NzArrayOverlapLayersz>ArrayOverlapLayer._dask_keys..keys..csg|] }|fqSrrro)rrmrrrps)r+range)rindresultr&rmrMr()rrrmsz*ArrayOverlapLayer._dask_keys..keys)rPrMr&r()rrsrrtr _dask_keyss  zArrayOverlapLayer._dask_keysFcCs |j}|j}|j}|}d|j}d|j}|rtd}nddlm}tt t |} t j t | |d} t|tt | t ttjt} i} i} | D]4}t|f||}|f||kra|| |f|<qH|f|| |f|<|t| d||dff| |f|<qHt| | }|S) z/Construct graph for a simple overlap operation.zgetitem-zoverlap-zdask.array.core.concatenate3r) concatenate3)dimsrNrrR)rNr&rMrurOrZdask.array.corervlistrr+ functoolspartial_expand_keys_around_centertoolzpiperconcatfractional_slicermerge)r deserializingrNr&rMZ dask_keysZ getitem_nameZ overlap_namervrwZ expand_key2Z interior_keysZinterior_slicesZoverlap_blockskZ frac_slicer\rrrrZs8     z"ArrayOverlapLayer._construct_graphF)r r!r"r#rrTpropertyr]r4rfrhrlrnrurZrKrrrArrLZs   rLc sfddg}t|ddD]\}}d}|dkr|d7}||dkr*|d7}||qfddt|ddD}|durJ|gg|}tt|} fddt|D} t| | } | S) aGet all neighboring keys around center Parameters ---------- k: Key The key around which to generate new keys dims: Sequence[int] The number of chunks in each dimension name: Option[str] The name to include in the output keys, or none to include no name axes: Dict[int, int] The axes active in the expansion. We don't expand on non-active axes Examples -------- >>> _expand_keys_around_center(('x', 2, 3), dims=[5, 5], name='y', axes={0: 1, 1: 1}) # noqa: E501 # doctest: +NORMALIZE_WHITESPACE [[('y', 1.1, 2.1), ('y', 1.1, 3), ('y', 1.1, 3.9)], [('y', 2, 2.1), ('y', 2, 3), ('y', 2, 3.9)], [('y', 2.9, 2.1), ('y', 2.9, 3), ('y', 2.9, 3.9)]] >>> _expand_keys_around_center(('x', 0, 4), dims=[5, 5], name='y', axes={0: 1, 1: 1}) # noqa: E501 # doctest: +NORMALIZE_WHITESPACE [[('y', 0, 3.1), ('y', 0, 4)], [('y', 0.9, 3.1), ('y', 0.9, 4)]] csNg}|ddkr||d|||d|dkr%||d|S)Ng?rrD)append)r:rrrv)rwrrindss  z(_expand_keys_around_center..indsrDNrcs2g|]\}}t|dfr||n|gqS)ranyget)r,r:rr)rNrrrrps&z._expand_keys_around_center..cs*g|]\}}t|dfr|ndqS)rrDr)r,r:d)rNrrrps*) enumeraterrxr reshapelist) rrwrMrNshaper:rrnumrseqZshape2rsr)rNrwrrr{s$       r{csBtdkr t|Stt|d}fddt||DS)zgReshape iterator to nested shape >>> reshapelist((2, 3), range(6)) [[0, 1, 2], [3, 4, 5]] rDrcsg|] }tdd|qSrC)rr,partrrrrpszreshapelist..)r+rxintr| partition)rrnrrrr s rc Cs|dftdd|ddD}g}tt|dd|ddD]S\}\}}||d}t|tr=|d}|d} n|}|} ||krO|tdddq#||kr^| r^|td| q#||krn|rn|t| dq#|tddq#t|}tdd|Dr|Stj ||fS)a >>> fractional_slice(('x', 5.1), {0: 2}) (, ('x', 5), (slice(-2, None, None),)) >>> fractional_slice(('x', 3, 5.1), {0: 2, 1: 3}) (, ('x', 3, 5), (slice(None, None, None), slice(-3, None, None))) >>> fractional_slice(('x', 2.9, 5.1), {0: 2, 1: 3}) (, ('x', 3, 5), (slice(0, 2, None), slice(-3, None, None))) rcss|] }tt|VqdSr)rroundrorrrr.$r;z#fractional_slice..rDNcss |] }|tdddkVqdSrrF)r,rrrrrr.:s) r/rr<r isinstancerrGalloperatorgetitem) taskrNZroundedindexr:trdepthZ left_depthZ right_depthrrrrs($ *      rcseZdZdZ  d fdd ZddZddZd d Zd d Ze d dZ ddZ ddZ ddZ ddZd!ddZddZddZd"ddZZS)#SimpleShuffleLayeraSimple HighLevelGraph Shuffle layer High-level graph layer for a simple shuffle operation in which each output partition depends on all input partitions. Parameters ---------- name : str Name of new shuffled output collection. column : str or list of str Column(s) to be used to map rows to output partitions (by hashing). npartitions : int Number of output partitions. npartitions_input : int Number of partitions in the original (un-shuffled) DataFrame. ignore_index: bool, default False Ignore index during shuffle. If ``True``, performance may improve, but index values will not be preserved. name_input : str Name of input collection. meta_input : pd.DataFrame-like object Empty metadata of input collection. parts_out : list of int (optional) List of required output-partition indices. annotations : dict (optional) Layer annotations Nc sv||_||_||_||_||_||_||_|pt||_d|j|_ | p%i} d|_ d| vr2|j | d<t j | ddS)Nsplit-priorityr)rMcolumn npartitionsnpartitions_input ignore_index name_input meta_inputrq parts_out split_nameZ _split_keys _key_priorityr@r) rrMrrrrrrrrrArrrds   zSimpleShuffleLayer.__init__cCs$t|tsJ|d|jkrdSdS)NrrD)rr/rrarrrrsz SimpleShuffleLayer._key_prioritycfddjDS)Nch|]}j|fqSrrRrrSrr z5SimpleShuffleLayer.get_output_keys..rrSrrSrrnz"SimpleShuffleLayer.get_output_keyscCsd|j|jS)Nz-SimpleShuffleLayer)formatrMrrSrrrrTszSimpleShuffleLayer.__repr__cCrirjrkrSrrrrlrz"SimpleShuffleLayer.is_materializedcCrUrVrXr[rrrr]r^zSimpleShuffleLayer._dictcCr_rr`rarrrr4rzSimpleShuffleLayer.__getitem__cCrcrrdrSrrrrfrzSimpleShuffleLayer.__iter__cCrcrrgrSrrrrhrzSimpleShuffleLayer.__len__c CJt}|D]}z|\}}Wn tyYqw||jkrq||q|Sz4Simple utility to convert keys to partition indices.set ValueErrorrMaddrrmpartsrb_name_partrrr_keys_to_parts    z!SimpleShuffleLayer._keys_to_partscsNtt}|p |}|D]}|j|ffddtjDO<q |S)zDetermine the necessary dependencies to produce `keys`. For a simple shuffle, output partitions always depend on all input partitions. This method does not require graph materialization. crrrrorSrrr z8SimpleShuffleLayer._cull_dependencies..)rrrrMrqr)rrmrdepsrrrSr_cull_dependenciess z%SimpleShuffleLayer._cull_dependenciesc Cs&t|j|j|j|j|j|j|j|dSNr)rrMrrrrrrrrrrr_cullszSimpleShuffleLayer._cullcC@||}|j||d}|t|jkr||}||fS||fS)aCull a SimpleShuffleLayer HighLevelGraph layer. The underlying graph will only include the necessary tasks to produce the keys (indices) included in `parts_out`. Therefore, "culling" the layer only requires us to reset this parameter. rrrrrrrrmall_keysrZ culled_deps culled_layerrrrcull  zSimpleShuffleLayer.cullFc sdj}|rtd}td}n ddlm}ddlm}i}jD]LfddtjD}||j f|jf<|D]/\}}} t j || f|f|j || f<|| f|vrl|j | fjdjjj jf||| f<q=q!|S) z/Construct graph for a simple shuffle operation.group-dask.dataframe.core._concat$dask.dataframe.shuffle.shuffle_groupr_concat shuffle_groupcsg|]}j|fqSr)r)r,Zpart_inZpart_outrrrrps z7SimpleShuffleLayer._construct_graph..)rMrdask.dataframe.corerdask.dataframe.shufflerrrqrrrrrrrr) rrshuffle_group_name concat_funcshuffle_group_funcr\ _concat_list_Z _part_outZ_part_inrrrrZsF       z#SimpleShuffleLayer._construct_graphNNrr)r r!r"r#rrrnrTrlrr]r4rfrhrrrrrZrKrrrArrGs&%*    rcsJeZdZdZ  dfdd ZddZdddZd d Zdd d ZZ S) ShuffleLayera"Shuffle-stage HighLevelGraph layer High-level graph layer corresponding to a single stage of a multi-stage inter-partition shuffle operation. Stage: (shuffle-group) -> (shuffle-split) -> (shuffle-join) Parameters ---------- name : str Name of new (partially) shuffled collection. column : str or list of str Column(s) to be used to map rows to output partitions (by hashing). inputs : list of tuples Each tuple dictates the data movement for a specific partition. stage : int Index of the current shuffle stage. npartitions : int Number of output partitions for the full (multi-stage) shuffle. npartitions_input : int Number of partitions in the original (un-shuffled) DataFrame. k : int A partition is split into this many groups during each stage. ignore_index: bool, default False Ignore index during shuffle. If ``True``, performance may improve, but index values will not be preserved. name_input : str Name of input collection. meta_input : pd.DataFrame-like object Empty metadata of input collection. parts_out : list of int (optional) List of required output-partition indices. annotations : dict (optional) Layer annotations Nc s@||_||_||_tj|||||| | | ptt|| d dS)N)rr)inputsstagensplitsr@rrqr+) rrMrrrrrrrrrrrrArrrAs zShuffleLayer.__init__cCd|j|j|j|jS)Nz=ShuffleLayer)rrMrrrrSrrrrT_zShuffleLayer.__repr__c Cstt}|p ||}ddt|jD}|D]B}|j|}t|jD]5}t||j|}||} |jdkrK| |j krK||j |f d|j |dfq#||j |f |j | fq#q|S)zqDetermine the necessary dependencies to produce `keys`. Does not require graph materialization. cSi|]\}}||qSrrr,r:inprrr krz3ShuffleLayer._cull_dependencies..rrempty) rrrrrrqrrrrrMrr) rrmrr inp_part_maproutr_inprrrrrds "zShuffleLayer._cull_dependenciesc Cs2t|j|j|j|j|j|j|j|j|j |j |d Sr) rrMrrrrrrrrrrrrrrwszShuffleLayer._cullFc Csfd|j}|rtd}td}n ddlm}ddlm}i}ddt|jD}|jD]}|j|}g} t |j D]} t ||j | } ||j } | |j| | fq9|| |jf||j|f<| D]Q\} } } tj|| f| f||j| | f<|| f|vr|| }|j dkr||jkr|j|f}n|| d f}|j||<n|j|f}|||j|j |j |j|j|jf||| f<q^q+|S) z2Construct graph for a "rearrange-by-column" stage.rrrrrrcSrrrrrrrrrz1ShuffleLayer._construct_graph..r)rMrrrrrrrrrqrrrrrrrrrrrrr)rrrrrr\rrrrr:rZ_idxrrZ input_keyrrrrZs\              !zShuffleLayer._construct_graphrrr) r r!r"r#rrTrrrZrKrrrArrs0 rcseZdZdZ    d fdd ZddZddZd d Zed d Z d dZ ddZ ddZ ddZ eddZd!ddZddZddZd"ddZZS)#BroadcastJoinLayera;Broadcast-based Join Layer High-level graph layer for a join operation requiring the smaller collection to be broadcasted to every partition of the larger collection. Parameters ---------- name : str Name of new (joined) output collection. lhs_name: string "Left" DataFrame collection to join. lhs_npartitions: int Number of partitions in "left" DataFrame collection. rhs_name: string "Right" DataFrame collection to join. rhs_npartitions: int Number of partitions in "right" DataFrame collection. parts_out : list of int (optional) List of required output-partition indices. annotations : dict (optional) Layer annotations. **merge_kwargs : **dict Keyword arguments to be passed to chunkwise merge func. Nc  stj|d||_||_||_||_||_||_|p!tt |j|_ t | t r,t | n| |_t | t r8t | n| |_| |_|jd|_|j|jd<|j|jd<dS)Nrhowleft_onright_on)r@rrMrlhs_namelhs_npartitionsrhs_namerhs_npartitionsrrqrrrxr/rr merge_kwargsrr) rrMrrrrrrrrrrrArrrs zBroadcastJoinLayer.__init__cr)NcrrrRrrSrrrrz5BroadcastJoinLayer.get_output_keys..rrSrrSrrnrz"BroadcastJoinLayer.get_output_keyscCr)Nz5BroadcastJoinLayer)rrMrrrrSrrrrT rzBroadcastJoinLayer.__repr__cCrirjrkrSrrrrlrz"BroadcastJoinLayer.is_materializedcCrUrVrXr[rrrr]r^zBroadcastJoinLayer._dictcCr_rr`rarrrr4rzBroadcastJoinLayer.__getitem__cCrcrrdrSrrrrfrzBroadcastJoinLayer.__iter__cCrcrrgrSrrrrh!rzBroadcastJoinLayer.__len__c Crrrrrrrr$rz!BroadcastJoinLayer._keys_to_partscCs4|j|jkr|j|j|j|jfS|j|j|j|jfSr)rrrrrrrSrrr_broadcast_plan1s  z"BroadcastJoinLayer._broadcast_plancs||jdd\}}tt}|p||}|D]$}||j|ffddt|DO<||j|f||fhO<q|S)zDetermine the necessary dependencies to produce `keys`. For a broadcast join, output partitions always depend on all partitions of the broadcasted collection, but only one partition of the "other" collection. Ncsh|]}|fqSrrro bcast_namerrrXsz8BroadcastJoinLayer._cull_dependencies..)rrrrrMrq)rrmr bcast_size other_namerrrrrrKs( z%BroadcastJoinLayer._cull_dependenciescCs0t|j|j|j|j|j|jf|j|d|jS)N)rr) rrMrrrrrrrrrrrr^s zBroadcastJoinLayer._cullcCr)aCull a BroadcastJoinLayer HighLevelGraph layer. The underlying graph will only include the necessary tasks to produce the keys (indices) included in `parts_out`. Therefore, "culling" the layer only requires us to reset this parameter. rrrrrrrkrzBroadcastJoinLayer.cullFcCs2d|j}d|j}|rtd}td}td}nddlm}ddlm}dd lm}|j\}}} } |j|jkr:d nd } i} |j D]U} |j d krT|| | f| |f| || f<g}t |D]2}|j d krit j || f|fn| | f||fg}| d kry||| |f}t|||jf| |<||qZ||f| |j| f<qA| S) z/Construct graph for a broadcast join operation.zinter-rz%dask.dataframe.multi._split_partitionz$dask.dataframe.multi._concat_wrapperz)dask.dataframe.multi._merge_chunk_wrapperr)_concat_wrapper)_merge_chunk_wrapper)_split_partitionleftrightinner)rMrZdask.dataframe.multirrrrrrrrrqrrreverserrr)rrZ inter_namerZsplit_partition_funcrZmerge_chunk_funcrrrZother_onZ bcast_sider\r:rjZ _merge_argsZ inter_keyrrrrZ{sX             z#BroadcastJoinLayer._construct_graph)NNNNrr)r r!r"r#rrnrTrlrr]r4rfrhrrrrrrZrKrrrArrs,"     rcsFeZdZdZ    d fdd ZeddZdd Zd d ZZ S) DataFrameIOLayera DataFrame-based Blockwise Layer with IO Parameters ---------- name : str Name to use for the constructed layer. columns : str, list or None Field name(s) to read in as columns in the output. inputs : list or BlockwiseDep List of arguments to be passed to ``io_func`` so that the materialized task to produce partition ``i`` will be: ``(, inputs[i])``. Note that each element of ``inputs`` is typically a tuple of arguments. io_func : callable A callable function that takes in a single tuple of arguments, and outputs a DataFrame partition. Column projection will be supported for functions that satisfy the ``DataFrameIOFunction`` protocol. label : str (optional) String to use as a prefix in the place-holder collection name. If nothing is specified (default), "subset-" will be used. produces_tasks : bool (optional) Whether one or more elements of `inputs` is expected to contain a nested task. This argument in only used for serialization purposes, and will be deprecated in the future. Default is False. creation_info: dict (optional) Dictionary containing the callable function ('func'), positional arguments ('args'), and key-word arguments ('kwargs') used to produce the dask collection with this underlying ``DataFrameIOLayer``. annotations: dict (optional) Layer annotations to pass through to Blockwise. NFc s||_||_||_||_||_||_||_||_t|t s-t ddt |jD|jd} n|} |j|t dfi} t j|jd| | dfgi|ddS)NcSsi|]\}}|f|qSrrrrrrrsz-DataFrameIOLayer.__init__..)r*rr:)outputZoutput_indicesr\indicesr(r)rM_columnsrio_funclabelr*r creation_inforr r rr r@r) rrMcolumnsrrrr*r rZ io_arg_mapr\rArrrs.   zDataFrameIOLayer.__init__cCs|jS)z(Current column projection for this layer)rrSrrrr szDataFrameIOLayer.columnsc Csddlm}t|}|jdust|j|rCt|j|r$|j|}n|j}t |j p,ddt |j |||j ||j |j|jd}|S|S)zProduce a column projection for this IO layer. Given a list of required output columns, this method returns the projected layer. r)DataFrameIOFunctionNZsubset-)rr*r)Zdask.dataframe.io.utilsr rxr r issupersetrrproject_columnsrrr rMrr*r)rr r rlayerrrrrs"   z DataFrameIOLayer.project_columnscCsd|jt|j|jS)Nz3DataFrameIOLayer)rrMr+rr rSrrrrT<rzDataFrameIOLayer.__repr__)NFNN) r r!r"r#rrr rrTrKrrrArrs*( rcseZdZUdZded<ded<ded<ded<ded <d ed <ded <ded <ded<ded<ded<ded<      d;d<fdd ZddddZd=d!d"Zd#d$Zd%d&Z d'd(Z d)d*Z d+d,Z e d-d.Zd/d0Zd1d2Zd3d4Zd5d6Zd7d8Zd9d:ZZS)>DataFrameTreeReductionagDataFrame Tree-Reduction Layer Parameters ---------- name : str Name to use for the constructed layer. name_input : str Name of the input layer that is being reduced. npartitions_input : str Number of partitions in the input layer. concat_func : callable Function used by each tree node to reduce a list of inputs into a single output value. This function must accept only a list as its first positional argument. tree_node_func : callable Function used on the output of ``concat_func`` in each tree node. This function must accept the output of ``concat_func`` as its first positional argument. finalize_func : callable, optional Function used in place of ``tree_node_func`` on the final tree node(s) to produce the final output for each split. By default, ``tree_node_func`` will be used. split_every : int, optional This argument specifies the maximum number of input nodes to be handled by any one task in the tree. Defaults to 32. split_out : int, optional This argument specifies the number of output nodes in the reduction tree. If ``split_out`` is set to an integer >=1, the input tasks must contain data that can be indexed by a ``getitem`` operation with a key in the range ``[0, split_out)``. output_partitions : list, optional List of required output partitions. This parameter is used internally by Dask for high-level culling. tree_node_name : str, optional Name to use for intermediate tree-node tasks. strrMrrrrrtree_node_funcCallable | None finalize_func split_every split_outz list[int]output_partitionstree_node_namewidthsheightN int | Nonelist[int] | None str | Nonerdict[str, Any] | Nonec stj| d||_||_||_||_||_||_||_||_ | dur,t t |j p)dn| |_ | p5d|j|_ |j} | g|_| dkrVt| |j} |jt| | dksBt|j|_dS)NrrDz tree_node-)r@rrMrrrrrrrrxrqrrrmathceilrrr+r) rrMrrrrrrrrrrrrArrrus*zDataFrameTreeReduction.__init__rsplitcGs|jr||fS|Sr)r)rr#Z name_partsrrr _make_keysz DataFrameTreeReduction._make_keyFcCs(|r |jr |j}n|j}tj||j|fSr)rrr|r}r)r input_keys final_taskZ outer_funcrrr _define_tasks z#DataFrameTreeReduction._define_taskc si}js|Sjjr0d7jD]tjD]}tjj|ff|j|d<qqjdkrjD]{tdjD]rtj D]h}j d}j |}t |j |}dkrrfddt||D}nfddt||D}jdkr|dksJd |d j |d d |j f<qIj |d d |jj|d<qIq@q8|SjD]jddg}j |d d |j f<q|S)z%Construct graph for a tree reduction.z-splitr"rDcsg|] }j|dqS)r")r$r,p)name_input_userIrrrrpsz;DataFrameTreeReduction._construct_graph..cs$g|]}jj|ddqS)rDr")r$rr))rrIrrrrps rzgroup = z%, not 0 for final tree reduction taskT)r&F)rrrrqrrrr$rrrminr'rMr)rr\r*groupZp_maxZlstartZlstopr%r)rr+rIrrrZs^        , z'DataFrameTreeReduction._construct_graphcCsd|j|j|jS)Nz>DataFrameTreeReduction)rrMrrrSrrrrTs zDataFrameTreeReduction.__repr__cr)NcrrrRrHrSrrrrz6DataFrameTreeReduction._output_keys..)rrSrrSr _output_keysrz#DataFrameTreeReduction._output_keyscCrU)N_cached_output_keys)rYr/r.)rZ output_keysrrrrns z&DataFrameTreeReduction.get_output_keyscCrirjrkrSrrrrlrz&DataFrameTreeReduction.is_materializedcCrUrVrXr[rrrr]r^zDataFrameTreeReduction._dictcCr_rr`rarrrr4rz"DataFrameTreeReduction.__getitem__cCrcrrdrSrrrrfrzDataFrameTreeReduction.__iter__cCs>t|jddp d|jpd}|jr||jt|jS|S)NrD)sumrrrr+r)rZ tree_sizerrrrhs zDataFrameTreeReduction.__len__c Cr)z;Simple utility to convert keys to output partition indices.r)rrmZsplitsrbr_splitrrr_keys_to_output_partitionsrz1DataFrameTreeReduction._keys_to_output_partitionsc Cs2t|j|j|j|j|j|j|j|j||j |j d S)N)rrrrrr) rrMrrrrrrrrr)rrrrrr,szDataFrameTreeReduction._cullcsTjdffddtjDi}|}|tjkr&|}||fS|fS)z2Cull a DataFrameTreeReduction HighLevelGraph layerrcrrrrorSrrr>rz.DataFrameTreeReduction.cull..)rMrqrr2rrr)rrmrrrrrrSrr;s  zDataFrameTreeReduction.cull)NrNNNN)rMrrrrrrrrrrrrrrrrrrrrrr)r r!r"r#r8rr$r'rZrTr.rnrlrr]r4rfrhr2rrrKrrrArrBsH % ' J   rr). __future__rryr r collectionsrcollections.abcr itertoolsrtypingrZtlzr|Z tlz.curriedrZ dask.baser Zdask.blockwiser r r r Z dask.corerZdask.highlevelgraphrZ dask.utilsrrrrrr$r9r=rLr{rrrrrrrrrrrs>           z7 /V3~w