main.py

diagram.json

# (c) 2016-2021 Paul Sokolovsky, MIT license, https://github.com/pfalcon/pycopy-lib
import usocket
import select

class Request:
    pass


class Response:

    def __init__(self, f):
        self.raw = f
        self.encoding = "utf-8"
        self._cached = None

    def close(self):
        if self.raw:
            self.raw.close()
            self.raw = None
        self._cached = None

    @property
    def content(self):
        if self._cached is None:
            try:
                self._cached = self.raw.read()
            finally:
                self.raw.close()
                self.raw = None
        return self._cached

    @property
    def text(self):
        return str(self.content, self.encoding)

    def json(self):
        import ujson
        return ujson.loads(self.content)


def request(method, url, data=None, json=None, headers={}, auth=None, stream=None, parse_headers=True):
    redir_cnt = 1
    if json is not None:
        assert data is None
        import ujson
        data = ujson.dumps(json).encode("utf-8")

    while True:
        try:
            proto, dummy, host, path = url.split("/", 3)
        except ValueError:
            proto, dummy, host = url.split("/", 2)
            path = ""
        if proto == "http:":
            port = 80
        elif proto == "https:":
            import ussl
            port = 443
        else:
            raise ValueError("Unsupported protocol: " + proto)

        if ":" in host:
            host, port = host.split(":", 1)
            port = int(port)

        if auth is not None:
            req = Request()
            req.method = method
            req.url = url
            if not headers:
                # Fresh local dict, not a copy of anything.
                headers = {}
            req.headers = headers
            req = auth(req)
            headers = req.headers

        ai = usocket.getaddrinfo(host, port, 0, usocket.SOCK_STREAM)
        ai = ai[0]

        resp_d = None
        if parse_headers is not False:
            resp_d = {}

        s = usocket.socket(ai[0], ai[1], ai[2])
        s.setblocking(True)

        try:
            s.connect(ai[-1])

            poller = select.poll()
            poller.register(s)
            res = poller.poll(5000) # time in milliseconds
            #if not res:
            #    print(" ### socket not ready")
            #else:
            #    print(" ### socket ready: "+str(res))

            if proto == "https:":
                s = ussl.wrap_socket(s, server_hostname=host)
            s.write(b"%s /%s HTTP/1.0\r\n" % (method, path))
            if not "Host" in headers:
                s.write(b"Host: %s\r\n" % host)
            # Iterate over keys to avoid tuple alloc
            for k in headers:
                s.write(k)
                s.write(b": ")
                s.write(headers[k])
                s.write(b"\r\n")
            if json is not None:
                s.write(b"Content-Type: application/json\r\n")
            if data:
                s.write(b"Content-Length: %d\r\n" % len(data))
            s.write(b"Connection: close\r\n\r\n")
            if data:
                s.write(data)

            l = s.readline()
            #print(l)
            l = l.split(None, 2)
            status = int(l[1])
            reason = ""
            if len(l) > 2:
                reason = l[2].rstrip()
            while True:
                l = s.readline()
                if not l or l == b"\r\n":
                    break
                #print(l)

                if l.startswith(b"Transfer-Encoding:"):
                    if b"chunked" in l:
                        raise ValueError("Unsupported " + l.decode())
                elif l.startswith(b"Location:") and 300 <= status <= 399:
                    if not redir_cnt:
                        raise ValueError("Too many redirects")
                    redir_cnt -= 1
                    url = l[9:].decode().strip()
                    #print("redir to:", url)
                    status = 300
                    break

                if parse_headers is False:
                    pass
                elif parse_headers is True:
                    l = l.decode()
                    k, v = l.split(":", 1)
                    resp_d[k] = v.strip()
                else:
                    parse_headers(l, resp_d)
        except OSError:
            s.close()
            raise

        if status != 300:
            break

    resp = Response(s)
    resp.status_code = status
    resp.reason = reason
    if resp_d is not None:
        resp.headers = resp_d
    return resp


def head(url, **kw):
    return request("HEAD", url, **kw)

def get(url, **kw):
    return request("GET", url, **kw)

def post(url, **kw):
    return request("POST", url, **kw)

def put(url, **kw):
    return request("PUT", url, **kw)

def patch(url, **kw):
    return request("PATCH", url, **kw)

def delete(url, **kw):
    return request("DELETE", url, **kw)



myurequests.py

import math
import time

def forceRange(v, max ):
    # force v to be >= 0 and < max
    if v < 0:
        return v + max
    elif v >= max:
        return v - max

    return v

def decimalToHHMMSS(dec):
    hrs = int(dec)
    dec = (dec-hrs)*60
    mins = int(dec)
    dec = (dec-mins)*60
    secs = int(dec)
    return (hrs, mins, secs)

def calcSunTime(coords, isRiseTime=True, zenith = 90.8 ):

    # isRiseTime == False, returns sunsetTime

    longitude = coords['longitude']
    latitude = coords['latitude']

    TO_RAD = math.pi/180

    #1. first calculate the day of the year
    N = time.gmtime()[7]

    #2. convert the longitude to hour value and calculate an approximate time
    lngHour = longitude / 15

    if isRiseTime:
        t = N + ((6 - lngHour) / 24)
    else: #sunset
        t = N + ((18 - lngHour) / 24)

    #3. calculate the Sun's mean anomaly
    M = (0.9856 * t) - 3.289

    #4. calculate the Sun's true longitude
    L = M + (1.916 * math.sin(TO_RAD*M)) + (0.020 * math.sin(TO_RAD * 2 * M)) + 282.634
    L = forceRange( L, 360 ) #NOTE: L adjusted into the range [0,360)

    #5a. calculate the Sun's right ascension

    RA = (1/TO_RAD) * math.atan(0.91764 * math.tan(TO_RAD*L))
    RA = forceRange( RA, 360 ) #NOTE: RA adjusted into the range [0,360)

    #5b. right ascension value needs to be in the same quadrant as L
    Lquadrant  = (math.floor( L/90)) * 90
    RAquadrant = (math.floor(RA/90)) * 90
    RA = RA + (Lquadrant - RAquadrant)

    #5c. right ascension value needs to be converted into hours
    RA = RA / 15

    #6. calculate the Sun's declination
    sinDec = 0.39782 * math.sin(TO_RAD*L)
    cosDec = math.cos(math.asin(sinDec))

    #7a. calculate the Sun's local hour angle
    cosH = (math.cos(TO_RAD*zenith) - (sinDec * math.sin(TO_RAD*latitude))) / (cosDec * math.cos(TO_RAD*latitude))

    if cosH > 1:
        return {'status': False, 'msg': 'the sun never rises on this location (on the specified date)'}

    if cosH < -1:
        return {'status': False, 'msg': 'the sun never sets on this location (on the specified date)'}

    #7b. finish calculating H and convert into hours

    if isRiseTime:
        H = 360 - (1/TO_RAD) * math.acos(cosH)
    else: #setting
        H = (1/TO_RAD) * math.acos(cosH)

    H = H / 15

    #8. calculate local mean time of rising/setting
    T = H + RA - (0.06571 * t) - 6.622

    #9. adjust back to UTC
    UT = T - lngHour
    UT = forceRange( UT, 24) # UTC time in decimal format (e.g. 23.23)

    return UT



def hoursUntilSunset(coords):
    sset = calcSunTime(coords, False)
    now = time.gmtime()

    diff = sset - (now[3]+now[4]/60+now[5]/3600)

    return diff


sun.py

import time
import gc
import ujson
import myurequests as urequests


class ubot:

    def __init__(self, token):
        self.url = 'https://api.telegram.org/bot' + token
        self.commands = {}
        self.default_handler = None
        self.message_offset = 0
        self.sleep_btw_updates = 3

        messages = self.read_messages()
        if messages:
            for message in messages:
                if message['update_id'] > self.message_offset:
                    self.message_offset = message['update_id']


    def send(self, chat_id, text, parse_mode=None):
        data = {'chat_id': chat_id, 'text': text}
        if parse_mode:
            data["parse_mode"] = parse_mode

        headers = {'Content-type': 'application/json', 'Accept': 'text/plain'}
        while True:
            try:
                response = urequests.post(self.url + '/sendMessage', json=data, headers=headers)
                break
            except:
                print("Failed to do request")
                time.sleep(.5)
            
        
        status_code = response.status_code
        response.close()
        return status_code==200

    def read_messages(self):
        result = []
        self.query_updates = {
            'offset': self.message_offset,
            'limit': 10,
            'timeout': 30,
            'allowed_updates': ['message']}


        while True:
            try:
                update_messages = urequests.post(self.url + '/getUpdates', json=self.query_updates).json()
                break
            except:
                print("Failed to do request")
                time.sleep(.5)
                
        if 'result' in update_messages:
            for item in update_messages['result']:
                result.append(item)
        return result


    def listen(self):
        while True:
            self.read_once()
            time.sleep(self.sleep_btw_updates)
            gc.collect()

    def read_once(self):
        messages = self.read_messages()
        if messages:
            for message in messages:
                if message['update_id'] > self.message_offset:
                    if 'text' in message['message']:
                        self.message_handler(message)
                    self.message_offset = message['update_id']

    def register(self, command, handler):
        self.commands[command] = handler

    def set_default_handler(self, handler):
        self.default_handler = handler

    def set_sleep_btw_updates(self, sleep_time):
        self.sleep_btw_updates = sleep_time

    def message_handler(self, message):
        parts = message['message']['text'].split(' ')
        if parts[0] in self.commands:
            self.commands[parts[0]](message)
        else:
            if self.default_handler:
                self.default_handler(message)


utelegram.py

import utelegram
import sun
import time
import gc
import os
import machine
import network
import ntptime
import json


def connectNetwork():
    print("Connecting Network")
    sta_if = network.WLAN(network.STA_IF);
    sta_if.active(True)
    while not sta_if.isconnected():
        print("Not connected to Wifi, trying...")
        sta_if.disconnect()
        sta_if.connect(b"Wokwi-GUEST", "")
        for _ in range(50):
            if sta_if.isconnected():
                break
            time.sleep(0.1)
    print("Connected to Wifi!")
    try:
        ntptime.settime()
    except:
        print("Failed to retrieve current time from network")

connectNetwork()

LED_GRUEN = machine.Pin(25, machine.Pin.OUT)
LED_ROT = machine.Pin(33, machine.Pin.OUT)
LED_BLAU = machine.Pin(32, machine.Pin.OUT)
SWITCH = machine.Pin(27, machine.Pin.IN, machine.Pin.PULL_UP)

utelegram_config = {
    "token": "5052928525:AAH2hoX12jhz85_q4vx4aRPkacgXq_eiFtQ"
}

def hoursUntilSunset():
    coords = {'latitude':47.733333, 'longitude':10.316667 }
    return sun.hoursUntilSunset(coords)

class chatIdStore():
    def __init__(self, fileName):
        self.subscribedChats = []

        self.fileName = fileName
        if self.fileName in os.listdir("/"):
            self.readFromDisk()
        print("Current Subscribers in "+self.fileName+":", self.subscribedChats)

    def readFromDisk(self):
        with open("/"+self.fileName, "r") as f:
            self.subscribedChats = []
            for each in f.read().split("\n"):
                if len(each)>0:
                    try:
                        self.subscribedChats.append(int(each))
                    except:
                        print("couldn't convert", each, "to integer")

    def writeToDisk(self):
        with open("/"+self.fileName, "w") as f:
            f.write("\n".join([str(i) for i in self.subscribedChats]))

    def get(self):
        return self.subscribedChats

    def append(self, x):
        if not x in self.subscribedChats:
            self.subscribedChats.append(x)
            self.writeToDisk()

    def remove(self, x):
        if x in self.subscribedChats:
            self.subscribedChats.remove(x)
            self.writeToDisk()

subscribedChats = chatIdStore("subscribers.txt")
reminderChats = chatIdStore("reminder.txt")
authenticatedChats = chatIdStore("authenticated.txt")

class settingsStore():
    def __init__(self, fileName):
        self.values = {}
        
        self.defaults = {
            "reminderOffset": -90 # Minutes
        }
        
        self.fileName = fileName
        if self.fileName in os.listdir("/"):
            self.readFromDisk()
        
        print("Current setting in "+self.fileName+":", self.values)
        
    def readFromDisk(self):
        with open("/"+self.fileName, "r") as f:
            self.values = json.loads(f.read())
            
    def writeToDisk(self):
        with open("/"+self.fileName, "w") as f:
            f.write(json.dumps(self.values))
            
    def get(self, key):
        try:
            return self.values[key]
        except KeyError:
            return self.defaults[key]
        
    def set(self, key, val):
        self.values[key] = val
        self.writeToDisk()
        
        
settings = settingsStore("settings.json")
    
def catch_error_in_reply(reply):
    def wrapper(message):
        try:
            return reply(message)
        except Exception as e:
            bot.send(
                message['message']['chat']['id'],
                "Leider ist mir ein Fehler unterlaufen 💣 und ich kann dir die richtige Antwort nicht geben 😟"
            )
            bot.send(
                message['message']['chat']['id'],
                "*Fehlermeldung:*\n```\n"+repr(e)+"\n"+"\n```",
                parse_mode="MarkdownV2"
            )
    return wrapper
    
def with_auth(reply):
    def wrapper(message):
        if message["message"]["chat"]["id"] in authenticatedChats.get():
            return reply(message)
        else:
            bot.send(
                message['message']['chat']['id'],
                "Du musst mir erst das Codewort sagen, damit ich mit dir rede! 🤐"
            )
    return wrapper
    
@catch_error_in_reply
def reply_authenticate(message):
    chatId = message['message']['chat']['id']
    authenticatedChats.append(chatId)
    bot.send(
        chatId,
        "Du hast mir das richtige Codewort gesagt! 😀"
    )
    reply_greeting(message)

@with_auth
@catch_error_in_reply
def reply_greeting(message):
    chatId = message['message']['chat']['id']
    subscribedChats.append(chatId)
    reminderChats.append(chatId)
    bot.send(
        chatId,
        "Hallo "+message['message']['chat']['first_name']+""", ich bin der Hühnerstall-Bot! 🐔 Ich kann diese Befehle:
        
/liveEin und /liveAus:
Live-Benachrichtigungen einstellen, wenn die Tür geschlossen/geöffnet wird

/erinnerungEin und /erinnerungAus:
Erinnerung zum schließen der Tür verwalten

/frueher und /spaeter:
Erinnerungszeitpunkt um 15 Minuten verstellen

/status:
Türstatus und Einstellungen anzeigen
        """
    )

@with_auth
@catch_error_in_reply
def reply_subscribe_on(message):
    chatId = message['message']['chat']['id']
    subscribedChats.append(chatId)
    bot.send(
        chatId,
        "Du bekommst jetzt immer eine Benachrichtigung, wenn die Hühnerstalltür geöffnet oder geschlossen wird 🚪"
    )

@with_auth
@catch_error_in_reply
def reply_subscribe_off(message):
    chatId = message['message']['chat']['id']
    subscribedChats.remove(chatId)
    bot.send(
        chatId,
        "Du bekommst jetzt keine Benachrichtigungen mehr, wenn die Hühnerstalltür geöffnet oder geschlossen wird 🤫"
    )

@with_auth
@catch_error_in_reply
def reply_reminder_on(message):
    chatId = message['message']['chat']['id']
    reminderChats.append(chatId)
    bot.send(
        chatId,
        "Du bekommst jetzt immer eine Benachrichtigung, wenn es dunkel wird und die Hühnerstalltür noch offen ist 🦊🙅"
    )

@with_auth
@catch_error_in_reply
def reply_reminder_off(message):
    chatId = message['message']['chat']['id']
    reminderChats.remove(chatId)
    bot.send(
        chatId,
        "Du bekommst jetzt KEINE Benachrichtigungen mehr, wenn es dunkel wird und die Hühnerstalltür noch offen ist 🤫"
    )
    
@with_auth
@catch_error_in_reply
def reply_remind_later(message):
    settings.set("reminderOffset", settings.get("reminderOffset")+15)
    reply_status(message)
    
@with_auth
@catch_error_in_reply
def reply_remind_earlier(message):
    settings.set("reminderOffset", settings.get("reminderOffset")-15)
    reply_status(message)

@with_auth
@catch_error_in_reply
def reply_status(message):
    txt = "So schaut's aus:"
    
    #now = time.gmtime()
    #txt += f"\n\nGerade ({now[3]:02d}:{now[4]:02d}) ist die Tür "
    
    txt += "\n\nGerade ist die Tür "
    
    if SWITCH.value():
        txt += "offen. "
    else:
        txt += "zu. "
    
    offset = settings.get("reminderOffset")
    h,m,s = sun.decimalToHHMMSS(abs(offset/60))
    
    txt += "\n\nIch soll "
    
    if offset!=0:
        if h>0:
            txt += f"{h} Stunde(n) "
        if m>0:
            if h>0:
                txt += "und "
            txt += f"{m} Minuten "
    
    if offset<0:
        txt += "vor "
    elif offset>0:
        txt += "nach "
    else:
        txt += "zum "
        
    txt += "Sonnenuntergang bei offener Türe warnen. "
    
    
    t = hoursUntilSunset()
    #print(t)
    h,m,s = sun.decimalToHHMMSS(abs(t))
    if t>=0:
        tense = "geht unter in"
    else:
        tense = "ging unter vor"
    txt += "\n\nDie Sonne {} {} Stunde(n) und {} Minute(n) 🌅".format(tense,h,m)
    
    
    txt += "\n\nDu, "+message['message']['chat']['first_name']+", wirst "
    if not message['message']['chat']['id'] in reminderChats.get():
        txt += "NICHT "
    txt += "benachrichtigt, wenn es dunkel wird und die Hühnerstalltür noch offen ist"
    
    bot.send(
        message['message']['chat']['id'],
        txt
    )

while True:
    try:
        print("creating bot")
        bot = utelegram.ubot(utelegram_config['token'])
        break
    except Exception as e:
        print("Failed:", e)
        print("Trying again in 3s")
        time.sleep(3)
        continue
print("registering ping function")
bot.register('/liveEin', reply_subscribe_on)
bot.register('/liveAus', reply_subscribe_off)
bot.register('/erinnerungEin', reply_reminder_on)
bot.register('/erinnerungAus', reply_reminder_off)
bot.register('/spaeter', reply_remind_later)
bot.register('/frueher', reply_remind_earlier)
bot.register('/status', reply_status)
bot.register('/spiegelei', reply_authenticate)
print("setting default handler")
bot.set_default_handler(reply_greeting)
print("listening")

try:
    oldSwitchState = SWITCH.value()
except:
    oldSwitchState = None
    
lastDayDoorWasHandeled = None

while True:
    try:
        LED_BLAU.value(1)
        #print("reading bot...")
        bot.read_once()
        #print("done")
        for _ in range(6):
            nowSwitchState = SWITCH.value()
            
            # Check whether door needs to be closed
            if lastDayDoorWasHandeled != time.gmtime()[7]:
                # It is dark already
                if hoursUntilSunset()+settings.get("reminderOffset")/60 < 0:
                    if nowSwitchState: # Open, Remind!!
                        for chatId in reminderChats.get():
                            bot.send(
                                chatId,
                                "📢 Achtung Achtung ⚠️ Die Hühnerstalltür ist noch offen! Nicht, dass der Fuchs kommt! 🍗🦊"
                            )
                        lastDayDoorWasHandeled = time.gmtime()[7]
                    else:
                        # Door already closed, all good for today
                        lastDayDoorWasHandeled = time.gmtime()[7]
                        
            
            
            if nowSwitchState: # Open
                LED_GRUEN.value(0)
                LED_ROT.value(1)                
            else: # Closed
                LED_GRUEN.value(1)
                LED_ROT.value(0)
            if not nowSwitchState==oldSwitchState:
                for chatId in subscribedChats.get():
                    if nowSwitchState:
                        txt = "Tür ist gerade AUF gegangen!"
                    else:
                        txt = "Tür ist gerade ZU gegangen!"
                    bot.send(
                        chatId,
                        txt
                    )
                oldSwitchState = nowSwitchState
            time.sleep(.5)
        gc.collect()
    except Exception as e:
        print("Fail in main loop:", e, "trying again in 3 sec")
        connectNetwork()
        for _ in range(9):
            LED_BLAU.value(0)
            time.sleep(.333)
            LED_BLAU.value(1)
            time.sleep(.333)
