Metrix/InterferenceETL
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 邻近高干扰小区数按制式拆分统计

Browse Source
This commit is contained in:
nixevol
2026-08-11 16:59:48 +08:00
parent eca4c5cbb1
commit 3835a0b928
4 changed files with 126 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_pipeline.py
+57 -7
View File
@@ -57,7 +57,15 @@ class PipelineTest(unittest.TestCase):
"latitude",
"azimuth",
"nearby_count",
"nearby_26g",
"nearby_700m",
"nearby_tdd",
"nearby_fdd",
"prev_nearby_count",
"prev_nearby_26g",
"prev_nearby_700m",
"prev_nearby_tdd",
"prev_nearby_fdd",
],
)
rows_by_cgi = {row["cgi"]: row for row in rows}
@@ -220,16 +228,34 @@ class PipelineTest(unittest.TestCase):
def test_nearby_count_uses_high_interference_rows_with_coordinates(self) -> None:
rows = [
nearby_row("a", "113.000000", "22.000000"),
nearby_row("b", "113.005000", "22.000000"),
nearby_row("c", "113.020000", "22.000000"),
nearby_row("d", "113.000000", "22.000000"),
nearby_row("e", "", ""),
nearby_row("a", "113.000000", "22.000000", "2.6G"),
nearby_row("b", "113.005000", "22.000000", "700M"),
nearby_row("c", "113.020000", "22.000000", "TDD"),
nearby_row("d", "113.000000", "22.000000", "FDD"),
nearby_row("e", "", "", "TDD"),
]
main.populate_nearby_counts(rows)
self.assertEqual([row["nearby_count"] for row in rows], ["2", "2", "0", "2", "0"])
by_cgi = {row["cgi"]: row for row in rows}
self.assertEqual(
[by_cgi["a"][column] for column in ("nearby_26g", "nearby_700m", "nearby_tdd", "nearby_fdd")],
["0", "1", "0", "1"],
)
self.assertEqual(
[by_cgi["b"][column] for column in ("nearby_26g", "nearby_700m", "nearby_tdd", "nearby_fdd")],
["1", "0", "0", "1"],
)
self.assertEqual(
[by_cgi["d"][column] for column in ("nearby_26g", "nearby_700m", "nearby_tdd", "nearby_fdd")],
["1", "1", "0", "0"],
)
for row in rows:
per_type_total = sum(
int(row[column]) for column in ("nearby_26g", "nearby_700m", "nearby_tdd", "nearby_fdd")
)
self.assertEqual(per_type_total, int(row["nearby_count"]))
self.assertLess(main.haversine_distance_km(113.0, 22.0, 113.005, 22.0), 1.0)
self.assertGreater(main.haversine_distance_km(113.0, 22.0, 113.02, 22.0), 1.0)
@@ -241,13 +267,25 @@ class PipelineTest(unittest.TestCase):
previous["cgi"] = "matched"
previous["interference_dbm"] = "-106.25"
previous["nearby_count"] = "8"
previous["nearby_26g"] = "3"
previous["nearby_700m"] = "2"
previous["nearby_tdd"] = "2"
previous["nearby_fdd"] = "1"
main.populate_previous_values(rows, [previous])
self.assertEqual(rows[0]["prev_interference_dbm"], "-106.25")
self.assertEqual(rows[0]["prev_nearby_count"], "8")
self.assertEqual(rows[0]["prev_nearby_26g"], "3")
self.assertEqual(rows[0]["prev_nearby_700m"], "2")
self.assertEqual(rows[0]["prev_nearby_tdd"], "2")
self.assertEqual(rows[0]["prev_nearby_fdd"], "1")
self.assertEqual(rows[1]["prev_interference_dbm"], "")
self.assertEqual(rows[1]["prev_nearby_count"], "")
self.assertEqual(rows[1]["prev_nearby_26g"], "")
self.assertEqual(rows[1]["prev_nearby_700m"], "")
self.assertEqual(rows[1]["prev_nearby_tdd"], "")
self.assertEqual(rows[1]["prev_nearby_fdd"], "")
def test_duplicate_cgi_keeps_maximum_interference(self) -> None:
rows = [database_row(), database_row(), database_row()]
@@ -292,12 +330,15 @@ class PipelineTest(unittest.TestCase):
self.assertTrue(any("ADD COLUMN `latitude`" in query for query in queries))
self.assertTrue(any("ADD COLUMN `azimuth`" in query for query in queries))
self.assertTrue(any("ADD COLUMN `nearby_count`" in query for query in queries))
for column in ("nearby_26g", "nearby_700m", "nearby_tdd", "nearby_fdd"):
self.assertTrue(any(f"ADD COLUMN `{column}`" in query for query in queries))
self.assertTrue(any(f"ADD COLUMN `prev_{column}`" in query for query in queries))
self.assertTrue(any("ADD COLUMN `prev_interference_dbm`" in query for query in queries))
self.assertTrue(any("ADD COLUMN `prev_nearby_count`" in query for query in queries))
script = next(payload for endpoint, payload in client.posts if endpoint.endswith("/run-script"))
self.assertTrue(script["single_session"])
self.assertIn("START TRANSACTION", script["content"])
self.assertIn("NULL, NULL, NULL, 0, 0, NULL", script["content"])
self.assertIn("NULL, NULL, NULL, 0, 0, 0, 0, 0, 0, NULL, NULL, NULL, NULL, NULL", script["content"])
self.assertIn("CONVERT(0x", script["content"])
self.assertNotIn("source_type", script["content"])
self.assertNotIn("source_path", script["content"])
@@ -507,15 +548,24 @@ def database_row() -> dict[str, str]:
"latitude": "",
"azimuth": "0",
"nearby_count": "0",
"nearby_26g": "0",
"nearby_700m": "0",
"nearby_tdd": "0",
"nearby_fdd": "0",
"prev_nearby_count": "",
"prev_nearby_26g": "",
"prev_nearby_700m": "",
"prev_nearby_tdd": "",
"prev_nearby_fdd": "",
}
def nearby_row(cgi: str, longitude: str, latitude: str) -> dict[str, str]:
def nearby_row(cgi: str, longitude: str, latitude: str, network_type: str = "2.6G") -> dict[str, str]:
row = database_row()
row["cgi"] = cgi
row["longitude"] = longitude
row["latitude"] = latitude
row["network_type"] = network_type
return row