kdeppe
/
DMAC
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

expanded the sensor/device manager stuff to prep for fully fledged managers

Browse Source
main
KeeganForelight 2 years ago
parent 63811505eb
commit e64efad3fc
14 changed files with 542 additions and 257 deletions
Show Stats Download Patch File Download Diff File

13
internal/configs/reactor.yaml
Unescape Escape View File

@ -1,9 +1,10 @@
devices:
address: 112
name: DO Sensor
reactor:
name: "Dummy Reactor"
id: 2166136261
model: ""
name: Dummy Reactor
server:
ip: "192.168.100.2"
ip: 192.168.100.2
port: 2022
devices:
address: 112 # decimal
name: "DO Sensor"

15
internal/pkg/I2C/device.go
Unescape Escape View File

@ -20,11 +20,11 @@ type data struct {
}
func (d I2CDevice) String() string {
t := map[int]string{97:"DO Sensor",99:"pH Sensor",102:"Temperature Sensor",64:"DHT11 Sensor"}
t := map[int]string{97: "DO Sensor", 99: "pH Sensor", 102: "Temperature Sensor", 64: "DHT11 Sensor"}
return t[d.int]
}
func NewDevice(addr int,bus *I2CBus) *I2CDevice {
func NewDevice(addr int, bus *I2CBus) *I2CDevice {
d := &I2CDevice{}
d.I2CBus = bus
d.int = addr
@ -36,16 +36,9 @@ func (d *I2CDevice) GetAddr() int {
return d.int
}
func (d *I2CDevice) GetStatus() string {
func (d *I2CDevice) GetStatus() bool {
// TODO
s := d.I2CBus.GetStatus(d.int)
if s {
d.Data.Active()
return "[green]ACTIVE[white]"
} else {
d.Data.Killed()
return "[red]KILLED[white]"
}
return d.I2CBus.GetStatus(d.int)
}
func (d *I2CDevice) GetType() string {

31
internal/pkg/I2C/monitor.go
Unescape Escape View File

@ -1,9 +1,9 @@
package I2C
import (
"time"
_ "fmt"
"sync"
"time"
)
/*
@ -21,12 +21,12 @@ type devs struct {
m map[int]*I2CDevice
}
func NewMonitor(bus int,ch chan int) *I2CMonitor {
func NewMonitor(bus int, ch chan int) *I2CMonitor {
m := &I2CMonitor{}
b := NewBus(bus)
m.I2CBus = b
d := make(map[int]*I2CDevice)
m.Devices = &devs{m:d}
m.Devices = &devs{m: d}
m.DevChan = ch
return m
}
@ -36,7 +36,7 @@ func (m *I2CMonitor) Update() {
scans bus and adds new active devices
*/
devs := m.Scan()
chng := m.Devices.Parse(m.I2CBus,devs)
chng := m.Devices.Parse(m.I2CBus, devs)
for _, d := range chng {
go m.ConnectDevice(d)
}
@ -46,12 +46,12 @@ func (m *I2CMonitor) Monitor() {
// functon that updates the device list and notifies rlc of any changes to sensor composition
s := make(chan struct{})
t := 5 * time.Second
go func(signal chan struct{},to time.Duration) { // simple signal func to init scan
go func(signal chan struct{}, to time.Duration) { // simple signal func to init scan
for {
signal <-struct{}{}
signal <- struct{}{}
time.Sleep(to)
}
}(s,t)
}(s, t)
for {
<-s
@ -60,16 +60,21 @@ func (m *I2CMonitor) Monitor() {
}
func (m *I2CMonitor) ConnectDevice(addr int) {
m.DevChan <-addr
m.DevChan <- addr
}
func (m *I2CMonitor) GetDevice(addr int) interface{ GetAddr() int; GetData() string; GetStatus() string; GetType() string } {
func (m *I2CMonitor) GetDevice(addr int) interface {
GetAddr() int
GetData() string
GetStatus() bool
GetType() string
} {
m.Devices.Lock()
defer m.Devices.Unlock()
return m.Devices.m[addr]
}
func (d *devs) Parse(bus *I2CBus,devices map[int]bool) []int {
func (d *devs) Parse(bus *I2CBus, devices map[int]bool) []int {
d.Lock()
defer d.Unlock()
newdevs := []int{}
@ -79,17 +84,17 @@ func (d *devs) Parse(bus *I2CBus,devices map[int]bool) []int {
if status != dev.bool { // if device state changed
dev.bool = status
if status {
newdevs = append(newdevs,dev.GetAddr())
newdevs = append(newdevs, dev.GetAddr())
}
}
} else {
// device not seen yet
if status {
// active
newd := NewDevice(addr,bus)
newd := NewDevice(addr, bus)
newd.bool = status
d.m[addr] = newd
newdevs = append(newdevs,newd.GetAddr())
newdevs = append(newdevs, newd.GetAddr())
}
}
}

10
internal/pkg/reactor/monitoring.go
Unescape Escape View File

@ -3,7 +3,6 @@ package reactor
import (
"context"
"fmt"
"sync"
//"FRMS/internal/pkg/logging"
//"google.golang.org/grpc"
@ -31,8 +30,10 @@ func (c *Coordinator) DevStatus(ch chan *DeviceStatus, a int, dm DeviceManager)
func (c *Coordinator) GetStatus() []*pb.Device {
// db stuff
//api := client.WriteAPIBlocking(c.Org, c.Bucket)
var wg sync.WaitGroup
//var wg sync.WaitGroup
devs := []*pb.Device{}
return devs
/*
statusChan := make(chan *DeviceStatus)
c.Devices.Lock()
for a, dm := range c.Devices.Managers {
@ -70,18 +71,21 @@ func (c *Coordinator) GetStatus() []*pb.Device {
}
devs = append(devs,&pb.Device{Addr:int32(s.Addr),Type:s.Type,Status:s.Status,Data:s.Data})
*/
/*
wg.Done()
case <-allDone:
return devs
}
}
*/
}
// grpc status update handler
func (c *Coordinator) Ping() {
// sends all device status to central coordinator
fmt.Printf("Pinging cc\n")
devs := c.GetStatus()
req := &pb.ReactorStatusPing{Id: uint32(c.ID), Devices: devs}
req := &pb.ReactorStatusPing{Id: int32(c.ID), Devices: devs}
_, err := c.MonitoringClient.ReactorStatusHandler(context.Background(), req)
if err != nil {
c.Err <- err

54
internal/pkg/reactor/rlcoordinator.go
Unescape Escape View File

@ -7,7 +7,6 @@ import (
pb "FRMS/internal/pkg/grpc"
"FRMS/internal/pkg/influxdb"
"FRMS/internal/pkg/logging"
"FRMS/internal/pkg/sensor"
"FRMS/internal/pkg/system"
"context"
"errors"
@ -51,7 +50,9 @@ type Coordinator struct {
Config *viper.Viper
MonitoringClient pb.MonitoringClient
// connected devices
Devices *DeviceManagers // struct for locking
*Devices // struct for locking
*Sensors
// other stuff and things
Err chan error
mu sync.Mutex
HB time.Duration
@ -67,7 +68,19 @@ type active struct {
sync.Mutex
}
type DeviceManagers struct {
type Sensors struct {
Managers map[int]SensorManager
sync.Mutex
}
type SensorManager interface {
Start()
GetType() string
GetStatus() string
GetData() string
}
type Devices struct {
Managers map[int]DeviceManager
sync.Mutex
}
@ -84,19 +97,19 @@ type DeviceManager interface {
type I2CDev interface {
GetAddr() int
GetData() string
GetStatus() string
GetStatus() bool
GetType() string
}
func NewDeviceManager(i2c I2CDev) DeviceManager {
return sensor.NewDeviceManager(i2c)
}
//func NewDeviceManager(i2c I2CDev) DeviceManager {
//return sensor.NewDeviceManager(i2c)
//}
type I2CMonitor interface {
Monitor()
GetDevice(int) interface {
GetAddr() int
GetStatus() string
GetStatus() bool
GetData() string
GetType() string
}
@ -107,9 +120,13 @@ func NewI2CMonitor(b int, ch chan int) I2CMonitor {
}
func NewCoordinator(config *viper.Viper, ch chan error) *Coordinator {
sen := new(DeviceManagers)
sen.Managers = make(map[int]DeviceManager)
c := &Coordinator{Err: ch, Devices: sen, Config: config}
// sensor/device manager struct
dm := new(Devices)
dm.Managers = make(map[int]DeviceManager)
sm := new(Sensors)
sm.Managers = make(map[int]SensorManager)
c := &Coordinator{Err: ch, Devices: dm, Sensors: sm, Config: config}
c.HB = time.Duration(5 * time.Second)
// this is going to be scuffed
@ -204,13 +221,14 @@ func (c *Coordinator) DeviceConnect(i2c I2CDev) {
c.Devices.Lock()
defer c.Devices.Unlock()
addr := i2c.GetAddr()
if dm, exists := c.Devices.Managers[addr]; !exists {
dm := NewDeviceManager(i2c)
c.Devices.Managers[addr] = dm
go dm.Start()
} else {
go dm.Start()
}
fmt.Printf("Device %d (%x) found!\n", addr, addr)
//if dm, exists := c.Devices.Managers[addr]; !exists {
//dm := NewDeviceManager(i2c)
//c.Devices.Managers[addr] = dm
//go dm.Start()
//} else {
//go dm.Start()
//}
}
func (c *Coordinator) Discover() {

45
internal/pkg/sensor/do_sensor.go
Unescape Escape View File

@ -0,0 +1,45 @@
package sensor
// do sensor and methods
import (
pb "FRMS/internal/pkg/grpc"
"errors"
"sync"
"github.com/spf13/viper"
)
type DOSensorManager struct {
// do sensor manager
Sensor *pb.Sensor // for sending/updating
sync.RWMutex
}
func (s *DOSensorManager) SetName(name string) error {
// #TODO
return errors.New("UNIMPL")
}
func (s *DOSensorManager) SetStatus(status int) error {
// #TODO
return errors.New("UNIMPL")
}
func (s *DOSensorManager) SetSampleRate(rate int) error {
// #TODO
return errors.New("UNIMPL")
}
func (s *DOSensorManager) Update(sensor *pb.Sensor, config *viper.Viper) error {
// updates info
s.Lock()
defer s.Unlock()
s.Sensor = sensor
return errors.New("UNIMPL")
}
func (s *DOSensorManager) String() string {
// #TODO
return "TODO"
}

12
internal/pkg/sensor/manager.go
Unescape Escape View File

@ -3,12 +3,10 @@ package sensor
import (
_ "FRMS/internal/pkg/I2C"
_ "fmt"
"log"
"strings"
"sync"
"time"
)
/*
// I think most of this is unnessecary as hell
type Manager struct {
*Dev
I2CDevice
@ -26,14 +24,13 @@ type Dev struct {
// last known values
Addr int
Type string
Status string // could be more efficient but to hell with it
Status int // could be more efficient but to hell with it
Data string
}
type I2CDevice interface {
// basic device info
GetAddr() int
GetStatus() string
GetStatus() bool
GetType() string
GetData() string
}
@ -112,3 +109,4 @@ func (a *Active) IsActive() bool {
defer a.Unlock()
return a.bool
}
*/

68
internal/pkg/sensor/mappings.go
Unescape Escape View File

@ -1,36 +1,66 @@
package sensor
import (
pb "FRMS/internal/pkg/grpc"
"errors"
"fmt"
"sync"
"github.com/spf13/viper"
)
/*
this file serves as a map that the sensor library can use to determine which manager to call
Returns the correct manager for sensor/device
*/
type NewManager interface {
// serves as interface to restrict managers can be relocated
type SensorManager interface {
// basic sensor stuff
SetName(string) error // change display name
SetStatus(int) error // update status
SetSampleRate(int) error // update sample rate
Update(*pb.Sensor, *viper.Viper) error // write updates
String() string // printable
}
type DM struct {
DeviceManagers map[uint]NewManager
sync.Mutex
type DeviceManager interface {
// basic device stuff
SetName(string) error // change displayed name
SetStatus(int) error // status change
SetParameter(string, string) error // key, val
Update(*pb.Device, *viper.Viper) error // write updates
String() string // printable
}
func NewManagerDirectory() *DM {
m := map[uint]NewManager{
// map to set functions up
//112: NewDOManager(),
func NewSensorManager(sensor *pb.Sensor) (SensorManager, error) {
// returns correct sensor manager by ID
var sm SensorManager
var err error
switch id := sensor.GetAddr(); id {
case 97:
// DO
sm = &DOSensorManager{Sensor: sensor}
case 99:
// pH
sm = &PHSensorManager{Sensor: sensor}
case 102:
// RTD
sm = &RTDSensorManager{Sensor: sensor}
default:
err = errors.New(fmt.Sprintf("Error: sensor id %v unrecognized!", id))
}
return &DM{DeviceManagers: m}
return sm, err
}
func (d *DM) GetManager(addr uint) (NewManager, error) {
d.Lock()
defer d.Unlock()
if m, ok := d.DeviceManagers[addr]; ok {
return m, nil
func NewDeviceManager(device *pb.Device) (DeviceManager, error) {
// returns correct device manager by ID
var dm DeviceManager
var err error
switch id := device.GetAddr(); id {
case 256:
// PWM
dm = &PWMDeviceManager{Device: device}
default:
err = errors.New(fmt.Sprintf("Error: device id %v unrecognized!", id))
}
return &DM{}, fmt.Errorf("No manager found for address %d!", addr)
return dm, err
}

45
internal/pkg/sensor/ph_sensor.go
Unescape Escape View File

@ -0,0 +1,45 @@
package sensor
// do sensor and methods
import (
pb "FRMS/internal/pkg/grpc"
"errors"
"sync"
"github.com/spf13/viper"
)
type PHSensorManager struct {
// do sensor manager
Sensor *pb.Sensor // for sending/updating
sync.RWMutex
}
func (s *PHSensorManager) SetName(name string) error {
// #TODO
return errors.New("UNIMPL")
}
func (s *PHSensorManager) SetStatus(status int) error {
// #TODO
return errors.New("UNIMPL")
}
func (s *PHSensorManager) SetSampleRate(rate int) error {
// #TODO
return errors.New("UNIMPL")
}
func (s *PHSensorManager) Update(sensor *pb.Sensor, config *viper.Viper) error {
// updates info
s.Lock()
defer s.Unlock()
s.Sensor = sensor
return errors.New("UNIMPL")
}
func (s *PHSensorManager) String() string {
// #TODO
return "TODO"
}

45
internal/pkg/sensor/pwm_device.go
Unescape Escape View File

@ -0,0 +1,45 @@
package sensor
// do sensor and methods
import (
pb "FRMS/internal/pkg/grpc"
"errors"
"sync"
"github.com/spf13/viper"
)
type PWMDeviceManager struct {
// do sensor manager
Device *pb.Device // for sending/updating
sync.RWMutex
}
func (s *PWMDeviceManager) SetName(name string) error {
// #TODO
return errors.New("UNIMPL")
}
func (s *PWMDeviceManager) SetStatus(status int) error {
// #TODO
return errors.New("UNIMPL")
}
func (s *PWMDeviceManager) SetParameter(key, val string) error {
// #TODO
return errors.New("UNIMPL")
}
func (s *PWMDeviceManager) Update(device *pb.Device, config *viper.Viper) error {
// updates info
s.Lock()
defer s.Unlock()
s.Device = device
return errors.New("UNIMPL")
}
func (s *PWMDeviceManager) String() string {
// #TODO
return "TODO"
}

45
internal/pkg/sensor/rtd_sensor.go
Unescape Escape View File

@ -0,0 +1,45 @@
package sensor
// do sensor and methods
import (
pb "FRMS/internal/pkg/grpc"
"errors"
"sync"
"github.com/spf13/viper"
)
type RTDSensorManager struct {
// do sensor manager
Sensor *pb.Sensor // for sending/updating
sync.RWMutex
}
func (s *RTDSensorManager) SetName(name string) error {
// #TODO
return errors.New("UNIMPL")
}
func (s *RTDSensorManager) SetStatus(status int) error {
// #TODO
return errors.New("UNIMPL")
}
func (s *RTDSensorManager) SetSampleRate(rate int) error {
// #TODO
return errors.New("UNIMPL")
}
func (s *RTDSensorManager) Update(sensor *pb.Sensor, config *viper.Viper) error {
// updates info
s.Lock()
defer s.Unlock()
s.Sensor = sensor
return errors.New("UNIMPL")
}
func (s *RTDSensorManager) String() string {
// #TODO
return "TODO"
}

6
internal/pkg/sensor/sensor.go
Unescape Escape View File

@ -1,6 +0,0 @@
package sensor
import (
_ "fmt"
)

135
internal/pkg/server/reactormanager.go
Unescape Escape View File

@ -3,10 +3,13 @@ package server
import (
pb "FRMS/internal/pkg/grpc"
"FRMS/internal/pkg/logging"
"FRMS/internal/pkg/sensor"
"context"
"fmt"
_ "log"
"sync"
"github.com/spf13/viper"
)
// this package will implement a reactor manager and associated go routines
@ -14,21 +17,33 @@ import (
type ReactorManager struct {
*Manager
// StatusMon *StatusMonitor putting on pause
*devstatus
*ReactorSensors
*ReactorDevices
Config *viper.Viper // config to update
}
type ReactorSensors struct {
// sensor struct
Sensors map[int]SensorManager
sync.RWMutex
}
type devstatus struct {
// keeping this around but not using it to create status for status mon
sync.Mutex
Devs map[int]*DeviceInfo
type ReactorDevices struct {
// device struct
Devices map[int]DeviceManager
sync.RWMutex
}
func NewReactorManager(err chan error) *ReactorManager {
r := &ReactorManager{}
di := make(map[int]*DeviceInfo)
r.devstatus = &devstatus{Devs: di}
// sub managers
dm := make(map[int]DeviceManager)
sm := make(map[int]SensorManager)
r.ReactorDevices = &ReactorDevices{Devices: dm}
r.ReactorSensors = &ReactorSensors{Sensors: sm}
// core manager
r.Manager = NewManager(err)
//r.StatusMon = NewStatusMonitor("Reactor", c.Id, sys)
return r
}
@ -41,38 +56,94 @@ func (r *ReactorManager) Exit() {
r.Manager.Exit()
logging.Debug(logging.DExit, "RMA %v exiting", r.Id)
//go r.StatusMon.Send(&DeviceInfo{Id: r.Id, Type: "Reactor", Status: "[red]OFFLINE[white]", Data: fmt.Sprintf("Last Seen %v", time.Now().Format("Mon at 03:04:05pm MST"))}, "Reactor")
r.devstatus.Lock()
defer r.devstatus.Unlock()
// keeping this because it **COULD** be useful, maybe
for _, d := range r.Devs {
newd := d
newd.Status = "UNKOWN"
r.Devs[newd.Id] = newd
//go r.StatusMon.Send(newd, "Device")
}
}
func (r *ReactorManager) ReactorStatusHandler(ctx context.Context, req *pb.ReactorStatusPing) (*pb.ReactorStatusResponse, error) {
// function client will call to update reactor information
//go r.PingReset()
fmt.Printf("Recieved ping from %v!\n", req.GetId())
for _, dev := range req.GetDevices() {
d := &DeviceInfo{Id: int(dev.GetAddr()), Type: dev.GetName(), Status: dev.GetStatus().String(), Data: dev.GetData()}
go r.UpdateDevice(d)
}
// update devices/sensors
go r.UpdateSensors(req.GetSensors())
go r.UpdateDevices(req.GetDevices())
return &pb.ReactorStatusResponse{Id: int32(r.Id)}, nil
}
func (r *ReactorManager) UpdateDevice(d *DeviceInfo) {
r.devstatus.Lock()
defer r.devstatus.Unlock()
if olddev, ok := r.Devs[d.Id]; !ok {
// new device
r.Devs[d.Id] = d
//go r.StatusMon.Send(d, "Device")
} else if olddev.Status != d.Status || olddev.Data != d.Data {
// dev status or data has changed
r.Devs[d.Id] = d
//go r.StatusMon.Send(d, "Device")
// sensor/device stuff
type SensorManager interface {
SetName(string) error // change name
SetSampleRate(int) error // change sample rate
Update(*pb.Sensor, *viper.Viper) error // write updates
String() string // printing
}
type DeviceManager interface {
SetName(string) error // change name
SetParameter(string, string) error // key, val
Update(*pb.Device, *viper.Viper) error // write updates
String() string // printing
}
func NewSensorManager(sens *pb.Sensor) (SensorManager, error) {
// returns a manager struct
return sensor.NewSensorManager(sens)
}
func NewDeviceManager(device *pb.Device) (DeviceManager, error) {
// returns a manager struct
return sensor.NewDeviceManager(device)
}
func (r *ReactorManager) UpdateDevices(devs []*pb.Device) {
// pass updates to correct manager
r.ReactorDevices.RLock() // read lock only
defer r.ReactorDevices.RUnlock()
for _, dev := range devs {
// looping over devs
if dm, ok := r.ReactorDevices.Devices[int(dev.GetAddr())]; ok {
// device manager found
go dm.Update(dev, r.Config) // update dm
} else {
// not found
go r.AddDevice(dev, r.Err)
}
}
}
func (r *ReactorDevices) AddDevice(dev *pb.Device, errCh chan error) {
r.Lock() // write lock
defer r.Unlock()
dm, err := NewDeviceManager(dev)
if err != nil {
errCh <- err
}
r.Devices[int(dev.GetAddr())] = dm
}
func (r *ReactorManager) UpdateSensors(sensors []*pb.Sensor) {
// pass updates to correct manager
r.ReactorSensors.RLock() // read lock
defer r.ReactorSensors.RUnlock()
for _, sens := range sensors {
// looping over sensors
if sm, ok := r.ReactorSensors.Sensors[int(sens.GetAddr())]; ok {
// sensor manager found
go sm.Update(sens, r.Config) // update sm
} else {
// not found
go r.AddSensor(sens, r.Err)
}
}
}
func (r *ReactorSensors) AddSensor(sensor *pb.Sensor, errCh chan error) {
r.Lock() // write lock
defer r.Unlock()
sm, err := NewSensorManager(sensor)
if err != nil {
errCh <- err
}
r.Sensors[int(sensor.GetAddr())] = sm
}

11
internal/pkg/server/system.go
Unescape Escape View File

@ -2,18 +2,9 @@ package server
import (
_ "fmt"
// sensor components
)
// allows for multiple readers/writers
type DeviceInfo struct {
Id int
Type string
Status string
Data string
Index int
TransactionId uint32
}
/*
type StatusMonitor struct {

Write Preview
Loading…
Cancel
Save