package models import ( "fmt" "time" "github.com/fiskerinc/cloud-services/services/jetfire/utils" "github.com/fiskerinc/cloud-services/pkg/grpc/kafka_grpc" "github.com/fiskerinc/cloud-services/pkg/logger" "github.com/fiskerinc/cloud-services/pkg/utils/envtool" ) var ( maxVinCount = envtool.GetEnvInt("JETFIRE_MAX_VINS", 10000) timestampThreshold = 500 * time.Millisecond ) // Vehicle State stores latest state for a particular VIN. type VehicleState struct { VIN string Timestamp time.Time //data timestamp of last data sample TripStart time.Time //data timestamp of start of trip TripID string //trip id StateValues map[string]float64 //map of values in state. Keys correspond to CAN signal name StateTimes map[string]time.Time //map of timestamps in state. Keys correspond to CAN signal name InsertTime time.Time // local timestamp of last received data. Used only for removing old vehicle states from cache pollingMap map[uint]time.Time //map of last polling times. Key matches update flag // for linked list Next *VehicleState prev *VehicleState } func (v *VehicleState) TimeSincePolled(updateIndex uint) time.Duration { pollTime, ok := v.pollingMap[updateIndex] if !ok { pollTime = time.Unix(0, 0) } return v.Timestamp.Sub(pollTime) } func (v *VehicleState) SetPollTime(updateIndex uint) { v.pollingMap[updateIndex] = v.Timestamp } func (v *VehicleState) Clear(newVIN string) { v.VIN = newVIN cacheSize := len(v.StateValues) v.StateValues = make(map[string]float64, cacheSize) v.StateTimes = make(map[string]time.Time, cacheSize) v.Timestamp = time.Unix(0, 0) v.TripStart = v.Timestamp v.TripID = "" v.InsertTime = time.Now().UTC() v.pollingMap = make(map[uint]time.Time) } // VehicleCache is a table of vehicle states. type VehicleCache struct { Cache map[string]*VehicleState SignalsSet *map[string]bool LastTimestamp time.Time // linked list LIFO of States in order of update. StatesListHead *VehicleState //oldest update StatesListTail *VehicleState //newest update } func (cache *VehicleCache) Clear() { clear(cache.Cache) // disconnect linked list node := cache.StatesListHead for node != nil { next := node.Next node.Next = nil node.prev = nil node = next } cache.StatesListHead = nil cache.StatesListTail = nil } // removes the node from linked list and appends it to the right end func (cache *VehicleCache) ReinsertRight(node *VehicleState) { if cache.StatesListTail == node { // node is already the tail. don't do anything return } // move head ptr if we are moving the first node if cache.StatesListHead == node { cache.StatesListHead = node.Next } // remove from list if node.prev != nil && node.Next != nil { p := node.prev n := node.Next n.prev = p p.Next = n } else if node.prev != nil { node.prev.Next = nil } else if node.Next != nil { node.Next.prev = nil } node.prev = nil node.Next = nil // append to tail if cache.StatesListTail != nil { cache.StatesListTail.Next = node } node.prev = cache.StatesListTail cache.StatesListTail = node if cache.StatesListHead == nil { cache.StatesListHead = node } } // removes left node from the linked list AND from the cache map func (cache *VehicleCache) PopLeft() *VehicleState { if cache.StatesListHead == nil { return nil } node := cache.StatesListHead cache.StatesListHead = node.Next delete(cache.Cache, node.VIN) return node } // Insert CANSignal into vehicle cache. Allocates new vehicle state as necessary. func (cache *VehicleCache) UpdateSignal(signal *kafka_grpc.GRPC_CANSignal, updateFlag uint) error { if len(*cache.SignalsSet) > 0 { // check if signal is in signals set, skip if not in signals set _, contains := (*cache.SignalsSet)[signal.Name] if !contains { return nil } } // if VIN is not in cache, allocate new vehicle state for VIN and add to cache _, contains := cache.Cache[signal.Vin] if !contains { if len(cache.Cache) > maxVinCount { oldState := cache.PopLeft() logger.Debug().Msgf("repurposing state %s -> %s", oldState.VIN, signal.Vin) oldState.Clear(signal.Vin) cache.Cache[signal.Vin] = oldState } else { cache.Cache[signal.Vin] = NewVehicleState(signal.Vin, len(*cache.SignalsSet)) logger.Debug().Msgf("new vehicle state %s", signal.Vin) } } //update value cache.Cache[signal.Vin].UpdateSignal(signal, updateFlag) cache.ReinsertRight(cache.Cache[signal.Vin]) //move state to end of orderly linkedlist cache.LastTimestamp = time.Now().UTC() return nil } // constructs a new VehicleState func NewVehicleState(VIN string, cacheSize int) *VehicleState { newState := new(VehicleState) newState.VIN = VIN newState.StateValues = make(map[string]float64, cacheSize) newState.StateTimes = make(map[string]time.Time, cacheSize) newState.Timestamp = time.Unix(0, 0) newState.TripStart = newState.Timestamp newState.TripID = "" newState.InsertTime = time.Now().UTC() newState.pollingMap = make(map[uint]time.Time) return newState } // constructs a new VehicleState func NewVehicleStateDefault(VIN string) *VehicleState { return NewVehicleState(VIN, 10) } // UpdateSignal() updates the vehicle state cache func (state *VehicleState) UpdateSignal(signal *kafka_grpc.GRPC_CANSignal, updateFlag uint) { // Mark start of new trip if too much time has elapsed between updates signalTime := utils.FloatToTime(signal.Timestamp) ignitionTriggered := false // check for vehicle ignition rising edge as a trigger for a new trip. if signal.Name == "BCM_PwrMod" && signal.Value >= 2 && signal.Value <= 4 && !signalTime.Before(state.Timestamp) { pwrMod, ok := state.StateValues["BCM_PwrMod"] ignitionTriggered = !ok || (ok && pwrMod < 2) } if signalTime.Sub(state.Timestamp) >= utils.TripTimeout || ignitionTriggered { logger.Debug().Msgf("%s New TripStart: %d, old %d, delta %d", state.VIN, signalTime.Unix(), state.Timestamp.Unix(), signalTime.Sub(state.Timestamp)) state.TripStart = signalTime state.TripID = fmt.Sprintf("%s_%d", state.VIN, state.TripStart.Unix()) } // Update the vehicle timestamp oldTime, ok := state.StateTimes[signal.Name] if !ok { oldTime = state.Timestamp } if !signalTime.Add(timestampThreshold).Before(oldTime) { if !signalTime.Before(state.Timestamp) { state.Timestamp = signalTime } // Insert new signal value state.StateValues[signal.Name] = signal.Value state.StateTimes[signal.Name] = signalTime state.InsertTime = time.Now() } else { // if receiving message out of timestamp order, only accept new signal value if // cached value is empty for signal name _, hasSignal := state.StateValues[signal.Name] if !hasSignal { state.StateValues[signal.Name] = signal.Value state.StateTimes[signal.Name] = signalTime state.InsertTime = time.Now() } utils.LogOutOfOrderMsg(signal.Name, signal.Vin) } }