poobrains/example.py

#!/usr/bin/env python

import math
import random
import json
import requests
import datetime
import flask
import numpy
import poobrains

from flask import redirect


app = poobrains.app

@app.route('/')
def front():
    return redirect(News.url())


class TestSubForm(poobrains.form.Fieldset):

    oink = poobrains.form.fields.Text(label="OMGWTF")
    foo = poobrains.form.fields.Text(label="SUBfoo")
    submit = poobrains.form.Button('submit', label="SUBSUBMIT") 


@app.expose('/form')
class TestForm(poobrains.form.Form):

    foo = poobrains.form.fields.Text()
    bar = TestSubForm()
    optin = poobrains.form.fields.Checkbox(label="Opt-in", default=False, required=True)
    radio = poobrains.form.fields.Radio(type=poobrains.form.types.INT, choices=[(1, 'One'), (5, 'Five'), (23, 'Twentythree'), (42, 'Fortytwo')])
    multicheck = poobrains.form.fields.Checkbox(label="Check 'em", type=poobrains.form.types.STRING, choices=[('dubs', 'dubs'), ('trips', 'TRIPS'), ('quads', 'QUADS!1!!!!')], multi=True)
    completeme = poobrains.form.fields.Text(label="Lookit me, I can autocomplete without JS!", choices=[('Mr. Foo', 'foo'), ('Mr. Florb', 'florb'), ('Ms. Bar', 'bar')])
    ranged = poobrains.form.fields.Range()
    trigger = poobrains.form.Button('submit', label='Hit me!')
    fail = poobrains.form.Button('submit', label='Fail')

    def validate(self, submit):

        e = poobrains.errors.CompoundError()
        if submit == 'fail':
            e.append(poobrains.errors.ValidationError("FAIL!", self['foo']))

        if len(e):
            raise e

    def process(self, submit):

        flask.flash('TestForm.process called!')
        return self


@app.expose('/news', mode='full')
class News(poobrains.commenting.Commentable):

    """
    A *news article*.
    Has a nice human-readable title and markdown-enabled text.

    Also, here's a random list of things to test markdown in class docstrings:

    * Jane Fondas Legwarmers
    * Gary Buseys Glass-eye
    * Peter Thiels head on a pike
    """

    class Meta:
        search_fields = ['title', 'name', 'text']

    title = poobrains.storage.fields.CharField()
    text = poobrains.md.MarkdownField()


@app.expose('/paste', mode='full', title='Copypasta')
class Paste(poobrains.tagging.Taggable):

    type = poobrains.storage.fields.CharField()
    text = poobrains.storage.fields.TextField()


@app.site.box('menu_main')
def menu_main():

    menu = poobrains.rendering.Menu('main')

    try:
        News.permissions['read'].check(flask.g.user)
        menu.append(News.url(), 'News')
    except poobrains.auth.AccessDenied:
        pass
    
    try:
        Paste.permissions['read'].check(flask.g.user)
        menu.append(Paste.url(), 'Pastes')
    except poobrains.auth.AccessDenied:
        pass

    for url, caption in poobrains.auth.Page.main_menu_entries():
        menu.append(url, caption)

    return menu


class NonExposed(poobrains.auth.Administerable):

    text = poobrains.storage.fields.TextField()


class NonExposedB(NonExposed):
    pass

class AVeryVeryLongNameToTestMenuPositioning(poobrains.auth.Administerable):
    
    florp = poobrains.storage.fields.BooleanField()


#class MultiPK(poobrains.auth.Administerable):
#
#    class Meta:
#        primary_key = peewee.CompositeKey('pk_a', 'pk_b')
#
#    pk_a = poobrains.storage.fields.IntegerField()
#    pk_b = poobrains.storage.fields.IntegerField()
#
#
#class NestedHandle(poobrains.auth.Administerable):
#
#    class Meta:
#        primary_key = peewee.CompositeKey('foreign', 'local')
#
#    foreign = poobrains.storage.fields.ForeignKeyField(MultiPK)
#    local = poobrains.storage.fields.CharField()


@app.site.listing(NonExposedB, '/custom', mode='full', title='Custom Listing')
def list_nonexposed(listing):

    return listing


class NEO_Approaches(poobrains.analysis.EphemeralDataset):

    title='NASA Near Earth Orbits'

    @classmethod
    def load(cls, spkid: int = 2099942): # default spkid for apophis

        try:
            spkid = int(spkid)
        except ValueError:
            raise poobrains.errors.ValidationError("Invalid SPK ID; must be an integer.")

        response = requests.get('https://api.nasa.gov/neo/rest/v1/neo/%d?api_key=DEMO_KEY' % spkid)

        if response.status_code != 200:
            raise poobrains.errors.ValidationError("NASA API responded with error code %d." % response.status_code)

        data = json.loads(response.text)

        ds = cls()
        ds.title = "Close approaches of %s" % (data['name_limited'] if 'name_limited' in data else data['name'])
        ds.description: "**%s** belongs to orbit class **%s**; %s" % (data['name'], data['orbital_data']['orbit_class']['orbit_class_type'], data['orbital_data']['orbit_class']['orbit_class_description'])

        ds.plot_data = {
            'kind': 'LinePlot',
            'layers': {
                'approaches': {
                    'x': 'time',
                    'y': 'distance'
                }
            }
        }

        ds['time'] = {
            'title': 'Time',
            'description': 'Time of close approach',
            'dtype': 'datetime64',
            'color': None,
            'observations': {},
        }

        ds['distance'] = {
            'title': 'Distance (km)',
            'description': 'Distance of close approach',
            'dtype': 'float64',
            'color': None,
            'observations': {},
        }

        for index, observation in enumerate(data['close_approach_data']):

            observation_time = numpy.datetime64(datetime.datetime.fromtimestamp(observation['epoch_date_close_approach'] / 1000))
            observation_distance = numpy.float64(observation['miss_distance']['kilometers'])

            ds['time']['observations'][index] = observation_time
            ds['distance']['observations'][index] = observation_distance

        return ds


class Stock_Weekly(poobrains.analysis.EphemeralDataset):

    title = 'Stock (weekly)'

    @classmethod
    def load(cls, symbol='FB'):

        response = requests.get('https://www.alphavantage.co/query?function=TIME_SERIES_WEEKLY&symbol=%s&apikey=%s' % (symbol, app.config['ALPHAVANTAGE_API_KEY']))

        if response.status_code != 200:
            raise poobrains.errors.ValidationError("AlphaVantage API responded with error code %d." % response.status_code)

        data = json.loads(response.text)

        dates = [x for x in data['Weekly Time Series'].keys()]
        first_date = dates[-1]
        last_date = dates[0]

        ds = cls()
        ds.title = symbol
        ds.description = f"Data for stock symbol **{symbol}** from *{first_date}* to *{last_date}*."

        ds.plot_data = {
            'kind': 'LinePlot',
            'layers': {
                symbol: {
                    'x': 'time',
                    'y': 'price'
                }
            }
        }

        ds['time'] = {
            'title': 'Time',
            'description': f"Data for stock symbol **{symbol}** from *{first_date}* to *{last_date}*.",
            'dtype': 'datetime64',
            'color': None,
            'observations': {}
        }

        ds['price'] = {
            'title': 'Price ($)',
            'description': f"Data for stock symbol **{symbol}** from *{first_date}* to *{last_date}*.",
            'dtype': 'float64',
            'color': None,
            'observations': {}
        }

        for index, (datestring, datapoint) in enumerate(data['Weekly Time Series'].items()):

            y, m, d = [int(x) for x in datestring.split('-')]
            date = datetime.datetime(y, m, d)

            ds['time']['observations'][index] = numpy.datetime64(date)
            ds['price']['observations'][index] = numpy.float64(datapoint['4. close'])
            #error_lower = y - float(datapoint['3. low'])
            #error_upper = y - float(datapoint['2. high'])

        return ds


class ConstrainedRandom(poobrains.analysis.EphemeralDataset):

    title = 'Constrained random'

    @classmethod
    def load(cls, magnitude: int = 2, length: int = 16):

        magnitude, length = int(magnitude), int(length)

        ranges = []
        for i in range(0, magnitude):
            ranges.append(sorted((random.random(), random.random())))

        ds = cls()
        ds.title = f"Constrained random ({magnitude}, {length})"
        ds.description = f"ConstrainedRandom of magnitude {magnitude} and length {length}."

        ds.plot_data = {
            'kind': 'LinePlot',
            'layers': {
                'plot': {
                    'x': 'x',
                    'y': 'y'
                }
            }
        }

        ds['x'] = {
            'title': "X",
            'description': "Equidistant steps for X-axis.",
            'dtype': 'int64',
            'color': None,
            'observations': {},
        }

        ds['y'] = {
            'title': "Y",
            'description': "Random values for Y-axis.",
            'dtype': 'float64',
            'color': None,
            'observations': {},
        }

        for index, x in enumerate(range(0, length)):

            y = random.random()
            for r in ranges:
                y = r[0] + (r[1] - r[0]) * y

            r = ranges[round((len(ranges) - 1) * random.random())] # select random range
            y = r[0] + (r[1] - r[0]) * y

            y *= math.sin(2 * math.pi * float(i) / length)

            ds['x']['observations'][index] = numpy.int64(x)
            ds['y']['observations'][index] = numpy.float64(y)

        return ds


class Sine(poobrains.analysis.EphemeralDataset):

    title = 'Sine'

    @classmethod
    def load(cls, length: int=10):

        ds = cls()
        ds.title = f"Sine ({length})"
        ds.description = f"A full sine wave out of {length} points."
        
        ds.plot_data = {
            'kind': 'LinePlot',
            'layers': {
                'plot': {
                    'x': 'x',
                    'y': 'y'
                }
            }
        }

        inc = (2 * math.pi) / (length - 1)

        ds['x'] = {
            'title': "X",
            'description': f"x for `sin(x)`; spaced at equidistance intervals of {inc}.",
            'dtype': 'float64',
            'color': None,
            'observations': {}
        }

        ds['y'] = {
            'title': "Y",
            'description': "Result of `sin(x)`",
            'dtype': 'float64',
            'color': None,
            'observations': {}
        }

        x = 0
        for index in range(0, length):

            ds['x']['observations'][index] = numpy.float64(x)
            ds['y']['observations'][index] = numpy.float64(math.sin(x))
            x += inc

        return ds


#class RandomMap(poobrains.analysis.geo.EphemeralGeoData):
#
#    @classmethod
#    def load(cls):
#
#        ds = cls()
#
#        data = poobrains.analysis.geo.geojson.MultiPolygon()
#
#        for poly_num in range(0, random.randrange(1, 8)):
#
#            points = []
#
#            for point_num in range(0, random.randrange(3, 7)):
#                points.append((random.randrange(-180, 181),random.randrange(-75, 76)))
#
#            data.coordinates.append([points])
#
#        ds.data = poobrains.analysis.geo.geojson.dumps(data)
#
#        return ds


if __name__ == '__main__':

    #app.run()
    poobrains.app.main()