109 lines
3.5 KiB
Python
109 lines
3.5 KiB
Python
import inspect
|
|
import json
|
|
import statistics
|
|
import timeit
|
|
from dataclasses import asdict
|
|
from datetime import timedelta
|
|
from logging import DEBUG
|
|
from typing import List
|
|
|
|
import requests_cache
|
|
from flask import Flask, jsonify, logging, request
|
|
|
|
from .strategies import miloStrats, iss, cars_in_traffic, tide_strat, upstairs_neighbour, bing, svm_strat, battery, just_eat, steam
|
|
from .util import Context
|
|
|
|
app = Flask(__name__)
|
|
logger = logging.create_logger(app)
|
|
logger.setLevel(DEBUG)
|
|
|
|
requests_cache.install_cache("requests_cache", expire_after=timedelta(minutes=2))
|
|
|
|
|
|
strategies = {
|
|
# name: (weight, probability function)
|
|
"TV2 News": miloStrats.tv2newsStrat,
|
|
"Australia": miloStrats.australiaStrat,
|
|
"Camera Image": miloStrats.camImgStrat,
|
|
"The International Space Station": iss.night_on_iss,
|
|
"Nearby Traffic situation": cars_in_traffic.cars_in_traffic,
|
|
"Tidal Measurements": tide_strat.is_tide,
|
|
"Legends of Nighttime": upstairs_neighbour.check_games,
|
|
"Bing AI": bing.clock,
|
|
"ML Parking": svm_strat.perform_svm_pred,
|
|
"Phone Battery Level": battery.battery_level,
|
|
"Pizza Availability": just_eat.do_just_eat_strat,
|
|
"Steam Players Online": steam.dota2_players,
|
|
}
|
|
|
|
|
|
@app.route("/", methods=["GET", "POST"])
|
|
def probabilities():
|
|
if request.method == "GET":
|
|
logger.warning("GET request: using default context parameters")
|
|
context = Context()
|
|
else:
|
|
phone_data = request.get_json(force=True)
|
|
logger.debug("phone_data:\n%s", json.dumps(phone_data, indent=2))
|
|
context = Context(**phone_data["data"])
|
|
|
|
#logger.debug("Context: %s", context)
|
|
|
|
predictions: List[dict] = []
|
|
for name, strategy in strategies.items():
|
|
try:
|
|
logger.debug("Executing %s..", name)
|
|
start = timeit.default_timer()
|
|
prediction = strategy(context)
|
|
stop = timeit.default_timer()
|
|
logger.debug("Execution time for %s: %ss", name, round(stop - start, 3 ))
|
|
except Exception as e:
|
|
logger.warning("Strategy '%s' failed:", name)
|
|
logger.exception(e)
|
|
continue
|
|
|
|
predictions.append({
|
|
"name": name,
|
|
"description": inspect.getdoc(strategy),
|
|
"weight": prediction.weight,
|
|
"weighted_probability": prediction.probability * prediction.weight,
|
|
"night": prediction.probability > 0.5,
|
|
**asdict(prediction),
|
|
})
|
|
|
|
mean = statistics.mean(p["weighted_probability"] for p in predictions)
|
|
median = statistics.median(p["weighted_probability"] for p in predictions)
|
|
night = mean > 0.5
|
|
|
|
# Invert if we're in Australia
|
|
if context.in_australia:
|
|
night = not night
|
|
for prediction in predictions:
|
|
prediction["night"] = not prediction["night"]
|
|
|
|
# Calculate contributions of predictions
|
|
weight_sum = sum(p["weight"] for p in predictions)
|
|
for prediction in predictions:
|
|
prediction["contribution"] = prediction["weight"] / weight_sum
|
|
|
|
# If this prediction disagrees with the consensus it contributed negatively
|
|
if prediction["night"] != night:
|
|
prediction["contribution"] *= -1
|
|
|
|
predictions.sort(key=lambda p: (p["contribution"], p["probability"]), reverse=True)
|
|
|
|
return jsonify({
|
|
"predictions": predictions,
|
|
"weighted_probabilities_mean": mean,
|
|
"weighted_probabilities_median": median,
|
|
"night": night,
|
|
})
|
|
|
|
|
|
def main():
|
|
app.run(host='0.0.0.0', debug=True)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|