From 01d2ebd3a1a4345cddba51dbbd66decb714d93b9 Mon Sep 17 00:00:00 2001 From: David Kebler Date: Thu, 15 Aug 2019 14:05:30 -0700 Subject: [PATCH] 0.2.22 interrupts class extends base on interrupts class change collection of individual interrupts to a Map method: listenReset, for setting handler for emitted reset of each single interrupt method: addInterSocket, to add identical consumer socket to each interrupt interrupt class add resetEnabled option switch default edge to 'both' refactor status method refactor reset method No hook by default, registerHook without argument sets the default hook refactored examples update deps --- examples/client.js | 24 +++++++---- examples/multi.js | 37 ++++++++++++---- examples/single.js | 29 +++++++++---- examples/socket.js | 37 ++++++++++++++++ package.json | 28 +++++++------ readme.md | 14 +++---- src/interrupt.js | 102 +++++++++++++++++++++++---------------------- src/interrupts.js | 84 +++++++++++++++++++++++++------------ 8 files changed, 234 insertions(+), 121 deletions(-) create mode 100644 examples/socket.js diff --git a/examples/client.js b/examples/client.js index 2df53c1..a07c498 100644 --- a/examples/client.js +++ b/examples/client.js @@ -1,9 +1,9 @@ import Base from '@uci/base' // const HOST = 'localhost' -const HOST = 'sbc' -const PORT = 9024 -let processor = new Base({sockets:'inter#c>t', inter:{host:HOST, port:PORT}, id:'interrupt-processor', useRootNS:true}) +const HOST = process.env.HOST || 'localhost' +const PORT = process.env.PORT // default 8080 +let processor = new Base({ id:'remote-interrupt-processor', useRootNS:true}) processor.interrupt = async function (packet) { return new Promise((resolve) => { @@ -15,21 +15,31 @@ processor.interrupt = async function (packet) { processor.reply = async function (packet) { return new Promise((resolve) => { - - console.log('reply from interrupt with packet') + console.log('reply from interrupt for request', packet._header.request) console.dir(packet) resolve({status: 'processed'}) }) } +processor.on('status', ev => { + console.log(`STATUS: **${ev.level}** ${ev.msg}`)} +) + +processor.on('consumer-connection', ev => { + console.log(`client ${ev.name} was ${ev.state}`) +}) + +processor.on('reconnected', client => { + console.log('client reconnected:', client) +}) ; (async () => { + // processor.addSocket('inter','c','t', {host:HOST, port:PORT}) + processor.addSocket('inter','c','t',{host:HOST, port:PORT}) await processor.init() console.log('====sending fire command to interrupt===') await processor.send({cmd: 'fire'}) - console.log('====sending fire command to interrupt===') - await processor.send({cmd: 'fire'}) // process.kill(process.pid, 'SIGTERM') })().catch(err => { diff --git a/examples/multi.js b/examples/multi.js index bdc0039..7435159 100644 --- a/examples/multi.js +++ b/examples/multi.js @@ -1,26 +1,45 @@ import Interrupts from '../src/interrupts' -import Base from '@uci/base' -const PINS = [9,10,24] +const PINS = [4] -let hook = (packet) => +let interrupts = new Interrupts(PINS,{id:'multi-interrupt-example', resetInterval:1, resetEnabled:false, 4:{name:'mybutton'} }) + +let hook = function (packet) { - packet.cmd = 'pin.interrupt.find' - console.dir(packet) + packet.cmd = 'interrupt.find' return packet } -let interrupts = new Interrupts(PINS,{hook:true, 10:{wait:200} }) +// interrupts.registerHook(hook) -interrupts.setHook(hook) +interrupts.on('status', ev => { + console.log(`STATUS:'--${ev.level}--" ${ev.msg}`)} +) +interrupts.on('consumer-connection', ev => { + console.log(`client ${ev.name} was ${ev.state}`) +}) + +interrupts.listen(function (packet) { + console.log(`============== ${this.id}=========`) + console.log(`emitted packet from interrupt ${packet.id}, pin:${packet.pin}`) + console.dir(packet) + this.push(packet) + console.log('------------------------') +}) +interrupts.listenReset(function (packet) { + console.log(`============== ${this.id}=========`) + console.log('an interrupt reset request emitted') + console.dir(packet) + console.log('------------------------') +}) ; (async () => { - // console.log(await listener.init()) + interrupts.addSocket('server','s','t') await interrupts.init() - interrupts.fire() + // interrupts.fire() })().catch(err => { console.error('FATAL: UNABLE TO START SYSTEM!\n',err) diff --git a/examples/single.js b/examples/single.js index 3d531af..8159ed6 100644 --- a/examples/single.js +++ b/examples/single.js @@ -2,24 +2,37 @@ import Interrupt from '../src/interrupt' const delay = time => new Promise(res=>setTimeout(()=>res(),time)) -let interrupt = new Interrupt(24,{id:'test-interrupt', wait:0, hook:true, useRootNS:true}) +const IPIN = process.env.IPIN || 4 + +let interrupt = new Interrupt(IPIN,{id:'test-interrupt', wait:0, edge:'rising', resetInterval:1, reset:true}) + +interrupt.on('interrupt', packet => { + console.log('event: interrupt fired for',interrupt.pin_num) + console.log('count:', packet.count, 'state:',packet.state, 'cmd:',packet.cmd, 'data:',packet.data) +}) + +interrupt.on('interrupt.reset', packet => { + console.log('interrupt reset packet sent/emitted') + console.dir(packet) +}) ; (async () => { await interrupt.init() console.log('interrupt ready and waiting') - console.log('manual fire of interrupt via interrupt instance') + console.log('manual fire of interrupt with default hook') interrupt.fire() - console.log('manual fire of interrupt via interrupt instance after changing hook') - interrupt.hook = (packet) => { + console.log('manual fire of interrupt via after changing hook') + + interrupt.registerHook((packet) => { packet.data='some hook added data' - console.log('custom hook function modifies', packet) + console.log('custom hook data prop added:', packet.data) return packet - } + }) + interrupt.fire() - await delay(3000) - process.kill(process.pid, 'SIGTERM') + })().catch(err => { console.error('FATAL: UNABLE TO START SYSTEM!\n',err) diff --git a/examples/socket.js b/examples/socket.js new file mode 100644 index 0000000..2c2092c --- /dev/null +++ b/examples/socket.js @@ -0,0 +1,37 @@ +import Base from '@uci/base' + +// const HOST = 'localhost' +const HOST = process.env.HOST || 'sbc' +const PORT = process.env.PORT +let processor = new Base({sockets:'inter#c>t', inter:{host:HOST, port:PORT}, id:'interrupt-processor', useRootNS:true}) + +processor.interrupt = async function (packet) { + return new Promise((resolve) => { + console.log('interrupt occured') + console.dir(packet) + resolve({status: 'processed'}) + }) +} + +processor.reply = async function (packet) { + return new Promise((resolve) => { + + console.log('reply from interrupt with packet') + console.dir(packet) + resolve({status: 'processed'}) + }) +} + +; +(async () => { + + await processor.init() + console.log('====sending fire command to interrupt===') + await processor.send({cmd: 'fire'}) + console.log('====sending fire command to interrupt===') + await processor.send({cmd: 'fire'}) + // process.kill(process.pid, 'SIGTERM') + +})().catch(err => { + console.error('FATAL: UNABLE TO START SYSTEM!\n',err) +}) diff --git a/package.json b/package.json index c2b2f48..8faad15 100644 --- a/package.json +++ b/package.json @@ -1,14 +1,18 @@ { "name": "@uci/interrupt", "main": "src", - "version": "0.2.19", + "version": "0.2.22", "description": "a class for adding interrupt processesing for gpio pins on Raspberry Pi and Similar SBCs", "scripts": { "single": "node -r esm examples/single", + "single:dev": "UCI_ENV=dev ./node_modules/.bin/nodemon -r esm examples/single", + "single:debug": "UCI_LOG_LEVEL=debug npm run single:dev", + "multi": "node -r esm examples/multi", + "multi:dev": "UCI_ENV=dev ./node_modules/.bin/nodemon -r esm examples/multi", + "multi:debug": "UCI_LOG_LEVEL=debug npm run multi:dev", "client": "node -r esm examples/client", - "singlelog": "UCI_ENV=dev node -r esm examples/single", - "multi": "sudo node --require esm examples/multi", - "multilog": "UCI_ENV=dev node --require esm examples/multi" + "client:dev": "UCI_ENV=dev ./node_modules/.bin/nodemon -r esm examples/client", + "client:debug": "UCI_LOG_LEVEL=debug npm run client:dev" }, "author": "David Kebler", "license": "MIT", @@ -28,17 +32,15 @@ }, "homepage": "https://github.com/uCOMmandIt/uci-interrrupt#readme", "dependencies": { - "@uci-utils/logger": "0.0.14", - "@uci/base": "^0.1.21", + "@uci-utils/logger": "^0.0.15", + "@uci/base": "^0.1.30", "death": "^1.1.0", - "onoff": "^4.1.1" + "onoff": "^4.1.4" }, "devDependencies": { - "chai": "^4.1.2", - "codecov": "^3.0.0", - "esm": "^3.0.58", - "istanbul": "^0.4.5", - "mocha": "^5.0.1", - "nodemon": "^1.14.3" + "chai": "^4.2.0", + "esm": "^3.2.25", + "mocha": "^6.2.0", + "nodemon": "^1.19.2" } } diff --git a/readme.md b/readme.md index 1eb6e31..498d7ad 100644 --- a/readme.md +++ b/readme.md @@ -1,16 +1,14 @@ # uCOMmandIt Interrupt Package for SBC GPio Pins -[![Build Status](https://img.shields.io/travis/uCOMmandIt/uci-interrupt.svg?branch=master)](https://travis-ci.org/uCOMmandIt/uci-interrupt) + -This module creates an instance of UCI Packect Interrupt class for a SBC gpio pin supporting interrupt via epoll (kernel gpio/export) by extending the UCI base class. +This module creates an UCI Packect Interrupt classes for a single or multiple gpio pin(s) supporting interrupt via epoll (kernel gpio/export). By extending the UCI base class it allows communication to/from other processes -Each pin will create it's own socket (based on options pased) By default it will create a tcp socket at 9000+pin number. I can create a socket for all transports. - -When a gpio pin so set up is tripped this class pushes a UCI packet to all connected consumers. +By default there are NO sockets created. You can create them as you need. Every interrupt thown will emit a packet as well as send and push that UCI packet if any sockets are associated with that interrupt. You can pass a custom packet to push via the options and/or ammend the basic packet via a hook you provide. @@ -30,11 +28,11 @@ Give `gpio` group permission to reading/writing from/to pins. On raspbien insta SUBSYSTEM=="gpio*", PROGRAM="/bin/sh -c 'find -L /sys/class/gpio/ -maxdepth 2 -exec chown root:gpio {} \; -exec chmod 770 {} \; || true'" ``` -if you get permission errors then likely this rule is not effective. Take a look at /sys/class/gpio folder and see if gpio group has been added appropriately (e.g. to /export and /unexport). +if you get permission errors then likely this rule is not effective. Take a look at /sys/class/gpio folder and see if gpio group has been added appropriately (e.g. to /export and /unexport). ### Set hardware pull -You must tell the gpio bus hardware about which pins you want to use as interrupts and what pull state you want. This only needs to be done once and is somewhat specific to the sbc you are using. +Will need to do this with DTOs device tree overlays (except for Rasberry Pi) #### Raspberry Pi For raspberry pi with recent distro/kernel (e.g. Ubuntu 18.04,mainline 4.15) you can use add a line to config.txt in /boot or subdir therein. Add a line like this diff --git a/src/interrupt.js b/src/interrupt.js index 311ffce..60c370a 100644 --- a/src/interrupt.js +++ b/src/interrupt.js @@ -10,12 +10,13 @@ let log = logger({package:'@uci/interrupt', file:'/src/interrupt.js'}) // conPacket is for connecting consumers. This will send this conPacket command on connect, which may needed to initilize something on related hardware class Interrupt extends Base { constructor(pin, opts = {}) { - if (typeof pin !=='number') pin = parseInt(pin) // make sure pin is a number! + pin = Number(pin) // make sure pin is a number! super(opts) - this.id = (opts.id || 'interrupt') + ':' + pin + this.id = opts.name || (opts.id || 'interrupt') + ':' + pin log.debug({ pins: pin, opts: opts, method:'constructor', line:16, msg:'created interrupt with these opts'}) this.pin_num = pin this.resetCmd = opts.resetCmd || 'interrupt.reset' + this.resetEnabled = opts.reset|| opts. resetEnabled this.resetInterval = opts.resetInterval * 1000 // sets an interval timeout to check on status and send/emit reset command this.mock = opts.mock || process.env.MOCK this.wait = opts.wait || 0 // debounce is off by default @@ -24,15 +25,14 @@ class Interrupt extends Base { // pull down/up (down is default) can't be set here it is done by in DTOs or in RPI in config.txt // this is only used to monitor the status of the interrupt this.pull = opts.pull || 'down' - this.ready = false // true is interrupt is ready + this.ready = this.edge === 'both' ? true : false // true is interrupt is ready this.pin = {} - this.hook = opts.hook + this.count = 0 this.packet = opts.packet || {} this.packet.id = this.id this.packet.pin = this.pin_num this.packet.cmd = this.packet.cmd || opts.cmd || opts.interruptCmd || 'interrupt' - this.packet.count = 0 - this._hookFunc = defaultHook + this.packet.count = this.count this.commands = { fire:this.fire.bind(this), status:this.status.bind(this), @@ -49,32 +49,28 @@ class Interrupt extends Base { // TODO devel mock versions for testing on other than sbc with gpios this.pin = new Gpio(this.pin_num, 'in', this.edge, { debounceTimeout:this.wait }) - let res = await this.reset() - log.debug({msg:'initial connect interrupt reset packet sent', ressponse:res, method:'init', line:53}) + if (this.resetEnabled) log.debug({msg:'initial connect interrupt reset packet sent', ready:await this.reset(), method:'init', line:53}) + if (this.resetInterval && this.resetEnabled) setInterval(this.reset.bind(this),this.resetInterval) DeadJim( (signal,err) => { - log.warn({signal:signal, method:'init', line:56, error:err, msg:'Interrupt process was killed'}) + log.warn({signal:signal, method:'init', line:56, error:err, msg:'Interrupt process was killed, remove watchers, unexport'}) this.pin.unwatchAll() this.pin.unexport() // kill the kernel entry }) - log.debug({msg:'new interrupt pin created and watching', method:'init', line: 62, num:this.pin_num, status:await this.status() ? 'ready' : 'not ready', pin:this.pin, edge:this.edge,debounce:this.wait}) - - log.debug({method:'init', line: 62, msg: 'setting reconnect listener'}) - this.consumersListen('reconnected', async () => { - let res = await this.reset() - log.debug({msg:'reconnected, interrupt reset packet sent', ressponse:res, method:'init', line:67}) - }) - - if (this.resetInterval) setInterval(this.commands.reset,this.resetInterval) - this.pin.watch( function (err,value) { - log.debug('sbc interrupt tripped, value:', value, 'error:', err) + log.debug('interrupt tripped, value:', value, 'error:', err) this.count +=1 this._interruptProcess(value,err) }.bind(this)) + this.on('reconnected', this.reset.bind(this)) + this.on('connected', this.reset.bind(this)) + + log.debug({msg:'new interrupt pin created and watching', method:'init', line: 62, pin_num:this.pin_num, state:await this.status(), ready:this.ready, edge:this.edge,debounce:this.wait}) + + } // end init // manual firing for testing @@ -82,44 +78,55 @@ class Interrupt extends Base { log.debug({method:'fire', line:82, msg:`mock manually firing interrupt for pin ${this.pin_num}`}) await this._interruptProcess(1) packet.status = 'fired' - packet.ipin = this.pin_num + packet.pin = this.pin_num packet.cmd = 'reply' return packet } // returns true if pin is ready and waiting to trigger interrupt - async status(packet={}) { + async status(packet) { let status = await this.pin.read() - this.ready = this.pull==='down' ? !status : !!status - packet.pin = this.pin_num - packet.cmd = 'reply' - packet.ready = this.ready - return packet - } - - async reset(packet={}) { - if (!(await this.status()).ready) { - packet.cmd = this.resetCmd - packet.pin =this.pin_num - this.emit(this.resetCmd) // emit locally - await this.send(packet) - await this.push(packet) - log.error({msg: `interrupt was forced reset. ready now? ${(await this.status()).ready}`}) - return {cmd:'reply', reset:true, ready: (await this.status()).ready} + if (this.edge !=='both') this.ready = this.pull==='down' ? !status : !!status // ready is always true for 'both' + if (packet) { + packet.pin = this.pin_num + packet.state = status + if (this.edge !=='both') packet.ready = this.ready + packet.cmd = 'reply' + return packet } - return {cmd:'reply', reset:false, ready:true} + return status } + async reset(packet) { + let res = {} + if (this.edge !=='both' && this.resetEnabled) { + if (!this.ready) { + let reset = Object.assign({},this.packet) + reset.cmd = this.resetCmd + this.emit(this.resetCmd,reset) // emit locally + await this.send(reset) + await this.push(reset) + await this.status() + log.error({msg: `interrupt was forced reset. ready now? ${this.ready}`}) + res = {cmd:'reply', msg:`attempted interrupt reset ${this.ready? 'succeeded' : 'failed'}`, reset:true, ready:this.ready} + } + else res = {cmd:'reply', reset:false, ready:true, msg:'interrupt was ready, no action taken'} + } else res = {cmd:'reply', reset:false, ready:true, msg:'reset NA or disabled'} + if (packet) return Object.assign(packet,res) + return this.ready + } // use hook to do more processing async _interruptProcess(value,err) { let packet = Object.assign({},this.packet) + packet.id = this.id + packet.pin = this.pin_num packet.error = err packet.state = value packet.count = this.count packet.timeStamp = Date.now() packet.dateTime = new Date().toString() - if (this.hook) packet = await this._hookFunc.call(this,packet) + if (this._hookFunc) packet = await this._hookFunc.call(this,packet) log.debug({packet: packet, msg:'interrupt tripped, emit/send/push packet to all connected/listening'}) this.emit('interrupt',packet) // emit locally this.send(packet) // will send a packet via client to any socket @@ -127,7 +134,8 @@ class Interrupt extends Base { } registerHook(func) { - this._hookFunc = func + if (func) this._hookFunc = func + else this._hookFunc=defaultHook } } // end Class @@ -141,15 +149,9 @@ async function defaultHook(packet) { // new Promise((resolve) => { console.log('==========default hook =============') console.log(`pin ${packet.pin} on sbc gpio bus has thrown an interrupt`) - console.log(`emitting/sending/pushing to all connected socket client with cmd:${packet.cmd}`) - console.dir(packet) - console.log('replace by a new function with .registerHook(function) to overwrite this') - console.log('Must be async/promise returning if anything async happens in your hook') - console.log('This hook allows you to modify/add to the packet being pushed to connected clients') - console.log('the function will be bound to the instance for complete access') - console.log('if you pass a hash for .hook you can use it here as this.hook') - console.log('the hook options contains', this.hook) - console.log('by default the instance id will be attached to the packet before this') + console.log('can change anything in the packet in this hook') + console.log('to replace this use .registerHook(function)') + console.log('============================') return packet // resolve(packet) // }) diff --git a/src/interrupts.js b/src/interrupts.js index 09b9383..2268829 100644 --- a/src/interrupts.js +++ b/src/interrupts.js @@ -1,16 +1,19 @@ import Interrupt from './interrupt' +import Base from '@uci/base' import logger from '@uci-utils/logger' let log = {} // will more easily create a group of sbc pin interrupts -class Interrupts { +class Interrupts extends Base { constructor(pins, opts = {}) { + super(opts) this.id = this.id || 'interrupts' - this.pins = pins - this.interrupt = {} - this.s = { fire:this.fire.bind(this)} // make fire available via consumer packet send + this.pins = pins.map(pin => Number(pin)) // make sure actual numbers are passed + this._interrupts = new Map() + this._s = { fire:this.fire.bind(this)} // make fire available via consumer packet send + this.resetCmd = opts.resetCmd || 'interrupt.reset' log = logger({ name: 'interrupts', id: this.id, package:'@uci/interrupt', file:'src/interrupts.js'}) let pinopts = {} pins.forEach(pin => { @@ -19,48 +22,77 @@ class Interrupts { delete opts[pin] }) pins.forEach(pin => { - if (typeof pin !=='number') pin = parseInt(pin) pinopts[pin] = Object.assign({}, opts, pinopts[pin]) - pinopts[pin].id = (opts.id || 'interrupt') + ':' + pin + pinopts[pin].id = pinopts[pin].id || this.id + ':' + pin log.debug({ opts: pinopts[pin], method:'constructor', line:25, msg:`pin options for pin ${pin}`}) - this.interrupt[pin] = new Interrupt(pin, pinopts[pin]) + this._interrupts.set(pin, new Interrupt(pin, pinopts[pin])) + + // bubble up events from single interrupts to common + const EVENTS = ['status','consumer-connection'] + EVENTS.forEach(event => { + this.interrupt(pin).on(event, data => { + data.interrupt = { msg:'emitted event from single interrupt', pin:pin, id:pinopts[pin].id } + this.emit(event,data) + }) + }) }) } + interrupt(pin) { return this._interrupts.get(Number(pin)) } + async init() { + let res = await super.init() + if (res.errors) return Promise.reject(res.errors) return Promise.all( - this.pins.map(pin => { - return this.interrupt[pin].init() + Array.from(this._interrupts).map(inter => { + return inter[1].init() }) ) } + // combine all interrupt emits to one handler async listen(fn) { - this.pins.forEach(pin => { - if (fn==='stop') this.interrupt[pin].removeAllListeners('interrupt') - else this.interrupt[pin].on('interrupt', fn.bind(this)) + this._interrupts.forEach( inter => { + if (fn==='stop') inter.removeAllListeners(inter.packet.cmd) + else inter.on(inter.packet.cmd, fn.bind(this)) }) } - async addSocket() { - return Promise.all( - this.pins.map(pin => { - return this.interrupt[pin].addSocket(...arguments) - }) - ) + async listenReset(fn) { + this._interrupts.forEach( inter => { + if (fn==='stop') inter.removeAllListeners(inter.resetCmd) + else inter.on(inter.resetCmd, fn.bind(this)) + }) } - // manual firing for testing - async fire(packet) { - if (packet.pin) return await this.interrupt[packet.pin].fire(packet) - for ( let pin of packet.pins) { - packet[pin] = await this.interrupt[pin].fire(packet) - } return packet + // only adds consumer sockets to each interrupt to same socket/server + // alternatively use listen handler and single socket + async addInterSocket(name,type) { + if (type !=='s') { + return Promise.all( + Array.from(this._interrupts).map(inter => { + return inter[1].addSocket(...arguments) + }) + ) + } + } + + // manual firing of all pins for testing + async fire(packet={}) { + if (!packet.pin || packet.pin==='all') { + for (let inter of this._interrupts.entries()) { + packet[inter[0]] = await inter[1].fire({}) + } + packet.cmd='reply' + return packet + } + let pin = isNaN(Number(packet)) ? packet.pin : packet + if (this._interrupts.has(Number(pin))) return await this.interrupt(packet.pin).fire(packet) } registerHook(func) { - this.pins.forEach(async pin => { - this.interrupt[pin].registerHook(func) + this._interrupts.forEach(inter => { + inter.registerHook(func) }) }