Skip to content

Utils

danling.metrics.utils

flist

Bases: list

Python list that support __format__ and to.

Source code in danling/metrics/utils.py
Python
31
32
33
34
35
36
37
38
class flist(list):
    r"""Python `list` that support `__format__` and `to`."""

    def to(self, *args, **kwargs):
        return flist(i.to(*args, **kwargs) for i in self)

    def __format__(self, *args, **kwargs):
        return " ".join([x.__format__(*args, **kwargs) for x in self])

MetricsDict

Bases: DefaultDict

A MetricsDict for better support for AverageMeters.

Source code in danling/metrics/utils.py
Python
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
class MetricsDict(DefaultDict):
    r"""
    A `MetricsDict` for better support for `AverageMeters`.
    """

    def value(self) -> FlatDict[str, float]:
        return FlatDict({key: metric.value() for key, metric in self.all_items()})

    def batch(self) -> FlatDict[str, float]:
        return FlatDict({key: metric.batch() for key, metric in self.all_items()})

    def average(self) -> FlatDict[str, float]:
        return FlatDict({key: metric.average() for key, metric in self.all_items()})

    @property
    def val(self) -> FlatDict[str, float]:
        return self.value()

    @property
    def bat(self) -> FlatDict[str, float]:
        return self.batch()

    @property
    def avg(self) -> FlatDict[str, float]:
        return self.average()

    def reset(self) -> None:
        for metric in self.all_values():
            metric.reset()

    def __format__(self, format_spec) -> str:
        return "\t".join(f"{key}: {metric.__format__(format_spec)}" for key, metric in self.all_items())

MultiTaskDict

Bases: NestedDict

A MultiTaskDict for better multi-task support For MultiTaskAverageMeters and MultiTaskMetrics.

Source code in danling/metrics/utils.py
Python
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
class MultiTaskDict(NestedDict):
    r"""
    A `MultiTaskDict` for better multi-task support For `MultiTaskAverageMeters` and `MultiTaskMetrics`.
    """

    return_average: bool

    def __init__(self, *args, return_average: bool = False, **kwargs) -> None:
        super().__init__(*args, **kwargs)
        self.setattr("return_average", return_average)

    def value(self) -> NestedDict[str, float]:
        output = NestedDict({key: metric.value() for key, metric in self.all_items()})
        if self.getattr("return_average", False):
            average = DefaultDict(default_factory=list)
            for key, metric in output.all_items():
                average[key.rsplit(".", 1)[-1]].append(metric)
            output["average"] = NestedDict({key: sum(values) / len(values) for key, values in average.items()})
        return output

    def batch(self) -> NestedDict[str, float]:
        output = NestedDict({key: metric.batch() for key, metric in self.all_items()})
        if self.getattr("return_average", False):
            average = DefaultDict(default_factory=list)
            for key, metric in output.all_items():
                average[key.rsplit(".", 1)[-1]].append(metric)
            output["average"] = NestedDict({key: sum(values) / len(values) for key, values in average.items()})
        return output

    def average(self) -> NestedDict[str, float]:
        output = NestedDict({key: metric.average() for key, metric in self.all_items()})
        if self.getattr("return_average", False):
            average = DefaultDict(default_factory=list)
            for key, metric in output.all_items():
                average[key.rsplit(".", 1)[-1]].append(metric)
            output["average"] = NestedDict({key: sum(values) / len(values) for key, values in average.items()})
        return output

    @property
    def val(self) -> NestedDict[str, float]:
        return self.value()

    @property
    def bat(self) -> NestedDict[str, float]:
        return self.batch()

    @property
    def avg(self) -> NestedDict[str, float]:
        return self.average()

    def reset(self) -> None:
        for metric in self.all_values():
            metric.reset()

    def __format__(self, format_spec) -> str:
        return "\n".join(f"{key}: {metric.__format__(format_spec)}" for key, metric in self.all_items())

get_world_size

Python
1
get_world_size() -> int

Return the number of processes in the current process group.

Source code in danling/metrics/utils.py
Python
24
25
26
27
28
def get_world_size() -> int:
    r"""Return the number of processes in the current process group."""
    if dist.is_available() and dist.is_initialized():
        return dist.get_world_size()
    return 1