Productores, Consumidores y Tuberías: Desbloqueando el Poder de los Generadores

Introducción

Los generadores son una herramienta útil para resolver varios tipos de problemas de productor/consumidor y tuberías de flujo de datos. En esta sección se aborda eso.

Problemas de Productor-Consumidor

Los generadores están estrechamente relacionados con diversas formas de problemas de productor-consumidor.

## Productor
def follow(f):
 ...
    while True:
     ...
        yield line        ## Produce el valor en `line` a continuación
     ...

## Consumidor
for line in follow(f):    ## Consume el valor del `yield` anterior
 ...

yield produce valores que for consume.

Tubos de Generador

Puedes utilizar este aspecto de los generadores para configurar tuberías de procesamiento (como las tuberías de Unix).

productor → procesamiento → procesamiento → consumidor

Las tuberías de procesamiento tienen un productor de datos inicial, un conjunto de etapas de procesamiento intermedio y un consumidor final.

productor → procesamiento → procesamiento → consumidor

def producer():
 ...
    yield item
 ...

El productor es típicamente un generador. Aunque también podría ser una lista u otra secuencia. yield alimenta datos a la tubería.

productor → procesamiento → procesamiento → consumidor

def consumer(s):
    for item in s:
 ...

El consumidor es un bucle for. Recibe los elementos y hace algo con ellos.

productor → procesamiento → procesamiento → consumidor

def processing(s):
    for item in s:
 ...
        yield newitem
 ...

Las etapas de procesamiento intermedio consumen y producen elementos al mismo tiempo. Pueden modificar el flujo de datos. También pueden filtrar (descartando elementos).

productor → procesamiento → procesamiento → consumidor

def producer():
 ...
    yield item          ## produce el elemento que recibe el `procesamiento`
 ...

def processing(s):
    for item in s:      ## Viene del `productor`
 ...
        yield newitem   ## produce un nuevo elemento
 ...

def consumer(s):
    for item in s:      ## Viene del `procesamiento`
 ...

Código para configurar la tubería

a = producer()
b = processing(a)
c = consumer(b)

Notarás que los datos fluyen incrementalmente a través de las diferentes funciones.

Para este ejercicio, el programa stocksim.py debe seguir ejecutándose en segundo plano. Vas a utilizar la función follow() que escribiste en el ejercicio anterior.

Ejercicio 6.8: Configurar una tubería simple

Veamos en acción la idea de tuberías. Escribe la siguiente función:

>>> def filematch(lines, substr):
        for line in lines:
            if substr in line:
                yield line

>>>

Esta función es casi exactamente igual que el primer ejemplo de generador del ejercicio anterior, excepto que ya no está abriendo un archivo, sino que simplemente opera sobre una secuencia de líneas que se le pasa como argumento. Ahora, prueba esto:

>>> from follow import follow
>>> lines = follow('stocklog.csv')
>>> goog = filematch(lines, 'GOOG')
>>> for line in goog:
        print(line)

... espera a que aparezca la salida...

Puede tardar un poco en aparecer la salida, pero eventualmente deberías ver algunas líneas que contengan datos de GOOG.

Nota: Estos ejercicios deben realizarse simultáneamente en dos terminales separadas.

Ejercicio 6.9: Configurar una tubería más compleja

Sigue avanzando un par de pasos con la idea de tuberías, realizando más acciones.

>>> from follow import follow
>>> import csv
>>> lines = follow('stocklog.csv')
>>> rows = csv.reader(lines)
>>> for row in rows:
        print(row)

['GOOG', '1502.08', '2023-10-01', '09:37.19', '1.83', '1500.25', '1502.08', '1500.25', '731']
['AAPL', '252.33', '2023-10-01', '09:37.19', '1.83', '250.50', '252.33', '250.50', '681']
['GOOG', '1502.09', '2023-10-01', '09:37.21', '1.84', '1500.25', '1502.09', '1500.25', '734']
['AAPL', '252.34', '2023-10-01', '09:37.21', '1.84', '250.50', '252.34', '250.50', '684']
['GOOG', '1502.10', '2023-10-01', '09:37.23', '1.85', '1500.25', '1502.10', '1500.25', '738']
['AAPL', '252.35', '2023-10-01', '09:37.23', '1.85', '250.50', '252.35', '250.50', '688']
...

Bueno, eso es interesante. Lo que estás viendo aquí es que la salida de la función follow() se ha conectado a la función csv.reader() y ahora estamos obteniendo una secuencia de filas divididas.

Ejercicio 6.10: Creando más componentes de tubería

Vamos a extender toda la idea a una tubería más grande. En un archivo separado ticker.py, comienza creando una función que lea un archivo CSV como lo hiciste anteriormente:

## ticker.py

from follow import follow
import csv

def parse_stock_data(lines):
    rows = csv.reader(lines)
    return rows

if __name__ == '__main__':
    lines = follow('stocklog.csv')
    rows = parse_stock_data(lines)
    for row in rows:
        print(row)

Escribe una nueva función que seleccione columnas específicas:

## ticker.py

...
def select_columns(rows, indices):
for row in rows:
yield [row[index] for index in indices]
...
def parse_stock_data(lines):
rows = csv.reader(lines)
rows = select_columns(rows, [0, 1, 4])
return rows

Ejecuta tu programa nuevamente. Deberías ver la salida reducida de esta manera:

['GOOG', '1503.06', '2.81']
['AAPL', '253.31', '2.81']
['GOOG', '1503.07', '2.82']
['AAPL', '253.32', '2.82']
['GOOG', '1503.08', '2.83']

...

Escribe funciones generadoras que conviertan tipos de datos y construyan diccionarios. Por ejemplo:

## ticker.py
...

def convert_types(rows, types):
    for row in rows:
        yield [func(val) for func, val in zip(types, row)]

def make_dicts(rows, headers):
    for row in rows:
        yield dict(zip(headers, row))
...
def parse_stock_data(lines):
    rows = csv.reader(lines)
    rows = select_columns(rows, [0, 1, 4])
    rows = convert_types(rows, [str, float, float])
    rows = make_dicts(rows, ['name', 'price', 'change'])
    return rows
...

Ejecuta tu programa nuevamente. Ahora deberías ver una corriente de diccionarios como esta:

{'name': 'GOOG', 'price': 1503.4, 'change': 3.15}
{'name': 'AAPL', 'price': 253.65, 'change': 3.15}
{'name': 'GOOG', 'price': 1503.41, 'change': 3.16}
{'name': 'AAPL', 'price': 253.66, 'change': 3.16}
{'name': 'GOOG', 'price': 1503.42, 'change': 3.17}
{'name': 'AAPL', 'price': 253.67, 'change': 3.17}

...

Ejercicio 6.11: Filtrando datos

Escribe una función que filtre datos. Por ejemplo:

## ticker.py
...

def filter_symbols(rows, names):
    for row in rows:
        if row['GOOG'] in names:
            yield row

Utiliza esto para filtrar las acciones solo a aquellas en tu cartera:

import report
import ticker
import follow
portfolio = report.read_portfolio('portfolio.csv')
rows = ticker.parse_stock_data(follow.follow('stocklog.csv'))
rows = ticker.filter_symbols(rows, portfolio)
for row in rows:
    print(row)

Discusión

Algunas lecciones aprendidas: Puedes crear varias funciones generadoras y enlazarlas para realizar el procesamiento que implica tuberías de flujo de datos. Además, puedes crear funciones que empaquetan una serie de etapas de tubería en una sola llamada a función (por ejemplo, la función parse_stock_data()).

Resumen

¡Felicitaciones! Has completado el laboratorio de Productores, Consumidores y Tuberías. Puedes practicar más laboratorios en LabEx para mejorar tus habilidades.