124 lines
3.4 KiB
Python
Executable File
124 lines
3.4 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
from flask import Flask, request, jsonify, send_file
|
|
from datetime import datetime
|
|
import os
|
|
import json
|
|
import uuid
|
|
import shutil
|
|
import base64
|
|
from io import BytesIO
|
|
|
|
app = Flask(__name__)
|
|
|
|
DATA_DIR = os.environ.get('DATA_DIR', '/var/tmp/themblem-research')
|
|
os.makedirs(DATA_DIR, exist_ok=True)
|
|
|
|
def image_data_url_to_binary(image_data_url):
|
|
if image_data_url.startswith('data:image/jpeg;base64,'):
|
|
return base64.b64decode(image_data_url.split(',')[1]), 'image/jpeg'
|
|
elif image_data_url.startswith('data:image/png;base64,'):
|
|
return base64.b64decode(image_data_url.split(',')[1]), 'png'
|
|
else:
|
|
raise ValueError(f"Unsupported image data URL: {image_data_url}")
|
|
|
|
@app.route('/')
|
|
def index():
|
|
return "Emblem research API\n"
|
|
|
|
@app.route('/event/<category>', methods=['POST'])
|
|
def event(category):
|
|
d = os.path.join(DATA_DIR, category, datetime.now().strftime("%Y-%m-%d"))
|
|
os.makedirs(d, exist_ok=True)
|
|
fname = datetime.now().strftime("%H%M%S") + '-' + str(uuid.uuid4())
|
|
with open(os.path.join(d, fname), 'wb') as f:
|
|
data = request.get_data()
|
|
f.write(data)
|
|
return {
|
|
"ok": True,
|
|
"message": "Event saved",
|
|
}
|
|
|
|
def get_folders():
|
|
return [
|
|
x for x in os.listdir(DATA_DIR + "/json")
|
|
if os.path.isdir(os.path.join(DATA_DIR, "json", x))
|
|
]
|
|
|
|
@app.route('/api/folders')
|
|
def folders():
|
|
x = get_folders()
|
|
ret = []
|
|
for d in x:
|
|
ret.append({
|
|
"folder": d,
|
|
"count": len(os.listdir(os.path.join(DATA_DIR, "json", d))),
|
|
})
|
|
return {
|
|
"ok": True,
|
|
"folders": ret,
|
|
}
|
|
|
|
@app.route('/api/datapoints')
|
|
def datapoints():
|
|
folder = request.args.get('folder')
|
|
folder_dir = os.path.join(DATA_DIR, "json", folder)
|
|
x = os.listdir(folder_dir)
|
|
datapoints = []
|
|
for f in x:
|
|
try:
|
|
datapoints.append(load_datapoint(os.path.join(folder_dir, f)))
|
|
except Exception as e:
|
|
print(e)
|
|
return {
|
|
"ok": True,
|
|
"datapoints": datapoints,
|
|
}
|
|
|
|
def load_datapoint(path):
|
|
with open(path, 'rb') as f:
|
|
data = json.load(f)
|
|
data["path"] = path
|
|
return data
|
|
|
|
@app.route('/api/image')
|
|
def image():
|
|
folder = request.args.get('folder')
|
|
name = request.args.get('name')
|
|
path = os.path.join(DATA_DIR, "json", folder, name)
|
|
with open(path, 'rb') as f:
|
|
data = json.load(f)
|
|
binary, mimetype = image_data_url_to_binary(data["image_data_url"])
|
|
buf = BytesIO(binary)
|
|
return send_file(buf, mimetype=mimetype)
|
|
|
|
@app.route('/api/datapoint')
|
|
def datapoint():
|
|
folder = request.args.get('folder')
|
|
name = request.args.get('name')
|
|
path = os.path.join(DATA_DIR, "json", folder, name)
|
|
return {
|
|
"ok": True,
|
|
"datapoint": load_datapoint(path),
|
|
}
|
|
|
|
def do_copy_to_folder(paths, target_folder):
|
|
for path in paths:
|
|
src = os.path.join(DATA_DIR, "json", path)
|
|
dst = os.path.join(DATA_DIR, "json", target_folder)
|
|
os.makedirs(dst, exist_ok=True)
|
|
print(f"Copying {src} to {dst}")
|
|
shutil.copy(src, dst)
|
|
|
|
@app.route('/api/copy-to-folder', methods=['POST'])
|
|
def copy_to_folder():
|
|
data = request.get_json()
|
|
paths = data.get('paths')
|
|
target_folder = data.get('target_folder')
|
|
do_copy_to_folder(paths, target_folder)
|
|
return {
|
|
"ok": True,
|
|
"message": "Datapoints copied to folder",
|
|
}
|
|
|
|
if __name__ == '__main__':
|
|
app.run(host='0.0.0.0', port=26966, debug=True) |