Charlyplexing or matrix

  • 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