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-03 15:25:51 +08:00
parent 9776391ad1
commit 60e5c1ce6a
4 changed files with 71 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main.py
+38 -22
View File
@@ -160,6 +160,7 @@ SUMMARY_HEADER = (
"interference_dbm",
"longitude",
"latitude",
"azimuth",
)
class ProcessingError(RuntimeError):
@@ -270,6 +271,7 @@ class ApiSummaryStore:
interference_dbm DECIMAL(10,3) NOT NULL,
longitude DECIMAL(10,6) NULL,
latitude DECIMAL(10,6) NULL,
azimuth DECIMAL(6,2) NOT NULL DEFAULT 0,
PRIMARY KEY (metric_time, cgi)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
""",
@@ -282,10 +284,15 @@ class ApiSummaryStore:
if not isinstance(columns, list):
raise ProcessingError("Metrix Database API returned invalid column metadata")
existing_columns = {str(item.get("name")) for item in columns if isinstance(item, dict)}
for column in ("longitude", "latitude"):
migrations = {
"longitude": "DECIMAL(10,6) NULL",
"latitude": "DECIMAL(10,6) NULL",
"azimuth": "DECIMAL(6,2) NOT NULL DEFAULT 0",
}
for column, definition in migrations.items():
if column not in existing_columns:
self._query(
f"ALTER TABLE `{self.table}` ADD COLUMN `{column}` DECIMAL(10,6) NULL",
f"ALTER TABLE `{self.table}` ADD COLUMN `{column}` {definition}",
database=DATABASE_NAME,
)
@@ -311,7 +318,7 @@ class ApiSummaryStore:
START TRANSACTION;
DELETE FROM `{self.table}` WHERE metric_time = {metric_literal};
INSERT INTO `{self.table}`
(metric_time, cgi, cell_name, interference_dbm, longitude, latitude)
(metric_time, cgi, cell_name, interference_dbm, longitude, latitude, azimuth)
VALUES
{values};
DELETE FROM `{self.table}` WHERE metric_time <> {metric_literal};
@@ -364,6 +371,7 @@ class ApiSummaryStore:
sql_decimal_literal(row["interference_dbm"], "interference_dbm"),
sql_decimal_literal(row["longitude"], "longitude", nullable=True),
sql_decimal_literal(row["latitude"], "latitude", nullable=True),
sql_decimal_literal(row["azimuth"], "azimuth"),
)
) + ")"
@@ -475,7 +483,7 @@ def select_latest_cell_data_file(entries: list[dict[str, object]], directory: st
return max(dated_files, key=lambda entry: (entry[0], entry[1]))[2]
def parse_cell_data_workbook(raw_xlsx: bytes, path: str) -> dict[str, tuple[str, str]]:
def parse_cell_data_workbook(raw_xlsx: bytes, path: str) -> dict[str, tuple[str, str, str]]:
try:
with warnings.catch_warnings():
warnings.filterwarnings("ignore", message="Workbook contains no default style")
@@ -490,39 +498,46 @@ def parse_cell_data_workbook(raw_xlsx: bytes, path: str) -> dict[str, tuple[str,
header = tuple(normalize_cell(value) for value in next(rows))
except StopIteration as exc:
raise ProcessingError(f"CellData workbook is empty: {path}") from exc
required_columns = ("eNB/gNB", "CI", "经度", "纬度")
required_columns = ("eNB/gNB", "CI", "经度", "纬度", "方向角")
missing = [column for column in required_columns if column not in header]
if missing:
raise ProcessingError(f"CellData workbook missing columns {missing}: {path}")
indexes = {column: header.index(column) for column in required_columns}
coordinates: dict[str, tuple[str, str]] = {}
metadata: dict[str, tuple[str, str, str]] = {}
for row_number, row in enumerate(rows, start=2):
node = normalize_cell(row[indexes["eNB/gNB"]])
cell = normalize_cell(row[indexes["CI"]])
longitude = normalize_cell(row[indexes["经度"]])
latitude = normalize_cell(row[indexes["纬度"]])
if not node or not cell or not longitude or not latitude:
azimuth = normalize_cell(row[indexes["方向角"]]) or "0"
if not node or not cell:
continue
try:
azimuth_number = Decimal(azimuth)
except InvalidOperation as exc:
raise ProcessingError(f"Invalid CellData azimuth {azimuth!r} in {path}, row {row_number}") from exc
if not azimuth_number.is_finite():
raise ProcessingError(f"Invalid CellData azimuth {azimuth!r} in {path}, row {row_number}")
cgi = f"{LTE_PLMN}-{node}-{cell}"
value = (longitude, latitude)
existing = coordinates.get(cgi)
value = (longitude, latitude, format(azimuth_number, "f"))
existing = metadata.get(cgi)
if existing is not None and existing != value:
raise ProcessingError(f"Conflicting CellData coordinates for {cgi} in {path}, row {row_number}")
coordinates[cgi] = value
return coordinates
raise ProcessingError(f"Conflicting CellData metadata for {cgi} in {path}, row {row_number}")
metadata[cgi] = value
return metadata
finally:
workbook.close()
def load_cell_coordinates(workbooks: list[tuple[str, bytes]]) -> dict[str, tuple[str, str]]:
coordinates: dict[str, tuple[str, str]] = {}
def load_cell_metadata(workbooks: list[tuple[str, bytes]]) -> dict[str, tuple[str, str, str]]:
metadata: dict[str, tuple[str, str, str]] = {}
for path, raw_xlsx in workbooks:
for cgi, value in parse_cell_data_workbook(raw_xlsx, path).items():
existing = coordinates.get(cgi)
existing = metadata.get(cgi)
if existing is not None and existing != value:
raise ProcessingError(f"Conflicting CellData coordinates for {cgi} across workbooks")
coordinates[cgi] = value
return coordinates
raise ProcessingError(f"Conflicting CellData metadata for {cgi} across workbooks")
metadata[cgi] = value
return metadata
def select_window(candidates: list[Candidate], requested: str = "") -> tuple[str, dict[str, Candidate], list[str]]:
@@ -598,7 +613,7 @@ def process(
candidates = source.candidates(lookback_days)
window, selected, warnings_out = select_window(candidates, requested_window)
cell_data_workbooks = source.cell_data_workbooks()
cell_coordinates = load_cell_coordinates(cell_data_workbooks)
cell_metadata = load_cell_metadata(cell_data_workbooks)
temp_dir = output_root.resolve() / f".{window}.tmp-{os.getpid()}"
final_dir = output_root.resolve() / window
ensure_scoped(output_root.resolve(), temp_dir)
@@ -620,7 +635,7 @@ def process(
for row_number, row in enumerate(rows, start=2):
record = dict(zip(header, row, strict=True))
summary_rows.append(
summary_record(record, source_type, candidate.path, window, row_number, cell_coordinates)
summary_record(record, source_type, candidate.path, window, row_number, cell_metadata)
)
manifest_files.append(
{
@@ -685,7 +700,7 @@ def summary_record(
source_path: str,
window: str,
row_number: int,
cell_coordinates: dict[str, tuple[str, str]] | None = None,
cell_metadata: dict[str, tuple[str, str, str]] | None = None,
) -> dict[str, str]:
hour_start = normalize_cell(record["开始时间"])
expected_start, hour_end = window_bounds(window)
@@ -709,7 +724,7 @@ def summary_record(
cell_name = required(record, "E-UTRAN FDD小区名称", source_path, row_number)
interference = required(record, "载波平均噪声干扰(dBm)", source_path, row_number)
longitude, latitude = (cell_coordinates or {}).get(cgi, ("", ""))
longitude, latitude, azimuth = (cell_metadata or {}).get(cgi, ("", "", "0"))
return {
"hour_start": hour_start,
"hour_end": hour_end,
@@ -718,6 +733,7 @@ def summary_record(
"interference_dbm": interference,
"longitude": longitude,
"latitude": latitude,
"azimuth": azimuth,
}