1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
| """Create small, georeferenced validation cases for the risk-rule Demo."""
|
| from __future__ import annotations
|
| import argparse
| import json
| from pathlib import Path
|
| import geopandas as gpd
| from shapely.geometry import Point, box
|
|
| CRS = "EPSG:3857"
|
|
| def rules_payload() -> dict:
| return {
| "schema_version": 1,
| "score_cap": 100,
| "levels": {"medium": 30, "high": 60, "critical": 85},
| "rules": [
| {"id": "restricted_zone", "type": "zone_intersection", "zone_type": "restricted", "score": 65, "suggestion": "暂停自动处置，复核限制区边界和对象位置。"},
| {"id": "caution_zone", "type": "zone_intersection", "zone_type": "caution", "score": 25, "suggestion": "安排人工复核并确认现场作业条件。"},
| {"id": "low_confidence", "type": "attribute_threshold", "field": "confidence", "operator": "lt", "value": 0.6, "score": 20, "suggestion": "补充观测证据，不以低置信度对象单独定性。"},
| ],
| }
|
|
| def write_case(root: Path, name: str, records: list[dict]) -> None:
| directory = root / name
| directory.mkdir(parents=True, exist_ok=False)
| zones = gpd.GeoDataFrame(
| [
| {"zone_id": "caution-a", "zone_type": "caution", "geometry": box(0, 0, 600, 600)},
| {"zone_id": "restricted-a", "zone_type": "restricted", "geometry": box(200, 200, 400, 400)},
| ],
| crs=CRS,
| )
| gpd.GeoDataFrame(records, crs=CRS).to_file(directory / "observations.geojson", driver="GeoJSON")
| zones.to_file(directory / "zones.geojson", driver="GeoJSON")
| (directory / "rules.json").write_text(json.dumps(rules_payload(), ensure_ascii=False, indent=2), encoding="utf-8")
|
|
| def write_validation_inputs(output: Path) -> None:
| if output.exists() and any(output.iterdir()):
| raise ValueError(f"Validation output is not empty: {output}")
| output.mkdir(parents=True, exist_ok=True)
| write_case(output, "normal", [
| {"object_id": "normal-outside", "confidence": 0.92, "geometry": Point(750, 750)},
| {"object_id": "normal-caution", "confidence": 0.88, "geometry": Point(120, 120)},
| {"object_id": "normal-restricted", "confidence": 0.85, "geometry": Point(300, 300)},
| ])
| write_case(output, "difficult", [
| {"object_id": "boundary-overlap", "confidence": 0.82, "geometry": Point(200, 300)},
| {"object_id": "low-confidence-restricted", "confidence": 0.45, "geometry": Point(300, 300)},
| {"object_id": "low-confidence-outside", "confidence": 0.50, "geometry": Point(780, 720)},
| ])
|
|
| def main() -> int:
| parser = argparse.ArgumentParser(description="Create risk-rule validation GeoJSON and rules.")
| parser.add_argument("--output", type=Path, required=True)
| args = parser.parse_args()
| write_validation_inputs(args.output)
| print(args.output)
| return 0
|
|
| if __name__ == "__main__":
| raise SystemExit(main())
|
|
