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();