Temperature -> CouchDB in go

templisten.go

package main

import (
	couch "code.google.com/p/couch-go"
	"encoding/json"
	"flag"
	"fmt"
	"log"
	"net"
	"reflect"
	"runtime"
	"strconv"
	"strings"
	"time"
)

// 2011/07/17 02:01:33 2011/07/17 02:01:32.953335	10E8C214000000E4	18.93	l=12.00,h=23.00
type reading struct {
	when    time.Time
	sensor  string
	reading float32
}

var timeInFormat string = "2006/01/02 15:04:05"
var timeOutFormat string = "2006-01-02T15:04:05"

var couchURL = flag.String("couch", "http://localhost:5984/temperature",
	"URL of the CouchDB")

func (r reading) TS() string {
	return r.when.Format(timeOutFormat)
}

func (r reading) MarshalJSON() ([]byte, error) {
	doc := map[string]interface{}{
		"type":    reflect.TypeOf(r).Name(),
		"reading": r.reading,
		"ts":      r.TS(),
		"sensor":  r.sensor,
	}
	return json.Marshal(doc)
}

func fatal(format string, v ...interface{}) {
	log.Printf(format, v...)
	runtime.Goexit()
}

func process(msg reading) {
	id := fmt.Sprintf("%s_%s", msg.TS(), msg.sensor)
	db, err := couch.Connect(*couchURL)
	if err != nil {
		fatal("Error connecting to DB:  %v", err)
	}

	_, _, ierr := db.InsertWith(msg, id)
	if ierr != nil {
		fatal("Error inserting new item:  %v", ierr)
	}
}

func read(s *net.UDPConn) {
	b := make([]byte, 256)
	for {
		n, e := s.Read(b)
		if e != nil {
			log.Fatalf("Error reading from socket:  %s", e)
		}
		parts := strings.Split(string(b[0:n]), "\t")
		f, ferr := strconv.ParseFloat(parts[2], 32)
		if ferr != nil {
			log.Printf("Error parsing float:  %s", ferr)
			continue
		}
		t, terr := time.Parse(timeInFormat, strings.Split(parts[0], ".")[0])
		if terr != nil {
			log.Printf("Error parsing time: %s", terr)
			continue
		}

		// Do this to convert to UTC
		// t = time.SecondsToUTC(t.Seconds() - int64(time.LocalTime().ZoneOffset))

		record := reading{
			when:    t,
			sensor:  parts[1],
			reading: float32(f),
		}
		go process(record)
	}
}

func main() {
	flag.Parse()

	socket, err := net.ListenMulticastUDP("udp4",
		nil, &net.UDPAddr{
			IP:   net.IPv4(225, 0, 0, 37),
			Port: 6789,
		})

	if err != nil {
		log.Fatalf("listen %s", err)
	}

	defer socket.Close()
	read(socket)
}