"os"
"time"
- // prometheus imports
- "net/http"
- "github.com/prometheus/client_golang/prometheus"
- "github.com/prometheus/client_golang/prometheus/promauto"
- "github.com/prometheus/client_golang/prometheus/promhttp"
+ // prometheus imports
+ "github.com/prometheus/client_golang/prometheus"
+ "github.com/prometheus/client_golang/prometheus/promauto"
+ "github.com/prometheus/client_golang/prometheus/promhttp"
+ "net/http"
- "unpiege.net/rabbit_go.git/helper"
+ "unpiege.net/rabbit_go.git/helper"
)
var (
- tempGauge = promauto.NewGaugeVec(
- prometheus.GaugeOpts{
- Name: "temp_sensor",
- Help: "Used to measure temperatures in the real world",
- },
- []string{"scale", "location"},
- )
- //tempExpire = map[prometheus.Labels]time.Time
- //tempExpire = map[string]string
-
- tempExpire = make(map[[2]string]time.Time)
-
- staleTime = time.Duration(15 * time.Second)
-
+ tempGauge = promauto.NewGaugeVec(
+ prometheus.GaugeOpts{
+ Name: "temp_sensor",
+ Help: "Used to measure temperatures in the real world",
+ },
+ []string{"scale", "location"},
+ )
+ //tempExpire = map[prometheus.Labels]time.Time
+ //tempExpire = map[string]string
+
+ tempExpire = make(map[[2]string]time.Time)
+
+ staleTime = time.Duration(15 * time.Second)
)
-
type Device interface {
- getRoutingKey() string
+ getRoutingKey() string
}
type ZigDevice struct {
- routingKey string
+ routingKey string
}
type DIYDevice struct {
- *ZigDevice
+ *ZigDevice
}
type PowerDevice struct {
- *ZigDevice
- powerName string
- queryNeeded bool
+ *ZigDevice
+ powerName string
+ queryNeeded bool
}
-
-func (dev *DIYDevice) getRoutingKey() (string){
- return fmt.Sprintf("zigbee2mqtt.%s", dev.routingKey)
+func (dev *DIYDevice) getRoutingKey() string {
+ return fmt.Sprintf("zigbee2mqtt.%s", dev.routingKey)
}
-func NewDevice(routingKey string) (*ZigDevice){
- retval := ZigDevice{routingKey: routingKey}
- //retval.routingKey = routingKey
- return &retval
+func NewDevice(routingKey string) *ZigDevice {
+ retval := ZigDevice{routingKey: routingKey}
+ //retval.routingKey = routingKey
+ return &retval
}
-func NewDIYDevice(routingKey string) (*DIYDevice){
- return &DIYDevice{
- ZigDevice: NewDevice(routingKey),
- }
-
+func NewDIYDevice(routingKey string) *DIYDevice {
+ return &DIYDevice{
+ ZigDevice: NewDevice(routingKey),
+ }
+
}
//tempExpire := make(map[[2]string]time.Time)
}
}
-func handleTemp(obj map[string]interface{}){
- log.Printf("temp! %v", obj)
- celsius := obj["celsius"].(float64)
- location := obj["location"].(string)
- now := time.Now().UTC()
- // TODO: last update portion
-
- // do the label update here
- //log.Printf("celsius: %f location %s", celsius, location)
- tempGauge.With(prometheus.Labels{"scale": "celsius", "location": location}).Set(celsius)
- tempExpire[[2]string{"celsius", location}] = now
-
-}
-
-func handleDIY(obj map[string]interface{}, sendChannel chan helper.RabbitSend){
- log.Printf("diy! %v", obj)
- _, isAction := obj["action"]
- if isAction {
- // this is yet more JSON, so decode it
- data := obj["action"].(map[string]interface{})
-/* err := json.Unmarshal([]byte(obj["action"].(string)), &data)
- if err != nil {
- log.Printf("Unable to decode json action from diy, ignoring")
- return
- }*/
- var senseType string
-
- _, isTemp := data["celsius"]
- if isTemp {
- senseType = "temperature"
- } else {
- _, hasType := data["type"]
- if hasType {
- senseType = data["type"].(string)
- } else {
- log.Printf("Did not find any sense type, just returning")
- return
- }
- }
- // should have the senseType setup correctly now, if we're still here
- if senseType == "temperature" {
- tempMap := make(map[string]interface{}, 0)
- // these should all exist properly
- tempMap["fahrenheit"] = data["celsius"].(float64) * 9/5 + 32
- tempMap["celsius"] = data["celsius"].(float64)
- tempMap["location"] = data["location"].(string)
- log.Printf("would send %v", tempMap)
- // hardcoded temp routingKey for this type of measurement
- sendThis := helper.RabbitSend{Data: tempMap, RoutingKey: "lol", IncludeDate: true}
- sendChannel <- sendThis
- }
-
- } else {
- log.Printf("not action, ignoring")
- }
+func handleTemp(obj map[string]interface{}) {
+ log.Printf("temp! %v", obj)
+ celsius := obj["celsius"].(float64)
+ location := obj["location"].(string)
+ now := time.Now().UTC()
+ // TODO: last update portion
+
+ // do the label update here
+ //log.Printf("celsius: %f location %s", celsius, location)
+ tempGauge.With(prometheus.Labels{"scale": "celsius", "location": location}).Set(celsius)
+ tempExpire[[2]string{"celsius", location}] = now
+
+}
+
+func handleDIY(obj map[string]interface{}, sendChannel chan helper.RabbitSend) {
+ log.Printf("diy! %v", obj)
+ _, isAction := obj["action"]
+ if isAction {
+ // this is yet more JSON, so decode it
+ data := obj["action"].(map[string]interface{})
+ /* err := json.Unmarshal([]byte(obj["action"].(string)), &data)
+ if err != nil {
+ log.Printf("Unable to decode json action from diy, ignoring")
+ return
+ }*/
+ var senseType string
+
+ _, isTemp := data["celsius"]
+ if isTemp {
+ senseType = "temperature"
+ } else {
+ _, hasType := data["type"]
+ if hasType {
+ senseType = data["type"].(string)
+ } else {
+ log.Printf("Did not find any sense type, just returning")
+ return
+ }
+ }
+ // should have the senseType setup correctly now, if we're still here
+ dataMap := make(map[string]interface{}, 0)
+
+ if senseType == "temperature" {
+ // these should all exist properly
+ dataMap["fahrenheit"] = data["celsius"].(float64)*9/5 + 32
+ dataMap["celsius"] = data["celsius"].(float64)
+ dataMap["location"] = data["location"].(string)
+ log.Printf("would send %v", dataMap)
+ // hardcoded temp routingKey for this type of measurement
+ sendThis := helper.RabbitSend{Data: dataMap, RoutingKey: "lol", IncludeDate: true}
+ sendChannel <- sendThis
+ } else if senseType == "motion" {
+ dataMap["motion_detected"] = data["state"].(bool)
+ // copied from above, but it might change later so shrug
+ dataMap["location"] = data["location"].(string)
+ log.Printf("sending reprocessed motion: %v", dataMap)
+ sendThis := helper.RabbitSend{Data: dataMap, RoutingKey: "lol", IncludeDate: true}
+ sendChannel <- sendThis
+ } else {
+ log.Printf("error! sense type not detected, ignoring")
+ }
+
+ } else {
+ log.Printf("not action, ignoring")
+ }
}
func readLoop(channel chan helper.RabbitSend, devices []Device, rabbit helper.RabbitConfig) {
- for _, device := range devices {
- rabbit = helper.Bind(device.getRoutingKey(), rabbit)
- log.Printf("bound to %s", device.getRoutingKey())
- }
- rabbit = helper.Bind("temp", rabbit)
- log.Printf("bound to temp")
- deliveries := helper.StartConsuming(rabbit)
- for delivery := range deliveries {
- //log.Printf("got a delivery %+v", delivery)
- item := helper.DecodeDelivery(delivery)
- //log.Printf("%+v", item)
- if delivery.RoutingKey == "temp" {
- handleTemp(item)
- }
- for _, device := range devices {
- if device.getRoutingKey() == delivery.RoutingKey {
- log.Printf("Found device for routing key %s, %v", delivery.RoutingKey, device)
- switch t := device.(type) {
- case *DIYDevice:
- log.Printf("DIY device, sending to handle")
- handleDIY(item, channel)
- default:
- log.Printf("other device, type was %s", t)
- }
- }
- }
- //time.Sleep(time.Duration(0.5*float64(time.Second)))
- }
+ for _, device := range devices {
+ rabbit = helper.Bind(device.getRoutingKey(), rabbit)
+ log.Printf("bound to %s", device.getRoutingKey())
+ }
+ rabbit = helper.Bind("temp", rabbit)
+ log.Printf("bound to temp")
+ deliveries := helper.StartConsuming(rabbit)
+ for delivery := range deliveries {
+ //log.Printf("got a delivery %+v", delivery)
+ item := helper.DecodeDelivery(delivery)
+ //log.Printf("%+v", item)
+ if delivery.RoutingKey == "temp" {
+ handleTemp(item)
+ }
+ for _, device := range devices {
+ if device.getRoutingKey() == delivery.RoutingKey {
+ log.Printf("Found device for routing key %s, %v", delivery.RoutingKey, device)
+ switch t := device.(type) {
+ case *DIYDevice:
+ log.Printf("DIY device, sending to handle")
+ handleDIY(item, channel)
+ default:
+ log.Printf("other device, type was %s", t)
+ }
+ }
+ }
+ //time.Sleep(time.Duration(0.5*float64(time.Second)))
+ }
}
func timeLoop(channel chan helper.RabbitSend) {
for {
- time.Sleep(time.Duration(0.5*float64(time.Second)))
+ time.Sleep(time.Duration(0.5 * float64(time.Second)))
}
}
-func sendLoop(channel chan helper.RabbitSend, rabbit helper.RabbitConfig){
- for sendItem := range channel {
- log.Printf("would try to send here %v", sendItem)
- helper.SendData(sendItem, rabbit, true)
- //helper.SendData(sendData, rabbit, false)
- }
+func sendLoop(channel chan helper.RabbitSend, rabbit helper.RabbitConfig) {
+ for sendItem := range channel {
+ log.Printf("would try to send here %v", sendItem)
+ helper.SendData(sendItem, rabbit, true)
+ //helper.SendData(sendData, rabbit, false)
+ }
}
-func promSetup(){
- // prometheus setup
- http.Handle("/metrics", promhttp.Handler())
- http.ListenAndServe(":9099", nil)
+func promSetup() {
+ // prometheus setup
+ http.Handle("/metrics", promhttp.Handler())
+ http.ListenAndServe(":9099", nil)
}
+func expireStaleMetrics() {
+ now := time.Now().UTC()
+ // tempExpire
+ tempExpireValues := make([][2]string, 0)
+ for labels, lastUpdate := range tempExpire {
+ curStale := now.Sub(lastUpdate)
+ if curStale > staleTime {
+ log.Printf("stale time %v, expiring labels %v\tlastUpdate: %v", curStale, labels, lastUpdate)
+ tempExpireValues = append(tempExpireValues, labels)
+ }
+ }
-func expireStaleMetrics(){
- now := time.Now().UTC()
- // tempExpire
- tempExpireValues := make([][2]string, 0)
- for labels, lastUpdate := range tempExpire {
- curStale := now.Sub(lastUpdate)
- if curStale > staleTime {
- log.Printf("stale time %v, expiring labels %v\tlastUpdate: %v", curStale, labels, lastUpdate)
- tempExpireValues = append(tempExpireValues, labels)
- }
- }
+ // now we have the values to delete
+ for _, labels := range tempExpireValues {
+ complete := tempGauge.DeleteLabelValues(labels[:]...)
+ if complete {
+ delete(tempExpire, labels)
+ } else {
+ log.Panicf("oh no wasn't able to delete metrics")
+ }
+ }
- // now we have the values to delete
- for _, labels := range tempExpireValues {
- complete := tempGauge.DeleteLabelValues(labels[:]...)
- if complete {
- delete(tempExpire, labels)
- } else {
- log.Panicf("oh no wasn't able to delete metrics")
- }
- }
+}
+func setupLogging() {
+ log.SetFlags(log.Ldate | log.Ltime | log.Lshortfile)
}
func main() {
+ setupLogging()
+
rabbit := helper.SetupRabbit(os.Args[1], "", "reprocess") // config file, default routing key, source
const sleepTime time.Duration = 5 * time.Second
- devices := make([]Device, 0)
- devices = append(devices, NewDIYDevice("office_pico"))
+ devices := make([]Device, 0)
+ devices = append(devices, NewDIYDevice("office_pico"))
+ devices = append(devices, NewDIYDevice("stairway_pico"))
+ devices = append(devices, NewDIYDevice("stairway_pico"))
+ devices = append(devices, NewDIYDevice("downstairs_stairway_pico"))
+ devices = append(devices, NewDIYDevice("upstairs_pico"))
+ devices = append(devices, NewDIYDevice("data_closet_pico"))
//currentTemp, err := fetchTemp(weatherStation)
channel := make(chan helper.RabbitSend)
- go promSetup()
- go readLoop(channel, devices, rabbit)
+ go promSetup()
+ go readLoop(channel, devices, rabbit)
go timeLoop(channel)
- go sendLoop(channel, rabbit)
+ go sendLoop(channel, rabbit)
for true {
time.Sleep(sleepTime)
- go expireStaleMetrics()
+ go expireStaleMetrics()
}
}
"time"
//"github.com/leekchan/timeutil"
- "unpiege.net/rabbit_go.git/helper"
+ "unpiege.net/rabbit_go.git/helper"
)
// TimeFloat allows user to specify times in a more meaningful way
type TimeFloat struct {
- Hour int
- Minute int
- Second int
+ Hour int
+ Minute int
+ Second int
}
// TimeColor is one instance of RGB values changing over time
-type TimeColor struct{
- PeakTime TimeFloat
- Red float64
- Green float64
- Blue float64
- Extent TimeFloat
+type TimeColor struct {
+ PeakTime TimeFloat
+ Red float64
+ Green float64
+ Blue float64
+ Extent TimeFloat
}
-type RGB struct{
- Red float64
- Green float64
- Blue float64
+type RGB struct {
+ Red float64
+ Green float64
+ Blue float64
}
// make sure no value is greater than 1
func (rgb *RGB) Clamp() {
- rgb.Red = math.Min(1, rgb.Red)
- rgb.Green = math.Min(1, rgb.Green)
- rgb.Blue = math.Min(1, rgb.Blue)
+ rgb.Red = math.Min(1, rgb.Red)
+ rgb.Green = math.Min(1, rgb.Green)
+ rgb.Blue = math.Min(1, rgb.Blue)
}
func (rgb *RGB) Mult(mult float64) {
- rgb.Red = rgb.Red * mult
- rgb.Green = rgb.Green * mult
- rgb.Blue = rgb.Blue * mult
+ rgb.Red = rgb.Red * mult
+ rgb.Green = rgb.Green * mult
+ rgb.Blue = rgb.Blue * mult
}
+func (rgb *RGB) Add(newrgb RGB) RGB {
+ retval := RGB{
+ Red: rgb.Red + newrgb.Red,
+ Green: rgb.Green + newrgb.Green,
+ Blue: rgb.Blue + newrgb.Blue,
+ }
-func (rgb *RGB) Add(newrgb RGB) (RGB){
- retval := RGB{
- Red: rgb.Red+newrgb.Red,
- Green: rgb.Green+newrgb.Green,
- Blue: rgb.Blue+newrgb.Blue,
- }
-
- retval.Clamp()
- return retval
+ retval.Clamp()
+ return retval
}
func (tf *TimeFloat) Float() float64 {
- return float64(tf.Hour*3600 + tf.Minute*60 + tf.Second)/86400.0
+ return float64(tf.Hour*3600+tf.Minute*60+tf.Second) / 86400.0
}
// hack to get the same values from an actual time, since we can't define another method
func RealTimeFloat(t time.Time) float64 {
- return float64(t.Hour()*3600 + t.Minute()*60 + t.Second())/86400.0
+ return float64(t.Hour()*3600+t.Minute()*60+t.Second()) / 86400.0
}
-
-
-func NewTimeColor(peak TimeFloat, red float64, green float64, blue float64, extent TimeFloat) (TimeColor) {
- var retval TimeColor
- retval.PeakTime = peak
- retval.Red = red
- retval.Green = green
- retval.Blue = blue
- retval.Extent = extent
- return retval
+func NewTimeColor(peak TimeFloat, red float64, green float64, blue float64, extent TimeFloat) TimeColor {
+ var retval TimeColor
+ retval.PeakTime = peak
+ retval.Red = red
+ retval.Green = green
+ retval.Blue = blue
+ retval.Extent = extent
+ return retval
}
-func (tc *TimeColor) RGBValues(evaluationTime time.Time) (RGB){
- retval := RGB{tc.Red, tc.Green, tc.Blue}
- var multiplier float64
- // get the distance between the peak time, and where we are right now
- curFloat := RealTimeFloat(evaluationTime)
- tcFloat := tc.PeakTime.Float()
- // this accounts for all the wraparound issues when the date ticks over
- distances := []float64 {
- math.Abs(tcFloat - curFloat),
- math.Abs(tcFloat + 1 - curFloat),
- math.Abs(curFloat + 1 - tcFloat),
- }
- minDistance := distances[0]
- for _, val := range distances[1:] {
- if val < minDistance {
- minDistance = val
- }
- }
- //log.Printf("minDistance: %f", minDistance)
-
- if minDistance > tc.Extent.Float()/2 {
- //log.Printf("turned off")
- // multiplier is already initialized zero so do nothing
- } else {
- extentMultiplier := 1/tc.Extent.Float()
- multiplier = math.Cos(math.Pi * extentMultiplier * minDistance )
- }
- //log.Printf("multiplier is %f", multiplier)
- retval.Mult(multiplier)
-
- return retval
+func (tc *TimeColor) RGBValues(evaluationTime time.Time) RGB {
+ retval := RGB{tc.Red, tc.Green, tc.Blue}
+ var multiplier float64
+ // get the distance between the peak time, and where we are right now
+ curFloat := RealTimeFloat(evaluationTime)
+ tcFloat := tc.PeakTime.Float()
+ // this accounts for all the wraparound issues when the date ticks over
+ distances := []float64{
+ math.Abs(tcFloat - curFloat),
+ math.Abs(tcFloat + 1 - curFloat),
+ math.Abs(curFloat + 1 - tcFloat),
+ }
+ minDistance := distances[0]
+ for _, val := range distances[1:] {
+ if val < minDistance {
+ minDistance = val
+ }
+ }
+ //log.Printf("minDistance: %f", minDistance)
+
+ if minDistance > tc.Extent.Float()/2 {
+ //log.Printf("turned off")
+ // multiplier is already initialized zero so do nothing
+ } else {
+ extentMultiplier := 1 / tc.Extent.Float()
+ multiplier = math.Cos(math.Pi * extentMultiplier * minDistance)
+ }
+ //log.Printf("multiplier is %f", multiplier)
+ retval.Mult(multiplier)
+
+ return retval
}
func failOnError(err error, msg string) {
}
func timeLoop(channel chan helper.RabbitSend, timeColors []TimeColor) {
- loc, _ := time.LoadLocation("America/Los_Angeles")
+ loc, _ := time.LoadLocation("America/Los_Angeles")
for {
- //log.Printf("okay just looping here")
- curTime := time.Now().UTC().In(loc)
- //log.Printf("here's the current time %s", curTime)
- totalRGB := RGB{}
- for _, tc := range timeColors {
- //log.Printf("Evaluating RGB for %+v", tc)
- rgb := tc.RGBValues(curTime)
- //log.Printf("%+v", rgb)
- totalRGB = totalRGB.Add(rgb)
- }
- totalRGB.Mult(0.877)
- //log.Printf("TotalRGB: %+v", totalRGB)
-
- sendObj := make(map[string]interface{})
- sendObj["red"] = totalRGB.Red
- sendObj["green"] = totalRGB.Green
- sendObj["blue"] = totalRGB.Blue
- data := helper.RabbitSend{Data: sendObj}
-
- channel <- data
- time.Sleep(time.Duration(0.5*float64(time.Second)))
+ //log.Printf("okay just looping here")
+ curTime := time.Now().UTC().In(loc)
+ //log.Printf("here's the current time %s", curTime)
+ totalRGB := RGB{}
+ for _, tc := range timeColors {
+ //log.Printf("Evaluating RGB for %+v", tc)
+ rgb := tc.RGBValues(curTime)
+ //log.Printf("%+v", rgb)
+ totalRGB = totalRGB.Add(rgb)
+ }
+ totalRGB.Mult(0.877)
+ //log.Printf("TotalRGB: %+v", totalRGB)
+
+ sendObj := make(map[string]interface{})
+ sendObj["red"] = totalRGB.Red
+ sendObj["green"] = totalRGB.Green
+ sendObj["blue"] = totalRGB.Blue
+ data := helper.RabbitSend{Data: sendObj}
+
+ channel <- data
+ time.Sleep(time.Duration(0.5 * float64(time.Second)))
}
}
-func sendLoop(channel chan helper.RabbitSend, rabbit helper.RabbitConfig){
- for sendData := range channel {
- helper.SendData(sendData, rabbit, true)
- }
+func sendLoop(channel chan helper.RabbitSend, rabbit helper.RabbitConfig) {
+ for sendData := range channel {
+ helper.SendData(sendData, rabbit, false)
+ }
}
func main() {
//currentTemp, err := fetchTemp(weatherStation)
channel := make(chan helper.RabbitSend)
- all_tcs := make([]TimeColor, 0)
- all_tcs = append(all_tcs, NewTimeColor(TimeFloat{Hour: 12}, 0, 1, 0.5, TimeFloat{Hour:12}))
- all_tcs = append(all_tcs, NewTimeColor(TimeFloat{Hour: 2}, 1, 0.1, 0, TimeFloat{Hour:8}))
- all_tcs = append(all_tcs, NewTimeColor(TimeFloat{Hour: 20}, 0.25, 0.25, 0, TimeFloat{Hour:12}))
-
+ all_tcs := make([]TimeColor, 0)
+ all_tcs = append(all_tcs, NewTimeColor(TimeFloat{Hour: 12}, 0, 1, 0.5, TimeFloat{Hour: 12}))
+ all_tcs = append(all_tcs, NewTimeColor(TimeFloat{Hour: 2}, 1, 0.1, 0, TimeFloat{Hour: 8}))
+ all_tcs = append(all_tcs, NewTimeColor(TimeFloat{Hour: 20}, 0.25, 0.25, 0, TimeFloat{Hour: 12}))
go timeLoop(channel, all_tcs)
- go sendLoop(channel, rabbit)
+ go sendLoop(channel, rabbit)
for true {
time.Sleep(sleepTime)
}
projects / rabbit_go / commitdiff