Coverage for webapp/metrics/metrics.py : 96%
Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1import pycountry
2from operator import itemgetter
5def _calculate_colors(countries, max_users):
6 """Calculate the displayed colors for a list of countries depending on the
7 maximum number of users
9 :param countries: List of countries
10 :param max_users: Maximum number of users
12 :returns: The list of countries with a calculated color for each
13 """
14 for country_code in countries:
15 countries[country_code]["color_rgb"] = _calculate_color(
16 countries[country_code]["percentage_of_users"], max_users
17 )
19 return countries
22def _calculate_color(thisCountry, max_users):
23 """Calculate displayed color for a given country
25 :param thisCountry: The country
26 :param max_users: The number of users of the country with
27 the most users
29 :returns: The calculated rgb for the country
30 """
31 colors = [
32 [222, 235, 247],
33 [210, 227, 243],
34 [198, 219, 239],
35 [178, 210, 232],
36 [158, 202, 225],
37 [133, 188, 220],
38 [107, 174, 214],
39 [66, 146, 198],
40 [50, 130, 190],
41 [33, 113, 181],
42 [8, 81, 156],
43 [8, 65, 132],
44 [8, 48, 107],
45 ]
47 buckets = max_users / len(colors)
49 if thisCountry == 0.0:
50 return [218, 218, 218]
52 color_index = int(thisCountry / buckets) - 1
54 if color_index == -1:
55 color_index = 0
57 return colors[color_index]
60def _capitalize_os_name(os_name):
61 """Capitalize OS name
63 :returns: Capitalized OS name (if part of the list)
64 """
65 capitalized_oses = {
66 "solus": "Solus",
67 "raspbian": "Raspbian",
68 "kali": "Kali Linux",
69 "galliumos": "GalliumOS",
70 "opensuse-leap": "openSUSE Leap",
71 "fedora": "Fedora",
72 "ubuntu-core": "Ubuntu Core",
73 "linuxmint": "Linux Mint",
74 "parrot": "Parrot OS",
75 "centos": "CentOS",
76 "ubuntu": "Ubuntu",
77 "arch": "Arch Linux",
78 "debian": "Debian",
79 "elementary": "elementary OS",
80 "neon": "KDE Neon",
81 "manjaro": "Manjaro",
82 "zorin": "Zorin OS",
83 }
85 name, version = os_name.rsplit("/", 1)
87 if version != "-":
88 return " ".join([capitalized_oses.get(name, name), version])
89 else:
90 return capitalized_oses.get(name, name)
93class Metric(object):
94 """This is a basic class for metrics
96 :var name: The name of the metric
97 :var series: The series dictionary from the metric
98 :var buckets: The buckets dictionary from the metric
99 :var status: The status of the metric"""
101 def __init__(self, name, series, buckets, status):
102 self.name = name
103 self.series = series
104 self.buckets = buckets
105 self.status = status
107 def __iter__(self):
108 yield ("name", self.name)
109 yield ("series", self.series)
110 yield ("buckets", self.buckets)
112 def __bool__(self):
113 """Verifies if one of the metrics has no data
115 :return: True if the metric has data, False if not
116 """
118 return self.status == "OK"
121class ActiveDevices(Metric):
122 """Metrics for the active devices.
124 By default the series will be sorted by name.
126 :var name: The name of the metric
127 :var series: The series dictionary from the metric
128 :var buckets: The buckets dictionary from the metric
129 :var status: The status of the metric"""
131 def __init__(self, name, series, buckets, status):
132 series_sorted = sorted(series, key=itemgetter("name"))
134 super().__init__(name, series_sorted, buckets, status)
136 def get_number_latest_active_devices(self):
137 """Get the number of latest active devices from the list of
138 active devices.
140 :returns The number of lastest active devices
141 """
142 latest_active_devices = 0
144 for series_index, series in enumerate(self.series):
145 for index, value in enumerate(series["values"]):
146 if value is None:
147 self.series[series_index]["values"][index] = 0
148 values = series["values"]
149 if len(values) == len(self.buckets):
150 latest_active_devices += values[len(values) - 1]
152 return latest_active_devices
155class CountryDevices(Metric):
156 """Metrics for the devices in countries.
158 :var name: The name of the metric
159 :var series: The series dictionary from the metric
160 :var buckets: The buckets dictionary from the metric
161 :var status: The status of the metric
162 :var private: Boolean, True to add private information
163 displayed for publisher, False if not
164 :var users_by_country: Dictionary with additional metrics per country
165 :var country_data: The metrics on every country"""
167 def __init__(self, name, series, buckets, status, private):
168 super().__init__(name, series, buckets, status)
169 self.private = private
170 self.users_by_country = self._calculate_metrics_countries()
171 self.country_data = self._build_country_info()
173 def get_number_territories(self):
174 """Get the number of territories with users
176 :returns The number of territories with users
177 """
178 territories_total = 0
179 for data in self.country_data.values():
180 if data["number_of_users"] > 0:
181 territories_total += 1
183 return territories_total
185 def _calculate_metrics_countries(self):
186 """Calculate metrics per countries:
187 - Number of users
188 - Percentage of users
189 - Colors to display
191 Output:
192 ```
193 {
194 'FR': {
195 'number_of_users': 1.125,
196 'percentage_of_users': 0.1875,
197 'color_rgb': [8, 64, 129]
198 },
199 ...
200 }
201 ```
203 :returns: The transformed metrics
204 """
205 users_by_country = {}
206 max_users = 0.0
207 for country_counts in self.series:
208 country_code = country_counts["name"]
209 users_by_country[country_code] = {}
210 counts = []
211 for daily_count in country_counts["values"]:
212 if daily_count is not None:
213 counts.append(daily_count)
215 number_of_users = 0
216 percentage_of_users = 0
217 if len(counts) > 0:
218 percentage_of_users = sum(counts) / len(counts)
219 number_of_users = sum(counts)
221 users_by_country[country_code]["number_of_users"] = number_of_users
222 users_by_country[country_code][
223 "percentage_of_users"
224 ] = percentage_of_users
226 if max_users < percentage_of_users:
227 max_users = percentage_of_users
229 metrics_countries = _calculate_colors(users_by_country, max_users)
231 return metrics_countries
233 def _build_country_info(self):
234 """Build information for every country from a subset of information of
235 country.
237 Input:
238 ```
239 {
240 'FR': {
241 'number_of_users': 1.125,
242 'percentage_of_users': 0.1875,
243 'color_rgb': [8, 64, 129]
244 },
245 ...
246 }
247 ```
249 :returns: A dictionary with the country information for every country
250 """
251 if not self.users_by_country:
252 return {}
254 country_data = {}
255 for country in pycountry.countries:
256 country_info = self.users_by_country.get(country.alpha_2)
257 number_of_users = 0
258 percentage_of_users = 0
259 color_rgb = [218, 218, 218]
260 if country_info is not None:
261 if self.private:
262 number_of_users = country_info["number_of_users"] or 0
263 percentage_of_users = country_info["percentage_of_users"] or 0
264 color_rgb = country_info["color_rgb"] or color_rgb
266 # Use common_name if available to be less political
267 # offending (#310)
268 try:
269 country_name = country.common_name
270 except AttributeError:
271 country_name = country.name
273 country_data[country.numeric] = {
274 "name": country_name,
275 "code": country.alpha_2,
276 "percentage_of_users": percentage_of_users,
277 "color_rgb": color_rgb,
278 }
280 if self.private:
281 country_data[country.numeric][
282 "number_of_users"
283 ] = number_of_users
285 return country_data
288class OsMetric(Metric):
289 """Metrics for the devices per os.
291 :var name: The name of the metric
292 :var series: The series dictionary from the metric
293 :var buckets: The buckets dictionary from the metric
294 :var status: The status of the metric
295 :var os: Dictionary with informations per os"""
297 def __init__(self, name, series, buckets, status):
298 super().__init__(name, series, buckets, status)
300 self.os = self._build_os_info()
302 def _build_os_info(self):
303 """Build information for OS distro graph
305 :returns: A list with the sorted distros
306 """
307 oses = []
309 for distro in self.series:
310 if distro["values"][0]:
311 name = _capitalize_os_name(distro["name"])
312 oses.append({"name": name, "value": distro["values"][-1]})
314 oses.sort(key=lambda x: x["value"], reverse=True)
316 return oses