iampeter/pixelplanet
1
Fork 0
forked from ppfun/pixelplanet
Code Pull Requests Activity

use api/me for historyDownload.py

Browse Source 
camel_case canvasID in areaDownload.py
add parameters to USER_AGENT in both
This commit is contained in:
HF 2023-10-01 01:48:40 +02:00
parent d79636ae88
commit e108a23a83
3 changed files with 101 additions and 101 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
utils/README.md
View File

@ -36,6 +36,7 @@ Usage: `areaDownload.py startX_startY endX_endY filename.png`
downloads the history from an canvas area between two dates. downloads the history from an canvas area between two dates.
Useage: `historyDownload.py canvasId startX_startY endX_endY start_date end_date Useage: `historyDownload.py canvasId startX_startY endX_endY start_date end_date
This is used for creating timelapses, see the cmd help to know how This is used for creating timelapses, see the cmd help to know how
**Requires:** aiohttp, asyncio and PIL python3 packages
## pp-center\*.png ## pp-center\*.png
center logo of pixelplanet center logo of pixelplanet

22
utils/areaDownload.py
View File

@ -5,7 +5,7 @@ import sys, os, io, math
import asyncio import asyncio
import aiohttp import aiohttp
USER_AGENT = "ppfun areaDownload 1.0" USER_AGENT = "ppfun areaDownload 1.0 " + ' '.join(sys.argv[1:])
PPFUN_URL = "https://pixelplanet.fun" PPFUN_URL = "https://pixelplanet.fun"
class Color(object): class Color(object):
@ -102,8 +102,8 @@ async def fetchMe():
await asyncio.sleep(5) await asyncio.sleep(5)
pass pass
async def fetch(session, canvasID, canvasoffset, ix, iy, target_matrix): async def fetch(session, canvas_id, canvasoffset, ix, iy, target_matrix):
url = f"{PPFUN_URL}/chunks/{canvasID}/{ix}/{iy}.bmp" url = f"{PPFUN_URL}/chunks/{canvas_id}/{ix}/{iy}.bmp"
headers = { headers = {
'User-Agent': USER_AGENT 'User-Agent': USER_AGENT
} }
@ -140,7 +140,7 @@ async def fetch(session, canvasID, canvasoffset, ix, iy, target_matrix):
await asyncio.sleep(3) await asyncio.sleep(3)
pass pass
async def get_area(canvasID, canvas, x, y, w, h): async def get_area(canvas_id, canvas, x, y, w, h):
target_matrix = Matrix() target_matrix = Matrix()
target_matrix.add_coords(x, y, w, h) target_matrix.add_coords(x, y, w, h)
canvasoffset = math.pow(canvas['size'], 0.5) canvasoffset = math.pow(canvas['size'], 0.5)
@ -154,7 +154,7 @@ async def get_area(canvasID, canvas, x, y, w, h):
async with aiohttp.ClientSession() as session: async with aiohttp.ClientSession() as session:
for iy in range(yc, hc + 1): for iy in range(yc, hc + 1):
for ix in range(xc, wc + 1): for ix in range(xc, wc + 1):
tasks.append(fetch(session, canvasID, canvasoffset, ix, iy, target_matrix)) tasks.append(fetch(session, canvas_id, canvasoffset, ix, iy, target_matrix))
await asyncio.gather(*tasks) await asyncio.gather(*tasks)
return target_matrix return target_matrix
@ -206,20 +206,20 @@ async def main():
print("Usage: areaDownload.py canvasID startX_startY endX_endY filename.png") print("Usage: areaDownload.py canvasID startX_startY endX_endY filename.png")
print("(use R key on pixelplanet to copy coordinates)") print("(use R key on pixelplanet to copy coordinates)")
print("canvasID: ", end='') print("canvasID: ", end='')
for canvasID, canvas in apime['canvases'].items(): for canvas_id, canvas in apime['canvases'].items():
if 'v' in canvas and canvas['v']: if 'v' in canvas and canvas['v']:
continue continue
print(f"{canvasID} = {canvas['title']}", end=', ') print(f"{canvas_id} = {canvas['title']}", end=', ')
print() print()
return return
canvasID = sys.argv[1] canvas_id = sys.argv[1]
if canvasID not in apime['canvases']: if canvas_id not in apime['canvases']:
print("Invalid canvas selected") print("Invalid canvas selected")
return return
canvas = apime['canvases'][canvasID] canvas = apime['canvases'][canvas_id]
if 'v' in canvas and canvas['v']: if 'v' in canvas and canvas['v']:
print("Can\'t get area for 3D canvas") print("Can\'t get area for 3D canvas")
@ -238,7 +238,7 @@ async def main():
EnumColorPixelplanet.getColors(canvas) EnumColorPixelplanet.getColors(canvas)
filename = sys.argv[4] filename = sys.argv[4]
matrix = await get_area(canvasID, canvas, x, y, w, h) matrix = await get_area(canvas_id, canvas, x, y, w, h)
matrix.create_image(filename) matrix.create_image(filename)
print("Done!") print("Done!")

179
utils/historyDownload.py
View File

@ -5,7 +5,10 @@ import sys, io, os
import datetime import datetime
import asyncio import asyncio
import aiohttp import aiohttp
import json
USER_AGENT = "ppfun historyDownload 1.0 " + ' '.join(sys.argv[1:])
PPFUN_URL = "https://pixelplanet.fun"
PPFUN_STORAGE_URL = "https://storage.pixelplanet.fun"
# how many frames to skip # how many frames to skip
# 1 means none # 1 means none
@ -14,62 +17,35 @@ import json
# [...] # [...]
frameskip = 1 frameskip = 1
canvases = [ async def fetchMe():
{ url = f"{PPFUN_URL}/api/me"
"canvas_name": "earth", headers = {
"canvas_size": 256*256, 'User-Agent': USER_AGENT
"canvas_id": 0, }
"bkg": (202, 227, 255), async with aiohttp.ClientSession() as session:
}, attempts = 0
{ while True:
"canvas_name": "moon", try:
"canvas_size": 16384, async with session.get(url, headers=headers) as resp:
"canvas_id": 1, data = await resp.json()
"bkg": (49, 46, 47), return data
"historical_sizes" : [ except:
["20210417", 4096], if attempts > 3:
] print(f"Could not get {url} in three tries, cancelling")
}, raise
{ attempts += 1
}, print(f"Failed to load {url}, trying again in 5s")
{ await asyncio.sleep(5)
"canvas_name": "corona", pass
"canvas_size": 256,
"canvas_id": 3,
"bkg": (33, 28, 15),
},
{
"canvas_name": "compass",
"canvas_size": 1024,
"canvas_id": 4,
"bkg": (196, 196, 196),
},
{
},
{
},
{
"canvas_name": "1bit",
"canvas_size": 256*256,
"canvas_id": 7,
"bkg": (0, 0, 0),
},
{
"canvas_name": "top10",
"canvas_size": 2048,
"canvas_id": 8,
"bkg": (197, 204, 184),
"historical_sizes" : [
["20220626", 1024],
]
},
]
async def fetch(session, url, offx, offy, image, bkg, needed = False): async def fetch(session, url, offx, offy, image, bkg, needed = False):
attempts = 0 attempts = 0
headers = {
'User-Agent': USER_AGENT
}
while True: while True:
try: try:
async with session.get(url) as resp: async with session.get(url, headers=headers) as resp:
if resp.status == 404: if resp.status == 404:
if needed: if needed:
img = PIL.Image.new('RGB', (256, 256), color=bkg) img = PIL.Image.new('RGB', (256, 256), color=bkg)
@ -91,11 +67,9 @@ async def fetch(session, url, offx, offy, image, bkg, needed = False):
attempts += 1 attempts += 1
pass pass
async def get_area(canvas, x, y, w, h, start_date, end_date): async def get_area(canvas_id, canvas, x, y, w, h, start_date, end_date):
canvas_data = canvases[canvas] canvas_size = canvas["size"]
canvas_id = canvas_data["canvas_id"] bkg = tuple(canvas['colors'][0])
canvas_size = canvas_data["canvas_size"]
bkg = canvas_data["bkg"]
delta = datetime.timedelta(days=1) delta = datetime.timedelta(days=1)
end_date = end_date.strftime("%Y%m%d") end_date = end_date.strftime("%Y%m%d")
@ -110,8 +84,8 @@ async def get_area(canvas, x, y, w, h, start_date, end_date):
start_date = start_date + delta start_date = start_date + delta
fetch_canvas_size = canvas_size fetch_canvas_size = canvas_size
if 'historical_sizes' in canvas_data: if 'historicalSizes' in canvas:
for ts in canvas_data['historical_sizes']: for ts in canvas['historicalSizes']:
date = ts[0] date = ts[0]
size = ts[1] size = ts[1]
if iter_date <= date: if iter_date <= date:
@ -122,14 +96,14 @@ async def get_area(canvas, x, y, w, h, start_date, end_date):
wc = (x + w - offset) // 256 wc = (x + w - offset) // 256
yc = (y - offset) // 256 yc = (y - offset) // 256
hc = (y + h - offset) // 256 hc = (y + h - offset) // 256
print("Load from %s / %s to %s / %s" % (xc, yc, wc + 1, hc + 1)) print("Load from %s / %s to %s / %s with canvas size %s" % (xc, yc, wc + 1, hc + 1, fetch_canvas_size))
tasks = [] tasks = []
async with aiohttp.ClientSession() as session: async with aiohttp.ClientSession() as session:
image = PIL.Image.new('RGBA', (w, h)) image = PIL.Image.new('RGBA', (w, h))
for iy in range(yc, hc + 1): for iy in range(yc, hc + 1):
for ix in range(xc, wc + 1): for ix in range(xc, wc + 1):
url = 'https://storage.pixelplanet.fun/%s/%s/%s/%s/tiles/%s/%s.png' % (iter_date[0:4], iter_date[4:6] , iter_date[6:], canvas_id, ix, iy) url = '%s/%s/%s/%s/%s/tiles/%s/%s.png' % (PPFUN_STORAGE_URL, iter_date[0:4], iter_date[4:6] , iter_date[6:], canvas_id, ix, iy)
offx = ix * 256 + offset - x offx = ix * 256 + offset - x
offy = iy * 256 + offset - y offy = iy * 256 + offset - y
tasks.append(fetch(session, url, offx, offy, image, bkg, True)) tasks.append(fetch(session, url, offx, offy, image, bkg, True))
@ -143,10 +117,13 @@ async def get_area(canvas, x, y, w, h, start_date, end_date):
cnt += 1 cnt += 1
#frames.append(image.copy()) #frames.append(image.copy())
image.save('./timelapse/t%s.png' % (cnt)) image.save('./timelapse/t%s.png' % (cnt))
headers = {
'User-Agent': USER_AGENT
}
while True: while True:
async with session.get('https://pixelplanet.fun/history?day=%s&id=%s' % (iter_date, canvas_id)) as resp: async with session.get('%s/history?day=%s&id=%s' % (PPFUN_URL, iter_date, canvas_id), headers=headers) as resp:
try: try:
time_list = json.loads(await resp.text()) time_list = await resp.json()
break break
except: except:
print('Couldn\'t decode json for day %s, trying again' % (iter_date)) print('Couldn\'t decode json for day %s, trying again' % (iter_date))
@ -162,7 +139,7 @@ async def get_area(canvas, x, y, w, h, start_date, end_date):
image_rel = image.copy() image_rel = image.copy()
for iy in range(yc, hc + 1): for iy in range(yc, hc + 1):
for ix in range(xc, wc + 1): for ix in range(xc, wc + 1):
url = 'https://storage.pixelplanet.fun/%s/%s/%s/%s/%s/%s/%s.png' % (iter_date[0:4], iter_date[4:6] , iter_date[6:], canvas_id, time, ix, iy) url = '%s/%s/%s/%s/%s/%s/%s/%s.png' % (PPFUN_STORAGE_URL, iter_date[0:4], iter_date[4:6] , iter_date[6:], canvas_id, time, ix, iy)
offx = ix * 256 + offset - x offx = ix * 256 + offset - x
offy = iy * 256 + offset - y offy = iy * 256 + offset - y
tasks.append(fetch(session, url, offx, offy, image_rel, bkg)) tasks.append(fetch(session, url, offx, offy, image_rel, bkg))
@ -183,41 +160,63 @@ async def get_area(canvas, x, y, w, h, start_date, end_date):
#frames[0].save('timelapse.png', save_all=True, append_images=frames[1:], duration=100, loop=0, default_image=False, blend=1) #frames[0].save('timelapse.png', save_all=True, append_images=frames[1:], duration=100, loop=0, default_image=False, blend=1)
if __name__ == "__main__": async def main():
apime = await fetchMe()
if len(sys.argv) != 5 and len(sys.argv) != 6: if len(sys.argv) != 5 and len(sys.argv) != 6:
print("Download history of an area of pixelplanet - useful for timelapses") print("Download history of an area of pixelplanet - useful for timelapses")
print("") print("")
print("Usage: historyDownload.py canvasId startX_startY endX_endY start_date [end_date]") print("Usage: historyDownload.py canvasID startX_startY endX_endY start_date [end_date]")
print("") print("")
print("→start_date and end_date are in YYYY-MM-dd formate") print("→start_date and end_date are in YYYY-MM-dd formate")
print("→user R key on pixelplanet to copy coordinates)") print("→user R key on pixelplanet to copy coordinates)")
print("→images will be saved into timelapse folder)") print("→images will be saved into timelapse folder)")
print("canvasID: ", end='')
for canvas_id, canvas in apime['canvases'].items():
if 'v' in canvas and canvas['v']:
continue
print(f"{canvas_id} = {canvas['title']}", end=', ')
print()
print("-----------") print("-----------")
print("You can create a timelapse from the resulting files with ffmpeg like that:") print("You can create a timelapse from the resulting files with ffmpeg like that:")
print("ffmpeg -framerate 15 -f image2 -i timelapse/t%d.png -c:v libvpx-vp9 -pix_fmt yuva420p output.webm") print("ffmpeg -framerate 15 -f image2 -i timelapse/t%d.png -c:v libvpx-vp9 -pix_fmt yuva420p output.webm")
print("or lossless example:") print("or lossless example:")
print("ffmpeg -framerate 15 -f image2 -i timelapse/t%d.png -c:v libvpx-vp9 -pix_fmt yuv444p -qmin 0 -qmax 0 -lossless 1 -an output.webm") print("ffmpeg -framerate 15 -f image2 -i timelapse/t%d.png -c:v libvpx-vp9 -pix_fmt yuv444p -qmin 0 -qmax 0 -lossless 1 -an output.webm")
return
canvas_id = sys.argv[1]
if canvas_id not in apime['canvases']:
print("Invalid canvas selected")
return
canvas = apime['canvases'][canvas_id]
if 'v' in canvas and canvas['v']:
print("Can\'t get area for 3D canvas")
return
start = sys.argv[2].split('_')
end = sys.argv[3].split('_')
start_date = datetime.date.fromisoformat(sys.argv[4])
if len(sys.argv) == 6:
end_date = datetime.date.fromisoformat(sys.argv[5])
else: else:
canvas = int(sys.argv[1]) end_date = datetime.date.today()
start = sys.argv[2].split('_') x = int(start[0])
end = sys.argv[3].split('_') y = int(start[1])
start_date = datetime.date.fromisoformat(sys.argv[4]) w = int(end[0]) - x + 1
if len(sys.argv) == 6: h = int( end[1]) - y + 1
end_date = datetime.date.fromisoformat(sys.argv[5]) if not os.path.exists('./timelapse'):
else: os.mkdir('./timelapse')
end_date = datetime.date.today() await get_area(canvas_id, canvas, x, y, w, h, start_date, end_date)
x = int(start[0]) print("Done!")
y = int(start[1]) print("to create a timelapse from it:")
w = int(end[0]) - x + 1 print("ffmpeg -framerate 15 -f image2 -i timelapse/t%d.png -c:v libvpx-vp9 -pix_fmt yuva420p output.webm")
h = int( end[1]) - y + 1 print("example with scaling *3 and audio track:")
loop = asyncio.get_event_loop() print("ffmpeg -i ./audio.mp3 -framerate 8 -f image2 -i timelapse/t%d.png -map 0:a -map 1:v -vf scale=iw*3:-1 -shortest -c:v libvpx-vp9 -c:a libvorbis -pix_fmt yuva420p output.webm")
if not os.path.exists('./timelapse'): print("lossless example:")
os.mkdir('./timelapse') print("ffmpeg -framerate 15 -f image2 -i timelapse/t%d.png -c:v libvpx-vp9 -pix_fmt yuv444p -qmin 0 -qmax 0 -lossless 1 -an output.webm")
loop.run_until_complete(get_area(canvas, x, y, w, h, start_date, end_date))
print("Done!") if __name__ == "__main__":
print("to create a timelapse from it:") asyncio.run(main())
print("ffmpeg -framerate 15 -f image2 -i timelapse/t%d.png -c:v libvpx-vp9 -pix_fmt yuva420p output.webm")
print("example with scaling *3 and audio track:")
print("ffmpeg -i ./audio.mp3 -framerate 8 -f image2 -i timelapse/t%d.png -map 0:a -map 1:v -vf scale=iw*3:-1 -shortest -c:v libvpx-vp9 -c:a libvorbis -pix_fmt yuva420p output.webm")
print("lossless example:")
print("ffmpeg -framerate 15 -f image2 -i timelapse/t%d.png -c:v libvpx-vp9 -pix_fmt yuv444p -qmin 0 -qmax 0 -lossless 1 -an output.webm")