#!/usr/bin/env lua-agent
--[[
 - @file nmeamux
 - ABSTRACT HERE << 
 -
 - $Id$
 -
 - (C) Copyright 2011 Amplex, fm@amplex.dk
--]]

require "stdlib" .include { "seq", "list" }
require "posix"
require "socket"
require "Class"
require "trace"
local NIL = Class.NIL
local list = _G.list
local sampleConf

local iometa = getmetatable(io.stdin)
if iometa.fileno and not iometa.getfd then iometa.getfd = iometa.fileno end

local function getMsecs()
   local now, mSec = appc.systemTime()
   return now * 1000 + mSec
end

local L_OPEN_RETRY = 5 * 1000

local DutyCycle = Class(nil, "DutyCycle", {
   mInterval = 1,
   mPort = NIL,
   mPrefix = NIL,
   mTail = NIL,
   mLastLog = NIL,
})

function DutyCycle.new(class, port, pfx)
   local now = getMsecs()
   return class:_new():_merge { mPort = port, mPrefix = pfx , mTail = tnew{}, mLastLog = now }
end

---
-- Keep track of bandwidth etc.
function DutyCycle:add(str)
   local now = getMsecs()
   local start = now - math.floor(self.mInterval * 1000)
   local tail = self.mTail
   while #tail > 0 and tail[#tail].t < start  do tail:remove() end
   self.mTail:insert(1, {t = now, s = str:sub(1,6), l = #str})

   if self.mLastLog < start then
      local sum, sums = 0, {}
      for _,ent in ipairs(tail) do 
         sum = sum + ent.l
         sums[ent.s] = 1 + (sums[ent.s] or 0)
      end
      local names = List(seq.all_keys(sums)):sort()
      log.notice(("PORT:%-12s %s:%3d%% %3d %s"):_(self.mPort.mName, self.mPrefix,
         idiv(sum * 1000, self.mPort.mSpeed), sum,
         List(seq.map(|n|("%s:%d"):_(n, sums[n]), list.iter(names))):concat(",")))
      self.mLastLog = now
   end
end

local nodes = {}
local sysmon
local sysstat= {} -- pos, sog, cog, stw, hdg, dpt, wrv, wrb, 

--------------------------------------------------------------------
-- NmeaNode
-- Handle node graph and telnet connections
--------------------------------------------------------------------

local NmeaNode = Class(nil, "NmeaNode", {
   mName = NIL,
   mInputs = NIL,
   mSinks = NIL,
   mPortNo = NIL,
   mPortSock = NIL,
   mListeners = NIL,
   mSubscriptions = NIL,
})

---
-- Create a generic node
function NmeaNode.new(_class, name, cfg)
   return _class:_new():_merge {
      mName = name,
      mInputs = cfg.inputs,
      mSinks = {}, 
      mPortNo = cfg.port,
      mListeners = {},
      mSubscriptions = {},
   }
end

---
-- Data forwarding
function NmeaNode:processInput(str)
   -- Simple relay
   log.debug(||("NODE:%-12s READ: %t"):_(self.mName, str))
   for sink, ix in pairs(self.mSinks) do
      sink:processInput(str, ix)
   end
   for l,_ in pairs(self.mListeners) do
      l:send(str.."\r\n")
   end
end

---
-- Register a data sink
function NmeaNode:addSink(new, ix)
   -- Simple relay
   self.mSinks[new] = ix
end

---
-- Read input from a connected peer.
-- Override and embed in derived classes to read input
function NmeaNode:readClient(subscr, inp)
   local res, err = posix.read(inp:getfd())
   log.debugonly(||("PORT:%-12s NET:%d:%s: %s"):
                  _(self.mName, inp:getfd(), err or "OK", (res or ""):quotebin())) 
   if not res or res == "" then
      log.notice(("PORT:%-12s Connection %d closed"):_(self.mName, inp:getfd()))
      self.mListeners[inp] = nil
      self.mSubscriptions[subscr] = nil
      subscr:clear()
      inp:close()
      return nil
   end
   return res
end

---
-- Startup system interaction
function NmeaNode:start(str)
   for ix,s in ipairs(self.mInputs or {}) do
      assert(nodes[s], ("No input: %s"):_(s)):addSink(self, ix)
   end
   if not self.mPortNo then return end

   if self.mPortNo then
      local sock, err = socket.bind("0.0.0.0", self.mPortNo)
      self.mPortSock = assert(sock, ("Can not open listen on %d: %t"):_(self.mPortNo, err))
      local subscr = appc.addFileIO(self.mPortSock:getfd(), function(subscr)
         local inp = self.mPortSock:accept()
         log.notice(("PORT:%-12s Connection %d opened from %t"):
                   _(self.mName, inp:getfd(), inp:getpeername()))
         local subscr = appc.addFileIO(inp:getfd(), function(subscr)
            self:readClient(subscr, inp)
         end)
         self.mSubscriptions[subscr] = true
         self.mListeners[inp] = true
      end)
      self.mSubscriptions[subscr] = true
   end
end

---
-- Cleanup subscriptions & listeners
function NmeaNode:clear()
   for l,_ in pairs(self.mSubscriptions) do l:clear() end
   for l,_ in pairs(self.mListeners) do l:close() end
   self.mSubscriptions = NIL
   self.mListeners = NIL
   if self.mPortSock then self.mPortSock:close() end
   self.mPortSock = nil
end
function NmeaNode:__gc() return self:clear() end

--------------------------------------------------------------------
-- NmeaOutput 
--------------------------------------------------------------------

local NmeaOutput = Class(NmeaNode, "NmeaOutput", {
   mOutput = NIL,
})

---
-- Create an output node
function NmeaOutput.new(_class, name, cfg)
   return NmeaOutput:getSuper().new(_class, name, cfg):_merge {
      mOutput = assert(cfg.output)
   }
end

---
-- Startup system interaction
function NmeaOutput:start(str)
   NmeaOutput:getSuper().start(self, str)
   self:addSink(nodes[self.mOutput], 1)
end

--------------------------------------------------------------------
-- NmeaPort 
--------------------------------------------------------------------

local NmeaPort = Class(NmeaNode, "NmeaPort", {
   mPath = NIL,
   mSpeed = NIL,
   mDutyIn = NIL,
   mDutyOut = NIL,
   mSerialPort = NIL,
   mLineBuffer = NIL,
})
local RawPort

function NmeaPort.new(_class, name, cfg)
   local new = RawPort.new(_class, name, cfg)
   return new:_merge {
      mDutyOut = DutyCycle(new, "O"),
      mDutyIn = DutyCycle(new, "I"),
      mLineBuffer = tnew {},
   }
end

---
-- Startup system interaction
function NmeaPort:start(str)
   NmeaPort:getSuper().start(self, str)
   return self:reOpen()
end

local function nmeaReadPort(self, port, subscr) 
   local str, err = posix.read(port:getfd())
   if not str or str == "" then
      log.error(("PORT:%-12s REOPEN: Failed to read %t: %t"):_(self.mName, self.mPath, err))
      port:close()
      self.mSerialPort = nil
      subscr:clear()
      self.mSubscriptions[subscr] = nil
      appc.addTimer(L_OPEN_RETRY, function() self:reOpen() end)
      return
   end
   -- TODO: Reopen on error
   log.debugonly(||("PPPPORT:%-12s INP:%d:%s: %s"):
                  _(self.mName, port:getfd(), err or "OK", (str or ""):quotebin())) 
   if self.mLineBuffer then
      local where,whend = str:find("[\r\n]+")
      if where then
         while where do
            self.mLineBuffer:append(str:sub(1, where - 1))
            local line = self.mLineBuffer:concat()
            if #line > 0 then
               --NmeaPort:getSuper().processInput(self, ("%d%s"):_(#self.mLineBuffer,line))
               self.mDutyIn:add(str)
               NmeaPort:getSuper().processInput(self, line)
            end
            str = str:sub(whend + 1, -1)
            where,whend = str:find("[\r\n]+")
            self.mLineBuffer = tnew {}
         end
         self.mLineBuffer = tnew { str }
      else
         self.mLineBuffer:append(str)
      end
   else
      log.debugonly(||("PORT:%-12s INP:%d:%s: %s"):
                     _(self.mName, port:getfd(), err or "OK", (str or ""):quotebin())) 
      for l,_ in pairs(self.mListeners) do
         l:send(str)
      end
   end
end

---
-- Startup system interaction
function NmeaPort:reOpen()
   local port, err = io.open(self.mPath, "r+")
   log.debugonly(||("REOPEN:%s %s -> %t, %t"):_(self.mName, self.mPath, port, err))
   if not port then
      log.error(("PORT:%-12s Failed to open %s: %s"):_(self.mName, self.mPath, err))
      appc.addTimer(L_OPEN_RETRY, function() self:reOpen() end)
      return
   end
   self.mSerialPort = port
   if self.mSpeed and self.mSpeed > 0 then
      os.execute(("stty %d raw -echo -parenb cs8 min 50 time 1 < %s"):_(self.mSpeed, self.mPath))
   end
   local subscr = appc.addFileIO(self.mSerialPort:getfd(), function(subscr) 
      return nmeaReadPort(self, port, subscr)
   end)
   self.mSubscriptions[subscr] = true
end

---
-- Cleanup subscriptions & listeners
function NmeaPort:clear()
   NmeaPort:getSuper().clear(self)
   if self.mSerialPort then self.mSerialPort:close() end
   self.mSerialPort = nil
end


---
-- Data forwarding
function NmeaPort:processInput(str)
   if self.mDutyOut then self.mDutyOut:add(str) end
   log.debug(||("OUTP:%-12s W: %t"):_(self.mName, str))
   posix.write(self.mSerialPort:getfd(), str.."\r\n")
   -- if self.mLineBuffer then sysmon:addOut(self, str:sub(1,6)) end
end

---
-- Read input from a connected peer.
function NmeaPort:readClient(subscr, inp)
   local str = NmeaPort:getSuper().readClient(self, subscr, inp)
   local fd = self.mSerialPort:getfd()
   if str and not self.mLineBuffer then
      log.debugonly(||("PORT:%-12s OUT:%d:%s: %s"):
                     _(self.mName, fd, err or "OK", (str or ""):quotebin())) 
      posix.write(fd, str)
   end
end

-- Must delay 'til after NmeaPort methods dur to Class.lua idiosyncasies
-- Create a NmeaNode sans the mLineBufer etc
RawPort = Class(NmeaPort, "RawPort", {})
---
-- Create an output node
function RawPort.new(_class, name, cfg)
   return NmeaNode.new(_class, name, cfg):_merge {
      mPath = assert(cfg.path),
      mSpeed = assert(cfg.speed),
   }
end

function RawPort:processInput(str)
   log.debug(||("OUTRAW:%-12s W: %t"):_(self.mName, str))
   posix.write(self.mSerialPort:getfd(), str)
end

--------------------------------------------------------------------
-- NmeaFilter 
--------------------------------------------------------------------

local NmeaFilter = Class(NmeaNode, "NmeaFilter", {
   mPosList = NIL,
   mNegList = NIL,
   mAuxList = NIL,
   mPrioTout = NIL,
   mPrioList = NIL,
})

---
-- Create an output node
function NmeaFilter.new(_class, name, cfg)
   return NmeaFilter:getSuper().new(_class, name, cfg):_merge {
      mPosList = cfg.pos,
      mNegList = cfg.neg,
      mAuxList = cfg.aux,
      mPrioTout = cfg.prio_delay,
      mPrioList = {},
   }
end

---
-- Startup system interaction
function NmeaFilter:start(str)
   NmeaFilter:getSuper().start(self, str)
   self.mAuxList = List(seq.map(function(cls) return cls() end, self.mAuxList or {}))
end

---
-- Data forwarding
function NmeaFilter:processInput(str, ix)
   if self.mNegList then
      for _,p in ipairs(self.mNegList) do
         if str:find(p) then
            log.debug(||("NREJ:%-12s NEG:%s: %t"):_(self.mName, p, str))
            return
         end
      end
   end
   if self.mPosList then
      local pass
      for _,p in ipairs(self.mPosList) do
         if str:find(p) then
            pass = true
            break
         end
      end
      if not pass then
         log.debug(||("PREJ:%-12s READ: %t"):_(self.mName, str))
         return
      end
   end
   if self.mAuxList then
      for _,aux in ipairs(self.mAuxList) do
         aux:process(self, str, function(str) self:passItOn(str, ix) end)
      end
   else
      self:passItOn(str, ix)
   end
end

---
-- Implement priority shadowing
function NmeaFilter:passItOn(str, ix)
   if self.mPrioTout then
      local now = getMsecs()
      local stmt = str:sub(1,6)
      local old = self.mPrioList[stmt]
      if old and (now - old.timeStamp) < self.mPrioTout and old.prio < ix then
         log.debug(||("TREJ:%-12s READ: %t"):_(self.mName, str))
         return -- He wins
      end
      self.mPrioList[stmt] = { timeStamp = now, prio = ix }
   end
   return NmeaFilter:getSuper().processInput(self, str, ix)
end

--------------------------------------------------------------------
-- NilFilter
--------------------------------------------------------------------

local NilFilter = Class(nil, "NilFilter", {})

function NilFilter:process(port, str, out)
   out(str)
end

--------------------------------------------------------------------
-- AisFilter
--------------------------------------------------------------------

local AisFilter = Class(NilFilter, "AisFilter", {mDb = NIL})

--------------------------------------------------------------------
-- GpsComp
--------------------------------------------------------------------
--
local GpsComp = Class(NilFilter, "GpsComp", {mDb = NIL})

--------------------------------------------------------------------
-- MONITOR
--------------------------------------------------------------------

local SysMon= Class(nil, "SysMon", {
   mPort = NIL,
   mQueue = NIL,
   mTimer = NIL,
   mIndex = NIL,
})

function  SysMon:runTimer()
end

local SysMonTon = Class(nil, "SysMonTon", {
   mSysMons = NIL,
   mNames = NIL,
   mListeners = NIL,
   mSocket = NIL,
})

function SysMonTon:openPort(port)
end

function SysMonTon:addIn(port, snt, out)
   local q = self.mSysMons[port].mQueue[out and 2 or 1]
   q:insert(snt)
   sysmon:LineStat(self.mSysMons[port], out)
   self.mSysMons[port]:runTimer()
end

function SysMonTon:addOut(port, snt)
   return self:addIn(port, snt, true)
end

function SysMonTon:addPort(port)
   if self.mSysMons[port] then return end

   self.mNames:insert(port.mName)
   self.mNames:sort()
   for ix,nm in ipairs(self.mNames) do
      if nm == port.mName then 
         self.mSysMons[port] = SysMon { mIndex = ix, mQueue = {{}, {}}, mPort = port }
      end
   end
end

local function startMonitor(port)
   sysmon = SysMonTon { mSysMons = {}, mListeners = {}, mNames = tnew {} }
   sysmon:openPort(port)
end

--------------------------------------------------------------------
-- SYSTEM
--------------------------------------------------------------------

local testConf
--[[
= {
   outputs = {
      testout      = { inputs = { "testfilter" },  output = "rawport",    port=27002 }, 
   },
   nmeaports = {
      testgps      = { path = "/dev/ttyUSB0",                     speed=4800  , port=26001},
      testport     = { path = "/dev/ttyUSB1",                     speed=4800  , port=26002},
   },
   rawports = {
      rawport      = { path = "/dev/ttyUSB2",                     speed=38400 , port=25006},
   },
   filter   = {
      testfilter       = { inputs = { "testport", "testgps" }                 , port=28001,
                           neg = { "^...GGA", }, prio_delay=2500, aux = {AisFilter} },
   },
}
--]]

---
-- Startup system after all nodes are brought into existence
local function startSystem()
   for name,n in pairs(nodes) do
      n:start(name)
   end
end

local function getConf()
   -- TODO: Find a configuration
end

local function configure(config)
   for cfgNode, classFactory in pairs {
      nmeaports = NmeaPort,
      rawports = RawPort,
      outputs = NmeaOutput,
      filter = NmeaFilter,
   }
   do
      for name, cfg in pairs(config[cfgNode] or {}) do
         assert(nodes[name] == nil)
         nodes[name] = classFactory(name, cfg)
      end
   end
end

local function runGC()
   collectgarbage("collect")
   log.notice(("! MEM: %.2f"):_(collectgarbage("count")))
   appc.addTimer(10 * 1000, runGC)
end

local function main(args)
   --trace.on()
   local conf = getConf() or testConf or sampleConf
   if not log.debug() and args[1] and args[1] == "trace" then
      local traceLog = log.create("nmeamux-trace.log", 0, false, 8000)
      log.debugonly = function() end
      log.debug = function(str)
         if not str then return false end
         if type(str) == "function" then str = str() end
         return traceLog:log(str)
      end
   else
      log.debugonly = log.debug
   end
   configure(conf)
   startMonitor(117)
   startSystem()
   runGC()
end

local NM_GPS = { "^...GGA", "^...GSA", "^...RMC", "^...GSV", }
 sampleConf = {
   outputs = {
      instruments = { inputs = {"p_lospeed_1", "nogps", "autopilot"}          , port=27001,
                      output = "p_lospeed_1" },
      --sfeed     = { inputs = { "gps_bkhousemux" },  output = "p_lospeed_2",   port=27002 }, 
      gpsfeed     = { inputs = { "gpsmin" },          output = "p_lospeed_2",   port=27002 }, 
      garmin_nmea = { inputs = { "p_lospeed_1", "p_lospeed_2", "aisfilter", "nogps", "gps" },
                      output = "p_hispeed",                                     port=27003 },
   },
   nmeaports = {
      p_lospeed_1  = { path = "/dev/ttyusb/mos7840-9710-7840/0",  speed=4800  , port=26001},
      p_lospeed_2  = { path = "/dev/ttyusb/mos7840-9710-7840/1",  speed=4800  , port=26002},
      p_hispeed    = { path = "/dev/ttyusb/mos7840-9710-7840/2",  speed=38400 , port=26003},
      p_gps        = { path = "/dev/rfcomm0",                     speed=38400 , port=26004},
      p_ais        = { path = "/dev/ttyusb/ftdi_sio-0403-6001/0", speed=38400 , port=25001},
  --  p_ais        = { path = "/dev/ttyS1",                       speed=38400 , port=25001},
  --  p_gps        = { path = "/dev/ttyS1",                       speed=38400 , port=26004},
      p_bkhousemux = { path = "/dev/ttyusb/pl2303-067b-2303/bus_usb_001_009:00",
                                                                  speed=38400 , port=25005},
   },
   rawports = {
      p_pactor     = { path = "/dev/ttyusb/ftdi_sio-0403-d010/0", speed=9600  , port=25002},
      p_garmin1    = { path = "/dev/ttyusb/mos7840-9710-7840/3",  speed=9600  , port=25003},
      p_garmin2    = { path = "/dev/ttyS1",                       speed=9600  , port=25004},
      p_navtex     = { path = "/dev/ttyusb/pl2303-067b-2303/bus_usb_001_010:00",
                                                                  speed=38400 , port=25006},
   },
   filter   = {
      nogps     =  { inputs = { "p_bkhousemux" }, neg = NM_GPS          , port=28001, },
      gps       =  { inputs = { "p_gps", "p_bkhousemux", "aisfilter"}   , port=28002,
                     pos = NM_GPS, aux={GpsComp}, prio_delay=2500 },
      gpsmin    =  { inputs = { "gps", pos = { "^...RMC" } } },
      autopilot =  { inputs = { "p_hispeed" }                           , port=28003, 
                     pos = { "^...XTE", "^...WPT" } },
      aisfilter =  { inputs = { "p_ais" }                               , port=28004, 
                           aux = {AisFilter} }
   },
}

return ac.ADDR_TRANSIENT, main
-- vim: set sw=3 sts=3 et nu:
