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
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
|
package mountain_pass
import (
"context"
"database/sql"
"fmt"
"io"
"log/slog"
"os"
"strconv"
"strings"
"github.com/paulmach/osm"
"github.com/paulmach/osm/osmpbf"
)
const insertStatement = `
INSERT INTO mountain_passes (osm_id, name, elevation, latitude, longitude)
VALUES (?, ?, ?, ?, ?)
ON CONFLICT(osm_id) DO UPDATE SET
name = excluded.name,
elevation = excluded.elevation,
latitude = excluded.latitude,
longitude = excluded.longitude
`
// ExtractMountainPasses scans pbfPath for mountain pass nodes and upserts them
// into the database. An optional reporter is invoked as scanning progresses
// with bytes scanned and the total file size, so callers (e.g. the CLI) can
// render progress.
func ExtractMountainPasses(ctx context.Context, pbfPath string, db *sql.DB, reporters ...func(current, total int64)) (int, error) {
file, err := os.Open(pbfPath)
if err != nil {
return 0, fmt.Errorf("failed to open pbf file: %w", err)
}
defer file.Close()
info, err := file.Stat()
if err != nil {
return 0, fmt.Errorf("failed to stat pbf file: %w", err)
}
total := info.Size()
report := func(int64, int64) {}
if len(reporters) > 0 {
report = reporters[0]
}
reader := &countingReader{r: file}
scanner := osmpbf.New(ctx, reader, 4)
scanner.SkipWays = true
scanner.SkipRelations = true
scanner.FilterNode = isMountainPass
tx, err := db.BeginTx(ctx, nil)
if err != nil {
return 0, fmt.Errorf("failed to begin transaction: %w", err)
}
defer tx.Rollback()
stmt, err := tx.PrepareContext(ctx, insertStatement)
if err != nil {
return 0, fmt.Errorf("failed to prepare insert: %w", err)
}
defer stmt.Close()
count := 0
for scanner.Scan() {
report(reader.n, total)
node, ok := scanner.Object().(*osm.Node)
if !ok {
continue
}
elevation, hasElevation := parseElevation(node.Tags.Find("ele"))
var elevationValue any
if hasElevation {
elevationValue = elevation
}
if _, err := stmt.ExecContext(ctx, node.ID, node.Tags.Find("name"), elevationValue, node.Lat, node.Lon); err != nil {
return count, fmt.Errorf("failed to insert osm pass %d: %w", node.ID, err)
}
count++
}
if err := scanner.Err(); err != nil && err != io.EOF {
return count, fmt.Errorf("failed to scan pbf: %w", err)
}
if err := tx.Commit(); err != nil {
return count, fmt.Errorf("failed to commit transaction: %w", err)
}
slog.Info("Extracted mountain passes", "count", count, "source", pbfPath)
return count, nil
}
// countingReader tracks the number of bytes read from an underlying reader,
// used to report scan progress against the PBF file size.
type countingReader struct {
r io.Reader
n int64
}
func (c *countingReader) Read(p []byte) (int, error) {
n, err := c.r.Read(p)
c.n += int64(n)
return n, err
}
func isMountainPass(node *osm.Node) bool {
if node.Tags.Find("mountain_pass") == "yes" {
return true
}
return node.Tags.Find("natural") == "mountain_pass"
}
func parseElevation(ele string) (int, bool) {
ele = strings.TrimSpace(ele)
if ele == "" {
return 0, false
}
for candidate := range strings.FieldsSeq(ele) {
if value, err := strconv.Atoi(candidate); err == nil {
return value, true
}
if value, err := strconv.ParseFloat(candidate, 64); err == nil {
return int(value), true
}
}
return 0, false
}
|
