From bca93079e6190e18788ce09ad4c94e6f312c553a Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 11 Nov 2026 15:32:48 +0000 Subject: [PATCH] feat(conflate): add matcher diagnostics to the spider stats Conflate now writes conflate.spider-diagnostics.tsv on the side: for each spider, why its missing_in_osm features found no match (no OSM feature of the kind close enough, all rejected, or taken by another feature of the spider), a histogram of the distance to the nearest rejected candidate, and chance pairs: how many of a sample of about 100 features per spider still match when moved 2 km away. compute_spider_stats adds these to spider-stats.parquet, together with the median offset of matched features from their OSM feature, per spider and per brand and country, and how many features share their position with another of the spider. Part of #924. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Ttfb2LhDRExioXkdHbxLno --- docs/TECHNICAL_DESIGN.md | 8 +- docs/outputs/SPIDER_STATS.md | 62 ++++- src/pipeline/conflate/diagnostics.rs | 374 +++++++++++++++++++++++++++ src/pipeline/conflate/mod.rs | 77 +++++- src/pipeline/spider_stats.rs | 167 +++++++++++- tests/integration_test.rs | 26 ++ 6 files changed, 685 insertions(+), 29 deletions(-) create mode 100644 src/pipeline/conflate/diagnostics.rs diff --git a/docs/TECHNICAL_DESIGN.md b/docs/TECHNICAL_DESIGN.md index 52222ab..36a20c7 100644 --- a/docs/TECHNICAL_DESIGN.md +++ b/docs/TECHNICAL_DESIGN.md @@ -227,6 +227,7 @@ graph TD OSMVIEWS_R --> CONFLATE QRANK_Q --> CONFLATE CONFLATE --> CONFLATED[conflated.parquet] + CONFLATE --> DIAGNOSTICS[conflate.spider-diagnostics.tsv] CONFLATED --> UPLOAD_CONFLATED(upload_conflated) --> S3A[("public S3")] CONFLATED --> EXTRACT_CONFLATED_LAYERS(extract_conflated_layers) @@ -241,6 +242,7 @@ graph TD CONFLATED --> SPIDER_STATS(compute_spider_stats) SPIDER_COUNTS --> SPIDER_STATS + DIAGNOSTICS --> SPIDER_STATS ATP_STATS --> SPIDER_STATS OSM_BRANDS --> SPIDER_STATS BRAND_NAMES --> SPIDER_STATS @@ -437,7 +439,11 @@ fetch away. matching evidence. Writes one row per ATP feature that a matcher handles — matched or not — to `conflated.parquet`, and logs how many features of which spiders no - matcher handles. Each row comes with an `importance` from 0 to 1 (see + matcher handles. On the side, it writes + `conflate.spider-diagnostics.tsv` for the spider stats + ([`diagnostics.rs`](../src/pipeline/conflate/diagnostics.rs)): why + features found no match, and how many of a sample per spider still + match when moved 2 km away. Each row comes with an `importance` from 0 to 1 (see [`src/pipeline/importance.rs`](../src/pipeline/importance.rs)): a weighted mean of the location's OSMViews quantile, the brand's QRank quantile and a coarse score for the kind of place diff --git a/docs/outputs/SPIDER_STATS.md b/docs/outputs/SPIDER_STATS.md index 9afbdf8..c051c93 100644 --- a/docs/outputs/SPIDER_STATS.md +++ b/docs/outputs/SPIDER_STATS.md @@ -47,7 +47,16 @@ same UINT64 non-null — rows with match.relation = "same" located_in UINT64 non-null — rows with match.relation = "located_in" missing_in_osm UINT64 non-null — rows with match.relation = "missing_in_osm" shared_osm_features UINT64 non-null — same rows whose OSM feature another feature of the spider also matched +shared_positions UINT64 non-null — features at the same position as another feature of the spider +offset_east_m FLOAT nullable — median of how far same rows lie east of their OSM feature +offset_north_m FLOAT nullable — ... and north distance_m LIST non-null — histogram of match distances, see below +missing_no_candidate UINT64 non-null — missing_in_osm: no OSM feature of the kind close enough +missing_rejected UINT64 non-null — missing_in_osm: the matcher found no evidence for any +missing_taken UINT64 non-null — missing_in_osm: another feature of the spider took them +rejected_distance_m LIST non-null — histogram of the distance to the nearest rejected one +chance_sampled UINT64 non-null — features moved 2 km to find chance pairs +chance_matched UINT64 non-null — ... and how many of them still matched opening_hours_top_share FLOAT nullable — share of features with the most common opening_hours rules LIST non-null rule UTF8 non-null — match.evidence.rule @@ -68,6 +77,8 @@ brands LIST non-null missing_in_osm UINT64 non-null osm_features UINT64 nullable — OSM features with this ID in this country deletion_candidates UINT64 nullable — deletion candidates with this ID in this country + offset_east_m FLOAT nullable — like offset_east_m above, for this brand and country + offset_north_m FLOAT nullable ``` ### What became of the features @@ -75,8 +86,9 @@ brands LIST non-null The counts add up like this: ``` -dump_features = not_for_osm + invalid + no_matcher + rows -rows = same + located_in + missing_in_osm +dump_features = not_for_osm + invalid + no_matcher + rows +rows = same + located_in + missing_in_osm +missing_in_osm = missing_no_candidate + missing_rejected + missing_taken ``` `atp_features` and `atp_errors` come from AllThePlaces' own statistics @@ -111,11 +123,57 @@ also matched another feature of the same spider. A high count hints at duplicates in the spider, or at OpenStreetMap having one feature where there are several. +`shared_positions` counts features of the spider at exactly the same +position as another one of its features. Some are right, such as the +departments of one store, but many hint at a spider that puts features +it can't place at a default position. + +`offset_east_m` and `offset_north_m` are the medians of how far the +AllThePlaces feature of a `same` row lies east and north of its +OpenStreetMap feature, from centroid to centroid, in meters (negative +for west and south). Single features scatter in all directions, so +the medians stay near zero, unless a spider shifts all its features +the same way. For example, `c_and_a` once placed all its stores +about 150 m west of where they are. The same medians per brand and +country show a shift in only one country. + `opening_hours_top_share` is the share of the spider's features with `opening_hours` that have its most common value. Near 1 for a large spider, it suggests that the spider writes the same hours everywhere, such as those of its head office. +### Why features are missing + +The `missing_*` columns say why the matcher found nothing for a +feature that ends up `missing_in_osm`: + +- `missing_no_candidate`: OpenStreetMap has no feature of the same kind + (such as a shop, for a shop) within the distance the matcher looks. + Either OpenStreetMap lacks it, or the spider's coordinates are far + off. +- `missing_rejected`: there are such features, but the matcher found no + evidence that any is the same, such as the same brand, phone number + or name. `rejected_distance_m` is a histogram of the distance to the + nearest of them, with the bins of `distance_m`. Many rejected + features close by hint at a matcher that misses some evidence, or at + tags that differ between the spider and OpenStreetMap. +- `missing_taken`: the matcher accepted some, but another feature of the + same spider took them first; see + [`CONFLATED_PARQUET.md`](CONFLATED_PARQUET.md#match). + +### Chance pairs + +To see how many matches may come from chance, conflate moves about +100 features of each spider (all of a smaller spider) 2 km in a random +direction and matches them again. `chance_matched` of +`chance_sampled` still find a match, which can only be wrong. Their +share estimates the share of wrong matches among the spider's matches, +and overstates it, since in the real matching a true partner usually +takes the feature first. The sample is chosen by hashing, so it stays +the same from run to run on the same input. See +[the notes on infrastructure matching](../notes/2026-11-05-infrastructure-matching.md) +for where this comes from. + ### How far tags agree `tags` compares the tags that `match.evidence.agreeing` compares, for diff --git a/src/pipeline/conflate/diagnostics.rs b/src/pipeline/conflate/diagnostics.rs new file mode 100644 index 0000000..38d2e52 --- /dev/null +++ b/src/pipeline/conflate/diagnostics.rs @@ -0,0 +1,374 @@ +//! What conflate finds out about each spider's matching on the side: +//! why features found no match in OpenStreetMap, and how often a +//! feature would match by chance. Written to +//! `conflate.spider-diagnostics.tsv` in the workdir for +//! `compute_spider_stats`, since `conflated.parquet` doesn't keep the +//! candidates that the matchers rejected. + +use super::s2_util::MergedCellRanges; +use crate::matchers::{ + OsmCandidate, brand_domains::BrandDomains, contact::Nondescript, create_matcher, match_distance, +}; +use crate::pipeline::spider_stats::DISTANCE_BINS_M; +use crate::places::{Place, PlaceReader}; +use crate::tables::{LocalFeatureRef, OsmFeatures}; +use anyhow::{Context, Result}; +use geo::{Destination, Haversine}; +use rayon::prelude::*; +use std::collections::HashMap; +use std::hash::{DefaultHasher, Hash, Hasher}; +use std::io::{BufRead, BufReader, BufWriter, Write}; +use std::path::{Path, PathBuf}; + +/// Filename, within `workdir`, of the diagnostics that conflate writes. +const FILENAME: &str = "conflate.spider-diagnostics.tsv"; + +const HEADER: &str = "spider\tno_candidate\trejected\ttaken\trejected_distance_m\ + \tchance_sampled\tchance_matched"; + +/// How far [chance_pairs] moves features. +const CHANCE_SHIFT_METERS: f64 = 2000.0; + +/// How many features per spider [chance_pairs] moves, roughly. +const CHANCE_SAMPLE: u64 = 100; + +/// At most this many rejected candidates are kept per ATP feature, to +/// find the nearest among them. +pub(super) const MAX_REJECTED: usize = 64; + +/// Path of the diagnostics that conflate writes in `workdir`. +pub(in crate::pipeline) fn path(workdir: &Path) -> PathBuf { + workdir.join(FILENAME) +} + +/// Why an ATP feature ended up `missing_in_osm`. +#[derive(Clone, Copy, Debug, PartialEq)] +pub(super) enum Missing { + /// No OSM feature of its kind within the matcher's distance. + NoCandidate, + /// OSM features of its kind were close enough, but the matcher + /// found no evidence for any of them. Holds the distance to the + /// nearest one in meters, if it could be computed. + Rejected(Option), + /// The matcher accepted some OSM features, but other features of + /// the same spider took them; see `conflate::assign_exclusive`. + Taken, +} + +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub(in crate::pipeline) struct SpiderDiagnostics { + pub spider: String, + /// `missing_in_osm` rows by [Missing]. + pub no_candidate: u64, + pub rejected: u64, + pub taken: u64, + /// Histogram of the distance to the nearest rejected candidate, + /// with the bins of [DISTANCE_BINS_M]. + pub rejected_distance_m: Vec, + /// Features moved by [CHANCE_SHIFT_METERS], and how many of them + /// still matched something. + pub chance_sampled: u64, + pub chance_matched: u64, +} + +impl SpiderDiagnostics { + fn new(spider: &str) -> Self { + SpiderDiagnostics { + spider: spider.to_string(), + rejected_distance_m: vec![0; DISTANCE_BINS_M.len() + 1], + ..Default::default() + } + } +} + +/// Diagnostics of all spiders, filled in while conflating. +#[derive(Debug, Default)] +pub(super) struct Diagnostics { + spiders: HashMap, +} + +impl Diagnostics { + fn spider(&mut self, spider: &str) -> &mut SpiderDiagnostics { + if !self.spiders.contains_key(spider) { + self.spiders + .insert(spider.to_string(), SpiderDiagnostics::new(spider)); + } + self.spiders.get_mut(spider).expect("inserted") + } + + pub(super) fn add_missing(&mut self, spider: &str, missing: Missing) { + let s = self.spider(spider); + match missing { + Missing::NoCandidate => s.no_candidate += 1, + Missing::Taken => s.taken += 1, + Missing::Rejected(distance) => { + s.rejected += 1; + if let Some(d) = distance { + let bin = DISTANCE_BINS_M.partition_point(|&upper| f64::from(upper) <= d); + s.rejected_distance_m[bin] += 1; + } + } + } + } + + fn add_chance(&mut self, spider: &str, matched: bool) { + let s = self.spider(spider); + s.chance_sampled += 1; + if matched { + s.chance_matched += 1; + } + } + + /// Writes the diagnostics, sorted by spider, to `out` via a + /// temporary file. + pub(super) fn write(&self, out: &Path) -> Result<()> { + let mut spiders: Vec<&SpiderDiagnostics> = self.spiders.values().collect(); + spiders.sort_by(|a, b| a.spider.cmp(&b.spider)); + let mut tmp = out.to_path_buf(); + tmp.add_extension("tmp"); + let mut w = BufWriter::new( + std::fs::File::create(&tmp).with_context(|| format!("could not create {tmp:?}"))?, + ); + writeln!(w, "{HEADER}")?; + for s in spiders { + let bins: Vec = s.rejected_distance_m.iter().map(u64::to_string).collect(); + writeln!( + w, + "{}\t{}\t{}\t{}\t{}\t{}\t{}", + s.spider, + s.no_candidate, + s.rejected, + s.taken, + bins.join(","), + s.chance_sampled, + s.chance_matched + )?; + } + w.into_inner()?.sync_all()?; + std::fs::rename(&tmp, out)?; + Ok(()) + } +} + +/// Reads diagnostics that [Diagnostics::write] wrote, by spider. +pub(in crate::pipeline) fn read(path: &Path) -> Result> { + let file = std::fs::File::open(path).with_context(|| format!("could not open {path:?}"))?; + let mut lines = BufReader::new(file).lines(); + if lines.next().transpose()?.as_deref() != Some(HEADER) { + anyhow::bail!("{path:?} doesn't start with the expected header"); + } + let mut result = HashMap::new(); + for line in lines { + let line = line?; + let cols: Vec<&str> = line.split('\t').collect(); + let [ + spider, + no_candidate, + rejected, + taken, + distances, + sampled, + matched, + ] = cols[..] + else { + anyhow::bail!("bad line in {path:?}: {line:?}"); + }; + let num = |s: &str| -> Result { + s.parse() + .with_context(|| format!("bad number in {path:?}: {line:?}")) + }; + let rejected_distance_m = distances.split(',').map(num).collect::>>()?; + anyhow::ensure!( + rejected_distance_m.len() == DISTANCE_BINS_M.len() + 1, + "wrong number of distance bins in {path:?}: {line:?}" + ); + result.insert( + spider.to_string(), + SpiderDiagnostics { + spider: spider.to_string(), + no_candidate: num(no_candidate)?, + rejected: num(rejected)?, + taken: num(taken)?, + rejected_distance_m, + chance_sampled: num(sampled)?, + chance_matched: num(matched)?, + }, + ); + } + Ok(result) +} + +/// The distance in meters from `atp` to the nearest of `rejected`. +pub(super) fn nearest( + atp: &Place, + rejected: &[LocalFeatureRef], + osm: &OsmFeatures, +) -> Result> { + let shape = atp.shape(); + let mut nearest: Option = None; + for &r in rejected { + let feature = osm.index.get_feature(r)?; + let candidate = OsmCandidate { + feature: &feature, + strings: &osm.strings, + }; + let Ok(geometry) = candidate.geometry() else { + continue; + }; + if let Some(d) = super::evidence::distance_meters(&shape, &geometry) + && nearest.is_none_or(|n| d < n) + { + nearest = Some(d); + } + } + Ok(nearest) +} + +fn hash(spider: &str, ordinal: u64) -> u64 { + let mut hasher = DefaultHasher::new(); + spider.hash(&mut hasher); + ordinal.hash(&mut hasher); + hasher.finish() +} + +/// `atp`, moved [CHANCE_SHIFT_METERS] in a direction that `seed` +/// picks. +fn shifted(atp: &Place, seed: u64) -> Place { + let center = s2::latlng::LatLng::from(s2::cellid::CellID(atp.s2_cell_id)); + let origin = geo::Point::new(center.lng.deg(), center.lat.deg()); + let bearing = (seed >> 11) as f64 / (1u64 << 53) as f64 * 360.0; + let moved = Haversine.destination(origin, bearing, CHANCE_SHIFT_METERS); + let lat_lng = s2::latlng::LatLng::from_degrees(moved.y(), moved.x()); + let mut place = atp.deep_clone(); + place.s2_cell_id = s2::cellid::CellID::from(lat_lng).0; + place.shape_wkb = crate::geometry::encode_wkb(&geo::Geometry::Point(moved)); + place +} + +/// Moves a sample of about [CHANCE_SAMPLE] features of each spider by +/// [CHANCE_SHIFT_METERS], and counts how many of them still match an +/// OSM feature. Those matches are by chance, so their share estimates +/// how many of the spider's matches may be wrong. It overstates it, as +/// in the real matching a true partner usually takes the feature +/// first; see `docs/notes/2026-11-05-infrastructure-matching.md`. +/// +/// `features` is the number of features per spider in `atp_reader`. +/// The sample depends only on the input, so it's the same each run. +pub(super) fn chance_pairs( + atp_reader: &PlaceReader, + features: &HashMap, + osm: &OsmFeatures, + nondescript: &Nondescript, + brand_domains: &BrandDomains, + diagnostics: &mut Diagnostics, +) -> Result<()> { + let coverer = s2::region::RegionCoverer { + max_cells: 16, + min_level: 12, + max_level: s2::cellid::MAX_LEVEL as u8, + level_mod: 1, + }; + let mut ordinal: u64 = 0; + for batch in atp_reader.read_all()? { + let batch = batch?; + let first = ordinal; + ordinal += batch.len() as u64; + let results: Vec> = batch + .par_iter() + .enumerate() + .map(|(i, atp)| -> Result> { + let n = features.get(&atp.spider).copied().unwrap_or(1); + let every = n.div_ceil(CHANCE_SAMPLE).max(1); + let seed = hash(&atp.spider, first + i as u64); + if !seed.is_multiple_of(every) { + return Ok(None); + } + let moved = shifted(atp, seed); + let Some(matcher) = create_matcher(&moved, nondescript, brand_domains) else { + return Ok(None); + }; + let covering = { + let cell = s2::cell::Cell::from(s2::cellid::CellID(moved.s2_cell_id)); + let radius = matcher + .max_distance() + .unwrap_or_else(|| match_distance(&moved.mask)); + let cap = s2::cap::Cap::from_center_chordangle(&cell.center(), &radius); + coverer.covering(&cap) + }; + for (lo, hi) in MergedCellRanges::new(covering) { + for r in osm.index.query(lo..=hi, moved.mask) { + let feature = osm.index.get_feature(r)?; + let candidate = OsmCandidate { + feature: &feature, + strings: &osm.strings, + }; + if matcher.score_osm_candidate(&candidate) > 0.0 { + return Ok(Some(true)); + } + } + } + Ok(Some(false)) + }) + .collect::>()?; + for (atp, matched) in batch.iter().zip(results) { + if let Some(matched) = matched { + diagnostics.add_chance(&atp.spider, matched); + } + } + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use geo::{Distance, Geodesic}; + + #[test] + fn test_write_read() -> Result<()> { + let dir = tempfile::tempdir()?; + let out = path(dir.path()); + let mut d = Diagnostics::default(); + d.add_missing("b", Missing::NoCandidate); + d.add_missing("b", Missing::Taken); + d.add_missing("b", Missing::Rejected(Some(12.0))); + d.add_missing("b", Missing::Rejected(Some(10.0))); + d.add_missing("b", Missing::Rejected(None)); + d.add_chance("a", true); + d.add_chance("a", false); + d.write(&out)?; + + let read = read(&out)?; + assert_eq!(read.len(), 2); + let b = &read["b"]; + assert_eq!((b.no_candidate, b.rejected, b.taken), (1, 3, 1)); + assert_eq!(b.rejected_distance_m, [0, 0, 2, 0, 0, 0, 0, 0]); + let a = &read["a"]; + assert_eq!((a.chance_sampled, a.chance_matched), (2, 1)); + assert_eq!(a.rejected_distance_m.iter().sum::(), 0); + Ok(()) + } + + #[test] + fn test_shifted() { + let lat_lng = s2::latlng::LatLng::from_degrees(47.37, 8.54); + let place = Place { + s2_cell_id: s2::cellid::CellID::from(lat_lng).0, + spider: "x".into(), + mask: crate::matchers::MatchMask(1), + tags: Vec::new(), + shape_wkb: Vec::new(), + fetched: crate::utils::UtcTimestamp(time::UtcDateTime::UNIX_EPOCH), + }; + for seed in [0, 1 << 62, u64::MAX] { + let moved = shifted(&place, seed); + let to = s2::latlng::LatLng::from(s2::cellid::CellID(moved.s2_cell_id)); + let d = Geodesic.distance( + geo::Point::new(8.54, 47.37), + geo::Point::new(to.lng.deg(), to.lat.deg()), + ); + assert!((d - CHANCE_SHIFT_METERS).abs() < 10.0, "{d}"); + assert!(matches!(moved.shape(), geo::Geometry::Point(_))); + } + } +} diff --git a/src/pipeline/conflate/mod.rs b/src/pipeline/conflate/mod.rs index b1232af..d3e101b 100644 --- a/src/pipeline/conflate/mod.rs +++ b/src/pipeline/conflate/mod.rs @@ -22,7 +22,7 @@ use std::{ hash::{Hash, Hasher}, path::{Path, PathBuf}, sync::{ - Mutex, + Mutex, OnceLock, atomic::{AtomicBool, AtomicU64, Ordering}, mpsc::{Receiver, SyncSender, sync_channel}, }, @@ -38,12 +38,14 @@ use std::{ const PROGRESS_LOG_INTERVAL: Duration = Duration::from_secs(30); mod deletion; +pub(super) mod diagnostics; pub(super) mod evidence; mod local_coverage; mod located_in; mod s2_util; mod writer; use deletion::{AtpCoverage, Candidate, Reason, deletion_reason, local_keys, place_country}; +use diagnostics::{Diagnostics, Missing}; use local_coverage::{LocalCoverage, LocalKey, MIN_LOCAL_SHARE}; use located_in::{Host, find_departments, located_in_tag}; use s2_util::MergedCellRanges; @@ -81,7 +83,11 @@ pub fn conflate( ])? .max(osm.modified()?); let out_path = workdir.join("conflated.parquet"); - if out_path.exists() && last_modified(&[&out_path])? >= input_modified { + let diagnostics_path = diagnostics::path(workdir); + if out_path.exists() + && diagnostics_path.exists() + && last_modified(&[&out_path])? >= input_modified + { return Ok(out_path); } @@ -103,7 +109,7 @@ pub fn conflate( make_progress_bar(progress, "conflate.match", num_atp_features, "ATP features"); let writer_progress = make_progress_bar(progress, "conflate.write", 0, "parquet rows"); - let mut producer_result = Ok(()); + let mut producer_result = Ok(Diagnostics::default()); let mut writer_result = Ok(()); thread::scope(|s| { let (tx, rx) = sync_channel::(8192); @@ -124,7 +130,7 @@ pub fn conflate( }); }); writer_result?; - producer_result?; + producer_result?.write(&diagnostics_path)?; Ok(out_path) } @@ -207,7 +213,7 @@ fn produce_rows( signals: &Signals, progress_bar: ProgressBar, out: SyncSender, -) -> Result<()> { +) -> Result { let coverer = s2::region::RegionCoverer { max_cells: 16, min_level: 12, @@ -229,6 +235,8 @@ fn produce_rows( // ATP features per spider that no matcher handles, and that get no // row in conflated.parquet. let mut no_matcher_by_spider: HashMap = HashMap::new(); + let mut features_by_spider: HashMap = HashMap::new(); + let mut diagnostics = Diagnostics::default(); let departments = find_departments(atp_reader)?; log::info!( departments = departments.len(), @@ -239,23 +247,27 @@ fn produce_rows( located_in_tag(atp).or_else(|| departments.get(&ordinal).cloned()) }; let located_in_rows = AtomicU64::new(0); - let send_unmatched = |atp: &Place, host: Option<&Host>| -> Result<()> { + // Sends the row of an ATP feature without a match, and returns + // whether it is `missing_in_osm`, rather than `located_in`. + let send_unmatched = |atp: &Place, host: Option<&Host>| -> Result { if let Some(host) = host && let Some(r) = find_host(atp, host, osm)? { // The host exists, whether or not ATP has it. matched.insert(r); located_in_rows.fetch_add(1, Ordering::Relaxed); - return send_row( + send_row( atp, Some(r), MatchInfo::located_in(host.rule()), osm, signals, &out, - ); + )?; + return Ok(false); } - send_row(atp, None, MatchInfo::missing_in_osm(), osm, signals, &out) + send_row(atp, None, MatchInfo::missing_in_osm(), osm, signals, &out)?; + Ok(true) }; let mut batch_start: u64 = 0; @@ -267,6 +279,9 @@ fn produce_rows( // don't match say nothing about unmatched OSM features. let has_matcher: Vec = batch.iter().map(|_| AtomicBool::new(false)).collect(); let no_matcher: Vec = batch.iter().map(|_| AtomicBool::new(false)).collect(); + // Why each ATP feature of this batch that is missing in OSM + // found no match, for the diagnostics. + let missing: Vec> = batch.iter().map(|_| OnceLock::new()).collect(); // Each batch is processed by the Rayon thread pool in parallel, // but the outer loop is sequential — so nearby places (within a // batch) always go to nearby workers, preserving spatial locality. @@ -301,6 +316,9 @@ fn produce_rows( coverer.covering(&cap) }; let mut best = Vec::<(f64, LocalFeatureRef)>::new(); + // OSM features of the kind that the matcher found no + // evidence for, for the diagnostics. + let mut rejected = Vec::new(); for (lo, hi) in MergedCellRanges::new(covering) { for r in osm.index.query(lo..=hi, atp.mask) { let feature = osm.index.get_feature(r)?; @@ -313,6 +331,9 @@ fn produce_rows( continue; } let score = matcher.score_osm_candidate(&candidate); + if score <= 0.0 && rejected.len() < diagnostics::MAX_REJECTED { + rejected.push(r); + } keep_best(&mut best, keep, score, r); } } @@ -350,7 +371,15 @@ fn produce_rows( return Ok(()); } let Some(&(score, r)) = best.first() else { - return send_unmatched(atp, host.as_ref()); + if send_unmatched(atp, host.as_ref())? { + let why = if rejected.is_empty() { + Missing::NoCandidate + } else { + Missing::Rejected(diagnostics::nearest(atp, &rejected, osm)?) + }; + missing[i].set(why).expect("set once"); + } + return Ok(()); }; matched.insert(r); let feature = osm.index.get_feature(r)?; @@ -385,6 +414,12 @@ fn produce_rows( if no_matcher.load(Ordering::Relaxed) { count_spider(&mut no_matcher_by_spider, &atp.spider); } + count_spider(&mut features_by_spider, &atp.spider); + } + for (atp, why) in batch.iter().zip(missing) { + if let Some(why) = why.into_inner() { + diagnostics.add_missing(&atp.spider, why); + } } batch_start += batch.len() as u64; } @@ -438,7 +473,11 @@ fn produce_rows( &out, )?; } - None => send_unmatched(&atp, host_of(ordinal, &atp).as_ref())?, + None => { + if send_unmatched(&atp, host_of(ordinal, &atp).as_ref())? { + diagnostics.add_missing(&atp.spider, Missing::Taken); + } + } } } ordinal += 1; @@ -447,6 +486,20 @@ fn produce_rows( send_deletion_candidates(osm, &coverage, &matched, signals, &out)?; + let chance_start = Instant::now(); + diagnostics::chance_pairs( + atp_reader, + &features_by_spider, + osm, + nondescript, + brand_domains, + &mut diagnostics, + )?; + log::info!( + elapsed_seconds = chance_start.elapsed().as_secs_f64(); + "conflate.chance_pairs: done" + ); + progress_bar.finish(); log::info!( elapsed_seconds = start.elapsed().as_secs_f64(), @@ -458,7 +511,7 @@ fn produce_rows( exclusive_matched = assigned.values().filter(|m| m.is_some()).count(); "conflate.match: done" ); - Ok(()) + Ok(diagnostics) } /// The OSM features that some ATP feature matched, as one bit per diff --git a/src/pipeline/spider_stats.rs b/src/pipeline/spider_stats.rs index b988fb6..a4f28cf 100644 --- a/src/pipeline/spider_stats.rs +++ b/src/pipeline/spider_stats.rs @@ -16,6 +16,7 @@ use super::atp::{run_stats, spider_counts}; use super::brand_totals; +use super::conflate::diagnostics::{self, SpiderDiagnostics}; use super::conflate::evidence::{compared_tags, has_compared_tag}; use super::wikidata_ids::WIKIDATA_KEYS; use crate::make_progress_bar; @@ -39,6 +40,7 @@ use parquet::file::properties::WriterProperties; use rayon::prelude::*; use std::collections::{BTreeSet, HashMap, HashSet}; use std::fs::File; +use std::hash::{Hash, Hasher}; use std::path::{Path, PathBuf}; use std::sync::Arc; use wkb::reader::read_wkb; @@ -53,7 +55,8 @@ const ROWS_PER_GROUP: usize = 64; /// Upper bounds, in meters, of the bins of the `distance_m` histogram; /// the last bin has no upper bound. -pub(super) const DISTANCE_BINS_M: [f32; 7] = [1.0, 10.0, 25.0, 50.0, 100.0, 250.0, 1000.0]; +pub(in crate::pipeline) const DISTANCE_BINS_M: [f32; 7] = + [1.0, 10.0, 25.0, 50.0, 100.0, 250.0, 1000.0]; /// Computes the stats of every spider from `conflated` and the side /// files in `workdir`, and writes them to `workdir/spider-stats.parquet`, @@ -76,6 +79,7 @@ pub fn compute_spider_stats( let run_stats = run_stats::read(workdir)?.unwrap_or_default(); let osm_brands = brand_totals::read_features(workdir)?; let names = brand_totals::read_names(workdir)?; + let diagnostics = diagnostics::read(&diagnostics::path(workdir))?; let stats = scan_conflated(conflated, progress)?; @@ -89,6 +93,7 @@ pub fn compute_spider_stats( spider, counts: counts.get(spider), run_stats: run_stats.get(spider), + diagnostics: diagnostics.get(spider), stats: stats.spiders.get(spider).unwrap_or(&empty), }) .collect(); @@ -128,6 +133,11 @@ struct RowInfo { opening_hours: Option, /// The OSM feature of a `same` row, as `id * 4 + type`. osm_ref: Option, + /// For a `same` row, how far the ATP feature lies east and north of + /// the OSM one, in meters, from centroid to centroid. + offset: Option<(f32, f32)>, + /// Hash of the ATP geometry, to find features at the same position. + position: Option, } #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] @@ -157,6 +167,7 @@ struct Brand { /// The main categories of the ATP features, so that the OSM total /// counts features of the same kinds only. categories: HashSet, + offsets: Vec<(f32, f32)>, } /// What we count for one spider from `conflated.parquet`. @@ -173,6 +184,8 @@ struct Spider { tags: Vec<[u64; 4]>, opening_hours: HashMap, osm_refs: HashMap, + offsets: Vec<(f32, f32)>, + positions: HashMap, /// By country and Wikidata ID; 0 for features without one. brands: HashMap<(Option, u64), Brand>, } @@ -213,6 +226,12 @@ impl Spider { if let Some(r) = row.osm_ref { *self.osm_refs.entry(r).or_default() += 1; } + if let Some(offset) = row.offset { + self.offsets.push(offset); + } + if let Some(p) = row.position { + *self.positions.entry(p).or_default() += 1; + } let ids = if row.wikidata.is_empty() { vec![0] } else { @@ -232,6 +251,9 @@ impl Spider { { brand.categories.insert(c.clone()); } + if let Some(offset) = row.offset { + brand.offsets.push(offset); + } } } @@ -241,6 +263,12 @@ impl Spider { self.osm_refs.values().filter(|&&n| n > 1).sum() } + /// ATP features at the same position as another feature of this + /// spider. + fn shared_positions(&self) -> u64 { + self.positions.values().filter(|&&n| n > 1).sum() + } + /// The share of ATP features with `opening_hours` that have the /// most common value, or `None` if none have any. fn opening_hours_top_share(&self) -> Option { @@ -250,6 +278,23 @@ impl Spider { } } +/// The median of the east and north components of `offsets`, each +/// on its own, or `None` if there are none. +fn median_offset(offsets: &[(f32, f32)]) -> Option<(f32, f32)> { + if offsets.is_empty() { + return None; + } + let median = |mut values: Vec| { + let mid = values.len() / 2; + let (_, m, _) = values.select_nth_unstable_by(mid, f32::total_cmp); + *m + }; + Some(( + median(offsets.iter().map(|o| o.0).collect()), + median(offsets.iter().map(|o| o.1).collect()), + )) +} + #[derive(Debug, Default)] struct Scanned { rows: u64, @@ -335,6 +380,11 @@ fn row_info(batch: &RecordBatch, row: usize) -> Result { let atp_tag_refs = || atp_tags.iter().map(|(k, v)| (k.as_str(), v.as_str())); let spider = get_string(get_struct(atp, "fetched")?, "spider", row)?; let geometry = get_binary(batch, "atp_geometry", row)?; + let position = { + let mut hasher = std::hash::DefaultHasher::new(); + geometry.hash(&mut hasher); + hasher.finish() + }; let mut tags = Vec::new(); if relation == Relation::Same { @@ -352,15 +402,19 @@ fn row_info(batch: &RecordBatch, row: usize) -> Result { tags.push((i, state)); } } - let osm_ref = if relation == Relation::Same { + let (osm_ref, offset) = if relation == Relation::Same { let kind = match get_string(osm, "type", row)?.as_str() { "node" => 1, "way" => 2, _ => 3, }; - Some(get_u64(osm, "id", row)? * 4 + kind) + let osm_geometry = get_binary(batch, "osm_geometry", row)?; + ( + Some(get_u64(osm, "id", row)? * 4 + kind), + wkb_offset(&geometry, &osm_geometry), + ) } else { - None + (None, None) }; Ok(RowInfo { spider: Some(spider), @@ -375,9 +429,22 @@ fn row_info(batch: &RecordBatch, row: usize) -> Result { .find(|(k, _)| *k == "opening_hours") .map(|(_, v)| v.to_string()), osm_ref, + offset, + position: Some(position), }) } +/// How far the centroid of `atp` lies east and north of the centroid +/// of `osm`, in meters. +fn wkb_offset(atp: &[u8], osm: &[u8]) -> Option<(f32, f32)> { + let a = read_wkb(atp).ok()?.to_geometry().centroid()?; + let o = read_wkb(osm).ok()?.to_geometry().centroid()?; + const METERS_PER_DEGREE: f64 = 111_320.0; + let east = (a.x() - o.x()) * METERS_PER_DEGREE * o.y().to_radians().cos(); + let north = (a.y() - o.y()) * METERS_PER_DEGREE; + Some((east as f32, north as f32)) +} + /// The distinct brand, operator and network Wikidata IDs in `tags`. fn wikidata_ids<'a>(tags: impl Iterator) -> Vec { let mut ids: Vec = tags @@ -441,6 +508,7 @@ struct Row<'a> { spider: &'a str, counts: Option<&'a spider_counts::SpiderCounts>, run_stats: Option<&'a run_stats::SpiderRunStats>, + diagnostics: Option<&'a SpiderDiagnostics>, stats: &'a Spider, } @@ -466,6 +534,8 @@ fn brand_fields() -> Fields { u64_field("missing_in_osm"), Field::new("osm_features", DataType::UInt64, true), Field::new("deletion_candidates", DataType::UInt64, true), + Field::new("offset_east_m", DataType::Float32, true), + Field::new("offset_north_m", DataType::Float32, true), ]) } @@ -486,6 +556,14 @@ fn rule_fields() -> Fields { ]) } +fn histogram_field(name: &str) -> Field { + Field::new( + name, + DataType::List(Arc::new(Field::new("item", DataType::UInt64, false))), + false, + ) +} + fn list_of(fields: Fields) -> DataType { DataType::List(Arc::new(Field::new( "item", @@ -510,11 +588,16 @@ pub(super) fn schema() -> Schema { u64_field("located_in"), u64_field("missing_in_osm"), u64_field("shared_osm_features"), - Field::new( - "distance_m", - DataType::List(Arc::new(Field::new("item", DataType::UInt64, false))), - false, - ), + u64_field("shared_positions"), + Field::new("offset_east_m", DataType::Float32, true), + Field::new("offset_north_m", DataType::Float32, true), + histogram_field("distance_m"), + u64_field("missing_no_candidate"), + u64_field("missing_rejected"), + u64_field("missing_taken"), + histogram_field("rejected_distance_m"), + u64_field("chance_sampled"), + u64_field("chance_matched"), Field::new("opening_hours_top_share", DataType::Float32, true), Field::new("rules", list_of(rule_fields()), false), Field::new("tags", list_of(tag_fields()), false), @@ -565,11 +648,23 @@ fn batch(rows: &[Row], lookups: &Lookups, schema: &Arc) -> Result) -> Result = s.rules.iter().collect(); @@ -685,6 +800,13 @@ fn batch(rows: &[Row], lookups: &Lookups, schema: &Arc) -> Result(8) .unwrap() .append_option(deletions); + let offset = median_offset(&brand.offsets); + b.field_builder::(9) + .unwrap() + .append_option(offset.map(|o| o.0)); + b.field_builder::(10) + .unwrap() + .append_option(offset.map(|o| o.1)); b.append(true); } brands.append(true); @@ -705,7 +827,16 @@ fn batch(rows: &[Row], lookups: &Lookups, schema: &Arc) -> Result().unwrap(); diff --git a/tests/integration_test.rs b/tests/integration_test.rs index 8dce78d..2068ce4 100644 --- a/tests/integration_test.rs +++ b/tests/integration_test.rs @@ -213,6 +213,32 @@ fn assert_spider_stats(path: &Path) -> Result<()> { // Two Winterthur features have no matcher, so no conflated row. assert_eq!(numbers("no_matcher"), [0, 0, 0, 0, 0, 2]); + // Why features are missing in OSM adds up to how many are. + let missing = numbers("missing_in_osm"); + let (no_candidate, rejected, taken) = ( + numbers("missing_no_candidate"), + numbers("missing_rejected"), + numbers("missing_taken"), + ); + for i in 0..spiders.len() { + assert_eq!(no_candidate[i] + rejected[i] + taken[i], missing[i]); + } + // One Misenso shop has a shop of the same kind right next to it in + // OSM, which the matcher rejected. + assert_eq!(rejected, [0, 0, 0, 1, 0, 0]); + // Small spiders move all their features for chance pairs. + assert_eq!(numbers("chance_sampled"), [1, 0, 1, 3, 1, 2]); + + let offset = column("offset_east_m"); + let offset = offset + .as_any() + .downcast_ref::() + .unwrap(); + let same = numbers("same"); + for (i, same) in same.iter().enumerate() { + assert_eq!(offset.is_valid(i), *same > 0, "offset of {}", spiders[i]); + } + // denner_ch matched 13.8 m away: bin 2, from 10 to 25 meters. let distance = column("distance_m"); let distance = distance.as_any().downcast_ref::().unwrap();