Content import sys import time from hashlib import sha1 import gc import system import ubinascii import network import wifi import display import buttons from umqtt.simple import MQTTClient import mch22 import nvs import machine import neopixel # catch everything and print error on display try: def setleds(r,g,b): for led in range(5): np[led] = (r, g, b) np.write() def connectToWifi(): display.drawFill(display.WHITE) display.drawText(30, 40, "Connecting to wifi...", 0x000000, "PermanentMarker22") display.flush() tries = 0 wifi.connect() while not wifi.status(): print(f'tries: {tries}') tries+=1 time.sleep(0.2) if wifi.wait(): time.sleep(0.2) print("Connected!") break else: display.drawFill(display.WHITE) display.drawText(30, 40, "No Wifi No Winks!", 0x000000, "PermanentMarker22") display.drawText(30, 100, f'Connecting { str(tries) } again...', 0x000000, "PermanentMarker22") display.flush() time.sleep(0.2) class WinkeKatze: # |\_/| # (. .) # =w= (\ # / ^ \// # (|| ||) # ,""_""_ . # |\_/| # (. .) # =w= # / ^ \\ # (|| ||)\\ # ,""_""_(/. def draw_cat_text(self, footer="https://devlol.org - press A to wink all other cats"): i=2 if footer: i +=1 display.drawText(8, 12*i, footer, 0x000000, self.font) nick_pos_x = 180 if self.clients > 0: display.drawText(0, 12, "%d online now" % (self.clients), 0x000000, self.font) if self.nick: display.drawText(nick_pos_x, 90, "%s winks" % (str(self.nick)), 0x000000, self.font) if self.nick1: display.drawText(nick_pos_x, 80, "%s winks" % (str(self.nick1)), 0x000000, self.font) if self.nick2: display.drawText(nick_pos_x, 70, "%s winks" % (str(self.nick2)), 0x000000, self.font) if self.nick3: display.drawText(nick_pos_x, 60, "%s winks" % (str(self.nick3)), 0x000000, self.font) if self.nick4: display.drawText(nick_pos_x, 50, "%s winks" % (str(self.nick4)), 0x000000, self.font) display.drawText(90, 15, self.owner, 0x000000, "PermanentMarker22") display.drawText(0, 0, str(self.motd), 0x000000, self.font) def draw_cat(self,cat='up',footer="https://devlol.org - press A to wink all other cats"): winkpos_x = 10 winkpos_y = 90 framesleep=0.2 for frame in [1, 4, 5, 3, 2, 3, 5, 4, 1]: display.drawFill(display.WHITE) self.draw_cat_text() display.drawPng(winkpos_x, winkpos_y, f'{ APP_PATH }/wink{frame}.png') display.flush() time.sleep(framesleep) def new_wink(self,msg): self.nick4 = self.nick3 self.nick3 = self.nick2 self.nick2 = self.nick1 self.nick1 = self.nick self.nick = msg.decode("utf-8") def sub_cb(self,topic,msg): print (f'received mqtt message topic: { topic }, payload: { msg }') if topic.endswith('/nick'): if msg.decode("utf-8") == self.nick: pass else: print ("setting received nick to %s" % msg.decode("utf-8")) self.new_wink(msg) list = [] hash = ubinascii.hexlify(sha1(self.nick).digest()) for i in range(6): # reverse because of reverse() list.append(0) list.append(int(int(hash[i*6+4:i*6+6], 16)/7)) list.append(int(int(hash[i*6+2:i*6+4], 16)/7)) list.append(int(int(hash[i*6:i*6+2], 16)/7)) list.reverse() colors = bytes(list) self.wink(colors) if topic.endswith('/motd'): self.motd = msg.decode("utf-8") self.draw_cat() for i in range(2): #badge.leds_send_data(bytes([0x50,0x00,0x00,0x00]*6)) setleds(0,150,0) time.sleep(0.1) #badge.leds_send_data(bytes([0x00,0x00,0x00,0x00]*6)) setleds(0,0,0) time.sleep(0.1) if topic.endswith('/num_clients'): try: self.clients = int(msg.decode("utf-8")) except: pass self.draw_cat() def wink(self, colors): #badge.leds_send_data(colors) for count in range(2): self.draw_cat(cat="down") #badge.vibrator_activate(127) #badge.leds_send_data(bytes([0x50,0x00,0x00,0x00]*6)) setleds(0,150,0) time.sleep(0.5) self.draw_cat() #badge.leds_send_data(bytes([0x00,0x00,0x00,0x00]*6)) setleds(0,0,0) time.sleep(0.5) def btn_a(self,pressed): if pressed: print (f'publish nick { self.owner } to topic { self.topic }/nick') self.c.publish(("%s/nick" % self.topic), self.owner) print ("publish WINK3 to topic %s" % ( self.topic)) self.c.publish(("%s" % self.topic), "WINK3") def btn_b(self,pressed): pass def btn_up(self,pressed): pass def btn_down(self,pressed): pass def btn_left(self,pressed): pass def btn_right(self,pressed): pass def btn_start(self,pressed): pass def btn_select(self,pressed): pass def btn_home(self,pressed): if pressed: mch22.exit_python() def __init__(self, font="roboto_regular12",server='mqtt.devlol.org',topic='devlol/h19/mainroom/winkekatze'): self.cat = [' |\_/| ', ' (. .) ', ' =w= (\ ', ' / ^ \// ', ' (|| ||) ', ' ,""_""_ .'] self.catdown =[' |\_/| ', ' (. .) ', ' =w= ', ' / ^ \\\\ ', ' (|| ||)\\', ' ,""_""_(/.'] self.font = font #self.clientname = 'SHA2017Badge' + str(ubinascii.hexlify(uos.urandom(5))) self.clientname = 'MCH2022Badge_cat' + ubinascii.hexlify(network.WLAN().config("mac")).decode("UTF-8") self.server = server self.topic = topic self.clients = 0 self.version = 46 self.nick = '' self.nick1 = '' self.nick2 = '' self.nick3 = '' self.nick4 = '' self.motd = 'Alle 5 Minuten verliebt sich ein Hacker ueber die WinkekatzenApp' print('starting') #badge.init() #badge.leds_enable() #workaround for emulator try: self.owner = nvs.nvs_getstr('owner','nickname') except: self.owner ="anonycat" self.statustopic = (f'{ self.topic }/clientstatus/{ self.owner }') print("connecting to wifi 1 ...") #so we can exit anytime while connecting buttons.attach(buttons.BTN_HOME, self.btn_home) connectToWifi() print (f'connecting to mqtt server:{self.server} clientname:{self.clientname}') self.c = MQTTClient(self.clientname,self.server) #,keepalive=60) # bug in umqtt, won't work self.c.set_callback(self.sub_cb) self.c.set_last_will(self.statustopic, '', retain=False) self.c.connect() self.c.subscribe(f'{self.topic}/nick') self.c.subscribe(f'{self.topic}/motd') self.c.subscribe(f'{self.topic}/num_clients') self.c.publish(self.statustopic, str(self.version), retain=True) #ugfx.clear(ugfx.BLACK) #ugfx.flush #ugfx.clear(ugfx.WHITE) #ugfx.flush buttons.attach(buttons.BTN_A, self.btn_a) buttons.attach(buttons.BTN_B, self.btn_b) buttons.attach(buttons.BTN_START, self.btn_start) buttons.attach(buttons.BTN_SELECT, self.btn_select) buttons.attach(buttons.BTN_DOWN, self.btn_down) buttons.attach(buttons.BTN_RIGHT, self.btn_right) buttons.attach(buttons.BTN_UP, self.btn_up) buttons.attach(buttons.BTN_LEFT, self.btn_left) self.draw_cat() # last_status_update = time.time() # status_update_interval = 60 while True: time.sleep(0.5) if wifi.status(): self.c.check_msg() else: break # if (time.time() > last_status_update + status_update_interval): # last_status_update = time.time() # self.c.publish(self.statustopic, str(self.version), retain=True) APP_PATH = "/".join(__file__.split("/")[:-1]) # Add Application path to sys path to import modules as if they were root modules sys.path.append(APP_PATH) np = neopixel.NeoPixel(machine.Pin(5), 5) wk = WinkeKatze() except Exception as e: print("ERROR : "+ str(e)) display.drawText(0, 0, str(e), 0x000000, "roboto_regular12") #display.drawText(0, 20, str(sys.exec_info()[2]), 0x000000, "roboto_regular12") display.flush() # still waiting anyway while True: time.sleep(1)