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-07 09:22:26 +08:00
parent 1c752e0f75
commit 5e4b58255f
4 changed files with 65 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main.py
+30 -2
View File
@@ -179,10 +179,12 @@ SUMMARY_HEADER = (
"cgi",
"cell_name",
"interference_dbm",
"prev_interference_dbm",
"longitude",
"latitude",
"azimuth",
"nearby_count",
"prev_nearby_count",
)
class ProcessingError(RuntimeError):
@@ -348,7 +350,7 @@ class ApiSummaryStore:
START TRANSACTION;
DELETE FROM `{self.table}` WHERE metric_time = {metric_literal};
INSERT INTO `{self.table}`
(metric_time, network_type, cgi, cell_name, interference_dbm, longitude, latitude, azimuth, nearby_count)
(metric_time, network_type, cgi, cell_name, interference_dbm, prev_interference_dbm, longitude, latitude, azimuth, nearby_count, prev_nearby_count)
VALUES
{values};
DELETE FROM `{self.table}` WHERE metric_time <> {metric_literal};
@@ -383,7 +385,8 @@ class ApiSummaryStore:
metric_literal = f"'{metric_time:%Y-%m-%d %H:%M:%S}'"
sql = (
"SELECT metric_time, network_type, cgi, cell_name, interference_dbm, "
f"longitude, latitude, azimuth, nearby_count FROM `{self.table}` "
"prev_interference_dbm, longitude, latitude, azimuth, nearby_count, prev_nearby_count "
f"FROM `{self.table}` "
f"WHERE metric_time = {metric_literal} ORDER BY cgi, cell_name"
)
result: list[dict[str, str]] = []
@@ -415,10 +418,12 @@ class ApiSummaryStore:
cgi VARCHAR(128) NOT NULL,
cell_name VARCHAR(255) NOT NULL,
interference_dbm DECIMAL(10,3) NOT NULL,
prev_interference_dbm DECIMAL(10,3) NULL,
longitude DECIMAL(10,6) NULL,
latitude DECIMAL(10,6) NULL,
azimuth DECIMAL(6,2) NOT NULL DEFAULT 0,
nearby_count INT NOT NULL DEFAULT 0,
prev_nearby_count INT NULL,
PRIMARY KEY (metric_time, cgi)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
""",
@@ -433,10 +438,12 @@ class ApiSummaryStore:
existing_columns = {str(item.get("name")) for item in columns if isinstance(item, dict)}
migrations = {
"network_type": "VARCHAR(16) NOT NULL DEFAULT ''",
"prev_interference_dbm": "DECIMAL(10,3) NULL",
"longitude": "DECIMAL(10,6) NULL",
"latitude": "DECIMAL(10,6) NULL",
"azimuth": "DECIMAL(6,2) NOT NULL DEFAULT 0",
"nearby_count": "INT NOT NULL DEFAULT 0",
"prev_nearby_count": "INT NULL",
}
for column, definition in migrations.items():
if column not in existing_columns:
@@ -467,10 +474,12 @@ class ApiSummaryStore:
sql_text_literal(row["cgi"]),
sql_text_literal(row["cell_name"]),
sql_decimal_literal(row["interference_dbm"], "interference_dbm"),
sql_decimal_literal(row["prev_interference_dbm"], "prev_interference_dbm", nullable=True),
sql_decimal_literal(row["longitude"], "longitude", nullable=True),
sql_decimal_literal(row["latitude"], "latitude", nullable=True),
sql_decimal_literal(row["azimuth"], "azimuth"),
str(int(row["nearby_count"])),
sql_decimal_literal(row["prev_nearby_count"], "prev_nearby_count", nullable=True),
)
) + ")"
@@ -787,6 +796,8 @@ def process(
print(f"missing_sources={','.join(missing)}")
return None
latest_time: datetime | None = None
previous_rows: list[dict[str, str]] = []
if store is not None:
latest_time = store.latest_metric_time()
if latest_time is not None and latest_time >= metric_time:
@@ -803,6 +814,8 @@ def process(
print(f"target_time={metric_time_text}")
print(f"database_metric_time={latest_time:%Y-%m-%d %H:%M:%S}")
return None
if latest_time is not None:
previous_rows = store.rows_for_time(latest_time)
cell_data_workbooks = source.cell_data_workbooks()
cell_metadata = load_cell_metadata(cell_data_workbooks)
@@ -846,6 +859,7 @@ def process(
)
populate_nearby_counts(summary_rows)
populate_previous_values(summary_rows, previous_rows)
summary_rows.sort(key=lambda item: (item["cgi"], item["cell_name"]))
summary_name = f"interference_summary_{output_key}.csv"
write_dict_csv(temp_dir / summary_name, SUMMARY_HEADER, summary_rows)
@@ -941,10 +955,12 @@ def summary_record(
"cgi": cgi,
"cell_name": cell_name,
"interference_dbm": interference,
"prev_interference_dbm": "",
"longitude": longitude,
"latitude": latitude,
"azimuth": azimuth,
"nearby_count": "0",
"prev_nearby_count": "",
}
@@ -989,6 +1005,18 @@ def populate_nearby_counts(rows: list[dict[str, str]]) -> None:
rows[index]["nearby_count"] = str(count)
def populate_previous_values(rows: list[dict[str, str]], previous_rows: list[dict[str, str]]) -> None:
previous_by_cgi = {row["cgi"]: row for row in previous_rows}
for row in rows:
previous = previous_by_cgi.get(row["cgi"])
if previous is None:
row["prev_interference_dbm"] = ""
row["prev_nearby_count"] = ""
continue
row["prev_interference_dbm"] = previous["interference_dbm"]
row["prev_nearby_count"] = previous["nearby_count"]
def spatial_bucket(longitude: float, latitude: float) -> tuple[int, int, int]:
longitude = math.radians(longitude)
latitude = math.radians(latitude)