markus
522fe2746a
Add additional functions to the repository to add or update devices, sensors or measured values. Furthermore the test has been adapt to the new functions.
938 lines
22 KiB
Go
938 lines
22 KiB
Go
package db
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"fmt"
|
|
"time"
|
|
|
|
"git.cryptic.systems/volker.raschek/flucky/pkg/types"
|
|
"git.cryptic.systems/volker.raschek/go-logger"
|
|
)
|
|
|
|
// Postgres implementation
|
|
type Postgres struct {
|
|
dbo *sql.DB
|
|
flogger logger.Logger
|
|
queries map[string]string
|
|
}
|
|
|
|
// Close closes the database and prevents new queries from starting. Close then
|
|
// waits for all queries that have started processing on the server to finish.
|
|
func (postgres *Postgres) Close() error {
|
|
return postgres.dbo.Close()
|
|
}
|
|
|
|
// DeleteDevices from the database
|
|
func (postgres *Postgres) DeleteDevices(ctx context.Context, deviceIDs ...string) error {
|
|
queryFile := "deleteDevice.sql"
|
|
query, present := postgres.queries[queryFile]
|
|
if !present {
|
|
return fmt.Errorf("Postgres-Backend: File %v not found", queryFile)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: false})
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to begin new transaction: %v", err)
|
|
}
|
|
|
|
stmt, err := tx.Prepare(query)
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to prepare statement: %v", err)
|
|
}
|
|
defer stmt.Close()
|
|
|
|
for _, deviceID := range deviceIDs {
|
|
_, err = stmt.Exec(deviceID)
|
|
if err != nil {
|
|
tx.Rollback()
|
|
return err
|
|
}
|
|
}
|
|
|
|
return tx.Commit()
|
|
}
|
|
|
|
// DeleteSensors from the database
|
|
func (postgres *Postgres) DeleteSensors(ctx context.Context, sensorIDs ...string) error {
|
|
queryFile := "deleteSensor.sql"
|
|
query, present := postgres.queries[queryFile]
|
|
if !present {
|
|
return fmt.Errorf("Postgres-Backend: File %v not found", queryFile)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: false})
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to begin new transaction: %v", err)
|
|
}
|
|
|
|
stmt, err := tx.Prepare(query)
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to prepare statement: %v", err)
|
|
}
|
|
defer stmt.Close()
|
|
|
|
for _, sensorID := range sensorIDs {
|
|
_, err = stmt.Exec(sensorID)
|
|
if err != nil {
|
|
tx.Rollback()
|
|
return err
|
|
}
|
|
}
|
|
|
|
return tx.Commit()
|
|
}
|
|
|
|
// InsertDevices into the database
|
|
func (postgres *Postgres) InsertDevices(ctx context.Context, devices ...*types.Device) error {
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: false})
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to begin new transaction: %v", err)
|
|
}
|
|
|
|
err = postgres.insertDevices(tx, devices...)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return tx.Commit()
|
|
}
|
|
|
|
func (postgres *Postgres) insertDevices(tx *sql.Tx, devices ...*types.Device) error {
|
|
queryFile := "insertDevice.sql"
|
|
query, present := postgres.queries[queryFile]
|
|
if !present {
|
|
return fmt.Errorf("Postgres-Backend: File %v not found", queryFile)
|
|
}
|
|
|
|
stmt, err := tx.Prepare(query)
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to prepare statement: %v", err)
|
|
}
|
|
defer stmt.Close()
|
|
|
|
for _, device := range devices {
|
|
|
|
if device.CreationDate.Equal(time.Time{}) {
|
|
device.CreationDate = time.Now()
|
|
}
|
|
|
|
_, err = stmt.Exec(
|
|
&device.ID,
|
|
&device.Name,
|
|
&device.Location,
|
|
&device.CreationDate,
|
|
&device.UpdateDate,
|
|
)
|
|
if err != nil {
|
|
tx.Rollback()
|
|
return fmt.Errorf("Failed to execute statement: %v", err)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// InsertMeasuredValues into the database
|
|
func (postgres *Postgres) InsertMeasuredValues(ctx context.Context, measuredValues ...*types.MeasuredValue) error {
|
|
splittedMeasuredValues := make(map[types.MeasuredValueType][]*types.MeasuredValue, 0)
|
|
|
|
for _, measuredValue := range measuredValues {
|
|
if _, ok := splittedMeasuredValues[measuredValue.ValueType]; !ok {
|
|
splittedMeasuredValues[measuredValue.ValueType] = make([]*types.MeasuredValue, 0)
|
|
}
|
|
splittedMeasuredValues[measuredValue.ValueType] = append(splittedMeasuredValues[measuredValue.ValueType], measuredValue)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: false})
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to begin new transaction: %v", err)
|
|
}
|
|
|
|
// General insert function
|
|
insert := func(tx *sql.Tx, queryFile string, measuredValues []*types.MeasuredValue) error {
|
|
query, present := postgres.queries[queryFile]
|
|
if !present {
|
|
return fmt.Errorf("Postgres-Backend: File %v not found", queryFile)
|
|
}
|
|
|
|
stmt, err := tx.Prepare(query)
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to prepare statement: %v", err)
|
|
}
|
|
defer stmt.Close()
|
|
|
|
for _, measuredValue := range measuredValues {
|
|
|
|
if measuredValue.CreationDate.Equal(time.Time{}) {
|
|
measuredValue.CreationDate = time.Now()
|
|
}
|
|
|
|
_, err := stmt.Exec(
|
|
&measuredValue.ID,
|
|
&measuredValue.Value,
|
|
&measuredValue.Date,
|
|
&measuredValue.SensorID,
|
|
&measuredValue.CreationDate,
|
|
&measuredValue.UpdateDate,
|
|
)
|
|
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to execute statement: %v", err)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
for measuredValueType, measuredValues := range splittedMeasuredValues {
|
|
var queryFile string
|
|
|
|
switch measuredValueType {
|
|
case types.Humidity:
|
|
queryFile = "insertHumidity.sql"
|
|
case types.Pressure:
|
|
queryFile = "insertPressure.sql"
|
|
case types.Temperature:
|
|
queryFile = "insertTemperature.sql"
|
|
default:
|
|
tx.Rollback()
|
|
return fmt.Errorf("Measured value type %v not supported", measuredValueType)
|
|
}
|
|
|
|
err := insert(tx, queryFile, measuredValues)
|
|
if err != nil {
|
|
tx.Rollback()
|
|
return err
|
|
}
|
|
}
|
|
|
|
return tx.Commit()
|
|
}
|
|
|
|
func (postgres *Postgres) InsertOrUpdateDevices(ctx context.Context, devices ...*types.Device) error {
|
|
queryFile := "insertOrUpdateDevice.sql"
|
|
query, present := postgres.queries[queryFile]
|
|
if !present {
|
|
return fmt.Errorf("Postgres-Backend: File %v not found", queryFile)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: false})
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to begin new transaction: %v", err)
|
|
}
|
|
|
|
for _, device := range devices {
|
|
|
|
if device.CreationDate.Equal(time.Time{}) {
|
|
device.CreationDate = time.Now()
|
|
}
|
|
|
|
_, err = tx.Exec(
|
|
query,
|
|
&device.ID,
|
|
&device.Name,
|
|
&device.Location,
|
|
&device.CreationDate,
|
|
&device.UpdateDate,
|
|
)
|
|
if err != nil {
|
|
tx.Rollback()
|
|
return fmt.Errorf("Failed to execute statement: %v", err)
|
|
}
|
|
}
|
|
|
|
return tx.Commit()
|
|
}
|
|
|
|
// InsertOrUpdateMeasuredValues into the database
|
|
func (postgres *Postgres) InsertOrUpdateMeasuredValues(ctx context.Context, measuredValues ...*types.MeasuredValue) error {
|
|
splittedMeasuredValues := make(map[types.MeasuredValueType][]*types.MeasuredValue, 0)
|
|
|
|
for _, measuredValue := range measuredValues {
|
|
if _, ok := splittedMeasuredValues[measuredValue.ValueType]; !ok {
|
|
splittedMeasuredValues[measuredValue.ValueType] = make([]*types.MeasuredValue, 0)
|
|
}
|
|
splittedMeasuredValues[measuredValue.ValueType] = append(splittedMeasuredValues[measuredValue.ValueType], measuredValue)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: false})
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to begin new transaction: %v", err)
|
|
}
|
|
|
|
// General insert function
|
|
insert := func(tx *sql.Tx, queryFile string, measuredValues []*types.MeasuredValue) error {
|
|
query, present := postgres.queries[queryFile]
|
|
if !present {
|
|
return fmt.Errorf("Postgres-Backend: File %v not found", queryFile)
|
|
}
|
|
|
|
stmt, err := tx.Prepare(query)
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to prepare statement: %v", err)
|
|
}
|
|
defer stmt.Close()
|
|
|
|
for _, measuredValue := range measuredValues {
|
|
|
|
if measuredValue.CreationDate.Equal(time.Time{}) {
|
|
measuredValue.CreationDate = time.Now()
|
|
}
|
|
|
|
_, err := stmt.Exec(
|
|
&measuredValue.ID,
|
|
&measuredValue.Value,
|
|
&measuredValue.Date,
|
|
&measuredValue.SensorID,
|
|
&measuredValue.CreationDate,
|
|
&measuredValue.UpdateDate,
|
|
)
|
|
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to execute statement: %v", err)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
for measuredValueType, measuredValues := range splittedMeasuredValues {
|
|
var queryFile string
|
|
|
|
switch measuredValueType {
|
|
case types.Humidity:
|
|
queryFile = "insertOrUpdateHumidity.sql"
|
|
case types.Pressure:
|
|
queryFile = "insertOrUpdatePressure.sql"
|
|
case types.Temperature:
|
|
queryFile = "insertOrUpdateTemperature.sql"
|
|
default:
|
|
tx.Rollback()
|
|
return fmt.Errorf("Measured value type %v not supported", measuredValueType)
|
|
}
|
|
|
|
err := insert(tx, queryFile, measuredValues)
|
|
if err != nil {
|
|
tx.Rollback()
|
|
return err
|
|
}
|
|
}
|
|
|
|
return tx.Commit()
|
|
}
|
|
|
|
// InsertOrUpdateSensors into the database
|
|
func (postgres *Postgres) InsertOrUpdateSensors(ctx context.Context, sensors ...*types.Sensor) error {
|
|
queryFile := "insertOrUpdateSensor.sql"
|
|
query, present := postgres.queries[queryFile]
|
|
if !present {
|
|
return fmt.Errorf("Postgres-Backend: File %v not found", queryFile)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: false})
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to begin new transaction: %v", err)
|
|
}
|
|
|
|
stmt, err := tx.Prepare(query)
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to prepare statement: %v", err)
|
|
}
|
|
defer stmt.Close()
|
|
|
|
for _, sensor := range sensors {
|
|
|
|
if sensor.CreationDate.Equal(time.Time{}) {
|
|
sensor.CreationDate = time.Now()
|
|
}
|
|
|
|
_, err = stmt.Exec(
|
|
&sensor.ID,
|
|
&sensor.Name,
|
|
&sensor.Location,
|
|
&sensor.WireID,
|
|
&sensor.I2CBus,
|
|
&sensor.I2CAddress,
|
|
&sensor.GPIONumber,
|
|
&sensor.Model,
|
|
&sensor.Enabled,
|
|
&sensor.TickDuration,
|
|
&sensor.DeviceID,
|
|
&sensor.CreationDate,
|
|
&sensor.UpdateDate,
|
|
)
|
|
if err != nil {
|
|
tx.Rollback()
|
|
return fmt.Errorf("Failed to execute statement: %v", err)
|
|
}
|
|
}
|
|
|
|
return tx.Commit()
|
|
}
|
|
|
|
// InsertSensors into the database
|
|
func (postgres *Postgres) InsertSensors(ctx context.Context, sensors ...*types.Sensor) error {
|
|
queryFile := "insertSensor.sql"
|
|
query, present := postgres.queries[queryFile]
|
|
if !present {
|
|
return fmt.Errorf("Postgres-Backend: File %v not found", queryFile)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: false})
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to begin new transaction: %v", err)
|
|
}
|
|
|
|
stmt, err := tx.Prepare(query)
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to prepare statement: %v", err)
|
|
}
|
|
defer stmt.Close()
|
|
|
|
for _, sensor := range sensors {
|
|
|
|
if sensor.CreationDate.Equal(time.Time{}) {
|
|
sensor.CreationDate = time.Now()
|
|
}
|
|
|
|
_, err = stmt.Exec(
|
|
&sensor.ID,
|
|
&sensor.Name,
|
|
&sensor.Location,
|
|
&sensor.WireID,
|
|
&sensor.I2CBus,
|
|
&sensor.I2CAddress,
|
|
&sensor.GPIONumber,
|
|
&sensor.Model,
|
|
&sensor.Enabled,
|
|
&sensor.TickDuration,
|
|
&sensor.DeviceID,
|
|
&sensor.CreationDate,
|
|
&sensor.UpdateDate,
|
|
)
|
|
if err != nil {
|
|
tx.Rollback()
|
|
return fmt.Errorf("Failed to execute statement: %v", err)
|
|
}
|
|
}
|
|
|
|
return tx.Commit()
|
|
}
|
|
|
|
// Scheme creates all required tables if not exist
|
|
func (postgres *Postgres) Scheme(ctx context.Context) error {
|
|
for _, query := range []string{
|
|
postgres.queries["createTableDevices.sql"],
|
|
postgres.queries["createTableSensors.sql"],
|
|
postgres.queries["createTableHumidities.sql"],
|
|
postgres.queries["createTablePressures.sql"],
|
|
postgres.queries["createTableTemperatures.sql"],
|
|
} {
|
|
_, err := postgres.dbo.ExecContext(ctx, query)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// SelectDevice from database
|
|
func (postgres *Postgres) SelectDevice(ctx context.Context, id string) (*types.Device, error) {
|
|
queryFile := "selectDevice.sql"
|
|
query, present := postgres.queries[queryFile]
|
|
if !present {
|
|
return nil, fmt.Errorf("Postgres-Backend: File %v not found", queryFile)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to begin new transaction: %v", err)
|
|
}
|
|
|
|
devices, err := postgres.selectDevices(tx, query, id)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
err = tx.Commit()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if len(devices) == 0 {
|
|
return nil, nil
|
|
}
|
|
|
|
return devices[0], nil
|
|
}
|
|
|
|
// SelectDevices from the database
|
|
func (postgres *Postgres) SelectDevices(ctx context.Context) ([]*types.Device, error) {
|
|
queryFile := "selectDevices.sql"
|
|
query, present := postgres.queries[queryFile]
|
|
if !present {
|
|
return nil, fmt.Errorf("Postgres-Backend: File %v not found", queryFile)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to begin new transaction: %v", err)
|
|
}
|
|
|
|
devices, err := postgres.selectDevices(tx, query)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
err = tx.Commit()
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to commit transaction: %v", err)
|
|
}
|
|
|
|
return devices, nil
|
|
}
|
|
|
|
func (postgres *Postgres) selectDevices(tx *sql.Tx, query string, args ...interface{}) ([]*types.Device, error) {
|
|
stmt, err := tx.Prepare(query)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to prepare statement: %v", err)
|
|
}
|
|
defer stmt.Close()
|
|
|
|
rows, err := stmt.Query(args...)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to query statement: %v", err)
|
|
}
|
|
|
|
devices := make([]*types.Device, 0)
|
|
for rows.Next() {
|
|
device := new(types.Device)
|
|
err = rows.Scan(
|
|
&device.ID,
|
|
&device.Name,
|
|
&device.Location,
|
|
&device.CreationDate,
|
|
&device.UpdateDate,
|
|
)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to scan row: %v", err)
|
|
}
|
|
devices = append(devices, device)
|
|
}
|
|
|
|
return devices, nil
|
|
}
|
|
|
|
// SelectHumidity returns humidity from the database
|
|
func (postgres *Postgres) SelectHumidity(ctx context.Context, id string) (*types.MeasuredValue, error) {
|
|
queryFile := "selectHumidity.sql"
|
|
query, present := postgres.queries[queryFile]
|
|
if !present {
|
|
return nil, fmt.Errorf("Postgres-Backend: File %v not found", queryFile)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
measuredValues, err := postgres.selectMeasuredValue(tx, query, id)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
err = tx.Commit()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if measuredValues == nil {
|
|
return nil, nil
|
|
}
|
|
|
|
for _, measuredValue := range measuredValues {
|
|
measuredValue.ValueType = types.Humidity
|
|
}
|
|
|
|
return measuredValues[0], nil
|
|
}
|
|
|
|
// SelectHumidities returns humidities from the database
|
|
func (postgres *Postgres) SelectHumidities(ctx context.Context) ([]*types.MeasuredValue, error) {
|
|
queryFile := "selectHumidities.sql"
|
|
query, present := postgres.queries[queryFile]
|
|
if !present {
|
|
return nil, fmt.Errorf("Postgres-Backend: File %v not found", queryFile)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
measuredValues, err := postgres.selectMeasuredValue(tx, query)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
err = tx.Commit()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
for _, measuredValue := range measuredValues {
|
|
measuredValue.ValueType = types.Humidity
|
|
}
|
|
|
|
return measuredValues, nil
|
|
}
|
|
|
|
func (postgres *Postgres) selectMeasuredValue(tx *sql.Tx, query string, args ...interface{}) ([]*types.MeasuredValue, error) {
|
|
stmt, err := tx.Prepare(query)
|
|
if err != nil {
|
|
tx.Rollback()
|
|
return nil, err
|
|
}
|
|
defer stmt.Close()
|
|
|
|
rows, err := stmt.Query(args...)
|
|
if err != nil {
|
|
tx.Rollback()
|
|
return nil, err
|
|
}
|
|
|
|
measuredValues := make([]*types.MeasuredValue, 0)
|
|
for rows.Next() {
|
|
measuredValue := new(types.MeasuredValue)
|
|
err := rows.Scan(
|
|
&measuredValue.ID,
|
|
&measuredValue.Value,
|
|
&measuredValue.Date,
|
|
&measuredValue.SensorID,
|
|
&measuredValue.CreationDate,
|
|
&measuredValue.UpdateDate,
|
|
)
|
|
|
|
if err != nil {
|
|
tx.Rollback()
|
|
return nil, err
|
|
}
|
|
|
|
measuredValues = append(measuredValues, measuredValue)
|
|
}
|
|
|
|
return measuredValues, nil
|
|
}
|
|
|
|
// SelectPressure returns pressure from the database
|
|
func (postgres *Postgres) SelectPressure(ctx context.Context, id string) (*types.MeasuredValue, error) {
|
|
queryFile := "selectPressure.sql"
|
|
query, present := postgres.queries[queryFile]
|
|
if !present {
|
|
return nil, fmt.Errorf("Postgres-Backend: File %v not found", queryFile)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
measuredValues, err := postgres.selectMeasuredValue(tx, query, id)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
err = tx.Commit()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if measuredValues == nil {
|
|
return nil, nil
|
|
}
|
|
|
|
for _, measuredValue := range measuredValues {
|
|
measuredValue.ValueType = types.Pressure
|
|
}
|
|
|
|
return measuredValues[0], nil
|
|
}
|
|
|
|
// SelectPressures returns pressure from the database
|
|
func (postgres *Postgres) SelectPressures(ctx context.Context) ([]*types.MeasuredValue, error) {
|
|
queryFile := "selectPressures.sql"
|
|
query, present := postgres.queries[queryFile]
|
|
if !present {
|
|
return nil, fmt.Errorf("Postgres-Backend: File %v not found", queryFile)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
measuredValues, err := postgres.selectMeasuredValue(tx, query)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
err = tx.Commit()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
for _, measuredValue := range measuredValues {
|
|
measuredValue.ValueType = types.Pressure
|
|
}
|
|
|
|
return measuredValues, nil
|
|
}
|
|
|
|
// SelectSensor from database
|
|
func (postgres *Postgres) SelectSensor(ctx context.Context, id string) (*types.Sensor, error) {
|
|
queryFile := "selectSensor.sql"
|
|
query, present := postgres.queries[queryFile]
|
|
if !present {
|
|
return nil, fmt.Errorf("Postgres-Backend: File %v not found", queryFile)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to begin new transaction: %v", err)
|
|
}
|
|
|
|
sensors, err := postgres.selectSensors(tx, query, id)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
err = tx.Commit()
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to commit transaction: %v", err)
|
|
}
|
|
|
|
if len(sensors) == 0 {
|
|
return nil, nil
|
|
}
|
|
|
|
return sensors[0], nil
|
|
}
|
|
|
|
// SelectSensors from the database
|
|
func (postgres *Postgres) SelectSensors(ctx context.Context) ([]*types.Sensor, error) {
|
|
queryFile := "selectSensors.sql"
|
|
query, present := postgres.queries[queryFile]
|
|
if !present {
|
|
return nil, fmt.Errorf("Postgres-Backend: File %v not found", queryFile)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to begin new transaction: %v", err)
|
|
}
|
|
|
|
sensors, err := postgres.selectSensors(tx, query)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
err = tx.Commit()
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to commit transaction: %v", err)
|
|
}
|
|
|
|
return sensors, nil
|
|
}
|
|
|
|
func (postgres *Postgres) selectSensors(tx *sql.Tx, query string, args ...interface{}) ([]*types.Sensor, error) {
|
|
stmt, err := tx.Prepare(query)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to prepare statement: %v", err)
|
|
}
|
|
defer stmt.Close()
|
|
|
|
rows, err := stmt.Query(args...)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to query statement: %v", err)
|
|
}
|
|
|
|
sensors := make([]*types.Sensor, 0)
|
|
for rows.Next() {
|
|
sensor := new(types.Sensor)
|
|
err = rows.Scan(
|
|
&sensor.ID,
|
|
&sensor.Name,
|
|
&sensor.Location,
|
|
&sensor.WireID,
|
|
&sensor.I2CBus,
|
|
&sensor.I2CAddress,
|
|
&sensor.GPIONumber,
|
|
&sensor.Model,
|
|
&sensor.Enabled,
|
|
&sensor.TickDuration,
|
|
&sensor.DeviceID,
|
|
&sensor.CreationDate,
|
|
&sensor.UpdateDate,
|
|
)
|
|
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Failed to scan row: %v", err)
|
|
}
|
|
sensors = append(sensors, sensor)
|
|
}
|
|
|
|
return sensors, nil
|
|
}
|
|
|
|
// SelectTemperature returns temperatures from the database
|
|
func (postgres *Postgres) SelectTemperature(ctx context.Context, id string) (*types.MeasuredValue, error) {
|
|
queryFile := "selectTemperature.sql"
|
|
query, present := postgres.queries[queryFile]
|
|
if !present {
|
|
return nil, fmt.Errorf("Postgres-Backend: File %v not found", queryFile)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
measuredValues, err := postgres.selectMeasuredValue(tx, query, id)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
err = tx.Commit()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if measuredValues == nil {
|
|
return nil, nil
|
|
}
|
|
|
|
for _, measuredValue := range measuredValues {
|
|
measuredValue.ValueType = types.Temperature
|
|
}
|
|
|
|
return measuredValues[0], nil
|
|
}
|
|
|
|
// SelectTemperatures returns temperatures from the database
|
|
func (postgres *Postgres) SelectTemperatures(ctx context.Context) ([]*types.MeasuredValue, error) {
|
|
queryFile := "selectTemperatures.sql"
|
|
query, present := postgres.queries[queryFile]
|
|
if !present {
|
|
return nil, fmt.Errorf("Postgres-Backend: File %v not found", queryFile)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
measuredValues, err := postgres.selectMeasuredValue(tx, query)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
err = tx.Commit()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
for _, measuredValue := range measuredValues {
|
|
measuredValue.ValueType = types.Temperature
|
|
}
|
|
|
|
return measuredValues, nil
|
|
}
|
|
|
|
// UpdateDevices updates a device in the database
|
|
func (postgres *Postgres) UpdateDevices(ctx context.Context, devices ...*types.Device) error {
|
|
queryFile := "updateDevice.sql"
|
|
query, present := postgres.queries[queryFile]
|
|
if !present {
|
|
return fmt.Errorf("Postgres-Backend: File %v not found", queryFile)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: false})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
stmt, err := tx.Prepare(query)
|
|
if err != nil {
|
|
tx.Rollback()
|
|
return err
|
|
}
|
|
defer stmt.Close()
|
|
|
|
for _, device := range devices {
|
|
now := time.Now()
|
|
device.UpdateDate = &now
|
|
|
|
_, err := stmt.Exec(
|
|
&device.Name,
|
|
&device.Location,
|
|
&device.CreationDate,
|
|
&device.UpdateDate,
|
|
&device.ID,
|
|
)
|
|
if err != nil {
|
|
tx.Rollback()
|
|
return err
|
|
}
|
|
}
|
|
|
|
return tx.Commit()
|
|
}
|
|
|
|
// UpdateSensors updates a sensor in the database
|
|
func (postgres *Postgres) UpdateSensors(ctx context.Context, sensors ...*types.Sensor) error {
|
|
queryFile := "updateSensor.sql"
|
|
query, present := postgres.queries[queryFile]
|
|
if !present {
|
|
return fmt.Errorf("Postgres-Backend: File %v not found", queryFile)
|
|
}
|
|
|
|
tx, err := postgres.dbo.BeginTx(ctx, &sql.TxOptions{ReadOnly: false})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
stmt, err := tx.Prepare(query)
|
|
if err != nil {
|
|
tx.Rollback()
|
|
return err
|
|
}
|
|
defer stmt.Close()
|
|
|
|
for _, sensor := range sensors {
|
|
now := time.Now()
|
|
sensor.UpdateDate = &now
|
|
_, err := stmt.Exec(
|
|
&sensor.Name,
|
|
&sensor.Location,
|
|
&sensor.WireID,
|
|
&sensor.I2CBus,
|
|
&sensor.I2CAddress,
|
|
&sensor.GPIONumber,
|
|
&sensor.Model,
|
|
&sensor.Enabled,
|
|
&sensor.TickDuration,
|
|
&sensor.DeviceID,
|
|
&sensor.CreationDate,
|
|
&sensor.UpdateDate,
|
|
&sensor.ID,
|
|
)
|
|
if err != nil {
|
|
tx.Rollback()
|
|
return err
|
|
}
|
|
}
|
|
|
|
return tx.Commit()
|
|
}
|