add first osm attempt (missing ways and relations)

.github/workflows/go.yml

@@ -0,0 +1,31 @@
+name: Go
+
+on:
+  push:
+    tags:
+      - 'v*.*.*'
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - name: Set up Go
+        uses: actions/setup-go@v2
+        with:
+          go-version: 1.16
+
+      - name: Build Windows
+        run: env GOOS=windows GOARCH=amd64 go build -o experive.exe -v cmd/cli/main.go
+
+      - name: Build Linux x64
+        run: env GOOS=linux GOARCH=amd64 go build -o experive cmd/cli/main.go
+
+      - name: Release
+        uses: softprops/action-gh-release@v1
+        with:
+          files: |
+            experive.exe
+            experive
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}

core/commands/connectors/db_connector.go

@@ -68,11 +68,17 @@ func (p *PgConnector) Write(data []map[string]interface{}) error {
 			case Text:
 				vals[i] = row[c.Name]
 			case Jsonb:
-				marshal, err := json.Marshal(row[c.Name])
-				if err != nil {
-					return err
+				value := row[c.Name]
+				switch value.(type) {
+				case string: // already marshalled
+					vals[i] = value
+				default:
+					marshal, err := json.Marshal(value)
+					if err != nil {
+						return err
+					}
+					vals[i] = string(marshal)
 				}
-				vals[i] = string(marshal)
 			case Point:
 				if lat, ok := row["latitude"]; ok {
 					if lng, ok := row["longitude"]; ok {

core/commands/json/osm_parser.go

@@ -0,0 +1,6 @@
+package json
+
+type OsmParser struct {
+	node int64
+	ways int64
+}

core/commands/json/osm_parser_test.go

@@ -0,0 +1,49 @@
+package json
+
+import (
+	"fmt"
+	"github.com/meekyphotos/experive-cli/core/commands/osm"
+	"io"
+	"os"
+	"runtime"
+	"testing"
+)
+
+func (o *OsmParser) CountNodes() {
+	open, err := os.Open("netherlands.osm.pbf")
+	if err != nil {
+		panic(err)
+	}
+	d := osm.NewDecoder(open)
+	d.SetBufferSize(osm.MaxBlobSize)
+	d.Skip(false, true, true)
+	if err := d.Start(runtime.GOMAXPROCS(-1)); err != nil {
+		panic(err)
+	}
+	for {
+		v, err := d.Decode()
+		if err == io.EOF {
+			break
+		}
+		if err != nil {
+			panic(err)
+		}
+
+		switch v.(type) {
+		case *osm.Node:
+			o.node++
+			//fmt.Println(v)
+		case *osm.Way:
+			o.ways++
+		case *osm.Relation:
+		default:
+		}
+	}
+}
+
+// 9.991.103 in 14.69s only counting and skipping ways and relations
+func TestParsing(t *testing.T) {
+	parser := OsmParser{}
+	parser.CountNodes()
+	fmt.Println(parser.node, parser.ways)
+}

core/commands/load_osm.go

@@ -1,11 +1,74 @@
 package commands
 
 import (
-	"fmt"
+	"github.com/meekyphotos/experive-cli/core/commands/connectors"
+	"github.com/meekyphotos/experive-cli/core/commands/pipeline"
+	"github.com/meekyphotos/experive-cli/core/utils"
 	"github.com/urfave/cli/v2"
+	"sync"
+	"time"
 )
 
+type OsmRunner struct {
+	Connector connectors.Connector
+}
+
+var osmFields = []connectors.Column{
+	{Name: "id", Type: connectors.Snowflake},
+	{Name: "osm_id", Type: connectors.Bigint, Indexed: true},
+	{Name: "class", Type: connectors.Text},
+	{Name: "type", Type: connectors.Text},
+	{Name: "names", Type: connectors.Jsonb},
+}
+
+func determineOsmCols(c *utils.Config) []connectors.Column {
+	cols := make([]connectors.Column, 0)
+	cols = append(cols, osmFields...)
+	if c.InclKeyValues {
+		cols = append(cols, connectors.Column{Name: "metadata", Type: connectors.Jsonb})
+	}
+	if c.UseGeom {
+		cols = append(cols, geomFields...)
+	} else {
+		cols = append(cols, latLngFields...)
+	}
+	return cols
+}
+
+func (r OsmRunner) Run(c *utils.Config) error {
+	dbErr := r.Connector.Connect()
+	if dbErr != nil {
+		return dbErr
+	}
+	defer r.Connector.Close()
+	dbErr = r.Connector.Init(determineOsmCols(c))
+	if dbErr != nil {
+		return dbErr
+	}
+	channel, err := pipeline.ReadFromPbf(c.File, &pipeline.NoopBeat{})
+	if err != nil {
+		return err
+	}
+	requests := pipeline.BatchRequest(channel, 10000, time.Second)
+	var pgWorkers sync.WaitGroup
+	pgWorkers.Add(1)
+	go func() {
+		err := pipeline.ProcessChannel(requests, r.Connector)
+		if err != nil {
+			panic(err)
+		}
+		pgWorkers.Done()
+	}()
+	pgWorkers.Wait()
+	return r.Connector.CreateIndexes()
+}
+
 func LoadOsmMeta() *cli.Command {
+	stdAction := utils.DatabaseLoader{
+		Runner:           OsmRunner{},
+		PasswordProvider: utils.TerminalPasswordReader{},
+	}
+
 	return &cli.Command{
 		Name:        "osm",
 		Usage:       "Load a osm dataset into target postgres",
@@ -25,14 +88,11 @@ func LoadOsmMeta() *cli.Command {
 			&cli.BoolFlag{Name: "latlong", Value: false, Usage: "Store coordinates in degrees of latitude & longitude."},
 			&cli.StringFlag{Name: "t", Aliases: []string{"table"}, Value: "planet_data", Usage: "Output table name"},
 
-			&cli.BoolFlag{Name: "j", Aliases: []string{"json"}, Value: false, Usage: "Add tags without column to an additional json (key/value) column in the database tables."},
+			&cli.BoolFlag{Name: "j", Aliases: []string{"json"}, Value: true, Usage: "Add tags without column to an additional json (key/value) column in the database tables."},
 
 			&cli.StringFlag{Name: "schema", Value: "public", Usage: "Use PostgreSQL schema SCHEMA for all tables, indexes, and functions in the pgsql output (default is no schema, i.e. the public schema is used)."},
 		},
 		UseShortOptionHandling: true,
-		Action: func(context *cli.Context) error {
-			fmt.Println("work in progress")
-			return nil
-		},
+		Action:                 stdAction.DoLoad,
 	}
 }

core/commands/osm/data.go

@@ -0,0 +1,68 @@
+package osm
+
+import "time"
+
+type BoundingBox struct {
+	Left   float64
+	Right  float64
+	Top    float64
+	Bottom float64
+}
+
+type Header struct {
+	BoundingBox                      *BoundingBox
+	RequiredFeatures                 []string
+	OptionalFeatures                 []string
+	WritingProgram                   string
+	Source                           string
+	OsmosisReplicationTimestamp      time.Time
+	OsmosisReplicationSequenceNumber int64
+	OsmosisReplicationBaseUrl        string
+}
+
+type Info struct {
+	Version   int32
+	Uid       int32
+	Timestamp time.Time
+	Changeset int64
+	User      string
+	Visible   bool
+}
+
+type Node struct {
+	OsmId int64
+	Class string
+	Type  string
+	Lat   float64
+	Lon   float64
+	Tags  string
+	Names string
+}
+
+type Way struct {
+	ID      int64
+	Tags    map[string]string
+	NodeIDs []int64
+	Info    Info
+}
+
+type MemberType int
+
+const (
+	NodeType MemberType = iota
+	WayType
+	RelationType
+)
+
+type Member struct {
+	ID   int64
+	Type MemberType
+	Role string
+}
+
+type Relation struct {
+	ID      int64
+	Tags    map[string]string
+	Members []Member
+	Info    Info
+}