mtsatellite/cmd/mtwebmapper/tilesupdater.go

// Copyright 2014 by Sascha L. Teichmann
// Use of this source code is governed by the MIT license
// that can be found in the LICENSE file.

package main

import (
	"encoding/json"
	"fmt"
	"image"
	"image/draw"
	"log"
	"net/http"
	"path/filepath"
	"strconv"
	"sync"

	"github.com/nfnt/resize"

	"bitbucket.org/s_l_teichmann/mtsatellite/common"
)

type tileUpdater struct {
	changes      map[xz]bool
	mapDir       string
	redisAddress string
	colors       *common.Colors
	workers      int
	cond         *sync.Cond
	mu           sync.Mutex
}

type xz struct {
	X int16
	Z int16
}

type xzm struct {
	P    xz
	Mask uint16
}

func (c xz) quantize() xz {
	return xz{X: (c.X - -1933) / 16, Z: (c.Z - -1933) / 16}
}

func (c xz) dequantize() xz {
	return xz{X: c.X*16 + -1933, Z: c.Z*16 + -1933}
}

func (c xz) parent() xzm {
	xp, xr := c.X>>1, uint16(c.X&1)
	zp, zr := c.Z>>1, uint16(c.Z&1)
	return xzm{
		P:    xz{X: xp, Z: zp},
		Mask: 1 << (zr<<1 | xr)}
}

func newTileUpdater(mapDir, redisAddress string, colors *common.Colors, workers int) *tileUpdater {
	tu := tileUpdater{
		mapDir:       mapDir,
		redisAddress: redisAddress,
		changes:      map[xz]bool{},
		colors:       colors,
		workers:      workers}
	tu.cond = sync.NewCond(&tu.mu)
	return &tu
}

func (tu *tileUpdater) ServeHTTP(rw http.ResponseWriter, r *http.Request) {
	var err error
	var newChanges []xz
	decoder := json.NewDecoder(r.Body)
	if err = decoder.Decode(&newChanges); err != nil {
		log.Printf("WARN: JSON document broken: %s", err)
		http.Error(rw, http.StatusText(http.StatusBadRequest), http.StatusBadRequest)
		return
	}

	if len(newChanges) > 0 {
		tu.cond.L.Lock()
		for _, c := range newChanges {
			tu.changes[c.quantize()] = true
		}
		tu.cond.L.Unlock()
		tu.cond.Signal()
	}

	rw.WriteHeader(http.StatusOK)
}

func (tu *tileUpdater) doUpdates() {

	for {
		var changes map[xz]bool
		tu.cond.L.Lock()
		for len(tu.changes) == 0 {
			tu.cond.Wait()
		}
		changes = tu.changes
		tu.changes = map[xz]bool{}
		tu.cond.L.Unlock()

		baseDir := filepath.Join(tu.mapDir, "8")

		jobs := make(chan xz)
		var done sync.WaitGroup

		for i, n := 0, min(tu.workers, len(changes)); i < n; i++ {
			var client *common.RedisClient
			var err error
			if client, err = common.NewRedisClient("tcp", tu.redisAddress); err != nil {
				log.Printf("WARN: Cannot connect to redis server: %s", err)
				continue
			}
			btc := common.NewBaseTileCreator(client, tu.colors, baseDir, true)
			done.Add(1)
			go updateBaseTiles(jobs, btc, &done)
		}

		parentJobs := make(map[xz]uint16)

		for c, _ := range changes {
			//log.Printf("job: %+v", c)
			jobs <- c
			pxz := c.parent()
			parentJobs[pxz.P] |= pxz.Mask
		}
		close(jobs)
		done.Wait()

		for level := 7; level >= 0; level-- {
			pJobs := make(chan xzm)
			for i, n := 0, min(len(parentJobs), tu.workers); i < n; i++ {
				done.Add(1)
				go updatePyramidTiles(level, tu.mapDir, pJobs, &done)
			}
			ppJobs := make(map[xz]uint16)
			for c, mask := range parentJobs {
				pJobs <- xzm{P: c, Mask: mask}
				pxz := c.parent()
				ppJobs[pxz.P] |= pxz.Mask
			}
			close(pJobs)
			done.Wait()
			parentJobs = ppJobs
		}
	}
}

func min(a, b int) int {
	if a < b {
		return a
	}
	return b
}

func updatePyramidTiles(level int, baseDir string, jobs chan xzm, done *sync.WaitGroup) {
	defer done.Done()
	scratch := image.NewRGBA(image.Rect(0, 0, 256, 256))

	for job := range jobs {
		if err := updatePyramidTile(scratch, level, baseDir, job); err != nil {
			log.Printf("Updating pyramid tile failed: %s", err)
		}
	}
}

/*
  (0,0)    (128, 0)
  (0, 128) (128, 128)
*/

var dps = [4]image.Point{
	image.Pt(0, 128),
	image.Pt(128, 128),
	image.Pt(0, 0),
	image.Pt(128, 0),
}

var ofs = [4][2]int{
	{0, 0},
	{1, 0},
	{0, 1},
	{1, 1}}

var windowSize = image.Pt(128, 128)

func updatePyramidTile(scratch *image.RGBA, level int, baseDir string, j xzm) error {

	var orig image.Image

	origPath := filepath.Join(
		baseDir,
		strconv.Itoa(level),
		strconv.Itoa(int(j.P.X)),
		fmt.Sprintf("%d.png", j.P.Z))

	for i := uint16(0); i < 4; i++ {
		if j.Mask&(1<<i) != 0 {
			//log.Printf("level %d: modified %d", level, i)
			o := ofs[i]
			bx, bz := int(2*j.P.X), int(2*j.P.Z)
			path := filepath.Join(
				baseDir,
				strconv.Itoa(level+1),
				strconv.Itoa(bx+o[0]),
				fmt.Sprintf("%d.png", bz+o[1]))
			img := common.LoadPNG(path)
			img = resize.Resize(128, 128, img, resize.Lanczos3)
			sr := img.Bounds()
			r := sr.Sub(sr.Min).Add(dps[i])
			draw.Draw(scratch, r, img, sr.Min, draw.Src)
		} else {
			// Load lazy
			if orig == nil {
				orig = common.LoadPNG(origPath)
			}
			//log.Printf("level %d: copied %d", level, i)
			min := orig.Bounds().Min.Add(dps[i])
			r := image.Rectangle{min, min.Add(windowSize)}
			draw.Draw(scratch, r, orig, min, draw.Src)
		}
	}

	return common.SaveAsPNGAtomic(origPath, scratch)
}

func updateBaseTiles(jobs chan xz, btc *common.BaseTileCreator, done *sync.WaitGroup) {
	defer btc.Close()
	defer done.Done()
	for job := range jobs {
		xz := job.dequantize()
		//log.Printf("%d/%d %d/%d", x, z, job.X, job.Z)
		if err := btc.CreateTile(xz.X-1, xz.Z-1, int(job.X), int(job.Z)); err != nil {
			log.Printf("WARN: create tile failed: %s", err)
		}
	}
}