用Python监控NASA TV直播画面的实现步骤

fromioimportBytesIO
fromdatetimeimportdatetime,timedelta
frompathlibimportPath
importrequests
fromrequests.exceptionsimportTimeout
fromPILimportImage
importPySimpleGUIassg

FEED_URL='https://science.ksc.nasa.gov/shuttle/countdown/video/chan2large.jpg'
#Framesizewithoutandwith16:9aspectratiocorrection
WIDTH=704
HEIGHT=480
HEIGHT_16_9=396
#Minimum,default,andmaximumautoreloadintervalinseconds
MIN_DELTA=45
DELTA=MIN_DELTA
MAX_DELTA=300

classStillFrame():
"""Holdsastillframe.
TheimageisstoredasaPNGPIL.ImageandkeptinPNGformat.
Attributes
----------
image:PIL.Image
Astillframe
original:PIL.Image
Originalframewithwchichtheinstanceisinitialized,cachedincaseof
resizingtotheoriginalsize
Methods
-------
bytes:Returntherawbytes
resize:Resizethescreenshot
new_size:Calculatenewaspectratio
"""
def__init__(self,image):
"""ConverttheimagetoPNGandcachetheconvertedoriginal.
Parameters
----------
image:PIL.Image
Imagetostore
"""
self.image=image
self._topng()
self.original=self.image
def_topng(self):
"""ConvertimageformatofframetoPNG.
Returns
-------
StillFrame
FramewithimageinPNGformat
"""
ifnotself.image.format=='PNG':
png_file=BytesIO()
self.image.save(png_file,'png')
png_file.seek(0)
png_image=Image.open(png_file)
self.image=png_image
returnself
defbytes(self):
"""Returnrawbytesofaframeimage.
Returns
-------
bytes
Bytestreamoftheframeimage
"""
file=BytesIO()
self.image.save(file,'png')
file.seek(0)
returnfile.read()
defnew_size(self):
"""Returnimagesizetoggledbetweenoriginaland16:9.
Returns
-------
2-tuple
Newsize
"""
size=self.image.size
original_size=self.original.size
new_size=(WIDTH,HEIGHT_16_9)ifsize==original_sizeelse(WIDTH,HEIGHT)
returnnew_size
defresize(self,new_size):
"""Resizeframeimage.
Parameters
----------
new_size:2-tuple
Newsize
Returns
-------
StillFrame
Framewithimageresized
"""
ifnot(self.image.size==new_size):
self.image=self.image.resize(new_size)
returnself

defmake_blank_image(size=(WIDTH,HEIGHT)):
"""Createablankimagewithabluebackground.
Parameters
----------
size:2-tuple
Imagesize
Returns
-------
PIL.Image
Blankimage
"""
image=Image.new('RGB',size=size,color='blue')
returnimage

defdownload_image(url):
"""DownloadcurrentNASATVimage.
Parameters
----------
url:str
URLtodownloadtheimagefrom
Returns
-------
PIL.Image
Downloadedimageifnoerrors,otherwiseblankimage
"""
try:
response=requests.get(url,timeout=(0.5,0.5))
ifresponse.status_code==200:
image=Image.open(BytesIO(response.content))
else:
image=make_blank_image()
exceptTimeout:
image=make_blank_image()
returnimage

defrefresh(window,resize=False,feed=FEED_URL):
"""Displaythelateststillframeinwindow.
Parameters
----------
window:sg.Window
Windowtodisplaythestillto
feed:string
FeedURL
Returns
-------
StillFrame
Refreshedscreenshot
"""
still=StillFrame(download_image(feed))
ifresize:
still=change_aspect_ratio(window,still,new_size=(WIDTH,HEIGHT_16_9))
else:
window['-IMAGE-'].update(data=still.bytes())
returnstill

defchange_aspect_ratio(window,still,new_size=(WIDTH,HEIGHT_16_9)):
"""Changetheaspectratioofthestilldisplayedinwindow.
Parameters
----------
window:sg.Window
Windowcontainingthestill
new_size:2-tuple
Newsizeofthestill
Returns
-------
StillFrame
Framecontainingtheresizedimage
"""
resized_still=still.resize(new_size)
window['-IMAGE-'].update(data=resized_still.bytes())
returnresized_still

defsave(still,path):
"""Savestilltoafile.
Parameters
----------
still:StillFrame
Stilltosave
path:string
Filename
Returns
-------
Boolean
Trueiffilesavedwithnoerrors
"""
filename=Path(path)
try:
withopen(filename,'wb')asfile:
file.write(still.bytes())
saved=True
exceptOSError:
saved=False
returnsaved

defnext_timeout(delta):
"""Returnthemomentintimerightnow+deltasecondsfromnow.
Parameters
----------
delta:int
Timeinsecondsuntilthenexttimeout
Returns
-------
datetime.datetime
Momentintimeofthenexttimeout
"""
rightnow=datetime.now()
returnrightnow+timedelta(seconds=delta)

deftimeout_due(next_timeout):
"""ReturnTrueifthenexttimeoutisdue.
Parameters
----------
next_timeout:datetime.datetime
Returns
-------
bool
Trueifthenexttimeoutisdue
"""
rightnow=datetime.now()
returnrightnow>=next_timeout

defvalidate_delta(value):
"""Checkifvalueisanintwithintheproperrangeforatimedelta.
Parameters
----------
value:int
Timeinsecondsuntilthenexttimeout
Returns
-------
int
Timeinsecondsuntilthenexttimeout
bool
Trueiftheargumentisavalidtimedelta
"""
isinteger=False
try:
isinteger=type(int(value))isint
exceptException:
delta=DELTA
delta=int(value)ifisintegerelsedelta
isvalid=MIN_DELTA<=delta<=MAX_DELTA
delta=deltaifisvalidelseDELTA
returndelta,isintegerandisvalid

LAYOUT=[[sg.Image(key='-IMAGE-')],
[sg.Checkbox('Correctaspectratio',key='-RESIZE-',enable_events=True),
sg.Button('Reload',key='-RELOAD-'),
sg.Button('Save',key='-SAVE-'),
sg.Exit()],
[sg.Checkbox('Auto-reloadevery(seconds):',key='-AUTORELOAD-',
default=True),
sg.Input(DELTA,key='-DELTA-',size=(3,1),justification='right'),
sg.Button('Set',key='-UPDATE_DELTA-')]]

defmain(layout):
"""Runeventloop."""
window=sg.Window('Spacestills',layout,finalize=True)
current_still=refresh(window)
delta=DELTA
next_reload_time=datetime.now()+timedelta(seconds=delta)
whileTrue:
event,values=window.read(timeout=100)
ifeventin(sg.WIN_CLOSED,'Exit'):
break
elif((event=='-RELOAD-')or
(values['-AUTORELOAD-']andtimeout_due(next_reload_time))):
current_still=refresh(window,values['-RESIZE-'])
ifvalues['-AUTORELOAD-']:
next_reload_time=next_timeout(delta)
elifevent=='-RESIZE-':
current_still=change_aspect_ratio(
window,current_still,current_still.new_size())
elifevent=='-SAVE-':
filename=sg.popup_get_file(
'Filename',file_types=[('PNG','*.png')],save_as=True,
title='Saveimage',default_extension='.png')
iffilename:
saved=save(current_still,filename)
ifnotsaved:
sg.popup_ok('Errorwhilesavingfile:',filename,title='Error')
elifevent=='-UPDATE_DELTA-':
#Thecurrentcycleshouldcompleteatthealreadyscheduledtime.So
#don'tupdatenext_reload_timeyetbecauseit'llbetakencareofatthe
#next-AUTORELOAD-or-RELOAD-event.
delta,valid=validate_delta(values['-DELTA-'])
ifnotvalid:
window['-DELTA-'].update(str(DELTA))
window.close()
delwindow

if__name__=='__main__':
main(LAYOUT)