#!/usr/bin/python3 # -*- coding: utf-8 -*- # https://github.com/halcy/Mastodon.py # Command: python3 solar_bot.py import settings import os, time, datetime, subprocess, sys import asyncio from asgiref.sync import async_to_sync from forecast_solar import ForecastSolar from suntime import Sun, SunTimeException sun = Sun(settings.latitude, settings.longitude) from mastodon import Mastodon from apscheduler.schedulers.background import BackgroundScheduler scheduler = BackgroundScheduler() scheduler.start() # DB 1 is used as DB 0 is used by mastodon import redis r = redis.Redis(host='localhost', port=6379, db=1, decode_responses=True) import logging logging.basicConfig(encoding='utf-8', level=logging.INFO) gpio_lights = "4" # This needs to be uncommented on first run and then commented out after that - obviously there is a better way to do this. #Mastodon.create_app( # 'pytooterapp', # api_base_url = instance_url, # to_file = 'pytooter_clientcrednew.secret' #) def toggle_lights(state): if state == 'on': logging.info('on') subprocess.run(["/usr/local/bin/gpio", "mode", gpio_lights, "out"]) subprocess.run(["/usr/local/bin/gpio", "write", gpio_lights, "1"]) else: logging.info('off') subprocess.run(["/usr/local/bin/gpio", "write", gpio_lights, "0"]) subprocess.run(["/usr/local/bin/gpio", "mode", gpio_lights, "in"]) def turn_on_lights(): logging.info('Turning lights on') toggle_lights('on') def turn_off_lights(): logging.info('Turning lights off') toggle_lights('off') @async_to_sync async def get_solar_estimate(): async with ForecastSolar(latitude=52.53, longitude=-0.75, declination=45, azimuth='SE', kwp=0.3,) as forecast: estimate = await forecast.estimate() logging.info('Estimated Wh today: {}Wh'.format(estimate.energy_production_today)) logging.info('Estimated Wh tomorrow: {}Wh'.format(estimate.energy_production_tomorrow)) return estimate.energy_production_today, estimate.energy_production_tomorrow def mastodon_login(): mastodon = Mastodon(client_id = settings.access_client_id,) mastodon.log_in( settings.mastodon_user, settings.mastodon_password, to_file = settings.access_secret ) def get_shutdown(): logging.info('get_shutdown') try: data = subprocess.run(["/usr/bin/ssh", "-t", "root@192.168.1.231", "cat /run/systemd/shutdown/scheduled"], shell=False, stdin=None, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False) except subprocess.CalledProcessError as e: logging.info(e.output) logging.info('Data: {}'.format(data.stdout)) if 'No such file' in data.stdout.decode('UTF-8'): logging.info('No Shutdown') ts = int(time.time() + 86400) else: logging.info('File found') token = data.stdout.decode('UTF-8').split('WARN') shutdown_date = int(token[0][5:]) ### This is for local checks (these days a supervisor system is used) # try: # fp = open('/run/systemd/shutdown/scheduled') # data = fp.readlines() # fp.close() # shutdown_date = data[0].split('=')[1].rstrip() ts = int(shutdown_date) / 1000000 time_now = int(time.time()) shutdown_date = datetime.datetime.utcfromtimestamp(ts).strftime('%Y-%m-%d %H:%M:%S') turnoff_date = datetime.datetime.utcfromtimestamp(ts + 300).strftime('%Y-%m-%d %H:%M:%S') time_to_shutdown = ts - time_now time_to_date = datetime.datetime.utcfromtimestamp(time_to_shutdown).strftime('%H:%M:%S') warning_time = ts - 600 warning_to_date = datetime.datetime.utcfromtimestamp(warning_time).strftime('%Y-%m-%d %H:%M:%S') return shutdown_date, time_to_date, warning_to_date, turnoff_date def warn_shutdown(): try: mastodon = Mastodon(access_token = settings.access_secret, api_base_url = settings.instance_url) mastodon.toot('Warning\nSystem Shutdown in 10 minutes') except Exception as e: logging.info('Failed to send toot (warn_shutdown): {}'.format(e)) def send_toot(): logging.info('Preparing Toot') try: batt_v_V = int(r.get('batt_v')) / 1000 main_current = r.get('main_current') panel_voltage = int(r.get('panel_voltage')) / 1000 panel_power = r.get('panel_power') load_current = int(r.get('load_current')) / 1000 yield_today = int(r.get('yield_today')) / 100 max_power_today = r.get('max_power_today') load_power = load_current * batt_v_V final_date, time_date, warning_time, turnoff_date = get_shutdown() toot_to_send = 'Solarcene.community Power Data\nBattery Voltage: {}V\nBattery Current: {}mA\nPanel Voltage: {}V\nPanel Power: {}W\nLoad Current: {}A\nLoad Power: {:.2f}W\nYield Today: {}kWh\nMax Power Today: {}W\nUpdated every 60 minutes\n\nShutdown planned for {}UTC (in {})'.format(batt_v_V, main_current, panel_voltage, panel_power, load_current, load_power, yield_today, max_power_today, final_date, time_date) logging.info(toot_to_send) except: logging.info('Failed to construct toot') try: mastodon = Mastodon(access_token = settings.access_secret, api_base_url = settings.instance_url) mastodon.toot(toot_to_send) except Exception as e: logging.info('Failed to send toot (send_toot): {}'.format(e)) def start_toot(energy_production_today, energy_production_tomorrow): logging.info('Sleeping for 60 seconds to allow mastodon to fully start') # time.sleep(60) # Get today's sunrise and sunset in UTC today_sr = sun.get_sunrise_time() today_ss = sun.get_sunset_time() logging.info('Sunrise: {}UTC and Sunset: {}UTC'.format(today_sr.strftime('%H:%M'), today_ss.strftime('%H:%M'))) try: final_date, time_date, warning_time, turnoff_date = get_shutdown() logging.info('Debug 1') mastodon = Mastodon(access_token = settings.access_secret, api_base_url = settings.instance_url) mastodon.toot('System Online\n\nLocal Sunrise: {}UTC\nLocal Sunset: {}UTC\nEstimated Wh today: {}Wh\nEstimated Wh tomorrow: {}Wh\nEstimation from https://forecast.solar\n\nShutdown planned for: {}UTC (in {})'.format(today_sr.strftime('%H:%M'), today_ss.strftime('%H:%M'), energy_production_today, energy_production_tomorrow, final_date, time_date)) except Exception as e: logging.info('Failed to send toot (start_toot): {}'.format(e)) def check_shutdown(): logging.info('Check Shutdown') # Get solar estimation try: energy_production_today, energy_production_tomorrow = get_solar_estimate() except: energy_production_today = "-1" energy_production_tomorrow = "-1" r.set('energy_production_today', energy_production_today) r.set('energy_production_tomorrow', energy_production_tomorrow) logging.info('Energy {} Batt {}'.format(energy_production_today, r.get('batt_v'))) if int(energy_production_today) < 400 or int(r.get('batt_v')) <= 12000 : os.system("/usr/bin/ssh -t root@192.168.1.231 'shutdown -h 20:30'") final_date, time_date, warning_time, turnoff_date = get_shutdown() logging.info(warning_time) scheduler.add_job(warn_shutdown, trigger='date', run_date=warning_time ) logging.info(turnoff_date) scheduler.add_job(turnoff, trigger='date', run_date=turnoff_date ) logging.info('Setting up to start again in the morning') scheduler.add_job(startup, trigger='cron', hour=8 ) # else: # scheduler.add_job(check_shutdown, trigger='cron', hour=8 ) # os.system("shutdown -h 18:15") start_toot(energy_production_today, energy_production_tomorrow) def startup(): logging.info('Starting up') turn_off_lights() logging.info('Lights are off') time.sleep(10) turn_on_lights() logging.info('Power on, restarting script') time.sleep(60) # Remove all jobs logging.info('Clear all scheduled jobs') scheduler.remove_all_jobs() scheduler.add_job(send_toot, trigger='cron', minute=1) scheduler.add_job(check_shutdown, trigger='cron', hour=9 ) def turnoff(): logging.info('Shutting Down') turn_off_lights() logging.info('Lights are off') if __name__ == '__main__': # startup() # logging.info('Sleeping 30s to allow server to boot') # time.sleep(30) logging.info('Starting') # mastodon_login() logging.info('Logged In') scheduler.add_job(send_toot, trigger='cron', minute=1) check_shutdown() scheduler.add_job(check_shutdown, trigger='cron', hour=9 ) while True: time.sleep(1) logging.info('Exit')