LED Sign

http://wls.wwco.com/ledsigns/prolite/

Source Code

signDaemon.py By Textile

#support python 3.0 compliant print statement
from __future__ import print_function
#default imports
import socket,sys,threading,time,re
#pyserial imports
import serial
import os
class server(threading.Thread):
    def __init__(self,signNS,controllerNS,ip="0.0.0.0",port=9001):
        threading.Thread.__init__(self)
        threading.Thread.daemon = True
	self.ip = ip
        self.port = port
        self.sign = signNS #sign is the namespace of the led sign class 
        self.controller = controllerNS #namespace of the controller used catch request to kills
        self.die = False
    
    def OpenListener(self):
        """
        Open socket to lisen for messages
        """
        self.mySocket = socket.socket ( socket.AF_INET, socket.SOCK_STREAM )
        self.mySocket.bind((self.ip, self.port))
        self.mySocket.listen(5)
    
    def CloseListener(self):
        """
        Close Socket
        """
        self.mySocket.close()
        print("Listen socket closed")
    

    def GetMessage(self):
        """
        Get a message from a client and append it to the Sign Buffer to be displayed 
        """
        print("Waiting for Connections \r")
        channel, details = self.mySocket.accept()
        print("We have opened a Connection with "+str(details))
	try:
	    message = channel.recv(1024)
            if self.sign.SignCheck() == False:
                channel.sendall('The sign is turned off')
                return None
            self.sign.signBufHistory.append('%s,%s'%(details[0],message))
	    self.sign.signBuffer.append(message)
        except Exception:
	    return False
        print(len(self.sign.signBuffer))
	channel.sendall("Updating sign to "+message)
	os.system("python /uas/random/ringaling.py 40")
        return None
                
    def recover(self):
        """
        recover if the listener crashes
        """
        try:
            self.CloseListener()
            self.OpenListener()
        except Exception:
            print("waiting for port to close:")
            try:
                self.CloseListener()
                time.sleep(10)
                self.OpenListener()
            except Exception:
                print("you broke it you dumb mook")
                self.controller.cancel()

    def run(self):
        """
        Run the thread in a loop
        """
        try:
            self.OpenListener()
            while True:
                if not self.controller.running.is_set():
                    self.CloseListener()
                    break
                else:
                    if self.GetMessage() == False:
                        self.recover()    

        except KeyboardInterrupt:
            pass
            self.CloseListener()
            print("The Listener Closed")
            #self.cancel()
            self.exit(-1)

class ledSign:
    def __init__(self, tty='/dev/ttyUSB0'):
        self.tty  = tty
        self.baud = 9600
        self.bytesize = serial.EIGHTBITS
        self.parity   = serial.PARITY_NONE
        self.stopbit  = serial.STOPBITS_ONE
        self.timeout  = .5
        self.xonxoff  = 0 
        self.rtscts   = 0
        self.signBuffer = [] #buffer to be written
        self.signBufHistory = [] #buffer that was written
        self.formatRE = re.compile('<ID\d\d><\w\w>.*') #match something that looks like <ID01><PA>
    
    def formatMSG(self,msg,page='A'):
        """
        Format a message with page and propper headers and footers
        """
        if str(msg) == '': #clear each page
            fmsg = '<ID01><P%s> %s \r\n'[:1016] %(page,msg)
        elif self.formatRE.match(msg) != None: #use client supplied formatting
            fmsg = str(msg)+'\r\n'[:1016]
        else:
            fmsg = "<ID01><P%s><CL> {0} %s \r\n"[:1016] %(page,msg)
        return fmsg
	
    def SignCheck(self):
	"""
	Check to see if the sign is on, True if on False if off
	"""
	self.OpenSerial()
        self.signObj.write("<ID01>\r\n")
        responce = self.signObj.read(10)
	self.CloseSerial()
        if len(responce) >=8 and responce[:8] == "\x13<ID01>S":
            return True
        else:
            return False

    def OpenSerial(self):
       """
       Open led sign serial device
       """
       try:
            self.signObj  = serial.Serial(self.tty,self.baud,self.bytesize,
                self.parity,self.stopbit,self.timeout,self.xonxoff,self.rtscts)
            self.signObj.flush() #flush serial connection
       except Exception:
            print("Open Failed")
            return

    def ClearSign(self):
        """
        A function to blank out every page on the sign
        """
        print("Clearing all sign Pages")
        for x in range(65,91):
            #time.sleep(.2)
            self.CycleWriteMsg('',chr(x)) #1016 is max message len
          
    def SendSTDMessage(self,message,page='A'):
        """
        Send a standard text message to the LED screen
        """
        STDMessage = self.formatMSG(message,page)
        self.signObj.write(STDMessage) #1016 is max message len

    def CloseSerial(self):
        """
        Flush all data to serial device and then close connection
        """
        if self.signObj.isOpen() is True:
            self.signObj.flush()
            self.signObj.close()
    
    def CycleWriteMsg(self,message,page):
        """
        Open serial device write message then close it
        """
        if self.signObj.isOpen() is True:
            self.CloseSerial()
            self.OpenSerial()
            self.SendSTDMessage(message,page)
            self.CloseSerial()
        else:
            self.OpenSerial()
            self.SendSTDMessage(message,page)
            self.CloseSerial()

class controller:
    """
    Class to controll the LCD sign
    """
    def __init__(self,signConnector):
        self.running = threading.Event() #used to kill thread
        self.newMessage = ""
        self.spaceSign = ledSign(signConnector)
        self.spaceSign.OpenSerial()
        self.spaceSign.ClearSign()
        self.spaceSign.CloseSerial()
        self.ledServer = server(self.spaceSign,self)
        self.ledServer.start()
        self.running.set() #mark that the server thread is running
        self.safety = threading.RLock()

    def cancel(self):
        """
	exit the program
	"""
	with self.safety:
            if self.running.is_set():
                print("\nTerminating...")
                sys.exit(-1)
            else:
                return
    def exitJob(self):
        """
        Close out anything we had running and exit
        THIS FUNCTION IS ALL FUCKED UP IGNORE THIS CODE 
        Ill fix the control C break later
        """
        self.spaceSign.CloseSerial()
        self.ledServer.mySocket.connect((self.ledServer.ip,self.ledserver.port))
        self.ledServer.mySocket.close()
        return
    
    def writeHistory(self):
        """
        Write out log file of every item written to sign
        """
        #should check what page its written to?
        try:
           if len(self.spaceSign.signBufHistory) == 0:
		pass
	   else:
                current = self.spaceSign.signBufHistory[-1]
                history = open('SignHistory.csv','a')
                history.writelines(current)
                history.close()

                csign = open('/tmp/sign','w')# added by CP for bot stuff
                csign.write(current[current.find(',')+1:])
                csign.close()

		os.system("bash ./sign2wave.sh") # for the asterisk box

        except IOError:
            print('Sign History file not found\n')
    
    def readHistory(self):
        """
        Read in log file to display last message update
        """
        try:
           history = open('SignHistory.csv','r')
	   x = history.readlines()[-1].strip()
           if x == '':
               history.seek(0)
               x = history.readlines()[-2].strip()
           history.close()
           return x
        except IOError:
            print('Sign History file not found\n')
            return " , "

    def main(self):
        """
        Control the LED sign from this function
        """
	try:
            #display the last updated message
            self.spaceSign.OpenSerial()
            self.spaceSign.SendSTDMessage(self.readHistory().split(',')[1])
            self.spaceSign.CloseSerial()
            while True:
                if not self.running.is_set():
                    self.exitJob()
                    return
                    
                if len(self.spaceSign.signBuffer) != 0:
                    time.sleep(2)
                    self.spaceSign.OpenSerial()
                    self.newMessage = self.spaceSign.signBuffer.pop(0)
                    self.spaceSign.SendSTDMessage(self.newMessage)
                    print("Wrote to Sign: "+str(self.newMessage))
                    self.spaceSign.CloseSerial()
                    self.writeHistory()
                else:
                    time.sleep(5)
                
        except KeyboardInterrupt:
            pass
            self.cancel()

if __name__ == "__main__":
    #test handler script
    controller(sys.argv[1]).main()

The sign now "dings" with a physical alarm bell when updated.

import serial,time,sys
s = serial.Serial('/dev/ttyUSB0', baudrate = 9600, rtscts='rtscts', xonxoff='xonxoff', timeout=1000)
time.sleep(1.5)
s.setDTR(False)
time.sleep(0.022)
s.setDTR(True)

t=sys.argv[1]

s.write(str(t)+"\n")
time.sleep((float(t))/1000)