o }oiW@sddlZddlZddlZddlmZddlmZmZddlZ ddl Z ddl Z ddl m Z ddlmZddlmZmZddlmZddlmZmZd Zd Zd Zd Zd ZdZddZidddfdedede j de j dedede!de"de!de!de#fddZ$idfde j de j d edede"de#f d!d"Z%Gd#d$d$Z&d%d&Z'idddfdeded'ee j d(ee j de!de"de!de!de#fd)d*Z(Gd+d,d,e&Z)d-e j d.e j d/e!fd0d1Z*   d6deded-ee j d.ee j de!d/e!de!de!de#fd2d3Z+Gd4d5d5e&Z,dS)7N)partial)CallableOptional)entropy)logits_to_probs)Poolset_start_method)Metric)GlobalAverageLossMetric Perplexity g?cCsFdtjd<dtjd<tjrtjdvr!tjjd||ddSdSdS) z Setup ddp enviroment localhost MASTER_ADDR8088 MASTER_PORT)win32cygwingloorank world_sizeN)osenvirontorch distributed is_availablesysplatforminit_process_grouprr#U/home/ubuntu/.local/lib/python3.10/site-packages/tests/collections/common/pl_utils.py setup_ddp5s  r%T:0yE>r worldsizepredstarget metric_class sk_metricdist_sync_on_step metric_argscheck_dist_sync_on_step check_batchatolc s`|d d|i|} t| } t| } t|t|D][| } | jr]|dkr\tfddt|D}tfddt|D}|||}|r\tj | || ds\Jq|}| rttj | || dstJq| }t |tj sJtfddttD}tfddttD}|||}tj | || dsJd S) a Utility function doing the actual comparison between lightning class metric and reference metric. Args: rank: rank of current process worldsize: number of processes preds: torch tensor with predictions target: torch tensor with targets metric_class: lightning metric class that should be tested sk_metric: callable function that is used for comparison dist_sync_on_step: bool, if true will synchronize metric state across processes at each ``forward()`` metric_args: dict with additional arguments used for class initialization check_dist_sync_on_step: bool, if true will check if the metric is also correctly calculated per batch per device (and not just at the end) check_batch: bool, if true will check if the metric is also correctly calculated across devices for each batch (and not just at the end) r,rcg|]}|qSr#r#.0r)ir(r#r$ hz_class_test..cr1r#r#r2)r5r)r#r$r6ir7r0cg|]}|qSr#r#r3r5)r(r#r$r6xcr9r#r#r:)r)r#r$r6yr;Nr#)pickledumpsloadsrange NUM_BATCHESr,rstacknpallclosenumpycompute isinstanceTensor)rr'r(r)r*r+r,r-r.r/r0metricpickled_metric batch_result ddp_preds ddp_targetsk_batch_resultresult total_preds total_target sk_resultr#)r5r(r)r$ _class_test>s.    rRmetric_functionalc Cs^t|fi|}ttD] }|||||}|||||} tj|| |ds,Jq dS)a Utility function doing the actual comparison between lightning functional metric and reference metric. Args: preds: torch tensor with predictions target: torch tensor with targets metric_functional: lightning metric functional that should be tested sk_metric: callable function that is used for comparison metric_args: dict with additional arguments used for class initialization r8N)rr?r@rBrCrD) r(r)rSr+r-r0rHr5lightning_resultrQr#r#r$_functional_tests  rUc@seZdZdZdZddZddZifdejdejd e d e d e f d d Z iddfde dejdejde d e de d e de de fddZdS) MetricTesterae Class used for efficiently run alot of parametrized tests in ddp mode. Makes sure that ddp is only setup once and that pool of processes are used for all tests. All tests should subclass from this and implement a new method called `test_metric_name` where the method `self.run_metric_test` is called inside. r&csZztdWn tyYnwt_tjd_jtfddtjDdS)z Setup the metric class. This will spawn the pool of workers that are used for metric testing and setup_ddp spawn) processescg|]}|jfqSr#poolSizer3rselfr#r$r6z,MetricTester.setup_class..N) r RuntimeError NUM_PROCESSESr[rpoolstarmapr%r?r]r#r]r$ setup_classs  &zMetricTester.setup_classcCs|j|jdS)z Close pool of workers N)rbclosejoinr]r#r#r$teardown_classs zMetricTester.teardown_classr(r)rSr+r-cCst||||||jddS)a Main method that should be used for testing functions. Call this inside testing method Args: preds: torch tensor with predictions target: torch tensor with targets metric_functional: lightning metric class that should be tested sk_metric: callable function that is used for comparison metric_args: dict with additional arguments used for class initialization )r(r)rSr+r-r0N)rUr0)r^r(r)rSr+r-r#r#r$run_functional_metric_tests z'MetricTester.run_functional_metric_testTddpr*r,r.r/c s||r,tjdkr tdjtt|||||||| jd fddt j DdStdd|||||||| jd dS) a Main method that should be used for testing class. Call this inside testing methods. Args: ddp: bool, if running in ddp mode or not preds: torch tensor with predictions target: torch tensor with targets metric_class: lightning metric class that should be tested sk_metric: callable function that is used for comparison dist_sync_on_step: bool, if true will synchronize metric state across processes at each ``forward()`` metric_args: dict with additional arguments used for class initialization check_dist_sync_on_step: bool, if true will check if the metric is also correctly calculated per batch per device (and not just at the end) check_batch: bool, if true will check if the metric is also correctly calculated across devices for each batch (and not just at the end) rDDP not supported on windows) r(r)r*r+r,r-r.r/r0crYr#rZr\r]r#r$r6r_z6MetricTester.run_class_metric_test..rN) r r!pytestskiprbrcrrRr0r?r[) r^rir(r)r*r+r,r-r.r/r#r]r$run_class_metric_tests>    z"MetricTester.run_class_metric_testN)__name__ __module__ __qualname____doc__r0rdrgrrGrdictrhboolr rnr#r#r#r$rVsN   "  rVcCst|dd}t|}|S)N)axis)rrBexpmean)probsentpplr#r#r$reference_perplexity_funcs  r|rylogitsc std d|i|} dudukr.tt| WddS1s'wYdSt| } t| } t|t|D]}| durGdndurPdn} | j r|dkrdurpt fddt|D} nt fddt|D} t | dd} t | }|rtj| ||d sJq>durt dd}n}t |}|rtj| ||d sJq>duduksJ| }t|t jsJdurt ddt }tj|||d sJdS) a Utility function doing the actual comparison between lightning class metric and reference metric. Args: rank: rank of current process worldsize: number of processes probs: torch tensor with probabilities logits: torch tensor with logits. The function checks ``probs`` and ``logits are mutually exclusive for ``Perplexity`` metric. dist_sync_on_step: bool, if true will synchronize metric state across processes at each ``forward()`` metric_args: dict with additional arguments used for class initialization check_dist_sync_on_step: bool, if true will check if the metric is also correctly calculated per batch per device (and not just at the end) check_batch: bool, if true will check if the metric is also correctly calculated across devices for each batch (and not just at the end) r,Nrcr1r#r#r2)r5ryr#r$r6Br7z*_perplexity_class_test..cr1r#r#r2)r5r}r#r$r6Dr7F) is_binaryr8r#)r rlraises ValueErrorr<r=r>r?r@r,rrArr|rBrCrDrErFrG)rr'ryr}r,r-r.r/r0 perplexityrIrJ ddp_probs ddp_logitsrMprNrQr#)r5r}ryr$_perplexity_class_testsH     *   rc@sFeZdZiddfdedeejdeejdedededefd d Zd S) PerplexityTesterTriryr}r,r-r.r/c st|r*tjdkr tdjtt||||||jdfddt j DdStdd||||||jd dS) a Main method that should be used for testing class. Call this inside testing methods. Args: ddp: bool, if running in ddp mode or not probs: torch tensor with probabilities. logits: torch tensor with logits. This test checks that probs and logits are mutually exclusive for ``Perplexity`` metric. dist_sync_on_step: bool, if true will synchronize metric state across processes at each ``forward()`` metric_args: dict with additional arguments used for class initialization check_dist_sync_on_step: bool, if true will check if the metric is also correctly calculated per batch per device (and not just at the end) check_batch: bool, if true will check if the metric is also correctly calculated across devices for each batch (and not just at the end) rrj)ryr}r,r-r.r/r0crYr#rZr\r]r#r$r6r_z>PerplexityTester.run_class_perplexity_test..rrkN) r r!rlrmrbrcrrr0r?r[)r^riryr}r,r-r.r/r#r]r$run_class_perplexity_testbs6    z*PerplexityTester.run_class_perplexity_testN) rorprqrtrrrGrsrr#r#r#r$ras&rloss_sum_or_avgnum_measurements take_avg_losscCsD|}|r ||9}|}|drttdS||S)a% Returns average loss for data from``loss_sum_or_avg``. This function sums all losses from ``loss_sum_or_avg`` and divides the sum by the sum of ``num_measurements`` elements. If ``take_avg_loss`` is ``True`` then ``loss_sum_or_avg[i]`` elements are mean values of ``num_measurements[i]`` losses. In that case before computing sum of losses each element of ``loss_sum_or_avg`` is multiplied by corresponding element of ``num_measurements``. If ``num_measurements`` sum is zero then the function returns NaN tensor. The function is used for testing ``nemo.collections.common.metrics.GlobalAverageLossMetric`` class. Args: loss_sum_or_avg: a one dimensional float ``torch.Tensor``. Sums or mean values of loss. num_measurements: a one dimensional integer ``torch.Tensor``. Number of values on which sums of means in ``loss_sum_or_avg`` are calculated. take_avg_loss: if ``True`` then ``loss_sum_or_avg`` contains mean losses else ``loss_sum_or_avg`` contains sums of losses. rnan)clonedetachsumeqrtensorfloat)rrrnm_sumr#r#r$reference_loss_funcs   rc st||d} t| } t| } t|t|D]| } | jrs|dkrrtfddt|D} tfddt|D} t | | |}|rr| rY| sXJqt j | ||dsrJd| d|d qd }d }t |||}|r| r| sJqt j | ||dsJd| d|d q| }t|tjsJt |}| r| sJd St j | ||dsJd | d |d S)a Utility function doing the actual comparison between lightning class metric and reference metric. Args: rank: rank of current process worldsize: number of processes loss_sum_or_avg: a one dimensional float torch tensor with loss sums or means. num_measurements: a one dimensional integer torch tensor with number of values on which sums or means from ``loss_sum_or_avg`` were computed. dist_sync_on_step: bool, if true will synchronize metric state across processes at each call of the method :meth:`forward()` take_avg_loss: dict with additional arguments used for class initialization check_dist_sync_on_step: bool, if true will check if the metric is also correctly calculated per batch per device (and not just at the end) check_batch: bool, if true will check if the metric is also correctly calculated across devices for each batch (and not just at the end) )r,rrcr1r#r#r2)r5rr#r$r6r7z$_loss_class_test..cr1r#r#r2)r5rr#r$r6r7r8zbatch_result = z, sk_batch_result = z, i = rkz result = z, sk_result = N)r r<r=r>r?r@r,rrArisnanrBrCrDrErFrG)rr'rrr,rr.r/r0 loss_metricrIrJddp_loss_sum_or_avgddp_num_measurementsrMlsnmrNrQr#)r5rrr$_loss_class_testsJ        0rc@s<eZdZ  d dedejdejdedededefd d Zd S) LossTesterTrirrr,rr.r/c st|r*tjdkr tdjtt||||||jdfddt j DdStdd||||||jd dS)Nrrj)rrr,rr.r/r0crYr#rZr\r]r#r$r6r_z2LossTester.run_class_loss_test..rrk) r r!rlrmrbrcrrr0r?r[)r^rirrr,rr.r/r#r]r$run_class_loss_tests6    zLossTester.run_class_loss_testN)TT)rorprqrtrrGrr#r#r#r$rs$r)TTr&)-rr<r functoolsrtypingrrrDrBrlr scipy.statsrtorch.distributions.utilsrtorch.multiprocessingrr torchmetricsr nemo.collections.common.metricsr r rar@ BATCH_SIZE NUM_CLASSES EXTRA_DIM THRESHOLDr%intrGrtrsrrRrUrVr|rrrrrr#r#r#r$s       G u  K9$ H