592 lines
19 KiB
Python
592 lines
19 KiB
Python
# Copyright (C) 2021 Neythen Treloar neythen@fintic.io
|
|
#
|
|
# This file is part of stockTicker project for justinodunn.
|
|
#
|
|
# stockTicker can not be copied and/or distributed without the express
|
|
# permission of Neythen Treloar
|
|
|
|
from flask import Flask, render_template, request
|
|
from stockTicker import StockTicker
|
|
from werkzeug.utils import secure_filename
|
|
import os
|
|
import datetime
|
|
import threading
|
|
import csv
|
|
import pexpect
|
|
import time
|
|
import json
|
|
from multiprocessing import Process
|
|
from subprocess import Popen, PIPE
|
|
import numpy as np
|
|
#stock_ticker = StockTicker()
|
|
#print('API CALLER NOT STARTED')
|
|
api_caller = pexpect.spawn("sudo -E taskset -c 3 python3 api_caller.py")
|
|
displaying_screensaver = False
|
|
screensaver_p = None
|
|
professional = json.load(open('csv/display_settings.json', 'r'))[0] == "Professional"
|
|
|
|
command = 300
|
|
tickerList = 0
|
|
DelayTime = 20
|
|
LastCommand = ''
|
|
speedTime = 25
|
|
|
|
LOGO_FOLDER = 'logos/'
|
|
CSV_FOLDER = 'csv/new/'
|
|
ALLOWED_EXTENSIONS = {'csv', 'png'}
|
|
|
|
ticker = pexpect.spawn("sudo -E python3 stockTicker.py")
|
|
ticker.sendline('*') # run by default
|
|
ticker.sendline('A') # run by default
|
|
|
|
|
|
def allowed_file(filename):
|
|
return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS
|
|
|
|
def process_file(path, filename):
|
|
default_csv = csv.writer(open('csv/tickers.csv', 'w'))
|
|
new_csv = csv.reader(open((path), 'r'))
|
|
|
|
for row in new_csv:
|
|
default_csv.writerow(row)
|
|
|
|
|
|
|
|
|
|
app = Flask(__name__)
|
|
@app.route("/", methods=['GET', 'POST'])
|
|
def index():
|
|
global command
|
|
all_features = ['Current Weather','Daily Forecast','News', 'Sports (Upcoming Games)','Sports (Past Games)','Sports (Live Games)',
|
|
'Sports (Team Stats)','Custom Images', 'Custom GIFs', 'Custom Messages', 'Stocks', 'Crypto', 'Forex']
|
|
|
|
global professional
|
|
|
|
feature_settings = json.load(open('csv/display_settings.json', 'r'))
|
|
|
|
if not professional:
|
|
currently_displaying = json.load(open('csv/display_settings.json', 'r'))[1]
|
|
not_displaying = [f for f in all_features if f not in currently_displaying]
|
|
|
|
elif professional:
|
|
currently_displaying = json.load(open('csv/display_settings.json', 'r'))[1]
|
|
not_displaying = [f for f in all_features if f not in currently_displaying[0] and f not in currently_displaying[1]]
|
|
|
|
|
|
|
|
now = datetime.datetime.now()
|
|
timeString = now.strftime("%Y-%m-%d %H:%M")
|
|
logos_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'logos')
|
|
LogoList = os.listdir(logos_path)
|
|
|
|
stocks_settings = json.load(open('csv/stocks_settings.json', 'r'))
|
|
crypto_settings = json.load(open('csv/crypto_settings.json', 'r'))
|
|
forex_settings = json.load(open('csv/forex_settings.json', 'r'))
|
|
current_weather= json.load(open('csv/current_weather.json', 'r'))
|
|
daily_weather = json.load(open('csv/daily_weather.json', 'r'))
|
|
news_settings = json.load(open('csv/news_settings.json', 'r'))
|
|
upcoming_games = json.load(open('csv/upcoming_games.json', 'r'))
|
|
live_games = json.load(open('csv/live_games.json', 'r'))
|
|
past_games = json.load(open('csv/past_games.json', 'r'))
|
|
team_stats = json.load(open('csv/league_tables.json', 'r'))
|
|
image_settings = json.load(open('csv/image_settings.json', 'r'))
|
|
GIF_settings = json.load(open('csv/GIF_settings.json', 'r'))
|
|
message_settings = json.load(open('csv/message_settings.json', 'r'))
|
|
|
|
|
|
templateData = {
|
|
'currently_displaying': currently_displaying,
|
|
'not_displaying': not_displaying,
|
|
'stocks_settings': stocks_settings,
|
|
'crypto_settings': crypto_settings,
|
|
'forex_settings': forex_settings,
|
|
'current_weather': current_weather,
|
|
'daily_weather': daily_weather,
|
|
'news_settings': news_settings,
|
|
'upcoming_games': upcoming_games,
|
|
'past_games': past_games,
|
|
'live_games': live_games,
|
|
'team_stats': team_stats,
|
|
'image_settings':image_settings,
|
|
'GIF_settings':GIF_settings,
|
|
'message_settings':message_settings,
|
|
'professional':professional
|
|
}
|
|
return render_template('index.html', **templateData)
|
|
|
|
|
|
|
|
|
|
@app.route ("/start", methods = ['PUT', 'POST'])
|
|
def start():
|
|
global displaying_screensaver
|
|
global ticker
|
|
global api_caller
|
|
global professional
|
|
if displaying_screensaver:
|
|
screensaver_p.close()
|
|
api_caller = pexpect.spawn("sudo -E python3 api_caller.py")
|
|
ticker = pexpect.spawn("sudo -E taskset -c 3 python3 stockTicker.py")
|
|
displaying_screensaver = False
|
|
|
|
|
|
|
|
data = str(request.data)
|
|
|
|
all_settings = ['Stocks', 'Crypto', 'Forex', 'Current Weather', 'Daily Forecast', 'News', 'Sports (Upcoming Games)', 'Sports (Past Games)',
|
|
'Sports (Live Games)', 'Sports (Team Stats)', 'Custom Images', 'Custom GIFs', 'Custom Messages']
|
|
|
|
if professional:
|
|
all_settings = ['Stocks', 'Crypto', 'Forex', 'Current Weather', 'News']
|
|
|
|
print(professional)
|
|
positions = []
|
|
display_settings = []
|
|
data = str(request.data.decode('utf-8'))
|
|
|
|
input_settings = json.loads(data)
|
|
print(input_settings)
|
|
print(all_settings)
|
|
if professional:
|
|
input_settings[0] = [i for i in input_settings[0] if i in all_settings]
|
|
input_settings[1] = [i for i in input_settings[1] if i in all_settings]
|
|
|
|
print(input_settings)
|
|
|
|
s = "Professional" if professional else "Standard"
|
|
display_settings = [s] + [input_settings]
|
|
|
|
with open('csv/display_settings.json', 'w+') as f:
|
|
json.dump(list(display_settings), f)
|
|
|
|
ticker.sendline('K')
|
|
|
|
ticker.sendline('A')
|
|
return index()
|
|
|
|
@app.route("/stop")
|
|
def stop():
|
|
print('stop')
|
|
global ticker
|
|
global displaying_screensaver
|
|
if not displaying_screensaver:
|
|
ticker.sendline('K')
|
|
else:
|
|
screensaver_p.close()
|
|
|
|
|
|
if displaying_screensaver:
|
|
screensaver_p.close()
|
|
api_caller = pexpect.spawn("sudo -E python3 api_caller.py")
|
|
ticker = pexpect.spawn("sudo -E taskset -c 3 python3 stockTicker.py")
|
|
displaying_screensaver = False
|
|
return index()
|
|
|
|
@app.route("/update", methods=['PUT','POST'])
|
|
def update():
|
|
os.system("./update.sh")
|
|
|
|
@app.route("/display_format", methods = ['PUT', 'POST', 'GET'])
|
|
def display_format():
|
|
global professional
|
|
data = str(request.data)
|
|
if "Standard" in data:
|
|
professional = False
|
|
elif "Professional" in data:
|
|
professional = True
|
|
|
|
print(professional)
|
|
return index()
|
|
|
|
@app.route("/feature_settings", methods = ['PUT', 'POST', 'GET'])
|
|
def feature_settings():
|
|
|
|
data = str(request.data.decode('utf-8'))
|
|
|
|
input_settings = json.loads(data)
|
|
print(input_settings)
|
|
|
|
feature = input_settings['feature']
|
|
if feature in ['Stocks', 'Crypto', 'Forex']:
|
|
save_trade_settings(input_settings)
|
|
elif feature in ['Current Weather', 'Daily Forecast']:
|
|
save_weather_settings(input_settings)
|
|
elif feature == 'News':
|
|
save_news_settings(input_settings)
|
|
elif 'Sports' in feature:
|
|
save_sports_settings(input_settings)
|
|
|
|
elif feature in ['Custom GIFs', 'Custom Images']:
|
|
save_image_settings(input_settings)
|
|
|
|
elif feature == 'Custom Messages':
|
|
save_message_settings(input_settings)
|
|
|
|
return index()
|
|
|
|
# saves files uploaded to the webpage for images and GIFs
|
|
@app.route("/upload", methods = ['PUT', 'POST', 'GET'])
|
|
def upload():
|
|
try:
|
|
images = request.files
|
|
names = list(request.files.keys())
|
|
for name in names:
|
|
images[name].save('user_uploads/' +name)
|
|
|
|
except Exception as e:
|
|
print(e)
|
|
return index()
|
|
|
|
|
|
|
|
|
|
|
|
@app.route("/brightness", methods=['PUT','POST'])
|
|
def brightness():
|
|
global brightness
|
|
|
|
data= request.data.decode('utf-8')
|
|
settings = json.loads(data)
|
|
print(settings)
|
|
brightness =settings['brightness']
|
|
print(str(int(brightness) - 1))
|
|
ticker.sendline(str(int(brightness) - 1))
|
|
|
|
f = open('csv/settings.csv', 'r')
|
|
CSV = csv.reader(f)
|
|
next(CSV)
|
|
for line in CSV:
|
|
speed, _ = line
|
|
f.close()
|
|
|
|
f = open('csv/settings.csv', 'w+')
|
|
f.write('speed,brightness\n')
|
|
f.write(str(speed) + ',' + str(int(brightness) - 1))
|
|
f.close()
|
|
|
|
global LastCommand
|
|
LastCommand = 'Change Brightness'
|
|
return index()
|
|
|
|
|
|
def edit_wpa_sup(country, ssid, pwd):
|
|
current_wpa = open('/etc/wpa_supplicant/wpa_supplicant.conf')
|
|
|
|
wpa_lines = current_wpa.readlines()
|
|
|
|
wpa_lines[2] = 'country={}\n'.format(country)
|
|
|
|
current_wpa.close()
|
|
|
|
|
|
wpa_lines.append('\n')
|
|
wpa_lines.append('network={\n')
|
|
wpa_lines.append('\tssid="{}"\n'.format(ssid))
|
|
wpa_lines.append('\tpsk="{}"\n'.format(pwd))
|
|
wpa_lines.append('}\n')
|
|
|
|
wpa_string = ''.join(wpa_lines)
|
|
print(wpa_string)
|
|
current_wpa = open('/etc/wpa_supplicant/wpa_supplicant.conf', 'w+')
|
|
current_wpa.write(wpa_string)
|
|
|
|
|
|
@app.route("/wifi", methods = ['PUT', 'POST', 'GET'])
|
|
def wifi():
|
|
|
|
data= request.data.decode('utf-8')
|
|
print(str(data))
|
|
settings = json.loads(data)
|
|
|
|
country = settings['country'].upper()
|
|
ssid = settings['ssid']
|
|
pwd = settings['pwd']
|
|
|
|
edit_wpa_sup(country, ssid, pwd)
|
|
# resstart netoworking
|
|
os.system('wpa_cli -i wlan0 reconfigure')
|
|
return index()
|
|
|
|
def edit_hosts(hostname):
|
|
current_hosts = open('/etc/hosts')
|
|
|
|
hosts_lines = current_hosts.readlines()
|
|
print(hosts_lines[5])
|
|
|
|
hosts_lines[5] = '127.0.1.1 {}'.format(hostname)
|
|
|
|
current_hosts.close()
|
|
|
|
hosts_string = ''.join(hosts_lines)
|
|
|
|
current_hosts = open('/etc/hosts', 'w+')
|
|
current_hosts.write(hosts_string)
|
|
|
|
@app.route("/hostname", methods = ['PUT', 'POST', 'GET'])
|
|
def hostname():
|
|
data= request.data.decode('utf-8')
|
|
settings = json.loads(data)
|
|
hostname = settings['hostname']
|
|
print('hostname', hostname)
|
|
|
|
edit_hosts(hostname)
|
|
os.system("sudo hostnamectl set-hostname {}".format(hostname))
|
|
os.system("sudo systemctl restart avahi-daemon")
|
|
|
|
return index()
|
|
|
|
|
|
@app.route("/screensaver", methods = ['PUT', 'POST'])
|
|
def screensaver():
|
|
global displaying_screensaver
|
|
global ticker
|
|
global api_caller
|
|
global screensaver_p
|
|
api_caller.close()
|
|
ticker.close()
|
|
data = str(request.data)
|
|
|
|
if displaying_screensaver:
|
|
screensaver_p.close()
|
|
|
|
|
|
if "Pulsating Colors" in data:
|
|
screensaver_p = pexpect.spawn("sudo -E python3 ./rpi-rgb-led-matrix/bindings/python/samples/pulsing-colors.py")
|
|
|
|
elif "Rotating Square" in data:
|
|
screensaver_p = pexpect.spawn("sudo -E python3 ./rpi-rgb-led-matrix/bindings/python/samples/rotating-block-generator.py")
|
|
|
|
elif "Pulsating brightness" in data:
|
|
screensaver_p = pexpect.spawn("sudo -E python3 ./rpi-rgb-led-matrix/bindings/python/samples/pulsing-brightness.py")
|
|
|
|
elif "Game of Life" in data:
|
|
screensaver_p = pexpect.spawn("sudo -E python3 game_of_life.py")
|
|
|
|
displaying_screensaver = True
|
|
return index()
|
|
|
|
|
|
|
|
|
|
def combine_dict(current_settings, input_symbols, current_key):
|
|
# removes keys not in input from current_settings[current_key] and adds keys not in current from input
|
|
|
|
current_symbols = list(current_settings[current_key].keys())
|
|
|
|
# add any stock that arent current in the settings
|
|
for IS in input_symbols:
|
|
if IS not in current_symbols:
|
|
current_settings[current_key][IS] = []
|
|
|
|
# remove stocks not in settings
|
|
for CS in current_symbols:
|
|
if CS not in input_symbols:
|
|
del current_settings[current_key][CS]
|
|
|
|
return current_settings
|
|
|
|
def save_trade_settings(input_settings):
|
|
|
|
|
|
filename = input_settings['feature'].lower() + '_settings.json'
|
|
|
|
current_settings = json.load(open('csv/' + filename, 'r'))
|
|
|
|
current_settings['speed'] = input_settings['speed'].lower()
|
|
current_settings['animation'] = input_settings['animation'].lower()
|
|
current_settings['percent'] = input_settings['percent']
|
|
current_settings['point'] = input_settings['point']
|
|
current_settings['logos'] = input_settings['logos']
|
|
current_settings['chart'] = input_settings['chart']
|
|
current_settings['title'] = input_settings['title']
|
|
|
|
|
|
current_settings = combine_dict(current_settings, input_settings['symbols'], 'symbols')
|
|
|
|
json.dump(current_settings, open('csv/' + filename, 'w+'))
|
|
|
|
if input_settings['feature'].lower() == 'stocks':
|
|
api_caller.sendline('s')
|
|
elif input_settings['feature'].lower() == 'crypto':
|
|
api_caller.sendline('c')
|
|
elif input_settings['feature'].lower() == 'forex':
|
|
api_caller.sendline('f')
|
|
|
|
def save_weather_settings(input_settings):
|
|
|
|
|
|
filename = 'current_weather.json' if input_settings['feature'] == 'Current Weather' else 'daily_weather.json'
|
|
|
|
current_settings = json.load(open('csv/' + filename, 'r'))
|
|
|
|
current_settings['speed'] = input_settings['speed'].lower()
|
|
current_settings['animation'] = input_settings['animation'].lower()
|
|
current_settings['temp'] = input_settings['temp'].lower()
|
|
current_settings['wind_speed'] = input_settings['wind_speed'].lower()
|
|
current_settings['colour'] = input_settings['colour'].lower()
|
|
current_settings['city_colour'] = input_settings['city_colour'].lower()
|
|
current_settings['title'] = input_settings['title']
|
|
|
|
if input_settings['feature'] == 'Daily Forecast':
|
|
current_settings['current_weather'] = input_settings['current_weather']
|
|
|
|
|
|
current_settings = combine_dict(current_settings, input_settings['locations'], 'locations')
|
|
|
|
json.dump(current_settings, open('csv/' + filename, 'w+'))
|
|
|
|
api_caller.sendline('w')
|
|
|
|
def save_news_settings(input_settings):
|
|
filename = 'news_settings.json'
|
|
|
|
current_settings = json.load(open('csv/' + filename, 'r'))
|
|
|
|
current_settings['speed'] = input_settings['speed'].lower()
|
|
current_settings['animation'] = input_settings['animation'].lower()
|
|
current_settings['title'] = input_settings['title']
|
|
current_settings['category'] = input_settings['category'].lower()
|
|
current_settings['country'] = input_settings['country']
|
|
|
|
|
|
current_settings['sources'] = list(set(current_settings['sources'] + input_settings['sources']))
|
|
|
|
json.dump(current_settings, open('csv/' + filename, 'w+'))
|
|
api_caller.sendline('n')
|
|
|
|
def save_sports_settings(input_settings):
|
|
|
|
feature = input_settings['feature']
|
|
|
|
if feature == 'Sports (Upcoming Games)':
|
|
filename = 'upcoming_games.json'
|
|
elif feature == 'Sports (Past Games)':
|
|
filename = 'past_games.json'
|
|
elif feature == 'Sports (Live Games)':
|
|
filename = 'live_games.json'
|
|
elif feature == 'Sports (Team Stats)':
|
|
filename = 'league_tables.json'
|
|
|
|
current_settings = json.load(open('csv/' + filename, 'r'))
|
|
|
|
current_settings['speed'] = input_settings['speed'].lower()
|
|
current_settings['animation'] = input_settings['animation'].lower()
|
|
current_settings['title'] = input_settings['title']
|
|
current_settings['feature'] = input_settings['feature']
|
|
|
|
current_settings = combine_dict(current_settings, input_settings['leagues'], 'leagues')
|
|
|
|
json.dump(current_settings, open('csv/' + filename, 'w+'))
|
|
|
|
api_caller.sendline('S')
|
|
|
|
# for images and GIFs
|
|
def save_image_settings(input_settings):
|
|
filename = 'image_settings.json' if input_settings['feature'] == 'Custom Images' else 'GIF_settings.json'
|
|
|
|
current_settings = input_settings
|
|
|
|
current_settings['speed'] = input_settings['speed'].lower()
|
|
current_settings['animation'] = input_settings['animation'].lower()
|
|
current_settings['title'] = input_settings['title']
|
|
|
|
del current_settings['feature']
|
|
json.dump(current_settings, open('csv/' + filename, 'w+'))
|
|
|
|
def save_message_settings(input_settings):
|
|
|
|
json.dump(input_settings, open('csv/message_settings.json', 'w+'))
|
|
|
|
|
|
|
|
@app.route("/shutdown")
|
|
def shutdown():
|
|
os.system("sudo shutdown now")
|
|
return index()
|
|
|
|
|
|
|
|
|
|
|
|
@app.route("/matrix")
|
|
def matrix():
|
|
|
|
if "Run Stocks" in request.form:
|
|
ticker.sendline('K')
|
|
ticker.sendline('S')
|
|
elif "Run Crypto" in request.form:
|
|
ticker.sendline('K')
|
|
ticker.sendline('C')
|
|
|
|
elif "Run Forex" in request.form:
|
|
ticker.sendline('K')
|
|
ticker.sendline('F')
|
|
|
|
elif "Run News" in request.form:
|
|
|
|
ticker.sendline('K')
|
|
ticker.sendline('N')
|
|
|
|
elif "Run Weather" in request.form:
|
|
|
|
|
|
ticker.sendline('K')
|
|
ticker.sendline('W')
|
|
|
|
elif "Run Daily Weather" in request.form:
|
|
|
|
ticker.sendline('K')
|
|
ticker.sendline('D')
|
|
|
|
elif "Past NHL" in request.form:
|
|
ticker.sendline('K')
|
|
ticker.sendline('P')
|
|
|
|
elif "Future NHL" in request.form:
|
|
ticker.sendline('K')
|
|
ticker.sendline('l')
|
|
|
|
elif "Live NHL" in request.form:
|
|
ticker.sendline('K')
|
|
ticker.sendline('L')
|
|
|
|
elif "Premier league table" in request.form:
|
|
|
|
ticker.sendline('K')
|
|
ticker.sendline('t')
|
|
|
|
elif "Professional" in request.form:
|
|
|
|
ticker.sendline('K')
|
|
ticker.sendline('b')
|
|
|
|
elif "All" in request.form:
|
|
|
|
ticker.sendline('K')
|
|
ticker.sendline('A')
|
|
|
|
elif "Multiple" in request.form:
|
|
|
|
ticker.sendline('K')
|
|
ticker.sendline('+')
|
|
|
|
elif "Stop Display" in request.form:
|
|
|
|
|
|
LastCommand = 'Stop display at next checkpoint'
|
|
ticker.sendline('K')
|
|
|
|
elif "Shutdown the pi" in request.form:
|
|
|
|
try:
|
|
os.system("sudo shutdown now")
|
|
except:
|
|
print("couldn't shutdown")
|
|
return index()
|
|
|
|
if __name__ == "__main__":
|
|
|
|
app.run(host='0.0.0.0', port=1024, debug=False) # the debuggger causes flickering
|
|
|
|
#sudo ./demo -D1 final.ppm -t 50 -m 25 --led-gpio-mapping=adafruit-hat --led-rows=32 --led-cols=256 --led-slowdown-gpio=4
|