Implement retransmission checker, include unit test

2026-02-26 14:17:28 +01:00
parent fe464ec6d0
commit 707cb560d5
8 changed files with 228 additions and 77 deletions
@@ -38,3 +38,97 @@ class ContainsString:
 				if re.search(needle, line):
 					return True, ""
 		return False, f"could not find '{needle}' in logs"
+
+class RetxCheck():
+	def _parseList(s):
+		if not s:
+			return []
+		return [int(n) for n in s.split(",")]
+
+	def _get_ul_dl(s):
+		dl, ul = (None, None)
+		for o in s.split(";", 1):
+			if o.startswith("dl="):
+				dl = o[3:]
+			elif o.startswith("ul="):
+				ul = o[3:]
+			else:
+				raise ValueError(f"unrecognized option {s}: needs format '[ul=,dl=]X,Y,...[;ul=Z,V,...]'")
+		return dl, ul
+
+	def _parseRetxCheckers(options):
+		if not "=" in options:
+			dlul = RetxCheck._parseList(options)
+			return dlul, dlul
+		dl, ul = RetxCheck._get_ul_dl(options)
+		return RetxCheck._parseList(dl), RetxCheck._parseList(ul)
+
+	def _analyzeUeRetx(rounds, checkers, regex):
+		perc = list(0 for i in checkers) # results in %
+		stats = list(False for i in checkers) # status if succeeded
+		tmp = re.match(regex, rounds)
+		if tmp is None:
+			logging.error('_analyzeUeRetx: did not match regex for DL retx analysis')
+			return stats
+		retx_data = [float(x) for x in tmp.groups()]
+		for i in range(0, len(perc)):
+			#case where numerator > denumerator with denum ==0 is disregarded, cannot hapen in principle, will lead to 0%
+			perc[i] = 0 if (retx_data[i] == 0) else 100 * retx_data[i + 1] / retx_data[i]
+			#treating % > 100 , % > requirement
+			stats[i] = perc[i] <= 100 and perc[i] <= checkers[i]
+		return stats
+
+	def _filter_retx_rounds(l):
+		# map to round number if False, else None
+		l = [i+1 if x is False else None for i, x in enumerate(l)]
+		# filter out None
+		return [x for x in l if x is not None]
+
+	def run(file, options):
+		# aggregate statistics per UE
+		dlsch_ulsch_stats = {}
+		status = True
+		with open(file, "r") as f:
+			for line in f.readlines():
+				keys = {'dlsch_rounds','ulsch_rounds'}
+				for k in keys:
+					result = re.search(k, line)
+					if result is None:
+						continue
+					result = re.search('UE (?:RNTI )?([0-9a-f]{4})', line)
+					if result is None:
+						logging.error(f'did not find RNTI while matching key {k}')
+						status = False
+						continue
+					rnti = result.group(1)
+					if not rnti in dlsch_ulsch_stats:
+						dlsch_ulsch_stats[rnti] = {}
+					dlsch_ulsch_stats[rnti][k]=re.sub(r'^.*\]\s+', r'' , line.rstrip())
+
+		dl, ul = RetxCheck._parseRetxCheckers(options)
+		msgs = []
+		for ue in dlsch_ulsch_stats:
+			dlulstat = dlsch_ulsch_stats[ue]
+
+			for key in dlulstat:
+				msgs.append(dlulstat[key])
+
+			retx_dl = RetxCheck._analyzeUeRetx(dlulstat['dlsch_rounds'], dl, r'^.*dlsch_rounds\s+(\d+)\/(\d+)\/(\d+)\/(\d+),\s+dlsch_errors\s+(\d+)')
+			retx_ul = RetxCheck._analyzeUeRetx(dlulstat['ulsch_rounds'], ul, r'^.*ulsch_rounds\s+(\d+)\/(\d+)\/(\d+)\/(\d+),\s+ulsch_errors\s+(\d+)')
+
+			if False in retx_dl or False in retx_ul:
+				dlstr = RetxCheck._filter_retx_rounds(retx_dl)
+				ulstr = RetxCheck._filter_retx_rounds(retx_ul)
+				msg = f"UE {ue}: retx rounds"
+				if dlstr:
+					msg = f"{msg} DL {dlstr}"
+				if ulstr:
+					msg = f"{msg} UL {ulstr}"
+				msg = f"{msg} exceeded threshold"
+				msgs.append('!!! Failure: ' + msg)
+				logging.error(f'\u001B[1;37;41m {msg}\u001B[0m')
+				status = False
+			else:
+				msg = f"UE {ue} retransmissions within bounds"
+				logging.debug(msg)
+		return status, "\n".join(msgs)