#!/usr/bin/python
# tz_stream_graph_server.py
# --copyright-- Copyright 2011 (C) Tranzoa, Co. All rights reserved. Warranty: You're free and on your own here. This code is not necessarily up-to-date or of public quality.
# --url-- http://www.tranzoa.net/tzpython/
# --email-- pycode is the name to send to. tranzoa.com is the place to send to.
# --bodstamps--
# September 29, 2011 bar
# September 30, 2011 bar continue
# October 3, 2011 bar continue
# October 4, 2011 bar get working with pulse
# October 8, 2011 bar put the log in the current dir
# October 10, 2011 bar redundant code to set the client.when and client.ip_adr - to match such redundant code in tz_web_server
# get rid of the .py in the log file name
# October 13, 2011 bar get the new client response stuff back to how it should be
# November 7, 2011 bar close the log in main
# November 29, 2011 bar pyflake cleanup
# February 22, 2012 bar extra verbosity
# --eodstamps--
## \file
#
#
# This is a web server for streaming real time data for graphing.
#
# How it works:
#
# TODO:
#
#
import os
import random
import re
import sys
import threading
import time
import SocketServer
SocketServer.TCPServer.request_queue_size = 150 # how many clients can hit us at the same time
import tz_http_server
import tz_web_server
import tzlib
HTTP_PORT = 12711
LOG_FILE_NAME = os.path.splitext(os.path.basename(__file__))[0] + ".log"
WAIT_TIMEOUT = 24 # how long to keep clients hanging in no data situations
MAX_NUMS = 100000 # how many numbers in a stream we remember, max
MAX_DELIVERY_NUMS = 1000 # how many numbers we deliver to the client, max
SI_ARG = 'si' # field name for stream id
VC_ARG = 'vc' # field name for next unknown data 'index' - our .js sets it if appropriate
CSI_ARG = 'csi' # field name for the client's stream number (index in to his 'streams' memory)
#
#
# Helper strings. Are put in to the main page's if they should be (the callbacks return something).
#
#
FICO_MAIN_PAGE = """
"""
FPNG_MAIN_PAGE = """
"""
MAIN_FILE_NAME = "stream.htm"
CSS_FILE_NAME = "stream.css"
PAGE_HEAD_CSS = "\n" % CSS_FILE_NAME
JS_FILE_NAME = "stream.js"
JS_FILE_NAME_X = "tz_stream_graph_client.js"
MAIN_PAGE_TITLE = "Stream Graph"
MAIN_PAGE = """
""" + FICO_MAIN_PAGE + FPNG_MAIN_PAGE + """
%s
%s
Well, I seem to be having a hard time starting up. I blame global bugs.
"""
class a_num(object) :
def __init__(me, num, when = None) :
me.num = num
me.when = when or tzlib.elapsed_time()
# a_num
class a_stream(object) :
def __init__(me, si) :
me.si = si or 'anonymous'
me.vc = 0
me.nums = []
me.lock = threading.RLock()
def append(me, num) :
if not isinstance(num, a_num) :
num = a_num(num)
me.lock.acquire()
me.nums.append(num)
me.vc += 1
if len(me.nums) > MAX_NUMS :
me.nums = me.nums[-len(me.nums) / 3 : ]
me.lock.release()
def get_nums(me, vc) :
me.lock.acquire()
vc = vc or (me.vc - MAX_DELIVERY_NUMS)
vc = min(vc, me.vc)
vc = max(vc, me.vc - MAX_DELIVERY_NUMS)
i = max(0, int(len(me.nums) - (me.vc - vc)))
na = me.nums[i:]
vc = int(me.vc)
me.lock.release()
return(vc, na)
# a_stream
class a_test_stream_source(threading.Thread) :
def __init__(me, srv, *args, **kwargs) :
super(a_test_stream_source, me).__init__(args = args, kwargs = kwargs)
me.srv = srv
me.st = a_stream('test')
me.srv.add_stream(me.st)
me.setDaemon(True)
me.start()
def run(me) :
me._stop = False
pt = tzlib.elapsed_time()
while not me._stop :
t = tzlib.elapsed_time()
if t - pt > 2 :
pt = t
me.st.append(random.randint(0, 99))
time.sleep(.1)
me.st.lock.acquire()
me._stop = -1
me.st.lock.release()
def stop(me) :
me.st.lock.acquire()
if me._stop == -1 :
me.st.lock.release()
return(True)
me._stop = True
me.st.lock.release()
return(False)
# a_test_stream_source
class a_server(tz_web_server.a_client_server) :
def __init__(me, *args, **kwargs) :
super(a_server, me).__init__(*args, **kwargs)
me._stop = False
me.streams = {}
me.css_file = ""
me.main_file = ""
me.js_file = ""
me.ts = a_test_stream_source(me)
def add_stream(me, st) :
if st.si == 'test' :
if st.si in me.streams :
raise ValueError('Cannot add two test streams.')
pass
me.streams[st.si] = st
def make_js_from_data(me, hit, st) :
vc = tz_web_server.get_arg_number(hit.args, 'vc')
csi = hit.args.get(CSI_ARG, [''])[0]
if not csi :
st = None # client needs to tell us how to get to the stream data on-client in javascript
try :
csi = str(int(csi))
except ( TypeError, ValueError ) :
csi = "'" + csi + "'" # note: client can screw himself up if he has a name rather than number as he should and if the name has single quotes in it
js = ""
if not st :
if vc > 4 :
time.sleep(10) # throttle the blank hitters
vc += 1
else :
t = tzlib.elapsed_time()
while True :
( vc, na ) = st.get_nums(vc)
if len(na) or (tzlib.elapsed_time() - t >= WAIT_TIMEOUT) or me._stop :
break
time.sleep(0.05)
if len(na) :
js += "if (client.streams[%s]) {\n" % ( csi, )
if hit.is_new :
js += "client.streams[%s].nms = [];\n" % ( csi, )
js += "client.streams[%s].nms = client.streams[%s].nms.concat(%s);\n" % ( csi, csi, str([ n.num for n in na ]), )
js += "}\n"
pass
js += "ss.ci = '%s';\n" % str(hit.ci)
js += "ss.vc = %u;\n" % vc
return(js)
def generic_hit(me, hit) :
if super(a_server, me).generic_hit(hit) :
return(True);
if hit.handler.command == 'OPTIONS' :
if hit.handler.send_http_headers(hit.is_head, content_type = "text/javascript", hdrs = { 'Access-Control-Allow-Origin' : '*' }) :
hit.handler.wfile.write('')
return(True)
pth = tz_http_server.safe_path(hit.handler.path)
if me.verbose > 2 :
me.log("; generic_hit pth=[%s]" % ( pth ))
if (pth == MAIN_FILE_NAME) or (pth == '') :
fd = me.main_file or me.load_file(MAIN_FILE_NAME, ".htm") or me.load_file(MAIN_FILE_NAME, ".html")
if not fd :
protocol = "http://"
url = "%s%s:%u" % ( protocol, tz_http_server.get_local_ip_address(), me.http_port )
stname = hit.args.get(SI_ARG, ['default'])[0]
fd = MAIN_PAGE % ( url + "/", MAIN_PAGE_TITLE, MAIN_PAGE_TITLE, url + "/", stname )
if fd :
if hit.handler.send_http_headers(hit.is_head) :
hit.handler.write_data(fd)
return(True)
pass
if pth == CSS_FILE_NAME :
fd = me.css_file or me.load_file(CSS_FILE_NAME, ".css")
if fd :
if hit.handler.send_http_headers(hit.is_head, content_type = "text/css") :
hit.handler.write_data(fd)
return(True)
pass
if pth == JS_FILE_NAME :
fd = me.js_file or me.load_file(JS_FILE_NAME, ".js") or me.load_file(os.path.join(os.path.dirname(__file__), JS_FILE_NAME ), ".js", rootdir = "")
if fd :
if hit.handler.send_http_headers(hit.is_head, content_type = "text/javascript") :
hit.handler.write_data(fd)
return(True)
pass
if pth == JS_FILE_NAME_X :
fd = me.js_file or me.load_file(JS_FILE_NAME_X, ".js") or me.load_file(os.path.join(os.path.dirname(__file__), JS_FILE_NAME_X), ".js", root_dir = "")
if fd :
if hit.handler.send_http_headers(hit.is_head, content_type = "text/javascript") :
hit.handler.write_data(fd)
return(True)
pass
return(False)
def new_client_hit(me, hit) :
""" Our server thinks this is a new client hitting us for something. """
if hit.ci in me.clients :
return(me.client_hit(hit))
me.learn_new_client(hit)
return(me.client_hit(hit))
def client_hit(me, hit) :
if hit.ci not in me.clients :
return(me.new_client_hit(hit))
hit.cl.set_when()
hit.cl.set_ipadr(hit.handler.client_address[0])
if re.search(r"^/?stream(/|\?|$)", hit.handler.path) :
st = me.streams.get(hit.args.get(SI_ARG, ['default'])[0], None)
if hit.handler.send_http_headers(hit.is_head, content_type = "text/javascript", cache_control = "", hdrs = { 'Access-Control-Allow-Origin' : '*' }) :
js = me.make_js_from_data(hit, st)
# print "@@@@", js
hit.handler.wfile.write(js)
return(True)
return(False)
def final_hit(me, hit) :
hit.handler.send_404(hit.is_head, "Golly, I can't figure that out.")
return(True)
def stop(me) :
me._stop = True
while not me.ts.stop() :
time.sleep(0.1)
super(a_server, me).stop()
# a_server
help_str = """
python %s (options) work_dir
Options:
--http_port port_number Set the server port number.
Web server for point/line streaming to a browser.
"""
if __name__ == '__main__' :
import TZKeyReady
import tz_server_logger
import TZCommandLineAtFile
program_name = sys.argv.pop(0)
TZCommandLineAtFile.expand_at_sign_command_line_files(sys.argv)
http_port = HTTP_PORT
if tzlib.array_find(sys.argv, [ "--help", "-h", "-?", "/?", "-?" ] ) >= 0 :
print help_str % ( os.path.basename(program_name), )
sys.exit(254)
while True :
oi = tzlib.array_find(sys.argv, [ "--http_port", "--httpport", "--http-port", "--port", "-p", ])
if oi < 0 : break
del sys.argv[oi]
http_port = int(sys.argv.pop(oi))
logger = tz_server_logger.a_logger(http_port = http_port, file_name = LOG_FILE_NAME)
me = a_server( http_port = http_port, logger = logger, show_exceptions = True)
ipadr = tz_http_server.get_local_ip_address()
print "Connect to http://localhost:%u/ or http://%s:%u/" % ( me.http_port, ipadr, me.http_port )
me.start()
while True :
k = TZKeyReady.key_ready()
if k and (k.lower() in [ 'q', '\033', ]) :
break
time.sleep(0.1)
me.stop()
logger.close()
#
#
#
# eof