660 lines
28 KiB
Python
660 lines
28 KiB
Python
from __future__ import annotations
|
|
|
|
import csv
|
|
from datetime import datetime
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
import zipfile
|
|
|
|
import main
|
|
from openpyxl import Workbook
|
|
|
|
|
|
COMPLETE_WINDOW = "2026073110001100"
|
|
INCOMPLETE_WINDOW = "2026073111001200"
|
|
TARGET_KEY = "20260731100000"
|
|
|
|
|
|
class PipelineTest(unittest.TestCase):
|
|
def test_latest_group_is_converted_merged_stored_and_archived(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
root = Path(temp) / "source"
|
|
output = Path(temp) / "output"
|
|
windows = ["2026073110001100", "2026073110141100", "2026073110051100"]
|
|
for index, source_type in enumerate(main.EXPECTED_TYPES):
|
|
create_archive(root, source_type, windows[index % len(windows)], middle="任意粒度")
|
|
create_archive(root, main.EXPECTED_TYPES[0], "2026073110091100", middle="更新版本")
|
|
create_cell_data_sources(root)
|
|
store = RecordingStore()
|
|
history = RecordingHistory()
|
|
|
|
result = main.process(main.LocalSource(root), output, lookback_days=3, store=store, history=history)
|
|
|
|
self.assertIsNotNone(result)
|
|
assert result is not None
|
|
self.assertEqual(result.name, TARGET_KEY)
|
|
converted = sorted((result / "converted").glob("*.csv"))
|
|
self.assertEqual(len(converted), 7)
|
|
summary_path = result / f"interference_summary_{TARGET_KEY}.csv"
|
|
summary_bytes = summary_path.read_bytes()
|
|
self.assertTrue(summary_bytes.startswith(b"\xef\xbb\xbf"))
|
|
with summary_path.open(encoding="utf-8-sig", newline="") as file:
|
|
rows = list(csv.DictReader(file))
|
|
self.assertEqual(len(rows), 2)
|
|
self.assertEqual(
|
|
list(rows[0]),
|
|
[
|
|
"metric_time",
|
|
"network_type",
|
|
"cgi",
|
|
"cell_name",
|
|
"interference_dbm",
|
|
"prev_interference_dbm",
|
|
"longitude",
|
|
"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}
|
|
self.assertEqual(rows_by_cgi["460-00-200-1"]["network_type"], "2.6G")
|
|
self.assertEqual(rows_by_cgi["460-00-100-1"]["network_type"], "FDD")
|
|
self.assertTrue(all(row["metric_time"] == "2026-07-31 10:00:00" for row in rows))
|
|
self.assertEqual(rows_by_cgi["460-00-200-1"]["cell_name"], "5G干扰监控-小区")
|
|
self.assertTrue(all(row["interference_dbm"] == "-100.5" for row in rows))
|
|
self.assertTrue(all(row["prev_interference_dbm"] == "" for row in rows))
|
|
self.assertTrue(all(row["prev_nearby_count"] == "" for row in rows))
|
|
self.assertEqual(rows_by_cgi["460-00-200-1"]["longitude"], "113.123456")
|
|
self.assertEqual(rows_by_cgi["460-00-200-1"]["latitude"], "22.654321")
|
|
self.assertEqual(rows_by_cgi["460-00-200-1"]["azimuth"], "30")
|
|
self.assertEqual(rows_by_cgi["460-00-200-1"]["nearby_count"], "0")
|
|
self.assertEqual(rows_by_cgi["460-00-100-1"]["nearby_count"], "0")
|
|
self.assertFalse(rows_by_cgi["460-00-100-1"]["longitude"])
|
|
self.assertEqual(rows_by_cgi["460-00-100-1"]["azimuth"], "0")
|
|
|
|
manifest = json.loads((result / "manifest.json").read_text(encoding="utf-8"))
|
|
self.assertEqual(manifest["source_file_count"], 7)
|
|
self.assertEqual(manifest["cell_data_file_count"], 5)
|
|
self.assertEqual(manifest["summary_rows"], 2)
|
|
self.assertEqual(manifest["threshold_filtered_rows"], 0)
|
|
self.assertEqual(manifest["duplicate_cgi_rows"], 5)
|
|
self.assertEqual(manifest["coordinate_matched_rows"], 1)
|
|
self.assertEqual(manifest["coordinate_unmatched_rows"], 1)
|
|
self.assertEqual(manifest["metric_time"], "2026-07-31 10:00:00")
|
|
self.assertNotIn("hour_start", manifest)
|
|
self.assertNotIn("hour_end", manifest)
|
|
self.assertNotIn("source_types", manifest)
|
|
self.assertTrue(all("source_type" not in item and "source_path" not in item for item in manifest["files"]))
|
|
self.assertIn("2026073110091100", [item["source_window"] for item in manifest["files"]])
|
|
self.assertEqual(manifest["database"]["metric_time"], "2026-07-31 10:00:00")
|
|
self.assertEqual(len(store.rows), 2)
|
|
self.assertEqual(history.path, "/history/2026-07-31/干扰数据处理结果_20260731100000.csv")
|
|
archived_rows = list(csv.DictReader(io.StringIO(history.payload.decode("gbk"))))
|
|
self.assertEqual(list(archived_rows[0]), list(main.SUMMARY_HEADER))
|
|
self.assertEqual(len(archived_rows), 2)
|
|
|
|
def test_interference_thresholds_remove_only_lower_values(self) -> None:
|
|
for source_type, threshold in (
|
|
("5G干扰监控", "-107"),
|
|
("5G下FDD干扰监控", "-107"),
|
|
("700M干扰监控", "-110"),
|
|
("700M下FDD干扰监控", "-110"),
|
|
("SDR_FDD干扰监控", "-110"),
|
|
("SDR_TDD干扰监控", "-110"),
|
|
("反开RD干扰监控", "-110"),
|
|
):
|
|
row = database_row()
|
|
row["interference_dbm"] = threshold
|
|
self.assertTrue(main.passes_interference_threshold(row, source_type))
|
|
row["interference_dbm"] = str(float(threshold) - 0.1)
|
|
self.assertFalse(main.passes_interference_threshold(row, source_type))
|
|
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
root = Path(temp) / "source"
|
|
for source_type in main.EXPECTED_TYPES:
|
|
interference = -107.1 if source_type == main.EXPECTED_TYPES[0] else -100.5
|
|
create_archive(root, source_type, COMPLETE_WINDOW, interference_dbm=interference)
|
|
create_cell_data_sources(root)
|
|
|
|
result = main.process(main.LocalSource(root), Path(temp) / "output", 3)
|
|
manifest = json.loads((result / "manifest.json").read_text(encoding="utf-8"))
|
|
|
|
self.assertEqual(manifest["summary_rows"], 2)
|
|
self.assertEqual(manifest["threshold_filtered_rows"], 1)
|
|
self.assertEqual(manifest["duplicate_cgi_rows"], 4)
|
|
|
|
def test_latest_incomplete_group_waits_without_falling_back(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
root = Path(temp) / "source"
|
|
for source_type in main.EXPECTED_TYPES:
|
|
create_archive(root, source_type, COMPLETE_WINDOW)
|
|
create_archive(root, main.EXPECTED_TYPES[0], INCOMPLETE_WINDOW)
|
|
store = RecordingStore()
|
|
|
|
result = main.process(main.LocalSource(root), Path(temp) / "output", 3, store=store)
|
|
|
|
self.assertIsNone(result)
|
|
self.assertEqual(store.latest_calls, 0)
|
|
self.assertEqual(store.rows, [])
|
|
|
|
def test_no_source_files_waits_successfully(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
root = Path(temp) / "source"
|
|
root.mkdir()
|
|
store = RecordingStore()
|
|
|
|
result = main.process(main.LocalSource(root), Path(temp) / "output", 3, store=store)
|
|
|
|
self.assertIsNone(result)
|
|
self.assertEqual(store.latest_calls, 0)
|
|
|
|
def test_schema_change_is_rejected(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
root = Path(temp) / "source"
|
|
for source_type in main.EXPECTED_TYPES:
|
|
create_archive(root, source_type, COMPLETE_WINDOW, bad_header=source_type == main.EXPECTED_TYPES[0])
|
|
|
|
with self.assertRaisesRegex(main.ProcessingError, "Unexpected Sheet0 header"):
|
|
main.process(main.LocalSource(root), Path(temp) / "output", 3)
|
|
|
|
def test_workbook_row_outside_selected_quarter_is_rejected(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
root = Path(temp) / "source"
|
|
for source_type in main.EXPECTED_TYPES:
|
|
create_archive(
|
|
root,
|
|
source_type,
|
|
COMPLETE_WINDOW,
|
|
row_window="2026073110151100" if source_type == main.EXPECTED_TYPES[0] else "",
|
|
)
|
|
create_cell_data_sources(root)
|
|
|
|
with self.assertRaisesRegex(main.ProcessingError, "is outside metric group"):
|
|
main.process(main.LocalSource(root), Path(temp) / "output", 3)
|
|
|
|
def test_file_name_middle_is_flexible_and_time_uses_natural_quarter(self) -> None:
|
|
candidate = main.parse_candidate(
|
|
"/source/700M干扰监控_任意描述_2026073112141300.zip",
|
|
123,
|
|
)
|
|
|
|
self.assertIsNotNone(candidate)
|
|
assert candidate is not None
|
|
self.assertEqual(candidate.source_type, "700M干扰监控")
|
|
self.assertEqual(candidate.size, 123)
|
|
self.assertEqual(main.metric_time_for_window(candidate.window), datetime(2026, 7, 31, 12, 0))
|
|
self.assertEqual(main.metric_time_for_window("2026073112151300"), datetime(2026, 7, 31, 12, 15))
|
|
|
|
def test_latest_dated_cell_data_file_is_selected(self) -> None:
|
|
entries = [
|
|
{"name": "江门5G小区信息表20260727.xlsx", "path": "/old.xlsx", "is_dir": False},
|
|
{"name": "江门5G小区信息表20260728.xlsx", "path": "/latest.xlsx", "is_dir": False},
|
|
{"name": "说明.txt", "path": "/说明.txt", "is_dir": False},
|
|
]
|
|
|
|
selected = main.select_latest_cell_data_file(entries, "/cell-data")
|
|
|
|
self.assertEqual(selected["path"], "/latest.xlsx")
|
|
|
|
def test_cell_data_schema_change_is_rejected(self) -> None:
|
|
raw = create_cell_data_workbook(include_latitude=False)
|
|
|
|
with self.assertRaisesRegex(main.ProcessingError, "missing columns.*纬度"):
|
|
main.parse_cell_data_workbook(raw, "bad.xlsx")
|
|
|
|
def test_conflicting_cell_data_metadata_are_rejected(self) -> None:
|
|
first = create_cell_data_workbook(longitude=113.1)
|
|
second = create_cell_data_workbook(longitude=113.2)
|
|
|
|
with self.assertRaisesRegex(main.ProcessingError, "Conflicting CellData metadata"):
|
|
main.load_cell_metadata([("first.xlsx", first), ("second.xlsx", second)])
|
|
|
|
def test_empty_cell_data_azimuth_defaults_to_zero(self) -> None:
|
|
metadata = main.parse_cell_data_workbook(create_cell_data_workbook(azimuth=None), "cell-data.xlsx")
|
|
|
|
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", "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)
|
|
|
|
def test_previous_values_are_matched_by_cgi(self) -> None:
|
|
rows = [database_row(), database_row()]
|
|
rows[0]["cgi"] = "matched"
|
|
rows[1]["cgi"] = "new"
|
|
previous = database_row()
|
|
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()]
|
|
rows[0]["cgi"] = "duplicate"
|
|
rows[0]["interference_dbm"] = "-108.5"
|
|
rows[1]["cgi"] = "duplicate"
|
|
rows[1]["interference_dbm"] = "-101.25"
|
|
rows[2]["cgi"] = "unique"
|
|
|
|
removed = main.deduplicate_by_cgi(rows)
|
|
|
|
self.assertEqual(removed, 1)
|
|
self.assertEqual(len(rows), 2)
|
|
self.assertEqual({row["cgi"]: row["interference_dbm"] for row in rows}, {
|
|
"duplicate": "-101.25",
|
|
"unique": "-100.5",
|
|
})
|
|
|
|
def test_gbk_history_csv_normalizes_non_breaking_spaces(self) -> None:
|
|
payload = main.dict_csv_bytes(
|
|
("cell_name",),
|
|
[{"cell_name": "测试\u00a0小区"}],
|
|
main.HISTORY_CSV_ENCODING,
|
|
)
|
|
|
|
self.assertEqual(payload.decode("gbk"), "cell_name\r\n测试 小区\r\n")
|
|
|
|
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")
|
|
|
|
result = store.replace_latest([database_row()])
|
|
|
|
self.assertEqual(result["metric_time"], "2026-07-31 10:00:00")
|
|
self.assertEqual(result["inserted_rows"], 1)
|
|
self.assertEqual(result["refreshed_rows"], 7)
|
|
self.assertEqual(result["old_rows_deleted"], 20)
|
|
queries = [payload["sql"] for endpoint, payload in client.posts if endpoint.endswith("/query")]
|
|
self.assertTrue(any("CREATE DATABASE IF NOT EXISTS `interference_etl`" in query for query in queries))
|
|
self.assertTrue(any("ADD COLUMN `network_type`" in query for query in queries))
|
|
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))
|
|
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, 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"])
|
|
|
|
def test_api_store_refuses_to_replace_a_newer_hour(self) -> None:
|
|
client = FakeApiClient("2026-07-31T11:00:00")
|
|
store = main.ApiSummaryStore(client, "db_share_mysql")
|
|
|
|
with self.assertRaisesRegex(main.ProcessingError, "already contains newer metric time"):
|
|
store.replace_latest([database_row()])
|
|
|
|
self.assertFalse(any(endpoint.endswith("/run-script") for endpoint, _ in client.posts))
|
|
|
|
def test_api_store_propagates_script_failure(self) -> None:
|
|
client = FakeApiClient(None, fail_script=True)
|
|
store = main.ApiSummaryStore(client, "db_share_mysql")
|
|
|
|
with self.assertRaisesRegex(main.ProcessingError, "statement 3: insert failed"):
|
|
store.replace_latest([database_row()])
|
|
|
|
def test_api_store_rejects_invalid_decimal(self) -> None:
|
|
client = FakeApiClient(None)
|
|
store = main.ApiSummaryStore(client, "db_share_mysql")
|
|
row = database_row()
|
|
row["interference_dbm"] = "not-a-number"
|
|
|
|
with self.assertRaisesRegex(main.ProcessingError, "Invalid decimal value for interference_dbm"):
|
|
store.replace_latest([row])
|
|
|
|
def test_api_store_exports_normalized_rows_for_history(self) -> None:
|
|
client = FakeApiClient(None)
|
|
store = main.ApiSummaryStore(client, "db_share_mysql")
|
|
|
|
rows = store.rows_for_time(datetime(2026, 7, 31, 10, 0))
|
|
|
|
self.assertEqual(rows, [database_row()])
|
|
export_query = next(
|
|
payload for endpoint, payload in client.posts
|
|
if endpoint.endswith("/query") and str(payload["sql"]).startswith("SELECT metric_time")
|
|
)
|
|
self.assertEqual(export_query["page_size"], 1000)
|
|
|
|
def test_existing_database_time_repairs_missing_history_without_reprocessing(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp:
|
|
root = Path(temp) / "source"
|
|
for source_type in main.EXPECTED_TYPES:
|
|
create_archive(root, source_type, COMPLETE_WINDOW)
|
|
stored_row = database_row()
|
|
store = RecordingStore(datetime(2026, 7, 31, 10, 0), [stored_row])
|
|
history = RecordingHistory(exists=False)
|
|
|
|
result = main.process(main.LocalSource(root), Path(temp) / "output", 3, store=store, history=history)
|
|
|
|
self.assertIsNone(result)
|
|
self.assertEqual(store.rows_for_time_calls, 1)
|
|
self.assertEqual(history.path, "/history/2026-07-31/干扰数据处理结果_20260731100000.csv")
|
|
archived_rows = list(csv.DictReader(io.StringIO(history.payload.decode("gbk"))))
|
|
self.assertEqual(archived_rows, [stored_row])
|
|
|
|
def test_api_history_store_creates_date_directory_and_uploads_csv(self) -> None:
|
|
client = FakeHistoryApiClient()
|
|
history = main.ApiHistoryStore(client, "stg_test", "/history")
|
|
metric_time = datetime(2026, 8, 6, 12, 0)
|
|
|
|
self.assertFalse(history.exists(metric_time))
|
|
path = history.upload(metric_time, b"csv-data")
|
|
|
|
self.assertEqual(path, "/history/2026-08-06/干扰数据处理结果_20260806120000.csv")
|
|
self.assertEqual(client.mkdir_paths, ["/history/2026-08-06"])
|
|
self.assertEqual(client.uploads, [("/history/2026-08-06", "干扰数据处理结果_20260806120000.csv", b"csv-data")])
|
|
|
|
|
|
class RecordingStore:
|
|
def __init__(self, latest_time: datetime | None = None, stored_rows: list[dict[str, str]] | None = None) -> None:
|
|
self.rows: list[dict[str, str]] = []
|
|
self.latest_time = latest_time
|
|
self.stored_rows = stored_rows or []
|
|
self.latest_calls = 0
|
|
self.rows_for_time_calls = 0
|
|
|
|
def latest_metric_time(self) -> datetime | None:
|
|
self.latest_calls += 1
|
|
return self.latest_time
|
|
|
|
def replace_latest(self, rows: list[dict[str, str]]) -> dict[str, object]:
|
|
self.rows = list(rows)
|
|
return {
|
|
"enabled": True,
|
|
"table": main.DATABASE_TABLE,
|
|
"metric_time": rows[0]["metric_time"],
|
|
"inserted_rows": len(rows),
|
|
"refreshed_rows": 0,
|
|
"old_rows_deleted": 0,
|
|
}
|
|
|
|
def rows_for_time(self, metric_time: datetime) -> list[dict[str, str]]:
|
|
self.rows_for_time_calls += 1
|
|
self.asserted_metric_time = metric_time
|
|
return list(self.stored_rows)
|
|
|
|
|
|
class RecordingHistory:
|
|
def __init__(self, exists: bool = False) -> None:
|
|
self.exists_value = exists
|
|
self.path = ""
|
|
self.payload = b""
|
|
|
|
def exists(self, metric_time: datetime) -> bool:
|
|
self.checked_metric_time = metric_time
|
|
return self.exists_value
|
|
|
|
def upload(self, metric_time: datetime, payload: bytes) -> str:
|
|
self.payload = payload
|
|
self.path = f"/history/{metric_time:%Y-%m-%d}/干扰数据处理结果_{metric_time:%Y%m%d%H%M%S}.csv"
|
|
return self.path
|
|
|
|
|
|
class FakeApiClient:
|
|
def __init__(self, latest_time: str | None, fail_script: bool = False) -> None:
|
|
self.latest_time = latest_time
|
|
self.fail_script = fail_script
|
|
self.posts: list[tuple[str, dict[str, object]]] = []
|
|
|
|
def get_json(self, endpoint: str, query: dict[str, object] | None = None, timeout: int = 30) -> object:
|
|
del endpoint, query, timeout
|
|
return [
|
|
{"name": "metric_time"},
|
|
{"name": "cgi"},
|
|
{"name": "cell_name"},
|
|
{"name": "interference_dbm"},
|
|
]
|
|
|
|
def post_json(self, endpoint: str, payload: dict[str, object], timeout: int = 30) -> object:
|
|
del timeout
|
|
self.posts.append((endpoint, payload))
|
|
if endpoint.endswith("/query"):
|
|
sql = str(payload["sql"]).lstrip()
|
|
if sql.startswith("SELECT MAX"):
|
|
return {"rows": [{"latest_time": self.latest_time}]}
|
|
if sql.startswith("SELECT metric_time"):
|
|
row = database_row()
|
|
row["metric_time"] = "2026-07-31T10:00:00"
|
|
return {"rows": [row], "total": 1}
|
|
return {"affected_rows": 0}
|
|
if self.fail_script:
|
|
return {
|
|
"results": [
|
|
{"index": 1, "ok": True, "affected_rows": 0},
|
|
{"index": 2, "ok": True, "affected_rows": 7},
|
|
{"index": 3, "ok": False, "message": "insert failed", "affected_rows": 0},
|
|
],
|
|
"stopped": True,
|
|
}
|
|
return {
|
|
"results": [
|
|
{"index": 1, "ok": True, "affected_rows": 0},
|
|
{"index": 2, "ok": True, "affected_rows": 7},
|
|
{"index": 3, "ok": True, "affected_rows": 1},
|
|
{"index": 4, "ok": True, "affected_rows": 20},
|
|
{"index": 5, "ok": True, "affected_rows": 0},
|
|
],
|
|
"stopped": False,
|
|
}
|
|
|
|
|
|
class FakeHistoryApiClient:
|
|
def __init__(self) -> None:
|
|
self.mkdir_paths: list[str] = []
|
|
self.uploads: list[tuple[str, str, bytes]] = []
|
|
|
|
def get_json(self, endpoint: str, query: dict[str, object] | None = None, timeout: int = 30) -> object:
|
|
del endpoint, timeout
|
|
path = str((query or {}).get("path") or "")
|
|
if path == "/history":
|
|
return {"entries": []}
|
|
return {"entries": []}
|
|
|
|
def post_json(self, endpoint: str, payload: dict[str, object], timeout: int = 30) -> object:
|
|
del endpoint, timeout
|
|
path = str(payload["path"])
|
|
self.mkdir_paths.append(path)
|
|
return {"path": path}
|
|
|
|
def post_file(
|
|
self,
|
|
endpoint: str,
|
|
query: dict[str, object],
|
|
filename: str,
|
|
payload: bytes,
|
|
timeout: int = 120,
|
|
) -> object:
|
|
del endpoint, timeout
|
|
directory = str(query["path"])
|
|
self.uploads.append((directory, filename, payload))
|
|
return {"path": f"{directory}/{filename}"}
|
|
|
|
|
|
def database_row() -> dict[str, str]:
|
|
return {
|
|
"metric_time": "2026-07-31 10:00:00",
|
|
"network_type": "2.6G",
|
|
"cgi": "460-00-200-1",
|
|
"cell_name": "测试小区",
|
|
"interference_dbm": "-100.5",
|
|
"prev_interference_dbm": "",
|
|
"longitude": "",
|
|
"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, 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
|
|
|
|
|
|
def create_archive(
|
|
root: Path,
|
|
source_type: str,
|
|
window: str,
|
|
bad_header: bool = False,
|
|
interference_dbm: float = -100.5,
|
|
middle: str = "LWP_每小时_过滤110",
|
|
row_window: str = "",
|
|
) -> None:
|
|
date_dir = root / f"{window[:4]}-{window[4:6]}-{window[6:8]}"
|
|
date_dir.mkdir(parents=True, exist_ok=True)
|
|
filename = f"{source_type}_{middle}_{window}"
|
|
workbook = Workbook()
|
|
sheet = workbook.active
|
|
sheet.title = "Sheet0"
|
|
header = list(main.EXPECTED_HEADERS[source_type])
|
|
if bad_header:
|
|
header[-1] = "unexpected"
|
|
sheet.append(header)
|
|
sheet.append(mock_row(source_type, row_window or window, interference_dbm))
|
|
metadata = workbook.create_sheet("指标(计数器)")
|
|
metadata.append(["指标或计数器", "指标或计数器描述", "指标公式", "指标或计数器状态"])
|
|
content = io.BytesIO()
|
|
workbook.save(content)
|
|
workbook.close()
|
|
with zipfile.ZipFile(date_dir / f"{filename}.zip", "w", zipfile.ZIP_DEFLATED) as archive:
|
|
archive.writestr(f"{filename}.xlsx", content.getvalue())
|
|
|
|
|
|
def mock_row(source_type: str, window: str, interference_dbm: float = -100.5) -> list[object]:
|
|
header = main.EXPECTED_HEADERS[source_type]
|
|
values: dict[str, object] = {column: "mock" for column in header}
|
|
values.update(
|
|
{
|
|
"开始时间": datetime.strptime(window[:12], "%Y%m%d%H%M"),
|
|
"粒度": "1 小时",
|
|
"eNodeBId": 100,
|
|
"eNodeBID": 100,
|
|
"gNBId": 200,
|
|
"gNBplmn": "460-00",
|
|
"cellId": 1,
|
|
"小区ID": 1,
|
|
"masterOperatorId": "unused-source-value",
|
|
"E-UTRAN FDD小区名称": f"{source_type}-小区",
|
|
"E-UTRAN TDD小区名称": f"{source_type}-小区",
|
|
"CU小区配置名称": f"{source_type}-小区",
|
|
"小区名称": f"{source_type}-小区",
|
|
"载波平均噪声干扰(dBm)": interference_dbm,
|
|
"小区上行平均干扰电平(dBm)": interference_dbm,
|
|
}
|
|
)
|
|
return [values[column] for column in header]
|
|
|
|
|
|
def create_cell_data_sources(root: Path) -> None:
|
|
raw = create_cell_data_workbook()
|
|
for index, configured_directory in enumerate(main.CELL_DATA_DIRECTORIES, start=1):
|
|
directory = root / Path(configured_directory.lstrip("/"))
|
|
directory.mkdir(parents=True, exist_ok=True)
|
|
(directory / f"江门小区信息表{index}-20260728.xlsx").write_bytes(raw)
|
|
|
|
|
|
def create_cell_data_workbook(
|
|
longitude: float = 113.123456,
|
|
include_latitude: bool = True,
|
|
azimuth: float | None = 30,
|
|
) -> bytes:
|
|
workbook = Workbook()
|
|
sheet = workbook.active
|
|
sheet.title = "小区信息表"
|
|
header = ["小区名称", "eNB/gNB", "CI", "经度"]
|
|
if include_latitude:
|
|
header.append("纬度")
|
|
header.append("方向角")
|
|
sheet.append(header)
|
|
row: list[object] = ["测试小区", 200, 1, longitude]
|
|
if include_latitude:
|
|
row.append(22.654321)
|
|
row.append(azimuth)
|
|
sheet.append(row)
|
|
content = io.BytesIO()
|
|
workbook.save(content)
|
|
workbook.close()
|
|
return content.getvalue()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|