a  dx.@sddlmZmZmZmZmZddlmZmZddl m Z ddl m Z ddl mZddlmZmZddlmZesvdgZGd d d eZd S) )AnyDictOptionalSequenceUnion)Tensornn)MetricCollection)Metric)_MATPLOTLIB_AVAILABLE)_AX_TYPE_PLOT_OUT_TYPE) WrapperMetricMultitaskWrapper.plotcseZdZdZdZeeeee ffddfdd Z e eeeee ffddddZ eee feee fdd d d Zeeefd d dZeee feee feeefd ddZdd fdd ZdeeeeefeeeeedddZZS)MultitaskWrappera Wrapper class for computing different metrics on different tasks in the context of multitask learning. In multitask learning the different tasks requires different metrics to be evaluated. This wrapper allows for easy evaluation in such cases by supporting multiple predictions and targets through a dictionary. Note that only metrics where the signature of `update` follows the stardard `preds, target` is supported. Args: task_metrics: Dictionary associating each task to a Metric or a MetricCollection. The keys of the dictionary represent the names of the tasks, and the values represent the metrics to use for each task. Raises: TypeError: If argument `task_metrics` is not an dictionary TypeError: If not all values in the `task_metrics` dictionary is instances of `Metric` or `MetricCollection` Example (with a single metric per class): >>> import torch >>> from torchmetrics.wrappers import MultitaskWrapper >>> from torchmetrics.regression import MeanSquaredError >>> from torchmetrics.classification import BinaryAccuracy >>> >>> classification_target = torch.tensor([0, 1, 0]) >>> regression_target = torch.tensor([2.5, 5.0, 4.0]) >>> targets = {"Classification": classification_target, "Regression": regression_target} >>> >>> classification_preds = torch.tensor([0, 0, 1]) >>> regression_preds = torch.tensor([3.0, 5.0, 2.5]) >>> preds = {"Classification": classification_preds, "Regression": regression_preds} >>> >>> metrics = MultitaskWrapper({ ... "Classification": BinaryAccuracy(), ... "Regression": MeanSquaredError() ... }) >>> metrics.update(preds, targets) >>> metrics.compute() {'Classification': tensor(0.3333), 'Regression': tensor(0.8333)} Example (with several metrics per task): >>> import torch >>> from torchmetrics import MetricCollection >>> from torchmetrics.wrappers import MultitaskWrapper >>> from torchmetrics.regression import MeanSquaredError, MeanAbsoluteError >>> from torchmetrics.classification import BinaryAccuracy, BinaryF1Score >>> >>> classification_target = torch.tensor([0, 1, 0]) >>> regression_target = torch.tensor([2.5, 5.0, 4.0]) >>> targets = {"Classification": classification_target, "Regression": regression_target} >>> >>> classification_preds = torch.tensor([0, 0, 1]) >>> regression_preds = torch.tensor([3.0, 5.0, 2.5]) >>> preds = {"Classification": classification_preds, "Regression": regression_preds} >>> >>> metrics = MultitaskWrapper({ ... "Classification": MetricCollection(BinaryAccuracy(), BinaryF1Score()), ... "Regression": MetricCollection(MeanSquaredError(), MeanAbsoluteError()) ... }) >>> metrics.update(preds, targets) >>> metrics.compute() {'Classification': {'BinaryAccuracy': tensor(0.3333), 'BinaryF1Score': tensor(0.)}, 'Regression': {'MeanSquaredError': tensor(0.8333), 'MeanAbsoluteError': tensor(0.6667)}} FN) task_metricsreturncs$||tt||_dSN)_check_task_metrics_typesuper__init__rZ ModuleDictr)selfr __class__h/var/www/html/stable-diffusion-webui/venv/lib/python3.9/site-packages/torchmetrics/wrappers/multitask.pyras  zMultitaskWrapper.__init__cCsJt|tstd||D]$}t|ttfs tdt|q dS)NzDExpected argument `task_metrics` to be a dict. Found task_metrics = zYExpected each task's metric to be a Metric or a MetricCollection. Found a metric of type ) isinstancedict TypeErrorvaluesr r type)rmetricrrrris  z)MultitaskWrapper._check_task_metrics_type) task_preds task_targetsrcCs|j|kr"|ksLntd|d|d|j|jD]$\}}||}||}|||qVdS)zUpdate each task's metric with its corresponding pred and target. Args: task_preds: Dictionary associating each task to a Tensor of pred. task_targets: Dictionary associating each task to a Tensor of target. zExpected arguments `task_preds` and `task_targets` to have the same keys as the wrapped `task_metrics`. Found task_preds.keys() = z, task_targets.keys() = z and self.task_metrics.keys() = N)rkeys ValueErroritemsupdate)rr"r# task_namer!predtargetrrrr'us$zMultitaskWrapper.update)rcCsdd|jDS)zCompute metrics for all tasks.cSsi|]\}}||qSr)compute.0r(r!rrr z,MultitaskWrapper.compute..rr&)rrrrr+szMultitaskWrapper.computecsfdd|jDS)zTCall underlying forward methods for all tasks and return the result as a dictionary.cs$i|]\}}||||qSrrr,r"r#rrr.sz,MultitaskWrapper.forward..r0)rr"r#rr1rforwards zMultitaskWrapper.forwardcs&|jD] }|q tdS)zReset all underlying metrics.N)rrresetr)rr!rrrr3s zMultitaskWrapper.reset)valaxesrc s,|durpt|ts$tdt|tdd|Ds>tdt|t|jkrptdt|dt|jd|dur||n|}g}t |j D]\}\}|dur||nd}t|t r|j ||d \}}n@t|tr|j fd d |D|d \}}ntd t|| ||fq|S) az Plot a single or multiple values from the metric. All tasks' results are plotted on individual axes. 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. axes: Sequence of matplotlib axis objects. If provided, will add the plots to the provided axis objects. If not provided, will create them. Returns: Sequence of tuples with Figure and Axes object for each task. .. plot:: :scale: 75 >>> # Example plotting a single value >>> import torch >>> from torchmetrics.wrappers import MultitaskWrapper >>> from torchmetrics.regression import MeanSquaredError >>> from torchmetrics.classification import BinaryAccuracy >>> >>> classification_target = torch.tensor([0, 1, 0]) >>> regression_target = torch.tensor([2.5, 5.0, 4.0]) >>> targets = {"Classification": classification_target, "Regression": regression_target} >>> >>> classification_preds = torch.tensor([0, 0, 1]) >>> regression_preds = torch.tensor([3.0, 5.0, 2.5]) >>> preds = {"Classification": classification_preds, "Regression": regression_preds} >>> >>> metrics = MultitaskWrapper({ ... "Classification": BinaryAccuracy(), ... "Regression": MeanSquaredError() ... }) >>> metrics.update(preds, targets) >>> value = metrics.compute() >>> fig_, ax_ = metrics.plot(value) .. plot:: :scale: 75 >>> # Example plotting multiple values >>> import torch >>> from torchmetrics.wrappers import MultitaskWrapper >>> from torchmetrics.regression import MeanSquaredError >>> from torchmetrics.classification import BinaryAccuracy >>> >>> classification_target = torch.tensor([0, 1, 0]) >>> regression_target = torch.tensor([2.5, 5.0, 4.0]) >>> targets = {"Classification": classification_target, "Regression": regression_target} >>> >>> classification_preds = torch.tensor([0, 0, 1]) >>> regression_preds = torch.tensor([3.0, 5.0, 2.5]) >>> preds = {"Classification": classification_preds, "Regression": regression_preds} >>> >>> metrics = MultitaskWrapper({ ... "Classification": BinaryAccuracy(), ... "Regression": MeanSquaredError() ... }) >>> values = [] >>> for _ in range(10): ... values.append(metrics(preds, targets)) >>> fig_, ax_ = metrics.plot(values) Nz>Expected argument `axes` to be a Sequence. Found type(axes) = css|]}t|tVqdSr)rr )r-axrrr r/z(MultitaskWrapper.plot..zBExpected each ax in argument `axes` to be a matplotlib axis objectzfExpected argument `axes` to be a Sequence of the same length as the number of tasks.Found len(axes) = z and z tasks)r6csg|] }|qSrr)r-vr(rr r/z)MultitaskWrapper.plot..zWExpected argument `val` to be None or of type Dict or Sequence[Dict]. Found type(val)= )rrrr alllenrr%r+ enumerater&rplotappend) rr4r5Zfig_axsiZ task_metricr6farr9rr>s8D   "r)NN)__name__ __module__ __qualname____doc__Zis_differentiablerstrrr r r staticmethodrrr'rr+r2r3rrr r r> __classcell__rrrrrs A" "* rN)typingrrrrrZtorchrrZtorchmetrics.collectionsr Ztorchmetrics.metricr Ztorchmetrics.utilities.importsr Ztorchmetrics.utilities.plotr r Ztorchmetrics.wrappers.abstractrZ__doctest_skip__rrrrrs