86 lines
2.6 KiB
Python
86 lines
2.6 KiB
Python
import inspect
|
|
import logging
|
|
import statistics
|
|
from dataclasses import asdict
|
|
from datetime import timedelta
|
|
from typing import List
|
|
|
|
import requests_cache
|
|
from flask import Flask, jsonify
|
|
|
|
from .strategies import miloStrats, iss, cars_in_traffic, tide_strat, upstairs_neighbour
|
|
from .util import Context
|
|
|
|
logger = logging.getLogger(__name__)
|
|
app = Flask(__name__)
|
|
|
|
requests_cache.install_cache("requests_cache", expire_after=timedelta(minutes=10))
|
|
|
|
|
|
strategies = {
|
|
# name: (weight, probability function)
|
|
"tv2news": miloStrats.tv2newsStrat,
|
|
"australia": miloStrats.australiaStrat,
|
|
"camera": miloStrats.camImgStrat,
|
|
"iss": iss.night_on_iss,
|
|
"cars_in_traffic": cars_in_traffic.cars_in_traffic,
|
|
"tide": tide_strat.is_tide,
|
|
"upstairs_neighbour": upstairs_neighbour.check_games,
|
|
}
|
|
|
|
|
|
@app.route("/", methods=["GET", "POST"])
|
|
def probabilities():
|
|
phone_data = {} # TODO: get from POST request
|
|
context = Context(**phone_data)
|
|
|
|
predictions: List[dict] = []
|
|
for name, strategy in strategies.items():
|
|
try:
|
|
prediction = strategy(context)
|
|
except Exception as e:
|
|
logger.warning("Strategy %s failed: %s", name, e)
|
|
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
|
|
consensus_weight_sum = sum(p["weight"] for p in predictions if p["night"] == night)
|
|
for prediction in predictions:
|
|
# If this prediction agrees with the consensus it contributed
|
|
if prediction["night"] == night:
|
|
prediction["contribution"] = prediction["weight"] / consensus_weight_sum
|
|
else:
|
|
prediction["contribution"] = 0.0
|
|
|
|
return jsonify({
|
|
"predictions": predictions,
|
|
"weighted_probabilities_mean": mean,
|
|
"weighted_probabilities_median": median,
|
|
"night": night,
|
|
})
|
|
|
|
|
|
def main():
|
|
app.run(host='0.0.0.0')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|