refactored reactor side coordinators and managers. in process of adding monitoring framework
KeeganForelight
2 years ago
parent
53eabda3b7
commit
530413e9a1
Binary file not shown.
Binary file not shown.
@ -1,42 +1,42 @@
|
||||
package I2C
|
||||
|
||||
// file has general wrappers to interact with i2c-tools
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
_ "fmt"
|
||||
"log"
|
||||
"os/exec"
|
||||
"bytes"
|
||||
"strings"
|
||||
"time"
|
||||
"sync"
|
||||
"strconv"
|
||||
)
|
||||
|
||||
type I2Cdev struct {
|
||||
Active bool
|
||||
LastSeen time.Time
|
||||
type I2CBus struct {
|
||||
int
|
||||
sync.Mutex
|
||||
}
|
||||
|
||||
type I2CMonitor struct { // used in RLC to synchronize accesses
|
||||
Bus int // bus to use {0,1,2}
|
||||
Devices map[int]*I2Cdev // mapping to quickly index addresses to their device structs
|
||||
Scan <-chan struct{} // empty struct for efficient sends
|
||||
Connect chan<- int
|
||||
mu sync.Mutex
|
||||
func NewBus(bus int) *I2CBus {
|
||||
b := &I2CBus{}
|
||||
b.int = bus
|
||||
return b
|
||||
}
|
||||
|
||||
func I2Cconnected(bus int) map[int]bool {
|
||||
func (b *I2CBus) Scan() map[int]bool {
|
||||
/*
|
||||
I2CConnect takes an integer specifying the bus to search as input
|
||||
Returns a slice of type I2Cdev
|
||||
Returns all the connected devices
|
||||
*/
|
||||
b := strconv.Itoa(bus)
|
||||
cmd := exec.Command("i2cdetect", "-y", "-r", b)
|
||||
b.Lock()
|
||||
defer b.Unlock()
|
||||
bus := strconv.Itoa(b.int)
|
||||
cmd := exec.Command("i2cdetect", "-y", "-r", bus)
|
||||
var out bytes.Buffer
|
||||
cmd.Stdout = &out
|
||||
err:= cmd.Run()
|
||||
|
||||
if err != nil {
|
||||
fmt.Println(err)
|
||||
if err := cmd.Run(); err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
|
||||
outString := out.String()
|
||||
// could split by \n too
|
||||
split := strings.SplitAfter(outString,":")
|
||||
@ -0,0 +1,33 @@
|
||||
package I2C
|
||||
|
||||
import (
|
||||
_ "fmt"
|
||||
_ "sync"
|
||||
)
|
||||
|
||||
type I2CDevice struct {
|
||||
*I2CBus // embeds bus
|
||||
bool // stores whether dev is currently connected
|
||||
int // addr
|
||||
}
|
||||
|
||||
func NewDevice(addr int,bus *I2CBus) *I2CDevice {
|
||||
d := &I2CDevice{}
|
||||
d.I2CBus = bus
|
||||
d.int = addr
|
||||
return d
|
||||
}
|
||||
|
||||
func (d *I2CDevice) GetAddr() int {
|
||||
return d.int
|
||||
}
|
||||
|
||||
func (d *I2CDevice) GetStatus() string {
|
||||
// TODO
|
||||
return "Unknown"
|
||||
}
|
||||
|
||||
func (d *I2CDevice) GetType() string {
|
||||
// TODO
|
||||
return "Unknown"
|
||||
}
|
||||
@ -0,0 +1,97 @@
|
||||
package I2C
|
||||
|
||||
import (
|
||||
"time"
|
||||
_ "fmt"
|
||||
"sync"
|
||||
)
|
||||
|
||||
/*
|
||||
i2c monitor implements a long running monitor responsible for sending active devices to the rlc
|
||||
*/
|
||||
|
||||
type I2CMonitor struct {
|
||||
*I2CBus
|
||||
Devices *devs
|
||||
DevChan chan int
|
||||
}
|
||||
|
||||
type devs struct {
|
||||
sync.Mutex
|
||||
m map[int]*I2CDevice
|
||||
}
|
||||
|
||||
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.DevChan = ch
|
||||
return m
|
||||
}
|
||||
|
||||
func (m *I2CMonitor) Update() {
|
||||
/*
|
||||
scans bus and adds new active devices
|
||||
*/
|
||||
devs := m.Scan()
|
||||
chng := m.Devices.Parse(m.I2CBus,devs)
|
||||
for _, d := range chng {
|
||||
go m.ConnectDevice(d)
|
||||
}
|
||||
}
|
||||
|
||||
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
|
||||
for {
|
||||
signal <-struct{}{}
|
||||
time.Sleep(to)
|
||||
}
|
||||
}(s,t)
|
||||
|
||||
for {
|
||||
<-s
|
||||
m.Update()
|
||||
}
|
||||
}
|
||||
|
||||
func (m *I2CMonitor) ConnectDevice(addr int) {
|
||||
m.DevChan <-addr
|
||||
}
|
||||
|
||||
func (m *I2CMonitor) GetDevice(addr int) interface{GetAddr() int; GetStatus() string; 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 {
|
||||
d.Lock()
|
||||
defer d.Unlock()
|
||||
newdevs := []int{}
|
||||
for addr, status := range devices {
|
||||
if dev, exists := d.m[addr]; exists {
|
||||
// device seen
|
||||
if status != dev.bool { // if device state changed
|
||||
dev.bool = status
|
||||
if status {
|
||||
newdevs = append(newdevs,dev.GetAddr())
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// device not seen yet
|
||||
if status {
|
||||
// active
|
||||
newd := NewDevice(addr,bus)
|
||||
newd.bool = status
|
||||
d.m[addr] = newd
|
||||
newdevs = append(newdevs,newd.GetAddr())
|
||||
}
|
||||
}
|
||||
}
|
||||
return newdevs
|
||||
}
|
||||
@ -1,95 +0,0 @@
|
||||
package I2C
|
||||
|
||||
import (
|
||||
"time"
|
||||
"fmt"
|
||||
)
|
||||
|
||||
func NewMonitor(c chan<- int, b int) (*I2CMonitor, error) {
|
||||
m := &I2CMonitor{Bus:b,Connect:c}
|
||||
m.Devices = make(map[int]*I2Cdev)
|
||||
s := make(chan struct{})
|
||||
go scan(s)
|
||||
m.Scan = s
|
||||
go m.Monitor()
|
||||
return m, nil
|
||||
}
|
||||
|
||||
func (m *I2CMonitor) Update() []int {
|
||||
/*
|
||||
gets newly connected sensors and updates any currently dead sensors
|
||||
*/
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
connected := I2Cconnected(m.Bus)
|
||||
now := time.Now()
|
||||
newDevs := []int{}
|
||||
for k,v := range connected {
|
||||
if _, ok := m.Devices[k]; ok { // if address existed we need to update sm
|
||||
dev := m.Devices[k]
|
||||
dev.Active = v
|
||||
if v {
|
||||
dev.LastSeen = now
|
||||
}
|
||||
m.Devices[k] = dev
|
||||
} else if v { // new device connected
|
||||
fmt.Printf("Device %v found\n",k)
|
||||
newDevs = append(newDevs, k) // mark device for adding
|
||||
//m.Devices[k] = I2Cdev{Active:true, LastSeen:now} // add new sensors
|
||||
}
|
||||
}
|
||||
return newDevs
|
||||
}
|
||||
|
||||
func (m *I2CMonitor) Remove(addr int) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
delete(m.Devices, addr)
|
||||
}
|
||||
|
||||
func scan(ch chan<- struct{}) {
|
||||
// helper func to time the device updates
|
||||
for true {
|
||||
ch <-struct{}{} // weird syntax but just empty struct init
|
||||
time.Sleep(5 * time.Second)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *I2CMonitor) Monitor() {
|
||||
// functon that updates the device list and notifies rlc of any changes to sensor composition
|
||||
for {
|
||||
<-m.Scan
|
||||
fmt.Println("Scanning I2C bus")
|
||||
newdevs := m.Update()
|
||||
for _,v := range newdevs {
|
||||
go m.ConnectDevice(v)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (m *I2CMonitor) ConnectDevice(addr int) {
|
||||
m.Connect <-addr
|
||||
}
|
||||
|
||||
func (m *I2CMonitor) CreateDevice(addr int) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
dev := &I2Cdev{Active:true}
|
||||
m.Devices[addr] = dev
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *I2CMonitor) GetStatus(addr int) bool {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
_,exists := m.Devices[addr]
|
||||
if exists {
|
||||
return m.Devices[addr].Active
|
||||
} else {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func (d *I2Cdev) GetStatus() bool {
|
||||
return d.Active
|
||||
}
|
||||
@ -0,0 +1,29 @@
|
||||
package reactor
|
||||
|
||||
import (
|
||||
"sync"
|
||||
)
|
||||
|
||||
// implements grpc handler and device update handler
|
||||
type SystemUpdates struct {
|
||||
sync.Mutex
|
||||
pending map[int]*Dev
|
||||
}
|
||||
|
||||
type Dev struct {
|
||||
Status string
|
||||
Type string
|
||||
}
|
||||
|
||||
func (s *SystemUpdates) DeviceUpdateHandler(a int, st, t string) {
|
||||
s.Lock()
|
||||
defer s.Unlock()
|
||||
s.pending[a] = &Dev{Status:st,Type:t}
|
||||
}
|
||||
|
||||
func (s *SystemUpdates) GetPendingChanges() (map[int]*Dev, chan bool) {
|
||||
// calls chld routine to block and return w/ buffer contents
|
||||
// chan is used to send grpc res
|
||||
// true = empty buffer
|
||||
// false = release lock unchanged
|
||||
res := s.LockAndWait()
|
||||
@ -1,86 +1,111 @@
|
||||
package sensor
|
||||
|
||||
import (
|
||||
_"fmt"
|
||||
"time"
|
||||
"sync"
|
||||
_ "fmt"
|
||||
_ "FRMS/internal/pkg/I2C"
|
||||
"log"
|
||||
)
|
||||
|
||||
type Manager struct {
|
||||
Addr int
|
||||
Sensor *SensorInfo
|
||||
I2CMonitor I2Cmonitor
|
||||
Kill chan<- bool
|
||||
mu sync.Mutex
|
||||
*Dev
|
||||
I2CDevice
|
||||
SystemUpdates
|
||||
*Active
|
||||
Hb time.Duration
|
||||
}
|
||||
|
||||
// implementing sensor skeleton
|
||||
|
||||
type SensorStatus uint32
|
||||
|
||||
const (
|
||||
READY SensorStatus = iota
|
||||
ALIVE
|
||||
SENDING
|
||||
KILLED
|
||||
)
|
||||
type SystemUpdates interface {
|
||||
DeviceUpdateHandler(int, string, string)
|
||||
}
|
||||
|
||||
func (s SensorStatus) String() string {
|
||||
return [...]string{"Ready","Active","Sending","Disabled"}[s]
|
||||
type Active struct {
|
||||
sync.Mutex
|
||||
bool
|
||||
int
|
||||
}
|
||||
|
||||
type SensorInfo struct {
|
||||
Status SensorStatus
|
||||
type Dev struct {
|
||||
// last known values
|
||||
Addr int
|
||||
Type string
|
||||
mu sync.Mutex
|
||||
Status string // could be more efficient but to hell with it
|
||||
}
|
||||
|
||||
func (s *SensorInfo) GetStatus() uint32{
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
return uint32(s.Status)
|
||||
type I2CDevice interface {
|
||||
// basic device info
|
||||
GetAddr() int
|
||||
GetStatus() string
|
||||
GetType() string
|
||||
}
|
||||
|
||||
func (s *SensorInfo) GetType() string{
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
return s.Type
|
||||
func NewDeviceManager(i2c I2CDevice,sys SystemUpdates) *Manager {
|
||||
m := &Manager{Hb:time.Duration(1*time.Second)}
|
||||
m.I2CDevice = i2c
|
||||
m.SystemUpdates = sys
|
||||
m.Active = &Active{}
|
||||
m.Dev = &Dev{Addr:i2c.GetAddr(),Type:i2c.GetType(),Status:i2c.GetStatus()}
|
||||
return m
|
||||
}
|
||||
|
||||
// i2c stuff
|
||||
type I2Cmonitor interface {
|
||||
GetStatus(int) bool
|
||||
func (m *Manager) Start() {
|
||||
// goal is to start a long running monitoring routine
|
||||
if !m.Activate() {
|
||||
log.Fatal("Manager already running!")
|
||||
} // atomically activated if this runs
|
||||
go m.Monitor()
|
||||
}
|
||||
|
||||
func NewSensorManager(addr int,i I2Cmonitor) (*Manager,error) {
|
||||
m := &Manager{Addr:addr,I2CMonitor:i}
|
||||
go m.Start()
|
||||
return m, nil
|
||||
func (m *Manager) Exit() {
|
||||
if !m.Deactivate() {
|
||||
log.Fatal("Manager already exited!")
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) Start() {
|
||||
types := map[int]string{97:"DO Sensor",99:"pH Sensor",102:"RTD Sensor"}
|
||||
m.Sensor = &SensorInfo{Type:types[m.Addr]}
|
||||
func (m *Manager) Monitor() {
|
||||
for m.IsActive() {
|
||||
go m.DeviceStatus()
|
||||
time.Sleep(m.Hb)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) GetStatus() uint32 {
|
||||
b := m.I2CMonitor.GetStatus(m.Addr)
|
||||
m.Sensor.Update(b)
|
||||
return m.Sensor.GetStatus()
|
||||
func (m *Manager) DeviceStatus() {
|
||||
status := m.GetStatus()
|
||||
if status != m.Status { // changed
|
||||
go m.DeviceUpdateHandler(m.Addr,status,m.Type)
|
||||
m.Status = status
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) GetType() string {
|
||||
return m.Sensor.GetType()
|
||||
// atomic activation and deactivation
|
||||
func (a *Active) Activate() bool {
|
||||
// returns true if success, false otherwise
|
||||
a.Lock()
|
||||
defer a.Unlock()
|
||||
if a.bool { // already active
|
||||
return false
|
||||
} else {
|
||||
a.bool = true
|
||||
a.int = 0
|
||||
return a.bool
|
||||
}
|
||||
}
|
||||
|
||||
// I2C interface stuff
|
||||
|
||||
func (s *SensorInfo) Update(b bool) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if b {
|
||||
s.Status = ALIVE
|
||||
} else {
|
||||
s.Status = KILLED
|
||||
func (a *Active) Deactivate() bool {
|
||||
// returns true if success false otherise
|
||||
a.Lock()
|
||||
defer a.Unlock()
|
||||
if a.bool {
|
||||
a.bool = false
|
||||
return true
|
||||
} else { // already deactivated
|
||||
return a.bool // false
|
||||
}
|
||||
}
|
||||
|
||||
func (a *Active) IsActive() bool {
|
||||
a.Lock()
|
||||
defer a.Unlock()
|
||||
return a.bool
|
||||
}
|
||||
|
||||
@ -1,8 +0,0 @@
|
||||
package sensor
|
||||
|
||||
import (
|
||||
_ "fmt"
|
||||
)
|
||||
|
||||
|
||||
|
||||
@ -0,0 +1,103 @@
|
||||
this will be a living doc
|
||||
|
||||
starting with for connection management:
|
||||
|
||||
listener:
|
||||
- knows
|
||||
- ip:port to listen to
|
||||
- clients connect with{ip, port, clientType, model, id}
|
||||
- is able to
|
||||
- create coordinators for each clientType
|
||||
- send new clients to coordiantor handlers via chan
|
||||
- depends on
|
||||
- clients sending known data (gRPC)
|
||||
- NewCoordinator func
|
||||
- implements
|
||||
* shouldnt really have any callable methods
|
||||
|
||||
coordinator: (General)
|
||||
- knows
|
||||
- what client ids have already connected
|
||||
- which managers correlate to which clients
|
||||
- is able to
|
||||
- create managers for new clients
|
||||
- start managers for clients
|
||||
- depends on
|
||||
- listener for client structure
|
||||
- manager for NewManager() function
|
||||
- implements
|
||||
- client connection handling
|
||||
- general manager call
|
||||
|
||||
manager (general):
|
||||
- knows
|
||||
- client info
|
||||
- timeout
|
||||
- if it is active
|
||||
- is able to
|
||||
- establish a connection with a client
|
||||
- stop when the connection drops
|
||||
- depends on
|
||||
- coordinator for start calls
|
||||
- implements
|
||||
- client connection creation
|
||||
- client info storage
|
||||
|
||||
manager (reactor):
|
||||
* embedds the gm
|
||||
- knows
|
||||
- devices attached to the reactor
|
||||
- underlying client manager
|
||||
- is able to
|
||||
- maintain device struct
|
||||
- no pings only control logic (i.e remove device, restart etc)
|
||||
- depends on
|
||||
- gm for client conn
|
||||
- coordiantor for starts
|
||||
- implements
|
||||
- reactor structure tracking
|
||||
|
||||
manager (tui):
|
||||
* embedds the gm
|
||||
- knows
|
||||
- structure of the system (ie reactors:devices)
|
||||
- underlying client manager
|
||||
- is able to
|
||||
- keep track of system changes
|
||||
- updates client from buffer or concurrent grpc?
|
||||
- depends on
|
||||
- RM to get system info
|
||||
- coordinator for starts
|
||||
- implements
|
||||
- system tracking
|
||||
|
||||
reactor level coordinator: (right away this feels so fat compared)
|
||||
- knows
|
||||
- current connected devices
|
||||
- server to init to
|
||||
- its own hwinfo to establish itself as a client
|
||||
- is able to:
|
||||
- reach out to server on boot
|
||||
- transmit client details
|
||||
- keep reactor devices current
|
||||
- depends on
|
||||
- I2C package to notify of connected devices
|
||||
- hardware info to get its client info
|
||||
- server to handle connection
|
||||
- sm for new manager
|
||||
- implements
|
||||
- reactor status handler for updates to other coords/managers
|
||||
|
||||
device itself:
|
||||
- knows
|
||||
- probe status ( maybe)
|
||||
- data in buffer
|
||||
- is able to
|
||||
- clear buffer on request
|
||||
- respond to commands
|
||||
- implements
|
||||
- data collection
|
||||
- control execution
|
||||
- depends on
|
||||
- nothing its a driver
|
||||
- maybe the control logic??
|
||||
Loading…
Reference in New Issue