update more stuff
This commit is contained in:
Mats Tornberg
@@ -1,5 +1,5 @@
|
||||
{
|
||||
"name": "Go Telldus Matter Dev",
|
||||
"name": "Go Telldus Dev",
|
||||
"build": {
|
||||
"dockerfile": "Dockerfile",
|
||||
"context": ".."
|
||||
@@ -21,7 +21,8 @@
|
||||
"vscode": {
|
||||
"extensions": [
|
||||
"ms-vscode.Go",
|
||||
"ms-vscode.vscode-typescript-next"
|
||||
"ms-vscode.vscode-typescript-next",
|
||||
"golang.go"
|
||||
],
|
||||
"settings": {
|
||||
"go.toolsManagement.checkForUpdates": "local",
|
||||
|
||||
313
datastore/datastore.go
Normal file
313
datastore/datastore.go
Normal file
@@ -0,0 +1,313 @@
|
||||
package datastore
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"iter"
|
||||
"time"
|
||||
|
||||
_ "github.com/mattn/go-sqlite3"
|
||||
)
|
||||
|
||||
// DataStore handles all database operations
|
||||
type DataStore struct {
|
||||
db *sql.DB
|
||||
}
|
||||
|
||||
// New creates a new DataStore instance
|
||||
func New(dbPath string) (*DataStore, error) {
|
||||
db, err := sql.Open("sqlite3", dbPath)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
ds := &DataStore{db: db}
|
||||
if err := ds.initTables(); err != nil {
|
||||
db.Close()
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return ds, nil
|
||||
}
|
||||
|
||||
// Close closes the database connection
|
||||
func (ds *DataStore) Close() error {
|
||||
return ds.db.Close()
|
||||
}
|
||||
|
||||
// initTables creates the necessary database tables
|
||||
func (ds *DataStore) initTables() error {
|
||||
tables := []string{
|
||||
`CREATE TABLE IF NOT EXISTS devices (
|
||||
id INTEGER PRIMARY KEY,
|
||||
name TEXT,
|
||||
unique_id TEXT UNIQUE
|
||||
)`,
|
||||
`CREATE TABLE IF NOT EXISTS sensors (
|
||||
sensor_id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
protocol TEXT,
|
||||
model TEXT,
|
||||
id INTEGER,
|
||||
name TEXT,
|
||||
temperature_unique_id TEXT,
|
||||
humidity_unique_id TEXT,
|
||||
last_temperature TEXT,
|
||||
last_humidity TEXT,
|
||||
last_timestamp INTEGER,
|
||||
hidden INTEGER DEFAULT 0,
|
||||
UNIQUE(protocol, model, id)
|
||||
)`,
|
||||
`CREATE TABLE IF NOT EXISTS potential_devices (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
class TEXT,
|
||||
protocol TEXT,
|
||||
model TEXT,
|
||||
device_id TEXT,
|
||||
last_data TEXT,
|
||||
last_seen INTEGER
|
||||
)`,
|
||||
}
|
||||
|
||||
for _, table := range tables {
|
||||
if _, err := ds.db.Exec(table); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// UpsertDevice inserts or updates a device
|
||||
func (ds *DataStore) UpsertDevice(device *Device) error {
|
||||
_, err := ds.db.Exec(
|
||||
"INSERT OR REPLACE INTO devices (id, name, unique_id) VALUES (?, ?, ?)",
|
||||
device.ID, device.Name, device.UniqueID,
|
||||
)
|
||||
return err
|
||||
}
|
||||
|
||||
// GetDevice retrieves a device by ID
|
||||
func (ds *DataStore) GetDevice(id int) (*Device, error) {
|
||||
device := &Device{}
|
||||
err := ds.db.QueryRow(
|
||||
"SELECT id, name, unique_id FROM devices WHERE id = ?",
|
||||
id,
|
||||
).Scan(&device.ID, &device.Name, &device.UniqueID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return device, nil
|
||||
}
|
||||
|
||||
// ListDevices returns an iterator over all devices
|
||||
func (ds *DataStore) ListDevices() iter.Seq[*Device] {
|
||||
return func(yield func(*Device) bool) {
|
||||
rows, err := ds.db.Query("SELECT id, name, unique_id FROM devices ORDER BY id")
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
for rows.Next() {
|
||||
device := &Device{}
|
||||
if err := rows.Scan(&device.ID, &device.Name, &device.UniqueID); err != nil {
|
||||
continue
|
||||
}
|
||||
if !yield(device) {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// UpdateDeviceName updates a device's name
|
||||
func (ds *DataStore) UpdateDeviceName(id int, name string) error {
|
||||
_, err := ds.db.Exec("UPDATE devices SET name = ? WHERE id = ?", name, id)
|
||||
return err
|
||||
}
|
||||
|
||||
// UpsertSensor inserts or updates a sensor
|
||||
func (ds *DataStore) UpsertSensor(sensor *Sensor) error {
|
||||
_, err := ds.db.Exec(
|
||||
`INSERT OR IGNORE INTO sensors
|
||||
(protocol, model, id, name, temperature_unique_id, humidity_unique_id)
|
||||
VALUES (?, ?, ?, ?, ?, ?)`,
|
||||
sensor.Protocol, sensor.Model, sensor.ID, sensor.Name,
|
||||
sensor.TemperatureUniqueID, sensor.HumidityUniqueID,
|
||||
)
|
||||
return err
|
||||
}
|
||||
|
||||
// GetSensor retrieves a sensor by sensor_id
|
||||
func (ds *DataStore) GetSensor(sensorID int) (*Sensor, error) {
|
||||
sensor := &Sensor{}
|
||||
var lastTemp, lastHum sql.NullString
|
||||
var lastTs sql.NullInt64
|
||||
var hidden int
|
||||
|
||||
err := ds.db.QueryRow(
|
||||
`SELECT sensor_id, protocol, model, id, name,
|
||||
temperature_unique_id, humidity_unique_id,
|
||||
last_temperature, last_humidity, last_timestamp, hidden
|
||||
FROM sensors WHERE sensor_id = ?`,
|
||||
sensorID,
|
||||
).Scan(
|
||||
&sensor.SensorID, &sensor.Protocol, &sensor.Model, &sensor.ID, &sensor.Name,
|
||||
&sensor.TemperatureUniqueID, &sensor.HumidityUniqueID,
|
||||
&lastTemp, &lastHum, &lastTs, &hidden,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
sensor.LastTemperature = lastTemp.String
|
||||
sensor.LastHumidity = lastHum.String
|
||||
sensor.LastTimestamp = lastTs.Int64
|
||||
sensor.Hidden = hidden != 0
|
||||
|
||||
return sensor, nil
|
||||
}
|
||||
|
||||
// GetSensorByIdentity retrieves a sensor by protocol, model, and id
|
||||
func (ds *DataStore) GetSensorByIdentity(protocol, model string, id int) (*Sensor, error) {
|
||||
sensor := &Sensor{}
|
||||
var lastTemp, lastHum sql.NullString
|
||||
var lastTs sql.NullInt64
|
||||
var hidden int
|
||||
|
||||
err := ds.db.QueryRow(
|
||||
`SELECT sensor_id, protocol, model, id, name,
|
||||
temperature_unique_id, humidity_unique_id,
|
||||
last_temperature, last_humidity, last_timestamp, hidden
|
||||
FROM sensors WHERE protocol = ? AND model = ? AND id = ?`,
|
||||
protocol, model, id,
|
||||
).Scan(
|
||||
&sensor.SensorID, &sensor.Protocol, &sensor.Model, &sensor.ID, &sensor.Name,
|
||||
&sensor.TemperatureUniqueID, &sensor.HumidityUniqueID,
|
||||
&lastTemp, &lastHum, &lastTs, &hidden,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
sensor.LastTemperature = lastTemp.String
|
||||
sensor.LastHumidity = lastHum.String
|
||||
sensor.LastTimestamp = lastTs.Int64
|
||||
sensor.Hidden = hidden != 0
|
||||
|
||||
return sensor, nil
|
||||
}
|
||||
|
||||
// ListSensors returns an iterator over all sensors
|
||||
func (ds *DataStore) ListSensors() iter.Seq[*Sensor] {
|
||||
return func(yield func(*Sensor) bool) {
|
||||
rows, err := ds.db.Query(
|
||||
`SELECT sensor_id, protocol, model, id, name,
|
||||
temperature_unique_id, humidity_unique_id,
|
||||
last_temperature, last_humidity, last_timestamp, hidden
|
||||
FROM sensors ORDER BY sensor_id`,
|
||||
)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
for rows.Next() {
|
||||
sensor := &Sensor{}
|
||||
var lastTemp, lastHum sql.NullString
|
||||
var lastTs sql.NullInt64
|
||||
var hidden int
|
||||
|
||||
if err := rows.Scan(
|
||||
&sensor.SensorID, &sensor.Protocol, &sensor.Model, &sensor.ID, &sensor.Name,
|
||||
&sensor.TemperatureUniqueID, &sensor.HumidityUniqueID,
|
||||
&lastTemp, &lastHum, &lastTs, &hidden,
|
||||
); err != nil {
|
||||
continue
|
||||
}
|
||||
|
||||
sensor.LastTemperature = lastTemp.String
|
||||
sensor.LastHumidity = lastHum.String
|
||||
sensor.LastTimestamp = lastTs.Int64
|
||||
sensor.Hidden = hidden != 0
|
||||
|
||||
if !yield(sensor) {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// UpdateSensorName updates a sensor's name
|
||||
func (ds *DataStore) UpdateSensorName(sensorID int, name string) error {
|
||||
_, err := ds.db.Exec("UPDATE sensors SET name = ? WHERE sensor_id = ?", name, sensorID)
|
||||
return err
|
||||
}
|
||||
|
||||
// UpdateSensorValue updates a sensor's last value
|
||||
func (ds *DataStore) UpdateSensorValue(protocol, model string, id int, dataType int, value string) error {
|
||||
column := ""
|
||||
switch dataType {
|
||||
case 1: // Temperature
|
||||
column = "last_temperature"
|
||||
case 2: // Humidity
|
||||
column = "last_humidity"
|
||||
default:
|
||||
return fmt.Errorf("unsupported data type: %d", dataType)
|
||||
}
|
||||
|
||||
query := fmt.Sprintf(
|
||||
"UPDATE sensors SET %s = ?, last_timestamp = ? WHERE protocol = ? AND model = ? AND id = ?",
|
||||
column,
|
||||
)
|
||||
_, err := ds.db.Exec(query, value, time.Now().Unix(), protocol, model, id)
|
||||
return err
|
||||
}
|
||||
|
||||
// SetSensorHidden updates a sensor's hidden status
|
||||
func (ds *DataStore) SetSensorHidden(sensorID int, hidden bool) error {
|
||||
hiddenVal := 0
|
||||
if hidden {
|
||||
hiddenVal = 1
|
||||
}
|
||||
_, err := ds.db.Exec("UPDATE sensors SET hidden = ? WHERE sensor_id = ?", hiddenVal, sensorID)
|
||||
return err
|
||||
}
|
||||
|
||||
// UpsertPotentialDevice inserts or updates a potential device
|
||||
func (ds *DataStore) UpsertPotentialDevice(device *PotentialDevice) error {
|
||||
_, err := ds.db.Exec(
|
||||
`INSERT OR REPLACE INTO potential_devices
|
||||
(class, protocol, model, device_id, last_data, last_seen)
|
||||
VALUES (?, ?, ?, ?, ?, ?)`,
|
||||
device.Class, device.Protocol, device.Model, device.DeviceID,
|
||||
device.LastData, device.LastSeen,
|
||||
)
|
||||
return err
|
||||
}
|
||||
|
||||
// ListPotentialDevices returns an iterator over all potential devices
|
||||
func (ds *DataStore) ListPotentialDevices() iter.Seq[*PotentialDevice] {
|
||||
return func(yield func(*PotentialDevice) bool) {
|
||||
rows, err := ds.db.Query(
|
||||
`SELECT id, class, protocol, model, device_id, last_data, last_seen
|
||||
FROM potential_devices ORDER BY last_seen DESC`,
|
||||
)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
for rows.Next() {
|
||||
device := &PotentialDevice{}
|
||||
if err := rows.Scan(
|
||||
&device.ID, &device.Class, &device.Protocol, &device.Model,
|
||||
&device.DeviceID, &device.LastData, &device.LastSeen,
|
||||
); err != nil {
|
||||
continue
|
||||
}
|
||||
if !yield(device) {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
53
datastore/types.go
Normal file
53
datastore/types.go
Normal file
@@ -0,0 +1,53 @@
|
||||
package datastore
|
||||
|
||||
import "time"
|
||||
|
||||
// Device represents a Telldus device
|
||||
type Device struct {
|
||||
ID int `json:"id"`
|
||||
Name string `json:"name"`
|
||||
UniqueID string `json:"unique_id"`
|
||||
}
|
||||
|
||||
// Sensor represents a Telldus sensor
|
||||
type Sensor struct {
|
||||
SensorID int `json:"sensor_id"`
|
||||
Protocol string `json:"protocol"`
|
||||
Model string `json:"model"`
|
||||
ID int `json:"id"`
|
||||
Name string `json:"name"`
|
||||
TemperatureUniqueID string `json:"temperature_unique_id"`
|
||||
HumidityUniqueID string `json:"humidity_unique_id"`
|
||||
LastTemperature string `json:"last_temperature,omitempty"`
|
||||
LastHumidity string `json:"last_humidity,omitempty"`
|
||||
LastTimestamp int64 `json:"last_timestamp,omitempty"`
|
||||
Hidden bool `json:"hidden"`
|
||||
}
|
||||
|
||||
// PotentialDevice represents a detected but not yet configured device
|
||||
type PotentialDevice struct {
|
||||
ID int `json:"id"`
|
||||
Class string `json:"class"`
|
||||
Protocol string `json:"protocol"`
|
||||
Model string `json:"model"`
|
||||
DeviceID string `json:"device_id"`
|
||||
LastData string `json:"last_data"`
|
||||
LastSeen int64 `json:"last_seen"`
|
||||
}
|
||||
|
||||
// RawEvent represents a raw device event
|
||||
type RawEvent struct {
|
||||
Timestamp time.Time `json:"timestamp"`
|
||||
ControllerID int `json:"controller_id"`
|
||||
Data string `json:"data"`
|
||||
}
|
||||
|
||||
// SensorEvent represents a sensor data event
|
||||
type SensorEvent struct {
|
||||
Timestamp time.Time `json:"timestamp"`
|
||||
Protocol string `json:"protocol"`
|
||||
Model string `json:"model"`
|
||||
ID int `json:"id"`
|
||||
DataType int `json:"data_type"`
|
||||
Value string `json:"value"`
|
||||
}
|
||||
169
devices/manager.go
Normal file
169
devices/manager.go
Normal file
@@ -0,0 +1,169 @@
|
||||
package devices
|
||||
|
||||
import (
|
||||
"app/datastore"
|
||||
"app/mqtt"
|
||||
"app/telldus"
|
||||
"fmt"
|
||||
"log"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// EventManager handles telldus events and callbacks
|
||||
type EventManager struct {
|
||||
store *datastore.DataStore
|
||||
mqttClient *mqtt.Client
|
||||
rawEvents []datastore.RawEvent
|
||||
sensorEvents []datastore.SensorEvent
|
||||
mu sync.Mutex
|
||||
maxEvents int
|
||||
}
|
||||
|
||||
// NewEventManager creates a new event manager
|
||||
func NewEventManager(store *datastore.DataStore, mqttClient *mqtt.Client, maxEvents int) *EventManager {
|
||||
return &EventManager{
|
||||
store: store,
|
||||
mqttClient: mqttClient,
|
||||
maxEvents: maxEvents,
|
||||
}
|
||||
}
|
||||
|
||||
// GetRawEvents returns a copy of raw events
|
||||
func (em *EventManager) GetRawEvents() []datastore.RawEvent {
|
||||
em.mu.Lock()
|
||||
defer em.mu.Unlock()
|
||||
events := make([]datastore.RawEvent, len(em.rawEvents))
|
||||
copy(events, em.rawEvents)
|
||||
return events
|
||||
}
|
||||
|
||||
// GetSensorEvents returns a copy of sensor events
|
||||
func (em *EventManager) GetSensorEvents() []datastore.SensorEvent {
|
||||
em.mu.Lock()
|
||||
defer em.mu.Unlock()
|
||||
events := make([]datastore.SensorEvent, len(em.sensorEvents))
|
||||
copy(events, em.sensorEvents)
|
||||
return events
|
||||
}
|
||||
|
||||
// HandleDeviceEvent handles device state change events
|
||||
func (em *EventManager) HandleDeviceEvent(deviceID, method int, data string, callbackID int) {
|
||||
fmt.Printf("Device event: ID=%d, Method=%d, Data=%s\n", deviceID, method, data)
|
||||
|
||||
var state string
|
||||
switch method {
|
||||
case telldus.MethodTurnOn:
|
||||
state = "ON"
|
||||
case telldus.MethodTurnOff:
|
||||
state = "OFF"
|
||||
}
|
||||
|
||||
if state != "" {
|
||||
em.mqttClient.PublishDeviceState(deviceID, state)
|
||||
}
|
||||
}
|
||||
|
||||
// HandleSensorEvent handles sensor data events
|
||||
func (em *EventManager) HandleSensorEvent(protocol, model string, id, dataType int, value string, timestamp, callbackID int) {
|
||||
fmt.Printf("Sensor event: Protocol=%s, Model=%s, ID=%d, Type=%d, Value=%s, Timestamp=%d\n",
|
||||
protocol, model, id, dataType, value, timestamp)
|
||||
|
||||
// Publish to MQTT
|
||||
em.mqttClient.PublishSensorValue(protocol, model, id, dataType, value)
|
||||
|
||||
// Store in history
|
||||
em.mu.Lock()
|
||||
em.sensorEvents = append(em.sensorEvents, datastore.SensorEvent{
|
||||
Timestamp: time.Now(),
|
||||
Protocol: protocol,
|
||||
Model: model,
|
||||
ID: id,
|
||||
DataType: dataType,
|
||||
Value: value,
|
||||
})
|
||||
if len(em.sensorEvents) > em.maxEvents {
|
||||
em.sensorEvents = em.sensorEvents[1:]
|
||||
}
|
||||
em.mu.Unlock()
|
||||
|
||||
// Update last value in DB
|
||||
if err := em.store.UpdateSensorValue(protocol, model, id, dataType, value); err != nil {
|
||||
log.Printf("Error updating sensor %s %s %d: %v", protocol, model, id, err)
|
||||
}
|
||||
}
|
||||
|
||||
// HandleRawDeviceEvent handles raw device detection events
|
||||
func (em *EventManager) HandleRawDeviceEvent(data string, controllerID, callbackID int) {
|
||||
fmt.Printf("Raw device event: ControllerID=%d, Data=%s\n", controllerID, data)
|
||||
|
||||
// Parse data
|
||||
fields := strings.Split(data, ";")
|
||||
var class, protocol, model, deviceID string
|
||||
for _, field := range fields {
|
||||
kv := strings.SplitN(field, ":", 2)
|
||||
if len(kv) == 2 {
|
||||
key, val := kv[0], kv[1]
|
||||
switch key {
|
||||
case "class":
|
||||
class = val
|
||||
case "protocol":
|
||||
protocol = val
|
||||
case "model":
|
||||
model = val
|
||||
case "id":
|
||||
deviceID = val
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Store in potential_devices
|
||||
potentialDev := &datastore.PotentialDevice{
|
||||
Class: class,
|
||||
Protocol: protocol,
|
||||
Model: model,
|
||||
DeviceID: deviceID,
|
||||
LastData: data,
|
||||
LastSeen: time.Now().Unix(),
|
||||
}
|
||||
if err := em.store.UpsertPotentialDevice(potentialDev); err != nil {
|
||||
log.Printf("Error storing potential device: %v", err)
|
||||
}
|
||||
|
||||
// If sensor, ensure in sensors table
|
||||
if class == "sensor" {
|
||||
idInt, _ := strconv.Atoi(deviceID)
|
||||
sensor := &datastore.Sensor{
|
||||
Protocol: protocol,
|
||||
Model: model,
|
||||
ID: idInt,
|
||||
Name: fmt.Sprintf("%s %s %s", protocol, model, deviceID),
|
||||
TemperatureUniqueID: fmt.Sprintf("telldus_sensor_%s_%s_%s_temperature", protocol, model, deviceID),
|
||||
HumidityUniqueID: fmt.Sprintf("telldus_sensor_%s_%s_%s_humidity", protocol, model, deviceID),
|
||||
}
|
||||
if err := em.store.UpsertSensor(sensor); err != nil {
|
||||
log.Printf("Error inserting sensor from raw: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// Log the raw event data
|
||||
em.mu.Lock()
|
||||
em.rawEvents = append(em.rawEvents, datastore.RawEvent{
|
||||
Timestamp: time.Now(),
|
||||
ControllerID: controllerID,
|
||||
Data: data,
|
||||
})
|
||||
if len(em.rawEvents) > em.maxEvents {
|
||||
em.rawEvents = em.rawEvents[1:]
|
||||
}
|
||||
em.mu.Unlock()
|
||||
}
|
||||
|
||||
// RegisterCallbacks registers all event callbacks with telldus
|
||||
func (em *EventManager) RegisterCallbacks() {
|
||||
telldus.RegisterDeviceEvent(em.HandleDeviceEvent)
|
||||
telldus.RegisterSensorEvent(em.HandleSensorEvent)
|
||||
telldus.RegisterRawDeviceEvent(em.HandleRawDeviceEvent)
|
||||
}
|
||||
110
devices/sync.go
Normal file
110
devices/sync.go
Normal file
@@ -0,0 +1,110 @@
|
||||
package devices
|
||||
|
||||
import (
|
||||
"app/datastore"
|
||||
"app/telldus"
|
||||
"fmt"
|
||||
"log"
|
||||
)
|
||||
|
||||
// Syncer handles synchronization of devices and sensors to the database
|
||||
type Syncer struct {
|
||||
store *datastore.DataStore
|
||||
}
|
||||
|
||||
// NewSyncer creates a new device/sensor syncer
|
||||
func NewSyncer(store *datastore.DataStore) *Syncer {
|
||||
return &Syncer{store: store}
|
||||
}
|
||||
|
||||
// SyncDevices synchronizes all telldus devices to the database
|
||||
func (s *Syncer) SyncDevices() error {
|
||||
numDevices := telldus.GetNumberOfDevices()
|
||||
for i := 0; i < numDevices; i++ {
|
||||
deviceID := telldus.GetDeviceId(i)
|
||||
name := telldus.GetName(deviceID)
|
||||
device := &datastore.Device{
|
||||
ID: deviceID,
|
||||
Name: name,
|
||||
UniqueID: fmt.Sprintf("telldus_device_%d", deviceID),
|
||||
}
|
||||
if err := s.store.UpsertDevice(device); err != nil {
|
||||
log.Printf("Error upserting device %d: %v", deviceID, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// SyncSensors synchronizes all telldus sensors to the database
|
||||
func (s *Syncer) SyncSensors() error {
|
||||
var protocol, model string
|
||||
var id, dataTypes int
|
||||
ret := telldus.Sensor(&protocol, &model, &id, &dataTypes)
|
||||
|
||||
for ret == 0 {
|
||||
sensor := &datastore.Sensor{
|
||||
Protocol: protocol,
|
||||
Model: model,
|
||||
ID: id,
|
||||
Name: fmt.Sprintf("%s %s %d", protocol, model, id),
|
||||
}
|
||||
if dataTypes&telldus.DataTypeTemperature != 0 {
|
||||
sensor.TemperatureUniqueID = fmt.Sprintf("telldus_sensor_%s_%s_%d_temperature", protocol, model, id)
|
||||
}
|
||||
if dataTypes&telldus.DataTypeHumidity != 0 {
|
||||
sensor.HumidityUniqueID = fmt.Sprintf("telldus_sensor_%s_%s_%d_humidity", protocol, model, id)
|
||||
}
|
||||
if err := s.store.UpsertSensor(sensor); err != nil {
|
||||
log.Printf("Error upserting sensor %s %s %d: %v", protocol, model, id, err)
|
||||
}
|
||||
ret = telldus.Sensor(&protocol, &model, &id, &dataTypes)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ListDevices prints all devices to stdout
|
||||
func (s *Syncer) ListDevices() {
|
||||
numDevices := telldus.GetNumberOfDevices()
|
||||
if numDevices < 0 {
|
||||
errStr := telldus.GetErrorString(numDevices)
|
||||
fmt.Printf("Error fetching devices: %s\n", errStr)
|
||||
return
|
||||
}
|
||||
fmt.Printf("Number of devices: %d\n", numDevices)
|
||||
for i := 0; i < numDevices; i++ {
|
||||
deviceID := telldus.GetDeviceId(i)
|
||||
name := telldus.GetName(deviceID)
|
||||
protocol := telldus.GetProtocol(deviceID)
|
||||
model := telldus.GetModel(deviceID)
|
||||
fmt.Printf("%d\t%s\tProtocol: %s\tModel: %s\n", deviceID, name, protocol, model)
|
||||
}
|
||||
}
|
||||
|
||||
// ListSensors prints all sensors to stdout
|
||||
func (s *Syncer) ListSensors() {
|
||||
fmt.Println("\nSENSORS:")
|
||||
var protocol, model string
|
||||
var id, dataTypes int
|
||||
ret := telldus.Sensor(&protocol, &model, &id, &dataTypes)
|
||||
if ret == 0 {
|
||||
for {
|
||||
fmt.Printf("Protocol: %s, Model: %s, ID: %d, DataTypes: %d\n", protocol, model, id, dataTypes)
|
||||
// Fetch values if available
|
||||
if dataTypes&telldus.DataTypeTemperature != 0 {
|
||||
value, timestamp, _ := telldus.SensorValue(protocol, model, id, telldus.DataTypeTemperature)
|
||||
fmt.Printf(" Temperature: %s°C at %d\n", value, timestamp)
|
||||
}
|
||||
if dataTypes&telldus.DataTypeHumidity != 0 {
|
||||
value, timestamp, _ := telldus.SensorValue(protocol, model, id, telldus.DataTypeHumidity)
|
||||
fmt.Printf(" Humidity: %s%% at %d\n", value, timestamp)
|
||||
}
|
||||
ret = telldus.Sensor(&protocol, &model, &id, &dataTypes)
|
||||
if ret != 0 {
|
||||
break
|
||||
}
|
||||
}
|
||||
} else if ret != -6 { // Assuming -6 is TELLSTICK_ERROR_DEVICE_NOT_FOUND
|
||||
errStr := telldus.GetErrorString(ret)
|
||||
fmt.Printf("Error fetching sensors: %s\n", errStr)
|
||||
}
|
||||
}
|
||||
817
main.go
817
main.go
@@ -1,522 +1,137 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"app/datastore"
|
||||
"app/devices"
|
||||
"app/mqtt"
|
||||
"app/telldus"
|
||||
"bufio"
|
||||
"database/sql"
|
||||
daemon "app/telldus-daemon"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/exec"
|
||||
"os/signal"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
mqtt "github.com/eclipse/paho.mqtt.golang"
|
||||
_ "github.com/mattn/go-sqlite3"
|
||||
"github.com/fsnotify/fsnotify"
|
||||
)
|
||||
|
||||
const (
|
||||
httpPort = ":8080"
|
||||
)
|
||||
|
||||
var mqttDev *mqttDevice
|
||||
var db *sql.DB
|
||||
var telldusCmd *exec.Cmd
|
||||
var telldusMu sync.Mutex
|
||||
|
||||
type mqttDevice struct {
|
||||
client mqtt.Client
|
||||
}
|
||||
|
||||
func newMqttDevice() *mqttDevice {
|
||||
opts := mqtt.NewClientOptions().AddBroker(os.Getenv("MQTT_URL"))
|
||||
opts.SetUsername(os.Getenv("MQTT_USER"))
|
||||
opts.SetPassword(os.Getenv("MQTT_PASSWORD"))
|
||||
client := mqtt.NewClient(opts)
|
||||
if token := client.Connect(); token.Wait() && token.Error() != nil {
|
||||
log.Fatal(token.Error())
|
||||
}
|
||||
return &mqttDevice{client: client}
|
||||
}
|
||||
|
||||
func (m *mqttDevice) publishDiscovery() {
|
||||
// Publish discovery for devices
|
||||
numDevices := telldus.GetNumberOfDevices()
|
||||
for i := 0; i < numDevices; i++ {
|
||||
deviceId := telldus.GetDeviceId(i)
|
||||
var name, uniqueId string
|
||||
err := db.QueryRow("SELECT name, unique_id FROM devices WHERE id = ?", deviceId).Scan(&name, &uniqueId)
|
||||
if err != nil {
|
||||
log.Printf("Device %d not in DB, skipping", deviceId)
|
||||
continue
|
||||
}
|
||||
topic := fmt.Sprintf("homeassistant/switch/%s/config", uniqueId)
|
||||
payload := fmt.Sprintf(`{
|
||||
"name": "%s",
|
||||
"command_topic": "telldus/device/%d/set",
|
||||
"state_topic": "telldus/device/%d/state",
|
||||
"unique_id": "%s",
|
||||
"device": {
|
||||
"identifiers": ["telldus_%d"],
|
||||
"name": "%s",
|
||||
"manufacturer": "Telldus"
|
||||
}
|
||||
}`, name, deviceId, deviceId, uniqueId, deviceId, name)
|
||||
m.client.Publish(topic, 0, true, payload)
|
||||
}
|
||||
|
||||
// Publish discovery for sensors
|
||||
var protocol, model string
|
||||
var id, dataTypes int
|
||||
ret := telldus.Sensor(&protocol, &model, &id, &dataTypes)
|
||||
for ret == 0 {
|
||||
var sensorName, tempUniqueId, humUniqueId string
|
||||
err := db.QueryRow("SELECT name, temperature_unique_id, humidity_unique_id FROM sensors WHERE protocol = ? AND model = ? AND id = ? AND hidden = 0", protocol, model, id).Scan(&sensorName, &tempUniqueId, &humUniqueId)
|
||||
if err != nil {
|
||||
log.Printf("Sensor %s %s %d not in DB, skipping", protocol, model, id)
|
||||
ret = telldus.Sensor(&protocol, &model, &id, &dataTypes)
|
||||
continue
|
||||
}
|
||||
if dataTypes&telldus.DataTypeTemperature != 0 && tempUniqueId != "" {
|
||||
topic := fmt.Sprintf("homeassistant/sensor/%s/config", tempUniqueId)
|
||||
payload := fmt.Sprintf(`{
|
||||
"name": "%s Temperature",
|
||||
"state_topic": "telldus/sensor/%s/%s/%d/temperature",
|
||||
"unit_of_measurement": "°C",
|
||||
"device_class": "temperature",
|
||||
"unique_id": "%s",
|
||||
"device": {
|
||||
"identifiers": ["telldus_sensor_%s_%s_%d"],
|
||||
"name": "%s",
|
||||
"manufacturer": "Telldus"
|
||||
}
|
||||
}`, sensorName, protocol, model, id, tempUniqueId, protocol, model, id, sensorName)
|
||||
m.client.Publish(topic, 0, true, payload)
|
||||
}
|
||||
if dataTypes&telldus.DataTypeHumidity != 0 && humUniqueId != "" {
|
||||
topic := fmt.Sprintf("homeassistant/sensor/%s/config", humUniqueId)
|
||||
payload := fmt.Sprintf(`{
|
||||
"name": "%s Humidity",
|
||||
"state_topic": "telldus/sensor/%s/%s/%d/humidity",
|
||||
"unit_of_measurement": "%%",
|
||||
"device_class": "humidity",
|
||||
"unique_id": "%s",
|
||||
"device": {
|
||||
"identifiers": ["telldus_sensor_%s_%s_%d"],
|
||||
"name": "%s",
|
||||
"manufacturer": "Telldus"
|
||||
}
|
||||
}`, sensorName, protocol, model, id, humUniqueId, protocol, model, id, sensorName)
|
||||
m.client.Publish(topic, 0, true, payload)
|
||||
}
|
||||
// Add other sensor types if needed
|
||||
ret = telldus.Sensor(&protocol, &model, &id, &dataTypes)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *mqttDevice) subscribeCommands() {
|
||||
numDevices := telldus.GetNumberOfDevices()
|
||||
for i := 0; i < numDevices; i++ {
|
||||
deviceId := telldus.GetDeviceId(i)
|
||||
topic := fmt.Sprintf("telldus/device/%d/set", deviceId)
|
||||
m.client.Subscribe(topic, 0, func(client mqtt.Client, msg mqtt.Message) {
|
||||
payload := string(msg.Payload())
|
||||
if payload == "ON" {
|
||||
telldus.TurnOn(deviceId)
|
||||
} else if payload == "OFF" {
|
||||
telldus.TurnOff(deviceId)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func (m *mqttDevice) republishDevice(deviceId int) {
|
||||
var name, uniqueId string
|
||||
err := db.QueryRow("SELECT name, unique_id FROM devices WHERE id = ?", deviceId).Scan(&name, &uniqueId)
|
||||
if err != nil {
|
||||
log.Printf("Error querying device %d: %v", deviceId, err)
|
||||
return
|
||||
}
|
||||
topic := fmt.Sprintf("homeassistant/switch/%s/config", uniqueId)
|
||||
payload := fmt.Sprintf(`{
|
||||
"name": "%s",
|
||||
"command_topic": "telldus/device/%d/set",
|
||||
"state_topic": "telldus/device/%d/state",
|
||||
"unique_id": "%s",
|
||||
"device": {
|
||||
"identifiers": ["telldus_%d"],
|
||||
"name": "%s",
|
||||
"manufacturer": "Telldus"
|
||||
}
|
||||
}`, name, deviceId, deviceId, uniqueId, deviceId, name)
|
||||
m.client.Publish(topic, 0, true, payload)
|
||||
}
|
||||
|
||||
func (m *mqttDevice) republishSensor(protocol, model string, id int) {
|
||||
var sensorName, tempUniqueId, humUniqueId string
|
||||
err := db.QueryRow("SELECT name, temperature_unique_id, humidity_unique_id FROM sensors WHERE protocol = ? AND model = ? AND id = ?", protocol, model, id).Scan(&sensorName, &tempUniqueId, &humUniqueId)
|
||||
if err != nil {
|
||||
log.Printf("Error querying sensor %s %s %d: %v", protocol, model, id, err)
|
||||
return
|
||||
}
|
||||
if tempUniqueId != "" {
|
||||
topic := fmt.Sprintf("homeassistant/sensor/%s/config", tempUniqueId)
|
||||
payload := fmt.Sprintf(`{
|
||||
"name": "%s Temperature",
|
||||
"state_topic": "telldus/sensor/%s/%s/%d/temperature",
|
||||
"unit_of_measurement": "°C",
|
||||
"device_class": "temperature",
|
||||
"unique_id": "%s",
|
||||
"device": {
|
||||
"identifiers": ["telldus_sensor_%s_%s_%d"],
|
||||
"name": "%s",
|
||||
"manufacturer": "Telldus"
|
||||
}
|
||||
}`, sensorName, protocol, model, id, tempUniqueId, protocol, model, id, sensorName)
|
||||
m.client.Publish(topic, 0, true, payload)
|
||||
}
|
||||
if humUniqueId != "" {
|
||||
topic := fmt.Sprintf("homeassistant/sensor/%s/config", humUniqueId)
|
||||
payload := fmt.Sprintf(`{
|
||||
"name": "%s Humidity",
|
||||
"state_topic": "telldus/sensor/%s/%s/%d/humidity",
|
||||
"unit_of_measurement": "%%",
|
||||
"device_class": "humidity",
|
||||
"unique_id": "%s",
|
||||
"device": {
|
||||
"identifiers": ["telldus_sensor_%s_%s_%d"],
|
||||
"name": "%s",
|
||||
"manufacturer": "Telldus"
|
||||
}
|
||||
}`, sensorName, protocol, model, id, humUniqueId, protocol, model, id, sensorName)
|
||||
m.client.Publish(topic, 0, true, payload)
|
||||
}
|
||||
}
|
||||
|
||||
type RawEvent struct {
|
||||
Timestamp time.Time `json:"timestamp"`
|
||||
ControllerId int `json:"controller_id"`
|
||||
Data string `json:"data"`
|
||||
}
|
||||
|
||||
type SensorEvent struct {
|
||||
Timestamp time.Time `json:"timestamp"`
|
||||
Protocol string `json:"protocol"`
|
||||
Model string `json:"model"`
|
||||
Id int `json:"id"`
|
||||
DataType int `json:"data_type"`
|
||||
Value string `json:"value"`
|
||||
}
|
||||
|
||||
var rawEvents []RawEvent
|
||||
var sensorEvents []SensorEvent
|
||||
var mu sync.Mutex
|
||||
var mqttClient *mqtt.Client
|
||||
var store *datastore.DataStore
|
||||
var daemonMgr *daemon.Manager
|
||||
var eventMgr *devices.EventManager
|
||||
|
||||
const maxEvents = 1000
|
||||
|
||||
// startTelldusd starts the telldusd daemon and captures its output
|
||||
func startTelldusd() error {
|
||||
telldusMu.Lock()
|
||||
defer telldusMu.Unlock()
|
||||
|
||||
if telldusCmd != nil && telldusCmd.Process != nil {
|
||||
log.Println("Telldusd already running")
|
||||
return nil
|
||||
}
|
||||
|
||||
log.Println("Starting telldusd...")
|
||||
cmd := exec.Command("/usr/local/sbin/telldusd", "--nodaemon")
|
||||
|
||||
// Capture stdout
|
||||
stdout, err := cmd.StdoutPipe()
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to get stdout pipe: %v", err)
|
||||
}
|
||||
|
||||
// Capture stderr
|
||||
stderr, err := cmd.StderrPipe()
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to get stderr pipe: %v", err)
|
||||
}
|
||||
|
||||
// Start the command
|
||||
if err := cmd.Start(); err != nil {
|
||||
return fmt.Errorf("failed to start telldusd: %v", err)
|
||||
}
|
||||
|
||||
telldusCmd = cmd
|
||||
|
||||
// Log stdout in a goroutine
|
||||
go func() {
|
||||
scanner := bufio.NewScanner(stdout)
|
||||
for scanner.Scan() {
|
||||
log.Printf("[telldusd] %s", scanner.Text())
|
||||
}
|
||||
}()
|
||||
|
||||
// Log stderr in a goroutine
|
||||
go func() {
|
||||
scanner := bufio.NewScanner(stderr)
|
||||
for scanner.Scan() {
|
||||
log.Printf("[telldusd] ERROR: %s", scanner.Text())
|
||||
}
|
||||
}()
|
||||
|
||||
// Monitor process in a goroutine
|
||||
go func() {
|
||||
err := cmd.Wait()
|
||||
telldusMu.Lock()
|
||||
telldusCmd = nil
|
||||
telldusMu.Unlock()
|
||||
if err != nil {
|
||||
log.Printf("Telldusd exited with error: %v", err)
|
||||
} else {
|
||||
log.Println("Telldusd exited normally")
|
||||
}
|
||||
}()
|
||||
|
||||
// Give telldusd a moment to start
|
||||
time.Sleep(500 * time.Millisecond)
|
||||
log.Println("Telldusd started successfully")
|
||||
return nil
|
||||
}
|
||||
|
||||
// stopTelldusd stops the telldusd daemon
|
||||
func stopTelldusd() error {
|
||||
telldusMu.Lock()
|
||||
defer telldusMu.Unlock()
|
||||
|
||||
if telldusCmd == nil || telldusCmd.Process == nil {
|
||||
log.Println("Telldusd not running")
|
||||
return nil
|
||||
}
|
||||
|
||||
log.Println("Stopping telldusd...")
|
||||
|
||||
// Send SIGTERM
|
||||
if err := telldusCmd.Process.Signal(syscall.SIGTERM); err != nil {
|
||||
log.Printf("Failed to send SIGTERM to telldusd: %v", err)
|
||||
// Try SIGKILL as fallback
|
||||
if err := telldusCmd.Process.Kill(); err != nil {
|
||||
return fmt.Errorf("failed to kill telldusd: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// Wait for process to exit (with timeout)
|
||||
done := make(chan error, 1)
|
||||
go func() {
|
||||
done <- telldusCmd.Wait()
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-done:
|
||||
log.Println("Telldusd stopped successfully")
|
||||
case <-time.After(5 * time.Second):
|
||||
log.Println("Telldusd did not stop gracefully, killing...")
|
||||
telldusCmd.Process.Kill()
|
||||
}
|
||||
|
||||
telldusCmd = nil
|
||||
return nil
|
||||
}
|
||||
|
||||
// restartTelldusd restarts the telldusd daemon
|
||||
func restartTelldusd() error {
|
||||
log.Println("Restarting telldusd due to configuration change...")
|
||||
|
||||
if err := stopTelldusd(); err != nil {
|
||||
log.Printf("Error stopping telldusd: %v", err)
|
||||
}
|
||||
|
||||
// Give it a moment to fully stop
|
||||
time.Sleep(1 * time.Second)
|
||||
|
||||
// Close and reinitialize telldus library
|
||||
telldus.Close()
|
||||
time.Sleep(500 * time.Millisecond)
|
||||
|
||||
if err := startTelldusd(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Reinitialize telldus library
|
||||
telldus.Init()
|
||||
|
||||
log.Println("Telldusd restarted successfully")
|
||||
return nil
|
||||
}
|
||||
|
||||
// watchConfigFile watches for changes to the tellstick.conf file
|
||||
func watchConfigFile(configPath string) {
|
||||
watcher, err := fsnotify.NewWatcher()
|
||||
if err != nil {
|
||||
log.Printf("Failed to create file watcher: %v", err)
|
||||
return
|
||||
}
|
||||
defer watcher.Close()
|
||||
|
||||
// Watch the parent directory since file operations might replace the file
|
||||
configDir := filepath.Dir(configPath)
|
||||
if err := watcher.Add(configDir); err != nil {
|
||||
log.Printf("Failed to watch config directory: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
log.Printf("Watching for changes to %s", configPath)
|
||||
|
||||
for {
|
||||
select {
|
||||
case event, ok := <-watcher.Events:
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
// Check if the event is for our config file
|
||||
if event.Name == configPath && (event.Op&fsnotify.Write == fsnotify.Write || event.Op&fsnotify.Create == fsnotify.Create) {
|
||||
log.Printf("Configuration file changed: %s", event.Op.String())
|
||||
if err := restartTelldusd(); err != nil {
|
||||
log.Printf("Failed to restart telldusd: %v", err)
|
||||
}
|
||||
}
|
||||
case err, ok := <-watcher.Errors:
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
log.Printf("File watcher error: %v", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func main() {
|
||||
// Start telldusd daemon
|
||||
if err := startTelldusd(); err != nil {
|
||||
// Initialize daemon manager
|
||||
daemonMgr = daemon.New()
|
||||
if err := daemonMgr.Start(); err != nil {
|
||||
log.Fatalf("Failed to start telldusd: %v", err)
|
||||
}
|
||||
defer stopTelldusd()
|
||||
|
||||
// Start watching config file
|
||||
configPath := "/etc/tellstick.conf"
|
||||
go watchConfigFile(configPath)
|
||||
defer daemonMgr.Stop()
|
||||
|
||||
// Initialize Telldus
|
||||
telldus.Init()
|
||||
defer telldus.Close()
|
||||
|
||||
// Initialize SQLite DB
|
||||
// Initialize DataStore
|
||||
var err error
|
||||
db, err = sql.Open("sqlite3", "./db/telldus.db")
|
||||
store, err = datastore.New("./db/telldus.db")
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
defer db.Close()
|
||||
defer store.Close()
|
||||
|
||||
// Create tables
|
||||
_, err = db.Exec(`CREATE TABLE IF NOT EXISTS devices (
|
||||
id INTEGER PRIMARY KEY,
|
||||
name TEXT,
|
||||
unique_id TEXT UNIQUE
|
||||
)`)
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
// Sync devices and sensors
|
||||
syncer := devices.NewSyncer(store)
|
||||
if err := syncer.SyncDevices(); err != nil {
|
||||
log.Printf("Error syncing devices: %v", err)
|
||||
}
|
||||
_, err = db.Exec(`CREATE TABLE IF NOT EXISTS sensors (
|
||||
sensor_id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
protocol TEXT,
|
||||
model TEXT,
|
||||
id INTEGER,
|
||||
name TEXT,
|
||||
temperature_unique_id TEXT,
|
||||
humidity_unique_id TEXT,
|
||||
last_temperature TEXT,
|
||||
last_humidity TEXT,
|
||||
last_timestamp INTEGER,
|
||||
hidden INTEGER DEFAULT 0,
|
||||
UNIQUE(protocol, model, id)
|
||||
)`)
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
_, err = db.Exec(`CREATE TABLE IF NOT EXISTS potential_devices (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
class TEXT,
|
||||
protocol TEXT,
|
||||
model TEXT,
|
||||
device_id TEXT,
|
||||
last_data TEXT,
|
||||
last_seen INTEGER
|
||||
)`)
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
if err := syncer.SyncSensors(); err != nil {
|
||||
log.Printf("Error syncing sensors: %v", err)
|
||||
}
|
||||
|
||||
// Sync devices to DB
|
||||
numDevices := telldus.GetNumberOfDevices()
|
||||
for i := 0; i < numDevices; i++ {
|
||||
deviceId := telldus.GetDeviceId(i)
|
||||
name := telldus.GetName(deviceId)
|
||||
uniqueId := fmt.Sprintf("telldus_device_%d", deviceId)
|
||||
_, err = db.Exec("INSERT OR IGNORE INTO devices (id, name, unique_id) VALUES (?, ?, ?)", deviceId, name, uniqueId)
|
||||
if err != nil {
|
||||
log.Printf("Error inserting device %d: %v", deviceId, err)
|
||||
// Device reload function for config file changes
|
||||
reloadDevices := func() error {
|
||||
log.Println("Configuration file changed, restarting telldusd...")
|
||||
if err := daemonMgr.Restart(); err != nil {
|
||||
log.Printf("Failed to restart telldusd: %v", err)
|
||||
return err
|
||||
}
|
||||
log.Println("Reloading devices and sensors after config change...")
|
||||
if err := syncer.SyncDevices(); err != nil {
|
||||
log.Printf("Error syncing devices: %v", err)
|
||||
return err
|
||||
}
|
||||
if err := syncer.SyncSensors(); err != nil {
|
||||
log.Printf("Error syncing sensors: %v", err)
|
||||
return err
|
||||
}
|
||||
if err := mqttClient.PublishAllDiscovery(); err != nil {
|
||||
log.Printf("Error republishing discovery: %v", err)
|
||||
return err
|
||||
}
|
||||
if err := mqttClient.SubscribeToDeviceCommands(); err != nil {
|
||||
log.Printf("Error resubscribing to commands: %v", err)
|
||||
return err
|
||||
}
|
||||
log.Println("Successfully reloaded devices and sensors")
|
||||
return nil
|
||||
}
|
||||
|
||||
// Sync sensors to DB
|
||||
var protocol, model string
|
||||
var id, dataTypes int
|
||||
ret := telldus.Sensor(&protocol, &model, &id, &dataTypes)
|
||||
for ret == 0 {
|
||||
sensorName := fmt.Sprintf("%s %s %d", protocol, model, id)
|
||||
tempUniqueId := ""
|
||||
humUniqueId := ""
|
||||
if dataTypes&telldus.DataTypeTemperature != 0 {
|
||||
tempUniqueId = fmt.Sprintf("telldus_sensor_%s_%s_%d_temperature", protocol, model, id)
|
||||
// Start watching config file
|
||||
configPath := "/etc/tellstick.conf"
|
||||
watcher := daemon.NewWatcher(configPath, reloadDevices)
|
||||
go func() {
|
||||
if err := watcher.Watch(); err != nil {
|
||||
log.Printf("Config watcher error: %v", err)
|
||||
}
|
||||
if dataTypes&telldus.DataTypeHumidity != 0 {
|
||||
humUniqueId = fmt.Sprintf("telldus_sensor_%s_%s_%d_humidity", protocol, model, id)
|
||||
}
|
||||
_, err = db.Exec("INSERT OR IGNORE INTO sensors (protocol, model, id, name, temperature_unique_id, humidity_unique_id) VALUES (?, ?, ?, ?, ?, ?)", protocol, model, id, sensorName, tempUniqueId, humUniqueId)
|
||||
if err != nil {
|
||||
log.Printf("Error inserting sensor %s %s %d: %v", protocol, model, id, err)
|
||||
}
|
||||
ret = telldus.Sensor(&protocol, &model, &id, &dataTypes)
|
||||
}
|
||||
}()
|
||||
|
||||
// Initialize MQTT
|
||||
mqttDev = newMqttDevice()
|
||||
defer mqttDev.client.Disconnect(250)
|
||||
mqttConfig := mqtt.Config{
|
||||
BrokerURL: os.Getenv("MQTT_URL"),
|
||||
Username: os.Getenv("MQTT_USER"),
|
||||
Password: os.Getenv("MQTT_PASSWORD"),
|
||||
}
|
||||
mqttClient, err = mqtt.New(mqttConfig, store)
|
||||
if err != nil {
|
||||
log.Fatalf("Failed to connect to MQTT: %v", err)
|
||||
}
|
||||
defer mqttClient.Close()
|
||||
|
||||
// Publish Home Assistant discovery
|
||||
mqttDev.publishDiscovery()
|
||||
if err := mqttClient.PublishAllDiscovery(); err != nil {
|
||||
log.Printf("Error publishing discovery: %v", err)
|
||||
}
|
||||
|
||||
// Subscribe to command topics
|
||||
mqttDev.subscribeCommands()
|
||||
if err := mqttClient.SubscribeToDeviceCommands(); err != nil {
|
||||
log.Printf("Error subscribing to commands: %v", err)
|
||||
}
|
||||
|
||||
// List devices and sensors
|
||||
listDevices()
|
||||
listSensors()
|
||||
syncer.ListDevices()
|
||||
syncer.ListSensors()
|
||||
|
||||
// Register callbacks
|
||||
telldus.RegisterDeviceEvent(deviceEventHandler)
|
||||
telldus.RegisterSensorEvent(sensorEventHandler)
|
||||
telldus.RegisterRawDeviceEvent(rawDeviceEventHandler)
|
||||
// Initialize event manager and register callbacks
|
||||
eventMgr = devices.NewEventManager(store, mqttClient, maxEvents)
|
||||
eventMgr.RegisterCallbacks()
|
||||
|
||||
// Setup graceful shutdown
|
||||
c := make(chan os.Signal, 1)
|
||||
signal.Notify(c, os.Interrupt, syscall.SIGTERM)
|
||||
signal.Notify(c, os.Interrupt)
|
||||
go func() {
|
||||
<-c
|
||||
log.Println("Shutting down gracefully...")
|
||||
mqttDev.client.Disconnect(250)
|
||||
mqttClient.Close()
|
||||
telldus.Close()
|
||||
stopTelldusd()
|
||||
daemonMgr.Stop()
|
||||
os.Exit(0)
|
||||
}()
|
||||
|
||||
@@ -525,145 +140,22 @@ func main() {
|
||||
log.Fatal(http.ListenAndServe(httpPort, setupRoutes()))
|
||||
}
|
||||
|
||||
func deviceEventHandler(deviceId, method int, data string, callbackId int) {
|
||||
fmt.Printf("Device event: ID=%d, Method=%d, Data=%s\n", deviceId, method, data)
|
||||
// Publish state to MQTT
|
||||
var state string
|
||||
switch method {
|
||||
case telldus.MethodTurnOn:
|
||||
state = "ON"
|
||||
case telldus.MethodTurnOff:
|
||||
state = "OFF"
|
||||
}
|
||||
if state != "" {
|
||||
topic := fmt.Sprintf("telldus/device/%d/state", deviceId)
|
||||
mqttDev.client.Publish(topic, 0, false, state)
|
||||
}
|
||||
// Log or handle locally, no MQTT
|
||||
}
|
||||
|
||||
func sensorEventHandler(protocol, model string, id, dataType int, value string, timestamp, callbackId int) {
|
||||
fmt.Printf("Sensor event: Protocol=%s, Model=%s, ID=%d, Type=%d, Value=%s, Timestamp=%d\n",
|
||||
protocol, model, id, dataType, value, timestamp)
|
||||
// Publish to MQTT
|
||||
var topic string
|
||||
switch dataType {
|
||||
case telldus.DataTypeTemperature:
|
||||
topic = fmt.Sprintf("telldus/sensor/%s/%s/%d/temperature", protocol, model, id)
|
||||
case telldus.DataTypeHumidity:
|
||||
topic = fmt.Sprintf("telldus/sensor/%s/%s/%d/humidity", protocol, model, id)
|
||||
}
|
||||
if topic != "" {
|
||||
mqttDev.client.Publish(topic, 0, false, value)
|
||||
}
|
||||
// Store in history
|
||||
mu.Lock()
|
||||
sensorEvents = append(sensorEvents, SensorEvent{time.Now(), protocol, model, id, dataType, value})
|
||||
if len(sensorEvents) > maxEvents {
|
||||
sensorEvents = sensorEvents[1:]
|
||||
}
|
||||
mu.Unlock()
|
||||
// Update last value in DB
|
||||
var column string
|
||||
switch dataType {
|
||||
case telldus.DataTypeTemperature:
|
||||
column = "last_temperature"
|
||||
case telldus.DataTypeHumidity:
|
||||
column = "last_humidity"
|
||||
}
|
||||
if column != "" {
|
||||
_, err := db.Exec(fmt.Sprintf("UPDATE sensors SET %s = ?, last_timestamp = ? WHERE protocol = ? AND model = ? AND id = ?", column), value, time.Now().Unix(), protocol, model, id)
|
||||
if err != nil {
|
||||
log.Printf("Error updating sensor %s %s %d: %v", protocol, model, id, err)
|
||||
}
|
||||
}
|
||||
// Log or handle locally, no MQTT
|
||||
}
|
||||
|
||||
func rawDeviceEventHandler(data string, controllerId, callbackId int) {
|
||||
fmt.Printf("Raw device event: ControllerID=%d, Data=%s\n", controllerId, data)
|
||||
// Parse data
|
||||
fields := strings.Split(data, ";")
|
||||
var class, protocol, model, deviceId string
|
||||
for _, field := range fields {
|
||||
kv := strings.SplitN(field, ":", 2)
|
||||
if len(kv) == 2 {
|
||||
key, val := kv[0], kv[1]
|
||||
switch key {
|
||||
case "class":
|
||||
class = val
|
||||
case "protocol":
|
||||
protocol = val
|
||||
case "model":
|
||||
model = val
|
||||
case "id":
|
||||
deviceId = val
|
||||
}
|
||||
}
|
||||
}
|
||||
// Store in potential_devices
|
||||
_, err := db.Exec("INSERT OR REPLACE INTO potential_devices (class, protocol, model, device_id, last_data, last_seen) VALUES (?, ?, ?, ?, ?, ?)", class, protocol, model, deviceId, data, time.Now().Unix())
|
||||
if err != nil {
|
||||
log.Printf("Error storing potential device: %v", err)
|
||||
}
|
||||
// If sensor, ensure in sensors table
|
||||
if class == "sensor" {
|
||||
idInt, _ := strconv.Atoi(deviceId)
|
||||
sensorName := fmt.Sprintf("%s %s %s", protocol, model, deviceId)
|
||||
tempUniqueId := fmt.Sprintf("telldus_sensor_%s_%s_%s_temperature", protocol, model, deviceId)
|
||||
humUniqueId := fmt.Sprintf("telldus_sensor_%s_%s_%s_humidity", protocol, model, deviceId)
|
||||
_, err = db.Exec("INSERT OR IGNORE INTO sensors (protocol, model, id, name, temperature_unique_id, humidity_unique_id) VALUES (?, ?, ?, ?, ?, ?)", protocol, model, idInt, sensorName, tempUniqueId, humUniqueId)
|
||||
if err != nil {
|
||||
log.Printf("Error inserting sensor from raw: %v", err)
|
||||
}
|
||||
}
|
||||
// Log the raw event data
|
||||
mu.Lock()
|
||||
rawEvents = append(rawEvents, RawEvent{time.Now(), controllerId, data})
|
||||
if len(rawEvents) > maxEvents {
|
||||
rawEvents = rawEvents[1:]
|
||||
}
|
||||
mu.Unlock()
|
||||
}
|
||||
|
||||
func getRawEvents(w http.ResponseWriter, r *http.Request) {
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(rawEvents)
|
||||
json.NewEncoder(w).Encode(eventMgr.GetRawEvents())
|
||||
}
|
||||
|
||||
func getSensorEvents(w http.ResponseWriter, r *http.Request) {
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(sensorEvents)
|
||||
json.NewEncoder(w).Encode(eventMgr.GetSensorEvents())
|
||||
}
|
||||
|
||||
func getPotentialDevices(w http.ResponseWriter, r *http.Request) {
|
||||
rows, err := db.Query("SELECT class, protocol, model, device_id, last_data, last_seen FROM potential_devices ORDER BY last_seen DESC")
|
||||
if err != nil {
|
||||
http.Error(w, "Database error", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
defer rows.Close()
|
||||
devices := []map[string]interface{}{}
|
||||
for rows.Next() {
|
||||
var class, protocol, model, deviceId, lastData string
|
||||
var lastSeen int
|
||||
err := rows.Scan(&class, &protocol, &model, &deviceId, &lastData, &lastSeen)
|
||||
if err != nil {
|
||||
http.Error(w, "Scan error", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
devices = append(devices, map[string]interface{}{
|
||||
"class": class,
|
||||
"protocol": protocol,
|
||||
"model": model,
|
||||
"device_id": deviceId,
|
||||
"last_data": lastData,
|
||||
"last_seen": lastSeen,
|
||||
})
|
||||
devices := []*datastore.PotentialDevice{}
|
||||
for device := range store.ListPotentialDevices() {
|
||||
devices = append(devices, device)
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(devices)
|
||||
@@ -683,13 +175,14 @@ func renameDevice(w http.ResponseWriter, r *http.Request) {
|
||||
http.Error(w, "Bad request", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
_, err = db.Exec("UPDATE devices SET name = ? WHERE id = ?", req.Name, id)
|
||||
if err != nil {
|
||||
if err := store.UpdateDeviceName(id, req.Name); err != nil {
|
||||
http.Error(w, "Database error", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
// Republish discovery for this device
|
||||
mqttDev.republishDevice(id)
|
||||
if err := mqttClient.PublishDeviceDiscovery(id); err != nil {
|
||||
log.Printf("Error republishing device discovery: %v", err)
|
||||
}
|
||||
w.WriteHeader(http.StatusOK)
|
||||
}
|
||||
|
||||
@@ -707,20 +200,19 @@ func renameSensor(w http.ResponseWriter, r *http.Request) {
|
||||
http.Error(w, "Bad request", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
var protocol, model string
|
||||
var id int
|
||||
err = db.QueryRow("SELECT protocol, model, id FROM sensors WHERE sensor_id = ?", sensorId).Scan(&protocol, &model, &id)
|
||||
sensor, err := store.GetSensor(sensorId)
|
||||
if err != nil {
|
||||
http.Error(w, "Sensor not found", http.StatusNotFound)
|
||||
return
|
||||
}
|
||||
_, err = db.Exec("UPDATE sensors SET name = ? WHERE sensor_id = ?", req.Name, sensorId)
|
||||
if err != nil {
|
||||
if err := store.UpdateSensorName(sensorId, req.Name); err != nil {
|
||||
http.Error(w, "Database error", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
// Republish discovery for this sensor
|
||||
mqttDev.republishSensor(protocol, model, id)
|
||||
if err := mqttClient.PublishSensorDiscovery(sensor.Protocol, sensor.Model, sensor.ID); err != nil {
|
||||
log.Printf("Error republishing sensor discovery: %v", err)
|
||||
}
|
||||
w.WriteHeader(http.StatusOK)
|
||||
}
|
||||
|
||||
@@ -731,8 +223,7 @@ func hideSensor(w http.ResponseWriter, r *http.Request) {
|
||||
http.Error(w, "Invalid sensor ID", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
_, err = db.Exec("UPDATE sensors SET hidden = 1 WHERE sensor_id = ?", sensorId)
|
||||
if err != nil {
|
||||
if err := store.SetSensorHidden(sensorId, true); err != nil {
|
||||
http.Error(w, "Database error", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
@@ -746,8 +237,7 @@ func unhideSensor(w http.ResponseWriter, r *http.Request) {
|
||||
http.Error(w, "Invalid sensor ID", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
_, err = db.Exec("UPDATE sensors SET hidden = 0 WHERE sensor_id = ?", sensorId)
|
||||
if err != nil {
|
||||
if err := store.SetSensorHidden(sensorId, false); err != nil {
|
||||
http.Error(w, "Database error", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
@@ -775,25 +265,9 @@ func setupRoutes() *http.ServeMux {
|
||||
return mux
|
||||
}
|
||||
func getDevices(w http.ResponseWriter, r *http.Request) {
|
||||
rows, err := db.Query("SELECT id, name FROM devices ORDER BY id")
|
||||
if err != nil {
|
||||
http.Error(w, "Database error", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
defer rows.Close()
|
||||
devices := []map[string]interface{}{}
|
||||
for rows.Next() {
|
||||
var id int
|
||||
var name string
|
||||
err := rows.Scan(&id, &name)
|
||||
if err != nil {
|
||||
http.Error(w, "Scan error", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
devices = append(devices, map[string]interface{}{
|
||||
"id": id,
|
||||
"name": name,
|
||||
})
|
||||
devices := []*datastore.Device{}
|
||||
for device := range store.ListDevices() {
|
||||
devices = append(devices, device)
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(devices)
|
||||
@@ -842,39 +316,8 @@ func turnOffDevice(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
|
||||
func getSensors(w http.ResponseWriter, r *http.Request) {
|
||||
rows, err := db.Query("SELECT sensor_id, protocol, model, id, name, last_temperature, last_humidity, last_timestamp, hidden FROM sensors ORDER BY sensor_id")
|
||||
if err != nil {
|
||||
http.Error(w, "Database error", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
defer rows.Close()
|
||||
sensors := []map[string]interface{}{}
|
||||
for rows.Next() {
|
||||
var sensorId int
|
||||
var protocol, model, name, lastTemp, lastHum string
|
||||
var id, lastTs, hidden int
|
||||
err := rows.Scan(&sensorId, &protocol, &model, &id, &name, &lastTemp, &lastHum, &lastTs, &hidden)
|
||||
if err != nil {
|
||||
http.Error(w, "Scan error", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
sensor := map[string]interface{}{
|
||||
"sensor_id": sensorId,
|
||||
"protocol": protocol,
|
||||
"model": model,
|
||||
"id": id,
|
||||
"name": name,
|
||||
"hidden": hidden != 0,
|
||||
}
|
||||
if lastTemp != "" {
|
||||
sensor["last_temperature"] = lastTemp
|
||||
}
|
||||
if lastHum != "" {
|
||||
sensor["last_humidity"] = lastHum
|
||||
}
|
||||
if lastTs > 0 {
|
||||
sensor["last_timestamp"] = lastTs
|
||||
}
|
||||
sensors := []*datastore.Sensor{}
|
||||
for sensor := range store.ListSensors() {
|
||||
sensors = append(sensors, sensor)
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
@@ -888,17 +331,13 @@ func getDevice(w http.ResponseWriter, r *http.Request) {
|
||||
http.Error(w, "Invalid device ID", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
var name string
|
||||
err = db.QueryRow("SELECT name FROM devices WHERE id = ?", id).Scan(&name)
|
||||
device, err := store.GetDevice(id)
|
||||
if err != nil {
|
||||
http.Error(w, "Device not found", http.StatusNotFound)
|
||||
return
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(map[string]interface{}{
|
||||
"id": id,
|
||||
"name": name,
|
||||
})
|
||||
json.NewEncoder(w).Encode(device)
|
||||
}
|
||||
|
||||
func getSensor(w http.ResponseWriter, r *http.Request) {
|
||||
@@ -908,65 +347,13 @@ func getSensor(w http.ResponseWriter, r *http.Request) {
|
||||
http.Error(w, "Invalid sensor ID", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
var protocol, model, name string
|
||||
var id int
|
||||
err = db.QueryRow("SELECT protocol, model, id, name FROM sensors WHERE sensor_id = ?", sensorId).Scan(&protocol, &model, &id, &name)
|
||||
sensor, err := store.GetSensor(sensorId)
|
||||
if err != nil {
|
||||
http.Error(w, "Sensor not found", http.StatusNotFound)
|
||||
return
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
json.NewEncoder(w).Encode(map[string]interface{}{
|
||||
"sensor_id": sensorId,
|
||||
"protocol": protocol,
|
||||
"model": model,
|
||||
"id": id,
|
||||
"name": name,
|
||||
})
|
||||
json.NewEncoder(w).Encode(sensor)
|
||||
}
|
||||
|
||||
func listDevices() {
|
||||
numDevices := telldus.GetNumberOfDevices()
|
||||
if numDevices < 0 {
|
||||
errStr := telldus.GetErrorString(numDevices)
|
||||
fmt.Printf("Error fetching devices: %s\n", errStr)
|
||||
return
|
||||
}
|
||||
fmt.Printf("Number of devices: %d\n", numDevices)
|
||||
for i := 0; i < numDevices; i++ {
|
||||
deviceId := telldus.GetDeviceId(i)
|
||||
name := telldus.GetName(deviceId)
|
||||
protocol := telldus.GetProtocol(deviceId)
|
||||
model := telldus.GetModel(deviceId)
|
||||
fmt.Printf("%d\t%s\tProtocol: %s\tModel: %s\n", deviceId, name, protocol, model)
|
||||
}
|
||||
}
|
||||
|
||||
func listSensors() {
|
||||
fmt.Println("\nSENSORS:")
|
||||
var protocol, model string
|
||||
var id, dataTypes int
|
||||
ret := telldus.Sensor(&protocol, &model, &id, &dataTypes)
|
||||
if ret == 0 {
|
||||
for {
|
||||
fmt.Printf("Protocol: %s, Model: %s, ID: %d, DataTypes: %d\n", protocol, model, id, dataTypes)
|
||||
// Fetch values if available
|
||||
if dataTypes&telldus.DataTypeTemperature != 0 {
|
||||
value, timestamp, _ := telldus.SensorValue(protocol, model, id, telldus.DataTypeTemperature)
|
||||
fmt.Printf(" Temperature: %s°C at %d\n", value, timestamp)
|
||||
}
|
||||
if dataTypes&telldus.DataTypeHumidity != 0 {
|
||||
value, timestamp, _ := telldus.SensorValue(protocol, model, id, telldus.DataTypeHumidity)
|
||||
fmt.Printf(" Humidity: %s%% at %d\n", value, timestamp)
|
||||
}
|
||||
// Add other types as needed
|
||||
ret = telldus.Sensor(&protocol, &model, &id, &dataTypes)
|
||||
if ret != 0 {
|
||||
break
|
||||
}
|
||||
}
|
||||
} else if ret != -6 { // Assuming -6 is TELLSTICK_ERROR_DEVICE_NOT_FOUND
|
||||
errStr := telldus.GetErrorString(ret)
|
||||
fmt.Printf("Error fetching sensors: %s\n", errStr)
|
||||
}
|
||||
}
|
||||
|
||||
159
mqtt/discovery.go
Normal file
159
mqtt/discovery.go
Normal file
@@ -0,0 +1,159 @@
|
||||
package mqtt
|
||||
|
||||
import (
|
||||
"app/datastore"
|
||||
"app/telldus"
|
||||
"fmt"
|
||||
"log"
|
||||
)
|
||||
|
||||
// DeviceDiscovery represents Home Assistant device discovery payload
|
||||
type DeviceDiscovery struct {
|
||||
Name string `json:"name"`
|
||||
CommandTopic string `json:"command_topic"`
|
||||
StateTopic string `json:"state_topic"`
|
||||
UniqueID string `json:"unique_id"`
|
||||
Device map[string]interface{} `json:"device"`
|
||||
}
|
||||
|
||||
// SensorDiscovery represents Home Assistant sensor discovery payload
|
||||
type SensorDiscovery struct {
|
||||
Name string `json:"name"`
|
||||
StateTopic string `json:"state_topic"`
|
||||
UnitOfMeasurement string `json:"unit_of_measurement,omitempty"`
|
||||
DeviceClass string `json:"device_class,omitempty"`
|
||||
UniqueID string `json:"unique_id"`
|
||||
Device map[string]interface{} `json:"device"`
|
||||
}
|
||||
|
||||
// PublishAllDiscovery publishes Home Assistant discovery messages for all devices and sensors
|
||||
func (c *Client) PublishAllDiscovery() error {
|
||||
if err := c.publishDeviceDiscovery(); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := c.publishSensorDiscovery(); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// publishDeviceDiscovery publishes discovery for all devices
|
||||
func (c *Client) publishDeviceDiscovery() error {
|
||||
for device := range c.store.ListDevices() {
|
||||
if err := c.PublishDeviceDiscovery(device.ID); err != nil {
|
||||
log.Printf("Error publishing discovery for device %d: %v", device.ID, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// PublishDeviceDiscovery publishes Home Assistant discovery for a single device
|
||||
func (c *Client) PublishDeviceDiscovery(deviceID int) error {
|
||||
device, err := c.store.GetDevice(deviceID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("device %d not found: %w", deviceID, err)
|
||||
}
|
||||
|
||||
topic := fmt.Sprintf("homeassistant/switch/%s/config", device.UniqueID)
|
||||
payload := fmt.Sprintf(`{
|
||||
"name": "%s",
|
||||
"command_topic": "telldus/device/%d/set",
|
||||
"state_topic": "telldus/device/%d/state",
|
||||
"unique_id": "%s",
|
||||
"device": {
|
||||
"identifiers": ["telldus_%d"],
|
||||
"name": "%s",
|
||||
"manufacturer": "Telldus"
|
||||
}
|
||||
}`, device.Name, deviceID, deviceID, device.UniqueID, deviceID, device.Name)
|
||||
|
||||
c.client.Publish(topic, 0, true, payload)
|
||||
return nil
|
||||
}
|
||||
|
||||
// publishSensorDiscovery publishes discovery for all sensors
|
||||
func (c *Client) publishSensorDiscovery() error {
|
||||
var protocol, model string
|
||||
var id, dataTypes int
|
||||
ret := telldus.Sensor(&protocol, &model, &id, &dataTypes)
|
||||
|
||||
for ret == 0 {
|
||||
sensor, err := c.store.GetSensorByIdentity(protocol, model, id)
|
||||
if err != nil || sensor.Hidden {
|
||||
log.Printf("Sensor %s %s %d not in DB or hidden, skipping", protocol, model, id)
|
||||
ret = telldus.Sensor(&protocol, &model, &id, &dataTypes)
|
||||
continue
|
||||
}
|
||||
|
||||
if err := c.publishSensorDiscoveryForSensor(sensor, dataTypes); err != nil {
|
||||
log.Printf("Error publishing discovery for sensor %s %s %d: %v", protocol, model, id, err)
|
||||
}
|
||||
|
||||
ret = telldus.Sensor(&protocol, &model, &id, &dataTypes)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// PublishSensorDiscovery publishes Home Assistant discovery for a single sensor
|
||||
func (c *Client) PublishSensorDiscovery(protocol, model string, id int) error {
|
||||
sensor, err := c.store.GetSensorByIdentity(protocol, model, id)
|
||||
if err != nil {
|
||||
return fmt.Errorf("sensor %s %s %d not found: %w", protocol, model, id, err)
|
||||
}
|
||||
|
||||
// Get current data types from telldus
|
||||
var p, m string
|
||||
var sensorID, dataTypes int
|
||||
ret := telldus.Sensor(&p, &m, &sensorID, &dataTypes)
|
||||
|
||||
// Find matching sensor
|
||||
for ret == 0 {
|
||||
if p == protocol && m == model && sensorID == id {
|
||||
return c.publishSensorDiscoveryForSensor(sensor, dataTypes)
|
||||
}
|
||||
ret = telldus.Sensor(&p, &m, &sensorID, &dataTypes)
|
||||
}
|
||||
|
||||
return fmt.Errorf("sensor %s %s %d not found in telldus", protocol, model, id)
|
||||
}
|
||||
|
||||
// publishSensorDiscoveryForSensor publishes discovery messages for a sensor's data types
|
||||
func (c *Client) publishSensorDiscoveryForSensor(sensor *datastore.Sensor, dataTypes int) error {
|
||||
if dataTypes&telldus.DataTypeTemperature != 0 && sensor.TemperatureUniqueID != "" {
|
||||
topic := fmt.Sprintf("homeassistant/sensor/%s/config", sensor.TemperatureUniqueID)
|
||||
payload := fmt.Sprintf(`{
|
||||
"name": "%s Temperature",
|
||||
"state_topic": "telldus/sensor/%s/%s/%d/temperature",
|
||||
"unit_of_measurement": "°C",
|
||||
"device_class": "temperature",
|
||||
"unique_id": "%s",
|
||||
"device": {
|
||||
"identifiers": ["telldus_sensor_%s_%s_%d"],
|
||||
"name": "%s",
|
||||
"manufacturer": "Telldus"
|
||||
}
|
||||
}`, sensor.Name, sensor.Protocol, sensor.Model, sensor.ID, sensor.TemperatureUniqueID,
|
||||
sensor.Protocol, sensor.Model, sensor.ID, sensor.Name)
|
||||
c.client.Publish(topic, 0, true, payload)
|
||||
}
|
||||
|
||||
if dataTypes&telldus.DataTypeHumidity != 0 && sensor.HumidityUniqueID != "" {
|
||||
topic := fmt.Sprintf("homeassistant/sensor/%s/config", sensor.HumidityUniqueID)
|
||||
payload := fmt.Sprintf(`{
|
||||
"name": "%s Humidity",
|
||||
"state_topic": "telldus/sensor/%s/%s/%d/humidity",
|
||||
"unit_of_measurement": "%%",
|
||||
"device_class": "humidity",
|
||||
"unique_id": "%s",
|
||||
"device": {
|
||||
"identifiers": ["telldus_sensor_%s_%s_%d"],
|
||||
"name": "%s",
|
||||
"manufacturer": "Telldus"
|
||||
}
|
||||
}`, sensor.Name, sensor.Protocol, sensor.Model, sensor.ID, sensor.HumidityUniqueID,
|
||||
sensor.Protocol, sensor.Model, sensor.ID, sensor.Name)
|
||||
c.client.Publish(topic, 0, true, payload)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
102
mqtt/mqtt.go
Normal file
102
mqtt/mqtt.go
Normal file
@@ -0,0 +1,102 @@
|
||||
package mqtt
|
||||
|
||||
import (
|
||||
"app/datastore"
|
||||
"app/telldus"
|
||||
"fmt"
|
||||
|
||||
mqtt "github.com/eclipse/paho.mqtt.golang"
|
||||
)
|
||||
|
||||
// Client handles all MQTT operations
|
||||
type Client struct {
|
||||
client mqtt.Client
|
||||
store *datastore.DataStore
|
||||
subscriptions []string
|
||||
}
|
||||
|
||||
// Config holds MQTT connection configuration
|
||||
type Config struct {
|
||||
BrokerURL string
|
||||
Username string
|
||||
Password string
|
||||
}
|
||||
|
||||
// New creates a new MQTT client
|
||||
func New(cfg Config, store *datastore.DataStore) (*Client, error) {
|
||||
opts := mqtt.NewClientOptions().AddBroker(cfg.BrokerURL)
|
||||
opts.SetUsername(cfg.Username)
|
||||
opts.SetPassword(cfg.Password)
|
||||
|
||||
client := mqtt.NewClient(opts)
|
||||
if token := client.Connect(); token.Wait() && token.Error() != nil {
|
||||
return nil, token.Error()
|
||||
}
|
||||
|
||||
return &Client{
|
||||
client: client,
|
||||
store: store,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Close disconnects the MQTT client
|
||||
func (c *Client) Close() {
|
||||
c.client.Disconnect(250)
|
||||
}
|
||||
|
||||
// PublishDeviceState publishes a device state change
|
||||
func (c *Client) PublishDeviceState(deviceID int, state string) {
|
||||
topic := fmt.Sprintf("telldus/device/%d/state", deviceID)
|
||||
c.client.Publish(topic, 0, false, state)
|
||||
}
|
||||
|
||||
// PublishSensorValue publishes a sensor value
|
||||
func (c *Client) PublishSensorValue(protocol, model string, id int, dataType int, value string) {
|
||||
var topic string
|
||||
switch dataType {
|
||||
case telldus.DataTypeTemperature:
|
||||
topic = fmt.Sprintf("telldus/sensor/%s/%s/%d/temperature", protocol, model, id)
|
||||
case telldus.DataTypeHumidity:
|
||||
topic = fmt.Sprintf("telldus/sensor/%s/%s/%d/humidity", protocol, model, id)
|
||||
default:
|
||||
return
|
||||
}
|
||||
c.client.Publish(topic, 0, false, value)
|
||||
}
|
||||
|
||||
// UnsubscribeFromDeviceCommands unsubscribes from all tracked command topics
|
||||
func (c *Client) UnsubscribeFromDeviceCommands() {
|
||||
if len(c.subscriptions) > 0 {
|
||||
if token := c.client.Unsubscribe(c.subscriptions...); token.Wait() && token.Error() != nil {
|
||||
// Log error but don't fail
|
||||
}
|
||||
c.subscriptions = nil
|
||||
}
|
||||
}
|
||||
|
||||
// SubscribeToDeviceCommands subscribes to command topics for all devices
|
||||
func (c *Client) SubscribeToDeviceCommands() error {
|
||||
// Unsubscribe from existing subscriptions first
|
||||
c.UnsubscribeFromDeviceCommands()
|
||||
|
||||
numDevices := telldus.GetNumberOfDevices()
|
||||
for i := 0; i < numDevices; i++ {
|
||||
deviceID := telldus.GetDeviceId(i)
|
||||
topic := fmt.Sprintf("telldus/device/%d/set", deviceID)
|
||||
|
||||
// Capture deviceID in closure
|
||||
id := deviceID
|
||||
c.client.Subscribe(topic, 0, func(client mqtt.Client, msg mqtt.Message) {
|
||||
payload := string(msg.Payload())
|
||||
if payload == "ON" {
|
||||
telldus.TurnOn(id)
|
||||
} else if payload == "OFF" {
|
||||
telldus.TurnOff(id)
|
||||
}
|
||||
})
|
||||
|
||||
// Track subscription
|
||||
c.subscriptions = append(c.subscriptions, topic)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
156
telldus-daemon/daemon.go
Normal file
156
telldus-daemon/daemon.go
Normal file
@@ -0,0 +1,156 @@
|
||||
package daemon
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"fmt"
|
||||
"log"
|
||||
"os/exec"
|
||||
"sync"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"app/telldus"
|
||||
)
|
||||
|
||||
// Manager handles the telldusd daemon lifecycle
|
||||
type Manager struct {
|
||||
cmd *exec.Cmd
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
// New creates a new daemon manager
|
||||
func New() *Manager {
|
||||
return &Manager{}
|
||||
}
|
||||
|
||||
// Start starts the telldusd daemon and captures its output
|
||||
func (m *Manager) Start() error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
|
||||
if m.cmd != nil && m.cmd.Process != nil {
|
||||
log.Println("Telldusd already running")
|
||||
return nil
|
||||
}
|
||||
|
||||
log.Println("Starting telldusd...")
|
||||
cmd := exec.Command("/usr/local/sbin/telldusd", "--nodaemon")
|
||||
|
||||
// Capture stdout
|
||||
stdout, err := cmd.StdoutPipe()
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to get stdout pipe: %v", err)
|
||||
}
|
||||
|
||||
// Capture stderr
|
||||
stderr, err := cmd.StderrPipe()
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to get stderr pipe: %v", err)
|
||||
}
|
||||
|
||||
// Start the command
|
||||
if err := cmd.Start(); err != nil {
|
||||
return fmt.Errorf("failed to start telldusd: %v", err)
|
||||
}
|
||||
|
||||
m.cmd = cmd
|
||||
|
||||
// Log stdout in a goroutine
|
||||
go func() {
|
||||
scanner := bufio.NewScanner(stdout)
|
||||
for scanner.Scan() {
|
||||
log.Printf("[telldusd] %s", scanner.Text())
|
||||
}
|
||||
}()
|
||||
|
||||
// Log stderr in a goroutine
|
||||
go func() {
|
||||
scanner := bufio.NewScanner(stderr)
|
||||
for scanner.Scan() {
|
||||
log.Printf("[telldusd] ERROR: %s", scanner.Text())
|
||||
}
|
||||
}()
|
||||
|
||||
// Monitor process in a goroutine
|
||||
go func() {
|
||||
err := cmd.Wait()
|
||||
m.mu.Lock()
|
||||
m.cmd = nil
|
||||
m.mu.Unlock()
|
||||
if err != nil {
|
||||
log.Printf("Telldusd exited with error: %v", err)
|
||||
} else {
|
||||
log.Println("Telldusd exited normally")
|
||||
}
|
||||
}()
|
||||
|
||||
// Give telldusd a moment to start
|
||||
time.Sleep(500 * time.Millisecond)
|
||||
log.Println("Telldusd started successfully")
|
||||
return nil
|
||||
}
|
||||
|
||||
// Stop stops the telldusd daemon
|
||||
func (m *Manager) Stop() error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
|
||||
if m.cmd == nil || m.cmd.Process == nil {
|
||||
log.Println("Telldusd not running")
|
||||
return nil
|
||||
}
|
||||
|
||||
log.Println("Stopping telldusd...")
|
||||
|
||||
// Send SIGTERM
|
||||
if err := m.cmd.Process.Signal(syscall.SIGTERM); err != nil {
|
||||
log.Printf("Failed to send SIGTERM to telldusd: %v", err)
|
||||
// Try SIGKILL as fallback
|
||||
if err := m.cmd.Process.Kill(); err != nil {
|
||||
return fmt.Errorf("failed to kill telldusd: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// Wait for process to exit (with timeout)
|
||||
done := make(chan error, 1)
|
||||
go func() {
|
||||
done <- m.cmd.Wait()
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-done:
|
||||
log.Println("Telldusd stopped successfully")
|
||||
case <-time.After(5 * time.Second):
|
||||
log.Println("Telldusd did not stop gracefully, killing...")
|
||||
m.cmd.Process.Kill()
|
||||
}
|
||||
|
||||
m.cmd = nil
|
||||
return nil
|
||||
}
|
||||
|
||||
// Restart restarts the telldusd daemon
|
||||
func (m *Manager) Restart() error {
|
||||
log.Println("Restarting telldusd due to configuration change...")
|
||||
|
||||
if err := m.Stop(); err != nil {
|
||||
log.Printf("Error stopping telldusd: %v", err)
|
||||
}
|
||||
|
||||
// Give it a moment to fully stop
|
||||
time.Sleep(1 * time.Second)
|
||||
|
||||
// Close and reinitialize telldus library
|
||||
telldus.Close()
|
||||
time.Sleep(500 * time.Millisecond)
|
||||
|
||||
if err := m.Start(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Reinitialize telldus library
|
||||
telldus.Init()
|
||||
|
||||
log.Println("Telldusd restarted successfully")
|
||||
return nil
|
||||
}
|
||||
63
telldus-daemon/watcher.go
Normal file
63
telldus-daemon/watcher.go
Normal file
@@ -0,0 +1,63 @@
|
||||
package daemon
|
||||
|
||||
import (
|
||||
"log"
|
||||
"path/filepath"
|
||||
|
||||
"github.com/fsnotify/fsnotify"
|
||||
)
|
||||
|
||||
// Watcher watches for configuration file changes
|
||||
type Watcher struct {
|
||||
configPath string
|
||||
onReloadFunc func() error
|
||||
}
|
||||
|
||||
// NewWatcher creates a new config file watcher
|
||||
func NewWatcher(configPath string, onReload func() error) *Watcher {
|
||||
return &Watcher{
|
||||
configPath: configPath,
|
||||
onReloadFunc: onReload,
|
||||
}
|
||||
}
|
||||
|
||||
// Watch starts watching for changes to the configuration file
|
||||
func (w *Watcher) Watch() error {
|
||||
watcher, err := fsnotify.NewWatcher()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer watcher.Close()
|
||||
|
||||
// Watch the parent directory since file operations might replace the file
|
||||
configDir := filepath.Dir(w.configPath)
|
||||
if err := watcher.Add(configDir); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
log.Printf("Watching for changes to %s", w.configPath)
|
||||
|
||||
for {
|
||||
select {
|
||||
case event, ok := <-watcher.Events:
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
// Check if the event is for our config file
|
||||
if event.Name == w.configPath && (event.Op&fsnotify.Write == fsnotify.Write || event.Op&fsnotify.Create == fsnotify.Create) {
|
||||
log.Printf("Configuration file changed: %s", event.Op.String())
|
||||
// Call reload callback if provided
|
||||
if w.onReloadFunc != nil {
|
||||
if err := w.onReloadFunc(); err != nil {
|
||||
log.Printf("Failed to reload devices: %v", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
case err, ok := <-watcher.Errors:
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
log.Printf("File watcher error: %v", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user