o .wi[C@sddlmZmZddlmZddlmZmZmZddl m Z m Z ddl m Z ddlmZddlmZddlmZmZdd lmZesEd gZGd d d eZd S))IterableSequence)deepcopy)AnyOptionalUnion)Tensornn)MetricCollection)Metric)_MATPLOTLIB_AVAILABLE)_AX_TYPE_PLOT_OUT_TYPE) WrapperMetricMultitaskWrapper.plotc seZdZUdZdZeed<  d*deee e e ffde ede eddffd d Z d+d edeeeejffd dZd+d edeefddZd+d edeejfddZdeeefdeeefddfddZdeeefdeeeffddZdeeeffddZdeeefdeeefdeeeffddZd,fdd Zed e ed!ede efd"d#Zd*de ede eddfd$d%Z d*d&e e eeefd'e ee dee!fd(d)Z"Z#S)-MultitaskWrapperahWrapper 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 standard `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. prefix: A string to append in front of the metric keys. If not provided, will default to an empty string. postfix: A string to append after the keys of the output dict. If not provided, will default to an empty string. .. tip:: The use prefix and postfix allows for easily creating task wrappers for training, validation and test. The arguments are only changing the output keys of the computed metrics and not the input keys. This means that a ``MultitaskWrapper`` initialized as ``MultitaskWrapper({"task": Metric()}, prefix="train_")`` will still expect the input to be a dictionary with the key "task", but the output will be a dictionary with the key "train_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` ValueError: If `prefix` is not a string ValueError: If `postfix` is not a string 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)}} Example (with a prefix and postfix): >>> 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() ... }, prefix="train_") >>> metrics.update(preds, targets) >>> metrics.compute() {'train_Classification': tensor(0.3333), 'train_Regression': tensor(0.8333)} Fis_differentiableN task_metricsprefixpostfixreturncstt|tstd||D]}t|ttfs'tdt|qt ||_ |dur>t|t s>t d||pAd|_|durSt|t sSt d||pVd|_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 zCExpected argument `prefix` to either be `None` or a string but got zDExpected argument `postfix` to either be `None` or a string but got )super__init__ isinstancedict TypeErrorvaluesr r typer ModuleDictrstr ValueError_prefix_postfix)selfrrrmetric __class__\/home/ubuntu/sommelier/.venv/lib/python3.10/site-packages/torchmetrics/wrappers/multitask.pyrs$     zMultitaskWrapper.__init__Tflattenccsv|jD]2\}}|r+t|tr+|D]\}}|j|d||j|fVqq|j||j|fVqdS)zIterate over task and task metrics. Args: flatten: If True, will iterate over all sub-metrics in the case of a MetricCollection. If False, will iterate over the task names and the corresponding metrics. _Nritemsrr r"r#)r$r* task_namer%sub_metric_name sub_metricr(r(r)r-s"zMultitaskWrapper.itemsccsf|jD]*\}}|r%t|tr%|D]}|j|d||jVqq|j||jVqdS)zIterate over task names. Args: flatten: If True, will iterate over all sub-metrics in the case of a MetricCollection. If False, will iterate over the task names and the corresponding metrics. r+Nr,)r$r*r.r%r/r(r(r)keysszMultitaskWrapper.keysccs:|jD]}|rt|tr|EdHq|VqdS)zIterate over task metrics. Args: flatten: If True, will iterate over all sub-metrics in the case of a MetricCollection. If False, will iterate over the task names and the corresponding metrics. N)rrrr )r$r*r%r(r(r)rs zMultitaskWrapper.values task_preds task_targetscCs|j|kr|ks&ntd|d|d|j|jD]\}}||}||}|||q+dS)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)rr1r!r-update)r$r2r3r.r%predtargetr(r(r)r4s$zMultitaskWrapper.updateoutputcsfdd|DS)zYConvert the output of the underlying metrics to a dictionary with the task names as keys.cs&i|]\}}j|j|qSr()r"r#).0r. task_outputr$r(r) s&z4MultitaskWrapper._convert_output..)r-)r$r7r(r:r)_convert_outputsz MultitaskWrapper._convert_outputcCs|dd|jDS)zCompute metrics for all tasks.cSsi|] \}}||qSr()computer8r.r%r(r(r)r;sz,MultitaskWrapper.compute..r<rr-r:r(r(r)r=szMultitaskWrapper.computecs |fdd|jDS)zTCall underlying forward methods for all tasks and return the result as a dictionary.cs$i|]\}}||||qSr(r(r>r2r3r(r)r;sz,MultitaskWrapper.forward..r?)r$r2r3r(r@r)forwardszMultitaskWrapper.forwardcs&|jD]}|qtdS)zReset all underlying metrics.N)rrresetr)r$r%r&r(r)rBs zMultitaskWrapper.resetargnamecCs.|dus t|tr |Std|dt|)NzExpected input `z` to be a string, but got )rr r!r)rCrDr(r(r) _check_argszMultitaskWrapper._check_argcCs0t|}||dp d|_||dpd|_|S)zMake a copy of the metric. Args: prefix: a string to append in front of the metric keys postfix: a string to append after the keys of the output dict. rr)rrEr"r#)r$rrmultitask_copyr(r(r)cloneszMultitaskWrapper.clonevalaxesc s*|dur8t|tstdt|tdd|Dstdt|t|jkr8tdt|dt|jd|dur>|n|}g}t |j D]G\}\}|durY||nd}t|t rl|j ||d \}}nt|tr|j fd d |D|d \}}n td t|| ||fqK|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|tVqdS)N)rr )r8axr(r(r) Wsz(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)rJcsg|]}|qSr(r()r8vr.r(r) gsz)MultitaskWrapper.plot..zWExpected argument `val` to be None or of type Dict or Sequence[Dict]. Found type(val)= )rrrralllenrr!r= enumerater-rplotappend) r$rHrIfig_axsi task_metricrJfar(rMr)rRs8D   "r)NN)T)rN)$__name__ __module__ __qualname____doc__rbool__annotations__rr rr r rrrtupler Moduler-r1rrr4r<r=rrArB staticmethodrErGrr rrR __classcell__r(r(r&r)rsB  e"&".    rN)collections.abcrrcopyrtypingrrrtorchrr torchmetrics.collectionsr torchmetrics.metricr torchmetrics.utilities.importsr torchmetrics.utilities.plotr rtorchmetrics.wrappers.abstractr__doctest_skip__rr(r(r(r)s