feat: 制式标注改为2.6G/700M/TDD/FDD
This commit is contained in:
@@ -48,19 +48,25 @@ EXPECTED_TYPES = (
|
||||
)
|
||||
|
||||
NETWORK_TYPE_BY_SOURCE = {
|
||||
"5G下FDD干扰监控": "2.6G",
|
||||
"5G下FDD干扰监控": "FDD",
|
||||
"5G干扰监控": "2.6G",
|
||||
"700M下FDD干扰监控": "700M",
|
||||
"700M下FDD干扰监控": "FDD",
|
||||
"700M干扰监控": "700M",
|
||||
"SDR_FDD干扰监控": "4G",
|
||||
"SDR_TDD干扰监控": "4G",
|
||||
"反开RD干扰监控": "4G",
|
||||
"SDR_FDD干扰监控": "FDD",
|
||||
"SDR_TDD干扰监控": "TDD",
|
||||
"反开RD干扰监控": "TDD",
|
||||
}
|
||||
|
||||
MIN_INTERFERENCE_BY_NETWORK_TYPE = {
|
||||
"2.6G": Decimal("-107"),
|
||||
"700M": Decimal("-110"),
|
||||
"4G": Decimal("-110"),
|
||||
# Thresholds keep the original band grouping (the two 5G-directory sources use
|
||||
# -107, all others -110); they are independent of the network_type label.
|
||||
MIN_INTERFERENCE_BY_SOURCE = {
|
||||
"5G下FDD干扰监控": Decimal("-107"),
|
||||
"5G干扰监控": Decimal("-107"),
|
||||
"700M下FDD干扰监控": Decimal("-110"),
|
||||
"700M干扰监控": Decimal("-110"),
|
||||
"SDR_FDD干扰监控": Decimal("-110"),
|
||||
"SDR_TDD干扰监控": Decimal("-110"),
|
||||
"反开RD干扰监控": Decimal("-110"),
|
||||
}
|
||||
|
||||
EARTH_RADIUS_KM = 6371.0088
|
||||
@@ -848,7 +854,7 @@ def process(
|
||||
for row_number, row in enumerate(rows, start=2):
|
||||
record = dict(zip(header, row, strict=True))
|
||||
summary = summary_record(record, source_type, candidate.path, metric_time, row_number, cell_metadata)
|
||||
if passes_interference_threshold(summary):
|
||||
if passes_interference_threshold(summary, source_type):
|
||||
summary_rows.append(summary)
|
||||
else:
|
||||
threshold_filtered_rows += 1
|
||||
@@ -975,15 +981,14 @@ def summary_record(
|
||||
}
|
||||
|
||||
|
||||
def passes_interference_threshold(record: dict[str, str]) -> bool:
|
||||
network_type = record["network_type"]
|
||||
def passes_interference_threshold(record: dict[str, str], source_type: str) -> bool:
|
||||
try:
|
||||
interference = Decimal(record["interference_dbm"])
|
||||
except InvalidOperation as exc:
|
||||
raise ProcessingError(f"Invalid interference value: {record['interference_dbm']}") from exc
|
||||
if not interference.is_finite():
|
||||
raise ProcessingError(f"Invalid interference value: {record['interference_dbm']}")
|
||||
return interference >= MIN_INTERFERENCE_BY_NETWORK_TYPE[network_type]
|
||||
return interference >= MIN_INTERFERENCE_BY_SOURCE[source_type]
|
||||
|
||||
|
||||
def populate_nearby_counts(rows: list[dict[str, str]]) -> None:
|
||||
|
||||
Reference in New Issue
Block a user