o .wixE@sddlmZddlmZddlmZmZmZddlZddlm Z ddl m Z ddl m Z ddlmZdd lmZdd lmZmZmZdd lmZdd lmZesSd gZGddde ZdS))Sequence)deepcopy)AnyOptionalUnionN)Tensor) ModuleList)MetricCollection)Metric)_MATPLOTLIB_AVAILABLE)_AX_TYPE_PLOT_OUT_TYPEplot_single_or_multi_val)rank_zero_warn)ClasswiseWrapperMetricTracker.plotcseZdZUdZeeeefed<eee fed<d'deee fdeeeedfddffdd Z e de fd d Z d(d d Zd ededdfddZd ededdfddZdefddZdefddZd(ddZd(ddZ d)dededeeeee eefee eeffedeeeedffeeeeedffeeee dfffffddZd eddfd!d"Z dd#eeeeefd$eede fd%d&Z!Z"S)* MetricTrackeraA wrapper class that can help keeping track of a metric or metric collection over time. The wrapper implements the standard ``.update()``, ``.compute()``, ``.reset()`` methods that just calls corresponding method of the currently tracked metric. However, the following additional methods are provided: -``MetricTracker.n_steps``: number of metrics being tracked -``MetricTracker.increment()``: initialize a new metric for being tracked -``MetricTracker.compute_all()``: get the metric value for all steps -``MetricTracker.best_metric()``: returns the best value Out of the box, this wrapper class fully supports that the base metric being tracked is a single `Metric`, a `MetricCollection` or another `MetricWrapper` wrapped around a metric. However, multiple layers of nesting, such as using a `Metric` inside a `MetricWrapper` inside a `MetricCollection` is not fully supported, especially the `.best_metric` method that cannot auto compute the best metric and index for such nested structures. Args: metric: instance of a ``torchmetrics.Metric`` or ``torchmetrics.MetricCollection`` to keep track of at each timestep. maximize: either single bool or list of bool indicating if higher metric values are better (``True``) or lower is better (``False``). Example (single metric): >>> from torch import randint >>> from torchmetrics.wrappers import MetricTracker >>> from torchmetrics.classification import MulticlassAccuracy >>> tracker = MetricTracker(MulticlassAccuracy(num_classes=10, average='micro')) >>> for epoch in range(5): ... tracker.increment() ... for batch_idx in range(5): ... tracker.update(randint(10, (100,)), randint(10, (100,))) ... print(f"current acc={tracker.compute()}") current acc=0.1120000034570694 current acc=0.08799999952316284 current acc=0.12600000202655792 current acc=0.07999999821186066 current acc=0.10199999809265137 >>> best_acc, which_epoch = tracker.best_metric(return_step=True) >>> best_acc # doctest: +ELLIPSIS 0.1260... >>> which_epoch 2 >>> tracker.compute_all() tensor([0.1120, 0.0880, 0.1260, 0.0800, 0.1020]) Example (multiple metrics using MetricCollection): >>> from torch import randn >>> from torchmetrics.wrappers import MetricTracker >>> from torchmetrics import MetricCollection >>> from torchmetrics.regression import MeanSquaredError, ExplainedVariance >>> tracker = MetricTracker(MetricCollection([MeanSquaredError(), ExplainedVariance()]), maximize=[False, True]) >>> for epoch in range(5): ... tracker.increment() ... for batch_idx in range(5): ... tracker.update(randn(100), randn(100)) ... print(f"current stats={tracker.compute()}") # doctest: +NORMALIZE_WHITESPACE current stats={'MeanSquaredError': tensor(2.3292), 'ExplainedVariance': tensor(-0.9516)} current stats={'MeanSquaredError': tensor(2.1370), 'ExplainedVariance': tensor(-1.0775)} current stats={'MeanSquaredError': tensor(2.1695), 'ExplainedVariance': tensor(-0.9945)} current stats={'MeanSquaredError': tensor(2.1072), 'ExplainedVariance': tensor(-1.1878)} current stats={'MeanSquaredError': tensor(2.0562), 'ExplainedVariance': tensor(-1.0754)} >>> from pprint import pprint >>> best_res, which_epoch = tracker.best_metric(return_step=True) >>> pprint(best_res) # doctest: +ELLIPSIS {'ExplainedVariance': -0.951..., 'MeanSquaredError': 2.056...} >>> which_epoch {'MeanSquaredError': 4, 'ExplainedVariance': 0} >>> pprint(tracker.compute_all()) {'ExplainedVariance': tensor([-0.9516, -1.0775, -0.9945, -1.1878, -1.0754]), 'MeanSquaredError': tensor([2.3292, 2.1370, 2.1695, 2.1072, 2.0562])} maximize _base_metricNmetricreturncstt|ttfstd|||_|dur~t|tr6t|dddur1td|j j d|j |_ nt|tr}g|_ | D]:\}tdddurVtd|dttrrtjjtrrfddttjjD}nj g}|j |qBnDt|ttfstdt|trtd d |Dstd t|trt|trt|t|krtd t|trt|tstd ||_ d|_dS)Nz[Metric arg need to be an instance of a torchmetrics `Metric` or `MetricCollection` but got higher_is_betterz The metric 'zb' does not have a 'higher_is_better' attribute. Please provide the `maximize` argument explicitly.zz' in the MetricCollection does not have a 'higher_is_better' attribute. Please provide the `maximize` argument explicitly.csg|]}jqS)r).0_mrZ/home/ubuntu/sommelier/.venv/lib/python3.10/site-packages/torchmetrics/wrappers/tracker.py sz*MetricTracker.__init__..zBArgument `maximize` should either be a single bool or list of boolcss|]}t|tVqdSN) isinstancebool)rrrrr sz)MetricTracker.__init__..z1Argument `maximize` is list but not type of bool.zOThe len of argument `maximize` should match the length of the metric collectionzLArgument `maximize` should be a single bool when `metric` is a single MetricF)super__init__r r r TypeErrorrgetattrAttributeError __class____name__rritemsrr num_classesintrangeextendr!list ValueErroralllen_increment_called)selfrrnamem_higher_is_betterr(rrr$osF      $ zMetricTracker.__init__cCs t|dS)z=Returns the number of times the tracker has been incremented.)r2r4rrrn_stepss zMetricTracker.n_stepscCsd|_|t|jdS)zDCreate a new instance of the input metric that will be updated next.TN)r3appendrrr9rrr incrementszMetricTracker.incrementargskwargscOsF|dt|dttfstdt|dd|d|i|S)z1Call forward of the current metric being tracked.forwardCExpected the last item to be a Metric or MetricCollection, but got .)_check_for_incrementr r r r%typer4r=r>rrrr?s zMetricTracker.forwardcOsL|dt|dttfstdt|dd|dj|i|dS)z(Update the current metric being tracked.updater@rArBN)rCr r r r%rDrFrErrrrFs zMetricTracker.updatecCs@|dt|dttfstdt|dd|dS)z1Call compute of the current metric being tracked.computer@rArB)rCr r r r%rDrGr9rrrrGs  zMetricTracker.computecs|dddt|Dz2tdtr&d}fdd|DWStdtr9tddDdWStjddWStyKYSw) aCompute the metric value for all tracked metrics. Return: By default will try stacking the results from all increments into a single tensor if the tracked base object is a single metric. If a metric collection is provided a dict of stacked tensors will be returned. If the stacking process fails a list of the computed results will be returned. Raises: ValueError: If `self.increment` have not been called before this method is called. compute_allcSs g|] \}}|dkr|qS)r)rG)rirrrrrs z-MetricTracker.compute_all..rcs*i|]tjfddDddqS)csg|]}|qSrrrrkrrrsz8MetricTracker.compute_all...rdimtorchstack)rresrLr s*z-MetricTracker.compute_all..cSsg|] }tj|ddqS)rrNrPrJrrrrsrN) rC enumerater dictkeysr/rQrRr%)r4rXrrSrrHs   zMetricTracker.compute_allcCs:t|dttfstdt|dd|ddS)z'Reset the current metric being tracked.r@rArBNr r r r%rDresetr9rrrrZszMetricTracker.resetcCs8|D]}t|ttfstdt|d|qdS)z Reset all metrics being tracked.z?Expected all metrics to be Metric or MetricCollection, but got rBNrY)r4rrrr reset_alls  zMetricTracker.reset_allF return_stepNNc Cs|}t|trtd|rdSdSt|jtrgt|jtsg|jr%tj ntj }z||d\}}|r;| | fWS| WSt t fyf}ztd|dt|r[WYd}~dSWYd}~dSd}~wwt|jtrp|jnt||jg}ii}}t|D]P\}\} } z"||rtj ntj }|| d} | d | d || <|| <Wqt t fy}ztd| d |d td\|| <|| <WYd}~qd}~ww|r||fS|S) aReturn the highest metric out of all tracked. Args: return_step: If ``True`` will also return the step with the highest metric value. Returns: Either a single value or a tuple, depends on the value of ``return_step`` and the object being tracked. - If a single metric is being tracked and ``return_step=False`` then a single tensor will be returned - If a single metric is being tracked and ``return_step=True`` then a 2-element tuple will be returned, where the first value is optimal value and second value is the corresponding optimal step - If a metric collection is being tracked and ``return_step=False`` then a single dict will be returned, where keys correspond to the different values of the collection and the values are the optimal metric value - If a metric collection is being bracked and ``return_step=True`` then a 2-element tuple will be returned where each is a dict, with keys corresponding to the different values of th collection and the values of the first dict being the optimal values and the values of the second dict being the optimal step In addition the value in all cases may be ``None`` if the underlying metric does have a proper defined way of being optimal or in the case where a nested structure of metrics are being tracked. zEncountered nested structure. You are probably using a metric collection inside a metric collection, or a metric wrapper inside a metric collection, which is not supported by `.best_metric()` method. Returning `None` instead.r]NrzDEncountered the following error when trying to get the best metric: z^this is probably due to the 'best' not being defined for this metric.Returning `None` instead.r8zNEncountered the following error when trying to get the best metric for metric :z_ this is probably due to the 'best' not being defined for this metric.Returning `None` instead.)rHr r/rrr rrrQmaxminitemr0 RuntimeError UserWarningr2rVr*) r4r\rTfnvalueidxerrorrrIrMvoutrrr best_metricsZ!   "  & zMetricTracker.best_metricmethodcCs|js td|ddS)zSCheck that a metric that can be updated/used for computations has been initialized.`z9` cannot be called before `.increment()` has been called.N)r3r0)r4rkrrrrC6sz"MetricTracker._check_for_incrementvalaxcCs2|dur|n|}t|||jjd\}}||fS)aqPlot a single or multiple values from the metric. Args: val: Either a single result from calling `metric.forward` or `metric.compute` or a list of these results. If no value is provided, will automatically call `metric.compute` and plot that result. ax: An matplotlib axis object. If provided will add plot to that axis Returns: Figure and Axes object Raises: ModuleNotFoundError: If `matplotlib` is not installed .. plot:: :scale: 75 >>> # Example plotting a single value >>> import torch >>> from torchmetrics.wrappers import MetricTracker >>> from torchmetrics.classification import BinaryAccuracy >>> tracker = MetricTracker(BinaryAccuracy(), maximize=True) >>> for epoch in range(5): ... tracker.increment() ... for batch_idx in range(5): ... tracker.update(torch.randint(2, (10,)), torch.randint(2, (10,))) >>> fig_, ax_ = tracker.plot() # plot all epochs N)rnr5)rHrr(r))r4rmrnfigrrrplot;s  rr)rN)F)#r) __module__ __qualname____doc__rr!r/__annotations__r r r$propertyr,r:r<rr?rFrGrHrZr[floatrtuplerWstrrjrCrrr r rp __classcell__rrr7rr!sL J2.   * Sr)collections.abcrcopyrtypingrrrrQrtorch.nnrtorchmetrics.collectionsr torchmetrics.metricr torchmetrics.utilities.importsr torchmetrics.utilities.plotr r rtorchmetrics.utilities.printsrtorchmetrics.wrappersr__doctest_skip__rrrrrs