Compare commits
11 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
c37420a993
|
|||
|
ac0c417a48
|
|||
|
dc965aeba6
|
|||
|
b940f715c0
|
|||
|
8c5626942f
|
|||
|
6d0dc12ac1
|
|||
|
6a4aac4140
|
|||
|
77d23e39cf
|
|||
|
e28042f3be
|
|||
|
e1ad76f703
|
|||
|
6dac149a48
|
@@ -17,7 +17,7 @@ metadata:
|
|||||||
labels:
|
labels:
|
||||||
app: udi
|
app: udi
|
||||||
annotations:
|
annotations:
|
||||||
secret.reloader.stakater.com/reload: "%PRE%-udi-conf,%PRE%-udi-db-cred,%PRE%-mqtt-password"
|
secret.reloader.stakater.com/reload: "%PRE%-udi-conf,%PRE%-udi-db-cred,%PRE%-mqtt-password,%PRE%-udi-influxdb-cred"
|
||||||
spec:
|
spec:
|
||||||
replicas: 1
|
replicas: 1
|
||||||
selector:
|
selector:
|
||||||
@@ -36,6 +36,8 @@ spec:
|
|||||||
name: %PRE%-udi-db-cred
|
name: %PRE%-udi-db-cred
|
||||||
- secretRef:
|
- secretRef:
|
||||||
name: %PRE%-mqtt-password
|
name: %PRE%-mqtt-password
|
||||||
|
- secretRef:
|
||||||
|
name: %PRE%-udi-influxdb-cred
|
||||||
- configMapRef:
|
- configMapRef:
|
||||||
name: %PRE%-udi-conf
|
name: %PRE%-udi-conf
|
||||||
volumeMounts:
|
volumeMounts:
|
||||||
|
|||||||
@@ -67,6 +67,7 @@ for NAMESPACE_DIR in `find $INSTANCES_DIR -type d -mindepth 1 -maxdepth 1`; do
|
|||||||
DATABASE_VARIABLE=$VARIABLE_PREFIX"_PGDATABASE"
|
DATABASE_VARIABLE=$VARIABLE_PREFIX"_PGDATABASE"
|
||||||
NEW_UDI_DB_DATABASE="${!DATABASE_VARIABLE}"
|
NEW_UDI_DB_DATABASE="${!DATABASE_VARIABLE}"
|
||||||
NEW_UDI_DB_HOST=database.database1.svc.cluster.local
|
NEW_UDI_DB_HOST=database.database1.svc.cluster.local
|
||||||
|
INFLUXDB_URL=$VARIABLE_PREFIX"_INFLUXDB_URL"
|
||||||
|
|
||||||
kubectl create secret generic $INSTANCE-udi-db-cred \
|
kubectl create secret generic $INSTANCE-udi-db-cred \
|
||||||
--dry-run=client \
|
--dry-run=client \
|
||||||
@@ -79,6 +80,13 @@ for NAMESPACE_DIR in `find $INSTANCES_DIR -type d -mindepth 1 -maxdepth 1`; do
|
|||||||
--from-literal=PGSSLMODE="require" | \
|
--from-literal=PGSSLMODE="require" | \
|
||||||
kubectl apply -f - -n $NAMESPACE
|
kubectl apply -f - -n $NAMESPACE
|
||||||
|
|
||||||
|
kubectl create secret generic $INSTANCE-udi-influxdb-cred \
|
||||||
|
--dry-run=client \
|
||||||
|
-o yaml \
|
||||||
|
--save-config \
|
||||||
|
--from-literal=INFLUXDB_URL="${!INFLUXDB_URL}" | \
|
||||||
|
kubectl apply -f - -n $NAMESPACE
|
||||||
|
|
||||||
# set configuration as configMap
|
# set configuration as configMap
|
||||||
kubectl create configmap $INSTANCE-udi-conf \
|
kubectl create configmap $INSTANCE-udi-conf \
|
||||||
--from-literal=UDI_CONF="`cat $CONFIG_FILE`" \
|
--from-literal=UDI_CONF="`cat $CONFIG_FILE`" \
|
||||||
|
|||||||
@@ -105,6 +105,7 @@
|
|||||||
"devicePart": "3",
|
"devicePart": "3",
|
||||||
"valueFrom": "payload",
|
"valueFrom": "payload",
|
||||||
"valuePart": "1",
|
"valuePart": "1",
|
||||||
|
"valueType": "float",
|
||||||
"unitFrom": "payload",
|
"unitFrom": "payload",
|
||||||
"unitPart": "3"
|
"unitPart": "3"
|
||||||
}
|
}
|
||||||
@@ -165,7 +166,7 @@
|
|||||||
{
|
{
|
||||||
"topics": [ "shellyplusht/+/status/humidity:0" ],
|
"topics": [ "shellyplusht/+/status/humidity:0" ],
|
||||||
"handler": "SVEJ",
|
"handler": "SVEJ",
|
||||||
"id": "SVE4",
|
"id": "SVEJ3",
|
||||||
"config": {
|
"config": {
|
||||||
"databaseConnStr": "",
|
"databaseConnStr": "",
|
||||||
"attributes": {
|
"attributes": {
|
||||||
@@ -175,6 +176,48 @@
|
|||||||
"unitSelector": "C:%"
|
"unitSelector": "C:%"
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"topics": [ "shellies/sensor/+/status/temperature:0" ],
|
||||||
|
"handler": "SVEJ",
|
||||||
|
"id": "SVEJ4",
|
||||||
|
"config": {
|
||||||
|
"databaseConnStr": "",
|
||||||
|
"attributes": {
|
||||||
|
"application": "Shellies Sensor Temperature",
|
||||||
|
"deviceSelector": "T:2",
|
||||||
|
"valueSelector": "J:$.tC",
|
||||||
|
"unitSelector": "C:°C"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"topics": [ "shellies/sensor/+/status/humidity:0" ],
|
||||||
|
"handler": "SVEJ",
|
||||||
|
"id": "SVEJ5",
|
||||||
|
"config": {
|
||||||
|
"databaseConnStr": "",
|
||||||
|
"attributes": {
|
||||||
|
"application": "Shellies Sensor Humidity",
|
||||||
|
"deviceSelector": "T:2",
|
||||||
|
"valueSelector": "J:$.rh",
|
||||||
|
"unitSelector": "C:%"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"topics": [ "shellies/sensor/+/status/devicepower:0" ],
|
||||||
|
"handler": "SVEJ",
|
||||||
|
"id": "SVEJ6",
|
||||||
|
"config": {
|
||||||
|
"databaseConnStr": "",
|
||||||
|
"attributes": {
|
||||||
|
"application": "Shellies Sensor Power",
|
||||||
|
"deviceSelector": "T:2",
|
||||||
|
"valueSelector": "J:$.battery.percent",
|
||||||
|
"unitSelector": "C:%"
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
],
|
],
|
||||||
"archiver": {
|
"archiver": {
|
||||||
|
|||||||
@@ -1,7 +1,8 @@
|
|||||||
-----BEGIN PGP MESSAGE-----
|
-----BEGIN PGP MESSAGE-----
|
||||||
|
|
||||||
jA0ECQMIOpibMyX5/pH90mMBMdpIUJC/lR3BeUddyWJjwixQcePZIAQ/QEV6VO71
|
jA0ECQMIYUoTHR96Qfb90psBoxuk38UXPXTWPCmdW690bi2+w34S4NLHZvHfe3Ra
|
||||||
2cleQWnw8Ll5/bKtDv94UMLDZy86YdcNyeUXDP5cYPrGAFqWLaBmkLbpAJf7LRtb
|
nck319+PXvr0agfHGZ733hhTQv4sa8I2o6ICrgFqtKGfHmgnqL5kYNP9+NuV/IsF
|
||||||
nckgwR1+++GPvn5vNYcqGt92pB0=
|
x3dxwjEejsZ5GYn/zk+CQceItQ8nyyJc2ms1KwTu2r4hMzuHmnVtvKxNCzPrw2N5
|
||||||
=xy3G
|
SJIRhh41eequFkzELQqqXXu10raBFsttOemVhA==
|
||||||
|
=TySu
|
||||||
-----END PGP MESSAGE-----
|
-----END PGP MESSAGE-----
|
||||||
|
|||||||
@@ -40,7 +40,7 @@ func NewDatabaseHandle() *DatabaseHandle {
|
|||||||
})
|
})
|
||||||
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Printf("Unable to create InfluxDB client: %s", err)
|
log.Printf("Unable to create InfluxDB client (config: URL: %s, Username: %s, Password: %s): %s", influxDB, username, password, err)
|
||||||
db.initialized = false
|
db.initialized = false
|
||||||
return &db
|
return &db
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,16 +1,17 @@
|
|||||||
package svej
|
package svej
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"encoding/json"
|
||||||
|
"fmt"
|
||||||
"log"
|
"log"
|
||||||
"time"
|
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
"fmt"
|
"time"
|
||||||
"github.com/oliveagle/jsonpath"
|
|
||||||
"encoding/json"
|
|
||||||
"udi/config"
|
"udi/config"
|
||||||
"udi/handlers/handler"
|
|
||||||
"udi/database"
|
"udi/database"
|
||||||
|
"udi/handlers/handler"
|
||||||
|
|
||||||
|
"github.com/oliveagle/jsonpath"
|
||||||
)
|
)
|
||||||
|
|
||||||
type SingleValueExtractorJsonpathHandler struct {
|
type SingleValueExtractorJsonpathHandler struct {
|
||||||
@@ -33,7 +34,6 @@ T:TopicPartIndex
|
|||||||
C:ConstantValue
|
C:ConstantValue
|
||||||
*/
|
*/
|
||||||
|
|
||||||
|
|
||||||
func New(id string, config config.HandlerConfigT) handler.Handler {
|
func New(id string, config config.HandlerConfigT) handler.Handler {
|
||||||
t := &SingleValueExtractorJsonpathHandler{
|
t := &SingleValueExtractorJsonpathHandler{
|
||||||
ready: false,
|
ready: false,
|
||||||
@@ -76,12 +76,12 @@ func New(id string, config config.HandlerConfigT) handler.Handler {
|
|||||||
t.Id = id
|
t.Id = id
|
||||||
t.ready = true
|
t.ready = true
|
||||||
t.dbh = database.NewDatabaseHandle()
|
t.dbh = database.NewDatabaseHandle()
|
||||||
log.Printf("Handler SVEJ %d initialized", id)
|
log.Printf("Handler SVEJ %s initialized", id)
|
||||||
return t
|
return t
|
||||||
}
|
}
|
||||||
|
|
||||||
func (self *SingleValueExtractorJsonpathHandler) ExtractionHelper(subTopics []string, jPayload interface{}, selector string, jp *jsonpath.Compiled) (string, error) {
|
func (self *SingleValueExtractorJsonpathHandler) ExtractionHelper(subTopics []string, jPayload interface{}, selector string, jp *jsonpath.Compiled) (interface{}, error) {
|
||||||
var res string
|
var res interface{}
|
||||||
switch selector[:2] {
|
switch selector[:2] {
|
||||||
case "J:":
|
case "J:":
|
||||||
// extract using jsonpath from payload
|
// extract using jsonpath from payload
|
||||||
@@ -89,7 +89,7 @@ func (self *SingleValueExtractorJsonpathHandler) ExtractionHelper(subTopics []st
|
|||||||
if e != nil {
|
if e != nil {
|
||||||
return "", fmt.Errorf("jp.Lookup failed with %s", e)
|
return "", fmt.Errorf("jp.Lookup failed with %s", e)
|
||||||
}
|
}
|
||||||
res = fmt.Sprint(r)
|
res = r
|
||||||
case "T:":
|
case "T:":
|
||||||
// T: extract from topic
|
// T: extract from topic
|
||||||
i, e := strconv.Atoi(selector[2:])
|
i, e := strconv.Atoi(selector[2:])
|
||||||
@@ -109,7 +109,6 @@ func (self *SingleValueExtractorJsonpathHandler) ExtractionHelper(subTopics []st
|
|||||||
return res, nil
|
return res, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
func (self *SingleValueExtractorJsonpathHandler) Handle(message handler.MessageT) {
|
func (self *SingleValueExtractorJsonpathHandler) Handle(message handler.MessageT) {
|
||||||
if !self.ready {
|
if !self.ready {
|
||||||
self.Lost("Handler is not marked as ready", nil, message)
|
self.Lost("Handler is not marked as ready", nil, message)
|
||||||
@@ -149,12 +148,12 @@ func (self *SingleValueExtractorJsonpathHandler) Handle(message handler.MessageT
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
measurement.Device = device
|
measurement.Device = device.(string)
|
||||||
|
|
||||||
var variable database.VariableType
|
var variable database.VariableType
|
||||||
variable.Label = ""
|
variable.Label = ""
|
||||||
variable.Variable = ""
|
variable.Variable = ""
|
||||||
variable.Unit = unit
|
variable.Unit = unit.(string)
|
||||||
variable.Value = value
|
variable.Value = value
|
||||||
measurement.Values = make(map[string]database.VariableType)
|
measurement.Values = make(map[string]database.VariableType)
|
||||||
measurement.Values["Value"] = variable
|
measurement.Values["Value"] = variable
|
||||||
@@ -163,4 +162,3 @@ func (self *SingleValueExtractorJsonpathHandler) Handle(message handler.MessageT
|
|||||||
self.dbh.StoreMeasurement(&measurement)
|
self.dbh.StoreMeasurement(&measurement)
|
||||||
self.S()
|
self.S()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -1,17 +1,16 @@
|
|||||||
package sver
|
package sver
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"time"
|
"log"
|
||||||
|
"regexp"
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
"regexp"
|
"time"
|
||||||
"log"
|
|
||||||
"udi/config"
|
"udi/config"
|
||||||
"udi/handlers/handler"
|
|
||||||
"udi/database"
|
"udi/database"
|
||||||
|
"udi/handlers/handler"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
type SingleValueExtractorRegexHandler struct {
|
type SingleValueExtractorRegexHandler struct {
|
||||||
handler.CommonHandler
|
handler.CommonHandler
|
||||||
ready bool
|
ready bool
|
||||||
@@ -32,12 +31,12 @@ type localConfig struct {
|
|||||||
device string
|
device string
|
||||||
valueFrom string
|
valueFrom string
|
||||||
valuePart int
|
valuePart int
|
||||||
|
valueType string
|
||||||
unitFrom string
|
unitFrom string
|
||||||
unitPart int
|
unitPart int
|
||||||
unit string
|
unit string
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
func New(id string, config config.HandlerConfigT) handler.Handler {
|
func New(id string, config config.HandlerConfigT) handler.Handler {
|
||||||
t := &SingleValueExtractorRegexHandler{
|
t := &SingleValueExtractorRegexHandler{
|
||||||
ready: false,
|
ready: false,
|
||||||
@@ -88,6 +87,12 @@ func New(id string, config config.HandlerConfigT) handler.Handler {
|
|||||||
localConfig.valuePart = valuePart
|
localConfig.valuePart = valuePart
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if config.Attributes["valueType"] != "float" && config.Attributes["valueType"] != "string" {
|
||||||
|
log.Printf("Error: invalid value %s for valueType", config.Attributes["valueType"])
|
||||||
|
return t
|
||||||
|
}
|
||||||
|
localConfig.valueType = config.Attributes["valueType"]
|
||||||
|
|
||||||
if config.Attributes["unitFrom"] != PAYLOAD_SEL && config.Attributes["unitFrom"] != CONSTANT_SEL {
|
if config.Attributes["unitFrom"] != PAYLOAD_SEL && config.Attributes["unitFrom"] != CONSTANT_SEL {
|
||||||
log.Printf("Error: invalid value %s for unitFrom", config.Attributes["unitFrom"])
|
log.Printf("Error: invalid value %s for unitFrom", config.Attributes["unitFrom"])
|
||||||
return t
|
return t
|
||||||
@@ -161,6 +166,7 @@ func (self *SingleValueExtractorRegexHandler) Handle(message handler.MessageT) {
|
|||||||
variable.Label = ""
|
variable.Label = ""
|
||||||
variable.Variable = ""
|
variable.Variable = ""
|
||||||
|
|
||||||
|
var value string
|
||||||
switch self.config.valueFrom {
|
switch self.config.valueFrom {
|
||||||
case PAYLOAD_SEL:
|
case PAYLOAD_SEL:
|
||||||
if self.payloadRegex == nil {
|
if self.payloadRegex == nil {
|
||||||
@@ -171,9 +177,19 @@ func (self *SingleValueExtractorRegexHandler) Handle(message handler.MessageT) {
|
|||||||
self.Lost("valuePart out of range", nil, message)
|
self.Lost("valuePart out of range", nil, message)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
variable.Value = payloadMatches[self.config.valuePart]
|
value = payloadMatches[self.config.valuePart]
|
||||||
case PAYLOAD_FULL_SEL:
|
case PAYLOAD_FULL_SEL:
|
||||||
variable.Value = message.Payload
|
value = message.Payload
|
||||||
|
}
|
||||||
|
if self.config.valueType == "float" {
|
||||||
|
fValue, err := strconv.ParseFloat(value, 64)
|
||||||
|
if err != nil {
|
||||||
|
self.Lost("Unable to convert value to float", err, message)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
variable.Value = fValue
|
||||||
|
} else {
|
||||||
|
variable.Value = value
|
||||||
}
|
}
|
||||||
|
|
||||||
switch self.config.unitFrom {
|
switch self.config.unitFrom {
|
||||||
@@ -197,4 +213,3 @@ func (self *SingleValueExtractorRegexHandler) Handle(message handler.MessageT) {
|
|||||||
self.dbh.StoreMeasurement(&measurement)
|
self.dbh.StoreMeasurement(&measurement)
|
||||||
self.S()
|
self.S()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user