o $iA@sddlZddlmZddlmZmZmZmZmZm Z m Z ddl m Z ddl mZddlmZededeefd d Zeded edefd d ZGdddZGdddeZGdddeZGdddZdeefddZdS)N)deque)AnyCallableDequeDictListOptionalUnion)Dataset) AggregateFnV2) DeveloperAPIvaluereturncCs |dS)z*Tokenize a string using a split on spaces. )split)r rY/home/ubuntu/veenaModal/venv/lib/python3.10/site-packages/ray/data/preprocessors/utils.pysimple_split_tokenizer s r num_featurescCs,t|}t|}t|d}||S)z6Deterministically hash a value into the integer space.)strencodehashlibsha1int hexdigest)r r encoded_value hashed_valuehashed_value_intrrr simple_hashs  rc @sBeZdZdZddddeeefdedeegeffdd Zd S) BaseStatSpeczEEncapsulates a statistical computation with optional post-processing.cC|SNrxrrr zBaseStatSpec.post_process_fnstat_fnr( post_key_fncCs||_||_||_dSr"r)r(r*)selfr)r(r*rrr__init__s zBaseStatSpec.__init__N) __name__ __module__ __qualname____doc__r r rrr-rrrrr s  r c s^eZdZdZdddddeeeegeffdedeegefd eeffd d Z Z S) AggregateStatSpecz5Represents an AggregateFnV2 spec for a single column.cCr!r"rr#rrrr%/r&zAggregateStatSpec.N)r(column aggregator_fnr(r*r3cstj|||d||_dSNr+)superr-r3)r,r4r(r*r3 __class__rrr-+s  zAggregateStatSpec.__init__) r.r/r0r1r r rrrr- __classcell__rrr7rr2(s r2c s`eZdZdZddddedeeegefdeeegefded eef fd d ZZ S) CallableStatSpeczPRepresents a user-defined stat function that operates outside Dataset.aggregate.cCr!r"rr#rrrr%Dr&zCallableStatSpec.r'r) stat_key_fnr*r(columnscs"tj|||d||_||_dSr5)r6r-r<r;)r,r)r;r*r(r<r7rrr->s   zCallableStatSpec.__init__) r.r/r0r1rrrrr-r9rrr7rr:;sr:c @seZdZdZddZddZdddd d eegefd ed e eegefd e eddf ddZ dddd dege fd edeegefd e eegefd e eddf ddZ dedeee ffddZde efddZde efddZde efddZdd Zd!d"ZdS)#StatComputationPlana\ Encapsulates a set of aggregators (AggregateFnV2) and legacy stat functions to compute statistics over a Ray dataset. Supports two types of aggregations: 1. AggregateFnV2-based aggregators, which are batch-executed using `Dataset.aggregate(...)`. 2. Callable-based stat functions, executed sequentially (legacy use case). cCs t|_dSr")r _aggregatorsr,rrrr-Xs zStatComputationPlan.__init__cCs|jdSr")r>clearr?rrrreset[szStatComputationPlan.resetcCr!r"rr#rrrr%br&zStatComputationPlan.N)r(r*r4r(r*r<rc Cs.|D]}||}|jt||||dqdS)a Registers an AggregateFnV2 factory for one or more columns. Args: aggregator_fn: A callable (typically a lambda or class) that accepts a column name and returns an instance of AggregateFnV2. post_process_fn: Function to post-process the aggregated result. post_key_fn: Optional key generator to use to save aggregation results after post-processing. columns: List of column names to aggregate. )r4r(r*r3N)r>appendr2)r,r4r(r*r<r3 agg_instancerrradd_aggregator^sz"StatComputationPlan.add_aggregatorcCr!r"rr#rrrr%~r&r)r;c Cs"|jt|||||p |ddS)a Registers a custom stat function to be run sequentially. This supports legacy use cases where arbitrary callables are needed and cannot be run via Dataset.aggregate(). Args: stat_fn: A zero-argument callable that returns the stat. post_process_fn: Function to apply to the result. stat_key_fn: post_key_fn: columns: )r)r(r<r;r*N)r>rBr:)r,r)r(r;r*r<rrradd_callable_statzsz%StatComputationPlan.add_callable_statdatasetc Csi}|}|r.|j|}|D]}|jj}|jdur"||jn|}|||||<q|D]!}||j }|j D]} | | }|| }|||||<q=q2|S)aq Executes all registered aggregators and stat functions. AggregateFnV2-based aggregators are batched and executed via Dataset.aggregate(). Callable-based stat functions are run sequentially. Args: dataset: The Ray Dataset to compute statistics on. Returns: A dictionary of computed statistics. N) _get_aggregate_fn_list aggregate_get_aggregate_specsr)namer*r3r(_get_custom_stat_fn_specsr;r<) r,rFstats aggregators raw_resultspecstat_keypost_keyresultcolrrrcomputes&          zStatComputationPlan.computecCdd|jDS)NcSsg|] }t|tr|jqSr) isinstancer2r).0rOrrr s z>StatComputationPlan._get_aggregate_fn_list..r>r?rrrrGsz*StatComputationPlan._get_aggregate_fn_listcCrU)NcSg|] }t|tr|qSr)rVr2rWrrrrY  z.rZr?rrrrIz(StatComputationPlan._get_aggregate_specscCrU)NcSr[r)rVr:rWrrrrYr\zAStatComputationPlan._get_custom_stat_fn_specs..rZr?rrrrKr]z-StatComputationPlan._get_custom_stat_fn_specscCst|dkS)Nr)lenrKr?rrrhas_custom_stat_fnsz&StatComputationPlan.has_custom_stat_fncCs t|S)z4 Iterates over all AggregatorSpecs. )iterrIr?rrr__iter__s zStatComputationPlan.__iter__)r.r/r0r1r-rArrr rrrDrrEr rrTrGr2rIr:rKr_rarrrrr=NsL        % r= callbackscsfdd}|S)z Wraps a base post-processing function with a sequence of callback functions. Useful when multiple post-processing steps need to be applied in order. cs|}D]}||}q|Sr"r)rR processedcbbase_fnrbrrwrappers z$make_post_processor..wrapperr)rfrbrgrrermake_post_processorsrh)r collectionsrtypingrrrrrrr ray.datar ray.data.aggregater ray.util.annotationsr rrobjectrrr r2r:r=rhrrrrs  $