问题描述
我有一个 Pi 的天气显示，它使用 pygame 来显示数据 我遇到的问题是当互联网因任何原因死机时，没有更新并且显示屏显示空白数据 我想要做的是如果没有更新，那么它会在屏幕上保留以前的数据 这可能吗？
if forecastData.status == forecast.STATUS_OK:
ren = font.render("Solar Radiation: {} W/m2".format(forecastData.solar_radiation),1,pg.Color('black'),pg.Color(185,208,240))
else:
ren = font.render("Solar Radiation: ",240))
screen.blit(ren,(5*HRES//1600,430*VRES//900-ren.get_height()//2))
如果没有更新，则显示太阳辐射：仅 - 我希望使用以前的数据 - 即不要更新显示的这一部分
解决方法
似乎您应该做的是在更新 self.data 类中的 Forecast 之前检查状态：
#Now make it available to outside world.
if data.status == STATUS_OK:
self.lock.acquire()
self.data = data
self.lock.release()
无法确定，因为无法运行您的任何代码示例。
这是一个使用线程模拟数据检索的最小示例，偶尔会出现错误。
import random
import threading
import time
import pygame
class ForecastThread(threading.Thread):
def __init__(self):
threading.Thread.__init__(self)
self.daemon = True
self.start() # starts on instantiation!
def retrieve_data(self):
global data
# make request to forecast service and parse response
response = random.randint(1,5)
if response == 1:
# Request failed
data = """(○o◌!*^#@!@"""
else: # success
data = random.choice(("Cloudy","Rainy","Stormy","Sunny","Windy"))
def run(self):
while True:
if not running: # exit thread if main loop exits
return
else:
self.retrieve_data()
time.sleep(1) # sleep for a second
WIDTH = 480
HEIGHT = 240
FPS = 30
random.seed(98765432)
pygame.init()
window = pygame.display.set_mode((WIDTH,HEIGHT))
clock = pygame.time.Clock()
# grab the first installed font
sys_font = pygame.font.SysFont(pygame.font.get_fonts()[0],100)
data = "" # create globals before thread
running = True
forecaster_thread = ForecastThread() # note: thread is auto-starting
while running:
for event in pygame.event.get():
if event.type == pygame.QUIT:
running = False
elif event.type == pygame.KEYUP:
if event.key == pygame.K_ESCAPE:
running = False
# update title bar
pygame.display.set_caption(f"Forecaster FPS: {clock.get_fps():.1f}")
# fill background to clear
window.fill(pygame.Color("lightsteelblue2"))
# create image
image = sys_font.render(data,True,pygame.Color("black"))
window.blit(image,(50,50)) # blit close to centre
# show surface
pygame.display.update()
# limit frames
clock.tick(FPS)
pygame.quit()
运行此代码最初将显示 Cloudy，然后更改为表示请求失败的乱码数据。
如果您将 retrieve_data() 函数更改为在失败时不更新 data，则不会显示任何乱码。
def retrieve_data(self):
global data
# make request to forecast service and parse response
response = random.randint(1,5)
if response == 1:
# Request failed
print("Failed Request") # log the error
else: # success
data = random.choice(("Cloudy","Windy"))
“随机导入”我已经尝试了很多方法，但总是失败 这有帮助吗？...这是收集预测的完整代码
import json
import urllib.request
import urllib.error
import time
from datetime import datetime
from datetime import timedelta
import pdb
import threading
import pygame as pg
import logging
import os.path
import socket
FORECAST_URL = "https://swd.weatherflow.com/swd/rest/better_forecast?api_key=20c70eae-e62f-4d3b-b3a4-8586e90f3ac8&station_id=44303&lat=53.440&lon=-2.105"
TIMEOUT = 15
STATUS_TIMEOUT = "Timeout"
STATUS_OK = "OK"
def nullis0hook(d):
"""This is a hook for the JSON decoder,replacing null with 0."""
for k in d.keys():
if d[k]==None:
d[k]=0
return d
class ForecastData:
def __init__(self):
self.status = STATUS_OK
self.conditions=""
self.iconnow = ""
self.updateTime="946684800"
self.tempnow=0
self.templow=30
self.temphigh=-10
self.sea_level_pressure=0
self.station_pressure=0
self.pressure_trend=""
self.relative_humidity=0
self.wind_avg=0
self.wind_direction_cardinal=""
self.angle=0
self.wind_gust=0
self.solar_radiation=0
self.uv=0
self.feels_like=0
self.dew_point=0
self.wet_bulb_temperature=""
self.delta_t=0
self.air_density=0
self.lightning_strike_last_distance="0"
self.lightning1=""
self.lightning_strike_last_epoch="946684800"
self.precip_accum_local_yesterday="0"
self.precip_accum_local_day="0"
self.condday=[""]*6
self.icon=[""]*6
self.iconday_filename = [os.path.join("images",os.path.join("forecast_icons","--_"))]*6
self.iconday = [pg.image.load(os.path.join("images","--_1.bmp"))).convert()]*6
self.thighday=[0]*6
self.tlowday=[0]*6
self.sunriseday=[""]*6
self.sunsetday=[""]*6
self.precprday=[0]*6
self.precpiconday=[""]*6
self.preciptypeday=[""]*6
self.conditionshour=[""]*9
self.iconhour=[""]*9
self.precipprhour=[""]*9
self.preciptypehour=[""]*9
self.feelslikehour=[0]*9
self.conditionshour=[""]*9
self.iconhour=[""]*9
self.precipprhour=[""]*9
self.preciptypehour=[""]*9
self.feelslikehour=[0]*9
self.airtemphour=[0]*9
self.windavghour=[0]*9
self.forecast_icon_filename = os.path.join("images",os.path.join("weather_icons","--_.bmp"))
self.forecast_icon = pg.image.load(self.forecast_icon_filename).convert()
self.kia = pg.image.load(os.path.join("images","kia.png"))).convert()
#The Forecast class retrieves the weatherflow.com forecast and extracts relevant forecast data.
#Note: pygame display must be initialized before constructing a Forecast object
class ForeCastHour:
def __init__(self):
self.conditions = [""]*6
self.icon = ["--_"]*6
self.precipr = [""]*6
self.precip_type = [""]*6
self.feels_like = [""]*6
class Forecast:
def __init__(self):
self.data = ForecastData()
self.lock = threading.Lock()
self.max_wind_gust=0
self.templow=30
self.temphigh=-10
self.updateTime = ""
def getData(self):
"""Get most recently retrieved consistent set of data."""
self.lock.acquire()
data = self.data
self.lock.release()
return data
def midnightReset(self):
"""Reset any values that require resetting at midnight"""
self.lock.acquire()
self.data.wind_gust=0
self.lock.release()
def update(self):
"""Update the forecast data"""
data = ForecastData()
try:
req = urllib.request.Request(FORECAST_URL)
with urllib.request.urlopen(req,timeout=TIMEOUT) as response:
raw = response.read()
forecast_json = json.loads(raw,object_hook=nullis0hook)
data.status = STATUS_OK
self.updateTime = forecast_json["current_conditions"]["time"]
data.updateTime = time.strftime("%H:%M:%S",time.localtime(self.updateTime))
data.conditions = forecast_json["current_conditions"]["conditions"]
iconnow = forecast_json["current_conditions"]["icon"]
if iconnow == "null":
iconnow = "--_" #icon replacement
if iconnow == "":
iconnow = "--_" #icon replacement
data.iconnow_filename = os.path.join("images",iconnow+".bmp"))
if os.path.exists(data.iconnow_filename):
data.iconnow = pg.image.load(data.iconnow_filename).convert()
else:
logging.warning("Weather icon file {} not found.".format(data.iconnow_filename))
data.tempnow = forecast_json["current_conditions"]["air_temperature"]
if data.tempnow < self.templow:
self.templow = data.tempnow
if data.tempnow > self.temphigh:
self.temphigh = data.tempnow
data.templow = self.templow
data.temphigh = self.temphigh
data.sea_level_pressure = forecast_json["current_conditions"]["sea_level_pressure"]
data.station_pressure = forecast_json["current_conditions"]["station_pressure"]
data.pressure_trend = forecast_json["current_conditions"]["pressure_trend"]
data.relative_humidity = forecast_json["current_conditions"]["relative_humidity"]
data.wind_avg = forecast_json["current_conditions"]["wind_avg"]* 2.23694 #Convert mps to mph
data.wind_gust = forecast_json["current_conditions"]["wind_gust"] * 2.23694 #Convert mps to mph
data.angle = forecast_json["current_conditions"]["wind_direction"]
if data.angle <= 180:
data.angle = data.angle + 180
else:
data.angle = data.angle - 180
data.wind_direction_cardinal = forecast_json["current_conditions"]["wind_direction_cardinal"]
data.solar_radiation = forecast_json["current_conditions"]["solar_radiation"]
data.uv = forecast_json["current_conditions"]["uv"]
data.feels_like = forecast_json["current_conditions"]["feels_like"]
lightning_strike_last_distance = forecast_json["current_conditions"].get("lightning_strike_last_distance",0)
lightning1 = lightning_strike_last_distance*0.621371 #Convert kph to mph
data.lightning_strike_last_distance = "{0:.1f} miles away ".format(lightning1)
lightning_strike_last_epoch = forecast_json["current_conditions"].get("lightning_strike_last_epoch")
data.lightning_strike_last_epoch = time.strftime("%d %b",time.localtime(lightning_strike_last_epoch))
data.precip_accum_local_yesterday = forecast_json["current_conditions"]["precip_accum_local_yesterday"]
data.precip_accum_local_day = forecast_json["current_conditions"]["precip_accum_local_day"]
for day in range(6):
data.sunriseday[day] = forecast_json["forecast"]["daily"][day]["sunrise"]
data.sunriseday[day] = time.strftime("%H:%M:%S",time.localtime(data.sunriseday[day]))
data.sunsetday[day] = forecast_json["forecast"]["daily"][day]["sunset"]
data.sunsetday[day] = time.strftime("%H:%M:%S",time.localtime(data.sunsetday[day]))
data.condday[day] = forecast_json["forecast"]["daily"][day]["conditions"]
icon = forecast_json["forecast"]["daily"][day]["icon"]
data.iconday_filename[day] = os.path.join("images",icon+"1.bmp"))
if os.path.exists(data.iconday_filename[day]):
iconimage = pg.image.load(data.iconday_filename[day]).convert()
data.iconday[day] = iconimage
else:
logging.warning("Forecast icon file {} not found.".format(data.iconday_filename[day]))
data.thighday[day] = forecast_json["forecast"]["daily"][day]["air_temp_high"]
data.tlowday[day] = forecast_json["forecast"]["daily"][day]["air_temp_low"]
data.precprday[day] = forecast_json["forecast"]["daily"][day]["precip_probability"]
if data.precprday[day] != 0:
data.precpiconday[day] = forecast_json["forecast"]["daily"][day]["precip_icon"]
data.preciptypeday[day] = forecast_json["forecast"]["daily"][day]["precip_type"]
data.forecast_icon_filename = os.path.join("images",iconnow+".bmp"))
if os.path.exists(data.forecast_icon_filename):
data.forecast_icon = pg.image.load(data.forecast_icon_filename).convert()
else:
logging.warning("Forecast icon file {} not found.".format(data.forecast_icon_filename))
for hours in range(9):
ps = forecast_json["forecast"]["hourly"][hours]["conditions"]
if ps == "Wintry Mix Possible":
ps = "Winty-P"
if ps == "Wintry Mix Likely":
ps = "Winty-L"
if ps == "Rain Likely":
ps = "Rain-L"
if ps == "Rain Possible":
ps ="Rain-P"
if ps == "Snow Possible":
ps = "Snow-P"
if ps == "Thunderstorms Likely":
ps = "ThundrL"
if ps == "Thunderstorms Possible":
ps = "ThundrP"
if ps == "Partly Cloudy":
ps = "Clouds"
if ps == "Very Light Rain":
ps = "drizzle"
data.conditionshour[hours] = "{}".format(ps)
data.iconhour[hours] = forecast_json["forecast"]["hourly"][hours]["icon"]
pp = forecast_json["forecast"]["hourly"][hours]["precip_probability"]
data.precipprhour[hours] = "{}%".format(pp)
if pp == 0:
data.preciptypehour[hours] = "0"
else:
data.preciptypehour[hours] = forecast_json["forecast"]["hourly"][hours]["precip_type"]
data.feelslikehour[hours] = "{} C".format(forecast_json["forecast"]["hourly"][hours]["feels_like"])
data.airtemphour[hours] = forecast_json["forecast"]["hourly"][hours]["air_temperature"]
data.windavghour[hours] = forecast_json["forecast"]["hourly"][hours]["wind_avg"]*0.621371 #Convert kph to mph
#datetime object containing current date and time
now = datetime.now()
data.updateTime = now.strftime("%H:%M:%S")
self.updateTime = now
except (socket.timeout,socket.gaierror,urllib.error.URLError,json.decoder.JSONDecodeError,KeyError):
logging.warning("Error retrieving forecast data")
#declare timeout only after timeout period
if datetime.now() - self.updateTime > timedelta(seconds=TIMEOUT):
data.status = STATUS_TIMEOUT
data.updateTime = self.data.updateTime #Use old update time value
else: #If timeout period has not elapsed yet,use previous data
logging.info("Not timing out yet")
self.data = self.data #Use old value
if datetime.now() - self.updateTime > timedelta(seconds=TIMEOUT):
data.status = STATUS_TIMEOUT
else: #If timeout period has not elapsed yet,use previous data
logging.info("Not timing out yet")
data = self.data
#Now make it available to outside world.
self.lock.acquire()
self.data = data
self.lock.release()
当有互联网时，这是有效的，但是当互联网关闭时，它不使用旧数据
'''THIS IS THE FORECAST SECTION THAT WORKS WHEN THERE IS INTENET AND FORECAST UPDATES
BUT WHEN THERE IS NO UPDATE IT JUST DISPLAYS THE TEXT AND DOES NOT USE THE OLD DATA VALUE'''
if forecastData.status == forecast.STATUS_OK:
ren = font.render("battery voltage : " + "{} V".format(forecastData.battery),1,pg.Color('white'),pg.Color(162,160,160))
else:
ren = font.render("",160))
screen.blit(ren,(700*HRES//1600,VRES//60))
if forecastData.status == forecast.STATUS_OK:
ren = font.render("Conditions: {} ".format(forecastData.conditions),pg.Color('black'),pg.Color(185,208,240))
else:
ren = font.render("Conditions: ",240))
screen.blit(ren,(5*HRES//1600,70*VRES//900-ren.get_height()//2))
'''THIS IS THE ERROR MESSAGE I GET LOGGED'''
2021-04-15 13:28:41,057 - root - INFO - Updating forecast.
2021-04-15 13:28:41,087 - root - WARNING - Error retrieving forecast data
2021-04-15 13:29:11,049 - root - WARNING - Previous every-minute thread still running. Not relaunching.
2021-04-15 13:29:26,602 - root - INFO - t key pressed. Toggle fullscreen.
2021-04-15 13:29:41,085 - root - WARNING - Previous every-minute thread still running. Not relaunching.
2021-04-15 13:30:11,089 - root - WARNING - Previous every-minute thread still running. Not relaunching.
'''THIS IS THE FUNCTION THAT THE ERROR REFERS TO'''
def everyMinuteThreadFunction():
"""This thread function executes once every minute."""
global initialWeatherUpdateReceived,everyMinuteThreadRunning
everyMinuteThreadRunning = True
assert(forecastObj)
checkMidnightRollOver()
logging.info("Updating forecast.")
forecastObj.update()
forecastData = forecastObj.getData()
if forecastData.status == forecast.STATUS_OK:
logging.info("Forecast data: {}".format(vars(forecastData)))
#The first time we pass here is a good time to kick off the five minute task. We now have our first
#forecast evice data available
if not initialWeatherUpdateReceived:
#program a periodic timer used to kick off the everyFiveMinutesThreadFunction.
pg.time.set_timer(EVERY_FIVE_MINUTES_THREAD_FUNCTION_EVENT,5*60000)
#Kick off the task now,for the initial interval.
t = threading.Thread(target=everyFiveMinutesThreadFunction,args=())
t.daemon = True
t.start()
updateGauge()
initialWeatherUpdateReceived = True
everyMinuteThreadRunning = False