Matthew Frost
bb86a855eb
All checks were successful
Deploy powerbar.ti Frontend Portal / Deploy-Tinance2-Frontend-Portal-Production (push) Successful in 38s
63 lines
2.1 KiB
Python
63 lines
2.1 KiB
Python
from flask import Flask
|
|
from app.views import routes
|
|
from flask import jsonify
|
|
from app.settings import powerbars
|
|
from app.utils import get_baytech_status_outlet_telnet
|
|
from flask_socketio import SocketIO, emit
|
|
from gevent import monkey
|
|
from app.threads import generate_threads
|
|
|
|
socketio = SocketIO(async_mode="gevent", logger=True, engineio_logger=True, cors_allowed_origins="*")
|
|
|
|
@socketio.on('connect', namespace='/powerupdates')
|
|
def connect_handler():
|
|
print('connected')
|
|
emit('response', {'meta': 'WS connected'})
|
|
|
|
|
|
def powerbar_status():
|
|
for powerbar_name, powerbar_info in powerbars.items():
|
|
print(f"Checking powerbar state {powerbar_name}")
|
|
if powerbar_info['type'] == 'baytech':
|
|
try:
|
|
print("Is a baytech powerbar, getting outlet status")
|
|
if powerbar_info['method'] in ["telnet", "batch_telnet"]:
|
|
outlets = get_baytech_status_outlet_telnet(powerbar_name)
|
|
|
|
try:
|
|
for outlet_id, outlet_info in powerbar_info['outlets'].items():
|
|
powerbar_info['outlets'][outlet_id]['state'] = outlets[outlet_id].lower()
|
|
print(f" Powerbar: {powerbar_name} Outlet: {outlet_id}, State: {outlets[outlet_id].lower()}")
|
|
except:
|
|
print(f"Failed to get outlet status from powerbar {powerbar_name} outlet: {outlet_id}")
|
|
continue
|
|
except:
|
|
print(f"Failed to get outlet status from powerbar {powerbar_name}")
|
|
continue
|
|
return
|
|
|
|
def create_app():
|
|
# create and configure the app
|
|
app = Flask(__name__)
|
|
app.config.from_pyfile('settings.py')
|
|
app.register_blueprint(routes)
|
|
powerbar_status()
|
|
scheduler.init_app(app)
|
|
generate_threads()
|
|
socketio.init_app(app)
|
|
return app
|
|
|
|
app = create_app()
|
|
scheduler.start()
|
|
|
|
@app.errorhandler(403)
|
|
def not_authorised(e):
|
|
return jsonify(error="Not authorized"), 403
|
|
|
|
@app.errorhandler(500)
|
|
def internal_server_error(e):
|
|
return jsonify(error="Internal server error"), 500
|
|
|
|
if not app.config["DEBUG"]:
|
|
monkey.patch_all()
|
|
|