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-06 15:27:13 +08:00
parent 5f07fe361d
commit 3e8c0f335c
4 changed files with 105 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_pipeline.py
+30 -1
View File
@@ -48,6 +48,7 @@ class PipelineTest(unittest.TestCase):
"longitude",
"latitude",
"azimuth",
"nearby_count",
],
)
rows_by_type = {row["cell_name"].removesuffix("-小区"): row for row in rows}
@@ -62,8 +63,11 @@ class PipelineTest(unittest.TestCase):
self.assertEqual(rows_by_type["5G干扰监控"]["longitude"], "113.123456")
self.assertEqual(rows_by_type["5G干扰监控"]["latitude"], "22.654321")
self.assertEqual(rows_by_type["5G干扰监控"]["azimuth"], "30")
self.assertEqual(rows_by_type["5G干扰监控"]["nearby_count"], "1")
self.assertEqual(rows_by_type["700M干扰监控"]["nearby_count"], "1")
self.assertTrue(all(not row["longitude"] and not row["latitude"] for row in rows if row["cgi"] == "460-00-100-1"))
self.assertTrue(all(row["azimuth"] == "0" for row in rows if row["cgi"] == "460-00-100-1"))
self.assertTrue(all(row["nearby_count"] == "0" for row in rows if row["cgi"] == "460-00-100-1"))
manifest = json.loads((result / "manifest.json").read_text(encoding="utf-8"))
self.assertEqual(manifest["source_file_count"], 7)
@@ -153,6 +157,21 @@ class PipelineTest(unittest.TestCase):
self.assertEqual(metadata["460-00-200-1"], ("113.123456", "22.654321", "0"))
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", "", ""),
]
main.populate_nearby_counts(rows)
self.assertEqual([row["nearby_count"] for row in rows], ["2", "2", "0", "2", "0"])
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)
def test_api_store_replaces_same_hour_and_deletes_other_hours(self) -> None:
client = FakeApiClient("2026-07-31T10:00:00")
store = main.ApiSummaryStore(client, "db_share_mysql")
@@ -169,10 +188,11 @@ class PipelineTest(unittest.TestCase):
self.assertTrue(any("ADD COLUMN `longitude`" in query for query in queries))
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))
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, 0", script["content"])
self.assertIn("NULL, NULL, 0, 0", script["content"])
self.assertIn("CONVERT(0x", script["content"])
self.assertNotIn("source_type", script["content"])
self.assertNotIn("source_path", script["content"])
@@ -273,9 +293,18 @@ def database_row() -> dict[str, str]:
"longitude": "",
"latitude": "",
"azimuth": "0",
"nearby_count": "0",
}
def nearby_row(cgi: str, longitude: str, latitude: str) -> dict[str, str]:
row = database_row()
row["cgi"] = cgi
row["longitude"] = longitude
row["latitude"] = latitude
return row
def create_archive(
root: Path,
source_type: str,