使用 tweepy 的 MAX7219 推特流媒体

Question

我正在尝试使用 tweepy 和 Raspberry Pi 在 MAX7219 4 合 1 点阵上显示 twitter 流。到目前为止，我已经能够：

• 在矩阵上打印一条消息
• 创建一个流媒体过滤包含特定单词的推文。

但我不确定如何将所有内容统一到一个脚本中。

这是我用来在矩阵中打印的代码：

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import re
import time
import argparse

from luma.led_matrix.device import max7219
from luma.core.interface.serial import spi, noop
from luma.core.render import canvas
from luma.core.virtual import viewport
from luma.core.legacy import text, show_message
from luma.core.legacy.font import proportional, CP437_FONT, TINY_FONT, SINCLAIR_FONT, LCD_FONT

t_end = time.time() + 30

while time.time() < t_end:

    def demo(n, block_orientation, rotate):

        serial = spi(port=0, device=0, gpio=noop())
        device = max7219(serial, width=8, height=32, block_orientation=block_orientation, rotate=rotate or 1)    
        print("Matriz creada")

        msg = "FUNDACION HORIZONTAL"
        print("Mensaje: " + "'" + msg + "'" + " compartido")
        show_message(device, msg, fill="white", font=proportional(CP437_FONT))
        time.sleep(1)



    if __name__ == "__main__":
        parser = argparse.ArgumentParser(description='matrix_demo arguments',
            formatter_class=argparse.ArgumentDefaultsHelpFormatter)

        parser.add_argument('--cascaded', '-n', type=int, default=1, help='Number of cascaded MAX7219 LED matrices')
        parser.add_argument('--block-orientation', type=int, default=0, choices=[0, 90, -90], help='Corrects block orientation when wired vertically')
        parser.add_argument('--rotate', type=int, default=0, choices=[0, 1, 2, 3], help='Rotate display 0=0°, 1=90°, 2=180°, 3=270°')

        args = parser.parse_args()

        try:
            demo(args.cascaded, args.block_orientation, args.rotate)
        except KeyboardInterrupt:
            pass

这是我用来获取流的：

from __future__ import absolute_import, print_function

from tweepy.streaming import StreamListener
from tweepy import OAuthHandler
from tweepy import Stream

consumer_key=""
consumer_secret=""

access_token=""
access_token_secret=""

class StdOutListener(StreamListener):

    def on_status(self, status):
        print(status.text)
        return True

    def on_error(self, status):
        print(status)

if __name__ == '__main__':
    l = StdOutListener()
    auth = OAuthHandler(consumer_key, consumer_secret)
    auth.set_access_token(access_token, access_token_secret)

    stream = Stream(auth, l)
    stream.filter(track=['#livinglab']

我尝试用状态填充 msg：

#!/usr/bin/env python
# -*- coding: utf-8 -*-
from __future__ import absolute_import, print_function

import re
import time
import argparse

from tweepy.streaming import StreamListener
from tweepy import OAuthHandler
from tweepy import Stream

consumer_key=""
consumer_secret=""

access_token=""
access_token_secret=""

from luma.led_matrix.device import max7219
from luma.core.interface.serial import spi, noop
from luma.core.render import canvas
from luma.core.virtual import viewport
from luma.core.legacy import text, show_message
from luma.core.legacy.font import proportional, CP437_FONT, TINY_FONT, SINCLAIR_FONT, LCD_FONT

class StdOutListener(StreamListener):

    def on_status(self, status):
        print(status.text)
        msg = status.text.encode('utf-8')
        return True

    def demo(n, block_orientation, rotate):

        serial = spi(port=0, device=0, gpio=noop())
        device = max7219(serial, width=8, height=32, block_orientation=block_orientation, rotate=rotate or 1)    
        print("Matriz creada")        

        print("Mensaje: " + "'" + msg + "'" + " compartido")
        show_message(device, msg, fill="white", font=proportional(CP437_FONT))
        time.sleep(1)

    def on_error(self, status):
            print(status)

if __name__ == "__main__":
    l = StdOutListener()
    auth = OAuthHandler(consumer_key, consumer_secret)
    auth.set_access_token(access_token, access_token_secret)

    stream = Stream(auth, l)
    stream.filter(track=['#livinglab'])

    parser = argparse.ArgumentParser(description='matrix_demo arguments',
        formatter_class=argparse.ArgumentDefaultsHelpFormatter)

    parser.add_argument('--cascaded', '-n', type=int, default=1, help='Number of cascaded MAX7219 LED matrices')
    parser.add_argument('--block-orientation', type=int, default=0, choices=[0, 90, -90], help='Corrects block orientation when wired vertically')
    parser.add_argument('--rotate', type=int, default=0, choices=[0, 1, 2, 3], help='Rotate display 0=0°, 1=90°, 2=180°, 3=270°')

    args = parser.parse_args()

    try:
        demo(args.cascaded, args.block_orientation, args.rotate)
    except KeyboardInterrupt:
        pass

有什么想法吗？

Answer 1

我更改了我的 on_status 函数，从矩阵中调用了 demo 函数并且它起作用了：

def on_status(self, status):
  print(status.text)

  def demo(n, block_orientation, rotate):
    serial = spi(port=0, device=0, gpio=noop())
    device = max7219(serial, width=8, height=32, block_orientation=block_orientation, rotate=rotate or 1)
    print("Matriz creada")

    msg = status.text
    print("Mensaje: " + "'" + msg + "'" + "compartido")
    show_message(device, msg, fill="white", font=proportional(CP437_FONT))

  demo(1, 0, 0)
  return True

def on_error(self, status):
  print(status)