13. Trabajo en grupo 2015 (Datos AEMET)¶

%matplotlib inline

import urllib
import lxml.etree as ET
import matplotlib.pyplot as plt

class PrediccionAEMET(object):
    def __init__(self):
        self._provinciaToId = _leerCodigosProvincias()
        self._prediccion = None
        self._provincia = None

    def maxTemperatura(self, provincia):
        u'''Asume provincia cadena con nombre de provincia según AEMET.
            Devuelve tupla (T, L) siendo T temperatura máxima de la predicción
            y L lista de tuplas (municipio, fecha) en la que se espera T.'''
        por_temperatura = []
        for p in self.prediccion(provincia):
            por_temperatura += zip(p.temperaturaMax(),
                                   [p.nombre]*len(p.fechas),
                                   p.fechas)
        por_temperatura = sorted(por_temperatura, reverse=True)
        T = por_temperatura[0][0]
        L = sorted([(p[1], p[2]) for p in por_temperatura if T == p[0]])
        return T, L

    def graficaSensacionTermica(self, municipio, formato):
        u'''Asume municipio, formato cadenas.
            Genera una gráfica de sensación térmica para el municipio indicado. '''
        if not self._prediccion:
            raise ValueError('No hay provincia')
        prediccion = [ p for p in self._prediccion if p.nombre == municipio ]
        if len(prediccion) < 1:
            raise ValueError('Municipio no encontrado')

        prediccion = prediccion[0]
        plt.title(u'Sensación térmica en {0}'.format(prediccion.nombre),
                  fontsize='x-large')
        plt.xlabel('Fecha')
        plt.ylabel(u'Sensación térmica (ºC)')
        plt.xticks(range(12),
                   prediccion.fechas,
                   rotation=45)

        smax = plt.plot(prediccion.sensTermicaMax(), 'ro--', label=u'máxima')
        smin = plt.plot(prediccion.sensTermicaMin(), 'bo--', label=u'mínima')

        plt2 = plt.twiny()
        plt2.axis('off')
        plt2.set_autoscale_on(False)
        plt2.set_xlim(xmax=len(prediccion.sensTermica)-1)
        horas, por_horas = self.sensacionTermicaPorHoras(prediccion)

        shoras = plt2.plot(horas, por_horas, 'g^--', label=u'estimación')

        plt.legend(handles=smax+smin+shoras,
                   loc='upper right',
                   shadow=True,
                   fontsize='large')
        outfile = u'{0}_{1}.{2}'.format(prediccion.nombre,
                                        prediccion.fechas[0],
                                        formato)
        plt.savefig(outfile)
        print 'Almacenado archivo', outfile
        plt.show()

    def prediccion(self, provincia):
        u'''Asume provincia cadena con nombre de provincia según AEMET.
            Devuelve lista de predicciones diarias'''
        id_provincia = self._provinciaToId[provincia]
        if not self._prediccion or self._provincia != id_provincia:
            self._prediccion = _leerPrediccionProvincia(id_provincia)
            self._provincia = id_provincia
        return self._prediccion

    def sensacionTermicaPorHoras(self, prediccion):
        u'''Asume self._prediccion previamente actualizado.
            Devuelve tupla de dos listas, eje X e Y de la predicción
            de sensación térmica por horas'''
        frac_dia = { '00': 0., '06': 0.25, '12': 0.5, '18': 0.75, '24': 1. }
        sens_horas = [ x[2] for x in prediccion.sensTermica  ]
        horas, por_horas = [], []
        for dia in range(len(sens_horas)):
            for p in sens_horas[dia]:
                horas.append(dia + frac_dia[p[0]])
                por_horas.append(p[1])
        return horas, por_horas

class PrediccionMunicipio(object):
    def __init__(self, root):
        self.nombre = root.find('nombre').text
        self.fechas = [ dia.get('fecha') for dia in root.iter('dia') ]
        self.sensTermica = [ self._leerMagnitud('sens_termica', dia) for dia in root.iter('dia') ]
        self.temperatura = [ self._leerMagnitud('temperatura', dia) for dia in root.iter('dia') ]

    def sensTermicaMax(self):
        return [ p[0] for p in self.sensTermica ]

    def sensTermicaMin(self):
        return [ p[1] for p in self.sensTermica ]

    def temperaturaMax(self):
        return [ p[0] for p in self.temperatura ]

    def temperaturaMin(self):
        return [ p[1] for p in self.temperatura ]

    def _leerMagnitud(self, nombre, dia):
        magnitud = dia.find(nombre)
        mmax = float(magnitud.find('maxima').text)
        mmin = float(magnitud.find('minima').text)
        horas = [ (d.get('hora'), float(d.text)) for d in magnitud.iter('dato') if d.text ]
        return (mmax, mmin, horas)

    def __str__(self):
        return '''Prediccion ({0},
            {1},
            {2},
            {3})'''.format(self.nombre,
                           self.fechas,
                           self.sensTermica,
                           self.temperatura)

def _leerCodigosProvincias():
    u'''Devuelve un diccionario que hace corresponder el nombre de la provincia al
        código numérico correspondiente según AEMET'''
    f = urllib.urlopen('http://www.aemet.es/es/eltiempo/prediccion/municipios')
    encoding = f.headers['content-type'].split('charset=')[-1]
    root = ET.HTML(unicode(f.read(), encoding))
    provincias = root.findall(".//select[@id='provincia_selector']/option")[1:]
    return  { nombre:int(x.get('value')) for x in provincias \
                for nombre in x.text.split('/') }

def _leerPrediccionProvincia(id_provincia):
    u'''Asume provincia cadena con nombre de provincia según AEMET.
        Lee de AEMET la predicción para todos sus municipios y
        devuelve una lista con todas las predicciones diarias.'''
    prediccion = []
    fallos = 0
    for municipio in range(1,10000):
        try:
            prediccion.append(_leerPrediccionMunicipio(id_provincia, municipio))
            fallos = 0
        except:
            fallos += 1
        if fallos > 5: break
    return prediccion

def _leerPrediccionMunicipio(id_provincia, id_municipio):
    u'''Asume id_provincia, id_municipio enteros.
        Devuelve lista de predicciones diarias para ese municipio.'''
    url = 'http://www.aemet.es/xml/municipios/localidad_{0:02d}{1:03d}.xml'
    url = url.format(id_provincia, id_municipio)
    f = urllib.urlopen(url)
    if f.getcode() != 200:
        raise ValueError('Failed ({0}) {1}'.format(f.getcode(), url))
    root = ET.fromstring(f.read())
    return PrediccionMunicipio(root)

from IPython.html.widgets import interact

provincias = [''] + sorted(_leerCodigosProvincias().keys())
formatos = ['', 'png', 'pdf']
aemet = PrediccionAEMET()

@interact(provincia=provincias, formato=formatos)
def datos_AEMET(provincia, formato):
    if not provincia: return

    tmax, municipios = aemet.maxTemperatura(provincia)
    print u'La máxima temperatura para los próximos días es de {0}ºC y se espera en:'.format(tmax)
    for nombre,fecha in municipios:
        print u' - {0} el día {1}'.format(nombre, fecha)

    if not formato: return
    print
    for nombre in set([m[0] for m in municipios]):
        aemet.graficaSensacionTermica(nombre, formato)