您最多选择25个主题 主题必须以字母或数字开头,可以包含连字符 (-),并且长度不得超过35个字符
 
 
 
 

112 行
2.6 KiB

import importlib
import threading
import time
import traceback
import json
import logging
log = logging.getLogger('werkzeug')
log.setLevel(logging.ERROR)
from flask import Flask
from flask_cors import CORS
build_folder = 'web_interface/build'
app = Flask(__name__, template_folder=build_folder, static_folder=build_folder, static_url_path='')
CORS(app)
from munch import Munch
from watchdog.observers import Observer
from watchdog.events import PatternMatchingEventHandler
from mosfet import bot
global_state = Munch()
g = global_state
g.connection = False
g.name = None
g.mcdata = False
g.pos = False
g.dimension = None
g.item_lock = False
g.inv = {}
g.objects = {}
g.mobs = {}
g.players = {}
g.player_names = {}
g.window = None
g.job = None
g.correction_count = 0
g.holding = 0
g.afk = False
g.health = 20
g.food = 20
g.sand_origin = None
@app.route('/api/global')
def api_global():
data = json.dumps(g, default=lambda o: str(o), indent=4)
response = app.response_class(
response=data,
status=200,
mimetype='application/json'
)
return response
@app.route('/')
def root():
return app.send_static_file('index.html')
reload_timeout = time.time()
def main():
def reload_bot(event):
global reload_timeout
if time.time() - reload_timeout > 2.0:
reload_timeout = time.time()
print('Reloading...')
g.running = False
event_handler = PatternMatchingEventHandler(patterns=['*.py'], ignore_patterns=['./main.py'])
event_handler.on_any_event = reload_bot
observer = Observer()
observer.schedule(event_handler, 'mosfet', recursive=True)
observer.start()
try:
while True:
try:
g.running = True
bot.bot(global_state)
importlib.reload(bot)
except BaseException as e:
g.running = True
traceback.print_exc()
print('Locking...')
while g.running:
time.sleep(1)
importlib.reload(bot)
except KeyboardInterrupt:
observer.stop()
observer.join()
def run_api():
host = '0.0.0.0'
port = 3300
while True:
print('Trying to run web interface on port:', port)
print('If it works, go to http://localhost:' + str(port))
try:
app.run(host=host, port=port)
except OSError:
print()
print('Error: Port already taken.')
port += 1
if __name__ == '__main__':
threading.Thread(target=run_api).start()
time.sleep(1)
main()