Charlyplexing or matrix

  • Hello,
    I m building a home automation system and I was wondering if I could use the unipis digital input for charlyplexing or matrix setup. So I could use a lot of push buttons.

    Kind regards

  • administrators

    Hello @Wim-Stockman,

    charlieplexing is not possible with digital inputs of UniPi. However matrix is OK, you just need the SW that will take care of decoding the states of inputs.

  • Hello,

    Thx for the quick reply. I was thinking of using the keypad arduino library as a base for making the matrix scanner. Do you think this might work ?.
    It would be great if I get it do work. Would mean you have suddenly 49 inputs available.
    I think that will be enough for my house :-).

    Kind regards
    Wim Stockman

  • administrators

    @Wim-Stockman Basically yes. What interface to the DIs do you plan to use? Direct GPIOs on the RPi or websocket...?

  • @tomas_hora I 'll try direct gpio with the pigpio.I found something on the forum to get started. So I 'll try over the weekend if I get something working :-)

  • administrators

    @Wim-Stockman Good luck! Would be great it you share some results :)

  • Hi,
    I run into a stupid problem.
    trying the example of connecting with python through websocket.
    I get this error : File "", line 20, in <module>
    ws = websocket.WebSocketApp(url, on_message = on_message, on_error = on_error, on_close = on_close)
    AttributeError: 'module' object has no attribute 'WebSocketApp'

    I don't know what I'm doing wrong , just copy paste the example check the ip adres (just localhost) but I seem stuck.
    could you help me ?

  • @Wim-Stockman Ok found the solution
    uninstall websocket: sudo apt-get uninstall python-websocket
    and install : sudo pip install websocket-client

    this does the trick for me :-)

  • It seems the get attribute isn't implemented for the relays
    and if I do a full request it gets dump on de command line in instead of sent to the websocketclient.
    So I could initialise the position of the relays

    import websocket
    import json
    import time
    import threading
    url = "ws://"
    i = 0
    relaystatus = [0,0,0,0,0,0,0,0]
    #initiliaze relay 3 just for demo purpose
    wsinit = websocket.create_connection("ws://")
    result = json.loads(wsinit.recv())
    relaystatus[2] = result["value"]
    print relaystatus
    def on_message(ws, message):
        obj = json.loads(message);
        dev = obj['dev'];
        print obj['dev'];
        circuit = obj['circuit'];
        value = obj['value'];
        global i
        global relaystatus
        i += 1
        print i
        if dev == 'input' and i > 2:
            print "erin";
            ws1 = websocket.WebSocket();
            #sending messages
            print obj['value'];
            if relaystatus[2] == 0:
                    relaystatus[2] = 1
                    relaystatus[2] = 0
            i = 0
        print message
    def on_error(ws, error):
        print error
    def on_close(ws):
        print "Connection closed"
    #receiving messages
    ws = websocket.WebSocketApp(url, on_message = on_message, on_error = on_error, on_close = on_close)
    wst = threading.Thread(target=ws.run_forever())
    wst.daemon = True
    while not  ws.sock.connected:

    This demo work for a push button that's why I have the 2 counter to activate the relay , because rising edge triggers a message as well as the falling edge. But that will come in handy for detecting how long somebody presses de button.

    Somebody know where I can begin to omplent the Get for the relays.

    Kind regards,

  • administrators

    @Wim-Stockman Thanks for the report. See this modification of the on_message function:

                    device = Devices.by_name(dev, circuit)
                    # result = device.set(value)
                    func = getattr(device, cmd)
                    if value is not None:
                        result = func(value)
                        result = func()
                    if is_future(result):
                        result = yield result
                    #print json.dumps(result)
                except Exception, E:
                    print E

    It will send the result of the command to only the client that send the command. But in our test, you are using two clients, so you will not receive the result in the other client. If you want to send the result to all clients connected to the WebSocket, use the devevents.satus(device) insted.

    Working example of the test client:

    import websocket
    import json
    import time
    import threading
    url = "ws://"
    i = 0
    relaystatus = [0,0,0,0,0,0,0,0]
    #initiliaze relay 3 just for demo purpose
    wsinit = websocket.create_connection("ws://")
    while True:
      result = json.loads(wsinit.recv())
      if result['dev'] == 'relay':
        print result

    I will commit it in the first week of January.

  • Hi thx,

    This works !
    But now I get a weird exception on the on line 140 it throws an exception:
    I get "Expected bytes, unicode, or None; got <type 'int'>"

    Each time I sent a message : wsinit.send('{"cmd":"set","dev":"relay","circuit":"3","value":"1"}')

    It's like it iterates twice over the on_message loop

    my demo code is

    import websocket
    import json
    import time
    import threading
    url = "ws://"
    i = 0
    global relaystatus
    relaystatus = [0,0,0,0,0,0,0,0]
    #initiliaze relay 3 just for demo purpose
    wsinit = websocket.create_connection("ws://")
    result = json.loads(wsinit.recv())
    while True:
      result = json.loads(wsinit.recv())
      if result['dev'] == 'input':
            if relaystatus[2]==0 :
               relaystatus[2] = 1
               relaystatus[2] = 0
      print result

    Thanks for the great help already :-)

  • administrators

    @Wim-Stockman I saw your pull request.. Did you find out what it was with the Expected bytes, unicode, or None; got <type 'int'>"? On what code did it crash?

  • So I found the problem, solved it and put a pull request in github.

    made a little change in the

    #adding support to get state of relays using full msg in cmd 
               if cmd == "full":

  • @tomas_hora yes the on_events routing fired a new message for the status change of the relay which was captured by the on_message and since it just sented a 0 or 1 the on_message gave this weird thing

  • This post is deleted!

  • So I finished the hardware part for the input plexing.
    I used a variation on the charlyplexing.

    In the example breadboard I only used 4 digital inputs (symbolized by the Leds), the 12V Vcc is symbolized by the battery and 8 pushbutton :

    You could attach up to 15 pushbuttons on 4 digital inputs with this method. It is just some binary logic behind and a lot of diodes :-)
    The purple wires are connected to an input just directly. (you could place diodes for safety)
    and the next 6 pushbuttons are connected through 2 wires.
    If you want 4 more pushbuttons you should combine 3 wires
    and for the last one you need 4 wires.

    So in theory with this method you could make 16383 different input combinations with this method.
    I think it will not be practical but with 2 wires you can easily attach up to (14²-1) 195 pushbuttons with each different reactions.
    So know I will create a python class to handle these inputs using the websocket method :-)

  • Hello,

    Is there a possibility to receive the inputs via websocket with timestamps.
    I know the pigpio librabry does this, but since the new neurons don't work anymore with pigpio. I was wondering if this is possible would be easy to detect , long pressed buttons etc...

  • Hello,
    I have written some code to detect two inputs simultaniously activated through websocket.
    Here is the code

    import websocket
    import threading
    import datetime
    import json
    import time
    url = "ws://"
    class DI:
    	def __init__(self,circuit,value,timestamp):		
    		self.circuit = circuit
    		self.value = value
    		self.timestamp = timestamp						
    		self.longpress = 0
    		self.doubletap = 0
    class homecontrol:
    	def __init__(self,websock_evok_url):
    		self.status = 0
    		self.digitalinput_list = [[0,0],[0,0]] # create list with timestamp inputs
    		self.timeout = 1000				
    		self.websock_evok_url = websock_evok_url	
    		#connect to websocket and spawn message thread = websocket.WebSocketApp(url, on_message = self.on_message, on_error = self.on_error, on_close = self.on_close)
    		self.wst = threading.Thread(		
    		self.wst.daemon = True
    	def time_stamp_inputs(self):
    			t =
    			t = (t.second * 1000)+(t.microsecond / 1000)								
    			return t
    	def detect_if_pressed_together(self):				
    		d1 = self.digitalinput_list[0]
    		d2 = self.digitalinput_list[1]
    		# must be two different inputs and the same values you cannot press and unpress at the same time
    		if d1.circuit != d2.circuit and d1.value == d2.value:		
    				t = d1.timestamp -d2.timestamp		
    				if t < self.timeout:				
    					return 1			
    					return 0
    	def on_message(self,ws, message):		
    			self.obj = json.loads(message)
    			print self.obj = self.obj["dev"]
    			self.circuit = self.obj["circuit"]
    			self.value = self.obj['value']		
    			if == "input" and self.value == 1 :											
    				digitalinput = DI(self.circuit,self.value,self.time_stamp_inputs())			
    					l = (self.digitalinput_list[0].circuit,self.digitalinput_list[1].circuit)
    					l = sorted(l)					
    					x = str(l[0])+str(l[1])						
    					self.digitalinput_list[0].circuit = int(x)
    					x = self.digitalinput_list[0]
    					proc = threading.Thread(target=self.proceed_input,args=(x,))
    	def detect_long_press_of_button(self):
    	def proceed_input(self,I):# SETUP a method to declare what to do which each function , was thinking of using codeblocks for this part ?		  
    		  if I.circuit == 17:			  
    			  print I.circuit
    			  relay = 5
    			  t = threading.Thread(target=self.switch_rel,args=(relay,))
    			  #t = threading.Thread(target=switch_on_off_relay,args=(relay,))
    			  print "thread"
    		  if I.circuit == 26:	
    			  print "26"
    			  relay = 1
    			  c = threading.Thread(target=self.switch_rel,args=(relay,))			  
    			  print "thread2"
    	def on_error(self,ws,error):
    			print "error"
    			print error
    	def on_close(self,ws):
    			print "connection closed"
    	def switch_rel(self,c):		
    		relaynumber = c'{"cmd":"set","dev":"relay","circuit":"'+str(relaynumber)+'","value":"0"}')
    		#		self.status = 1
    # SETUP a method to declare what to do which each function , was thinking of using codeblocks for this part ?		  				
    def switch_on_off_relay(c):
    			print c
    			relaynumber = c
    			url = "ws://"		
    			ws = websocket.create_connection(url)		
    			print "swdf"
    x = homecontrol(url)	

    But I have a problem that some inputs when they are to fast one after each other get forgetten.
    and if I do none multithreading it does do al the steps but then one push may hold the others so not good.
    Can somebody help me ?

  • administrators

    @Wim-Stockman I will take a look at it during the next week. Thanks for the tip.

    I did not check the code much but there should be some time buffer while decoding the correct input. The messages does not always come at the same time, there might be a tens of ms gap between each and you shoudl tie those together...

Log in to reply

Looks like your connection to Forum was lost, please wait while we try to reconnect.