voussoir
/
pixelcanvas
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
pixelcanvas/pixelcanvas.py

593 lines
19 KiB
Python
Raw Blame History

import argparse
import datetime
import gzip
import PIL.Image
import random
import requests
import sqlite3
import sys
import time
from voussoirkit import threadpool
from voussoirkit import vlogging
log = vlogging.getLogger(__name__, 'pixelcanvasdl')
vlogging.getLogger('urllib3.connectionpool').setLevel(vlogging.CRITICAL)
WHITE = (255, 255, 255)
LIGHTGRAY = (228, 228, 228)
DARKGRAY = (136, 136, 136)
BLACK = (34, 34, 34)
PINK = (255, 167, 209)
RED = (229, 0, 0)
ORANGE = (229, 149, 0)
BROWN = (160, 106, 66)
YELLOW = (229, 217, 0)
LIGHTGREEN = (148, 224, 68)
DARKGREEN = (2, 190, 1)
LIGHTBLUE = (0, 211, 221)
MEDIUMBLUE = (0, 131, 199)
DARKBLUE = (0, 0, 234)
LIGHTPURPLE = (207, 110, 228)
DARKPURPLE = (130, 0, 128)
COLOR_MAP = {
0: WHITE,
1: LIGHTGRAY,
2: DARKGRAY,
3: BLACK,
4: PINK,
5: RED,
6: ORANGE,
7: BROWN,
8: YELLOW,
9: LIGHTGREEN,
10: DARKGREEN,
11: LIGHTBLUE,
12: MEDIUMBLUE,
13: DARKBLUE,
14: LIGHTPURPLE,
15: DARKPURPLE,
}
# The width and height of a chunk, in pixels.
CHUNK_SIZE_PIX = 64
# The number of bytes for a full chunk.
# They are 32x64 because each byte represents two 4-bit pixels.
CHUNK_SIZE_BYTES = int(CHUNK_SIZE_PIX * (CHUNK_SIZE_PIX / 2))
# The width and height of a bigchunk, in chunks.
BIGCHUNK_SIZE_CHUNKS = 15
# The width and height of a bigchunk, in pixels.
BIGCHUNK_SIZE_PIX = BIGCHUNK_SIZE_CHUNKS * CHUNK_SIZE_PIX
# The number of bytes for a full bigchunk.
BIGCHUNK_SIZE_BYTES = int(BIGCHUNK_SIZE_PIX * (BIGCHUNK_SIZE_PIX / 2))
# The chunk 0, 0 has a pixel coordinate of -448, -448 for some reason.
ORIGIN_OFFSET_X = 448
ORIGIN_OFFSET_Y = 448
DB_INIT = '''
BEGIN;
CREATE TABLE IF NOT EXISTS chunks (x INT, y INT, data BLOB, updated_at REAL);
CREATE INDEX IF NOT EXISTS chunks_x_y ON chunks(x, y);
COMMIT;
'''
sql = sqlite3.connect('pixelcanvas.db')
cur = sql.cursor()
cur.executescript(DB_INIT)
session = requests.Session()
# HELPER FUNCTIONS
################################################################################
def now():
n = datetime.datetime.now(datetime.timezone.utc)
return n.timestamp()
# DB FUNCTIONS
################################################################################
def db_commit():
log.debug('Committing')
sql.commit()
def get_chunk_from_db(chunk_x, chunk_y, as_of=None):
'''
Get the chunk from the database, and raise IndexError if it doesn't exist.
'''
query = f'''
SELECT x, y, data FROM chunks
WHERE x == ? AND y == ?
{'AND updated_at <= ?' if as_of is not None else ''}
ORDER BY updated_at DESC
LIMIT 1
'''
bindings = [chunk_x, chunk_y]
if as_of is not None:
bindings.append(as_of)
cur.execute(query, bindings)
fetch = cur.fetchone()
if fetch is None:
raise IndexError(f'{chunk_x}, {chunk_y}')
(x, y, data) = fetch
data = gzip.decompress(data)
return (x, y, data)
def get_chunk(chunk_x, chunk_y, *args, **kwargs):
'''
Get the chunk from the database if it exists, or else download it.
'''
try:
return get_chunk_from_db(chunk_x, chunk_y, *args, **kwargs)
except IndexError:
(bigchunk_x, bigchunk_y) = chunk_to_bigchunk(chunk_x, chunk_y)
chunks = download_bigchunk(bigchunk_x, bigchunk_y)
insert_chunks(chunks)
return get_chunk_from_db(chunk_x, chunk_y, *args, **kwargs)
def insert_chunk(chunk_x, chunk_y, data, commit=True):
try:
existing_chunk = get_chunk_from_db(chunk_x, chunk_y)
except IndexError:
pass
else:
if data == existing_chunk[2]:
return
log.debug('Updating chunk %s %s', chunk_x, chunk_y)
data = gzip.compress(data)
cur.execute('INSERT INTO chunks VALUES(?, ?, ?, ?)', [chunk_x, chunk_y, data, now()])
if commit:
db_commit()
def insert_chunks(chunks, commit=True):
for (index, chunk) in enumerate(chunks):
if index % 25000 == 0 and commit:
db_commit()
insert_chunk(*chunk, commit=False)
if commit:
db_commit()
# API FUNCTIONS
################################################################################
def url_for_bigchunk(bigchunk_x, bigchunk_y):
return f'http://api.pixelcanvas.io/api/bigchunk/{bigchunk_x}.{bigchunk_y}.bmp'
def request(url):
response = session.get(url)
response.raise_for_status()
return response
def download_bigchunk(bigchunk_x, bigchunk_y):
'''
Download a bigchunk and return the list of chunks.
'''
url = url_for_bigchunk(bigchunk_x, bigchunk_y)
log.info('Downloading %s', url)
response = request(url)
bigchunk_data = response.content
if len(bigchunk_data) != BIGCHUNK_SIZE_BYTES:
message = 'Received bigchunk does not matched the expected byte size!\n'
message += 'Got %d instead of %d' % (len(bigchunk_data), BIGCHUNK_SIZE_BYTES)
raise ValueError(message)
chunks = split_bigchunk(bigchunk_x, bigchunk_y, bigchunk_data)
return chunks
def download_bigchunk_range(bigchunk_xy1, bigchunk_xy2, shuffle=False, threads=1):
'''
Given (UPPERLEFT_X, UPPERLEFT_Y), (LOWERRIGHT_X, LOWERRIGHT_Y),
download multiple bigchunks, and yield all of the small chunks.
'''
if threads < 1:
raise ValueError(threads)
log.debug('Downloading bigchunk range %s-%s', bigchunk_xy1, bigchunk_xy2)
bigchunks = bigchunk_range_iterator(bigchunk_xy1, bigchunk_xy2)
if shuffle:
bigchunks = list(bigchunks)
random.shuffle(bigchunks)
if threads == 1:
for (x, y) in bigchunks:
chunks = download_bigchunk(x, y)
yield from chunks
else:
pool = threadpool.ThreadPool(size=threads)
job_gen = (
{'function': download_bigchunk, 'args': (x, y), 'name': (x, y),}
for (x, y) in bigchunks
)
pool.add_generator(job_gen)
for job in pool.result_generator():
job.join()
if job.exception:
raise job.exception
yield from job.value
# CHUNK FUNCTIONS
################################################################################
def chunk_range_iterator(chunk_xy1, chunk_xy2):
'''
Given (UPPERLEFT_X, UPPERLEFT_Y), (LOWERRIGHT_X, LOWERRIGHT_Y),
yield (x, y) pairs for chunks in this range, inclusive.
'''
for x in range(chunk_xy1[0], chunk_xy2[0] + 1):
for y in range(chunk_xy1[1], chunk_xy2[1] + 1):
yield (x, y)
def bigchunk_range_iterator(bigchunk_xy1, bigchunk_xy2):
'''
Given (UPPERLEFT_X, UPPERLEFT_Y), (LOWERRIGHT_X, LOWERRIGHT_Y),
yield (x, y) pairs for bigchunks in this range, inclusive.
'''
for x in range(bigchunk_xy1[0], bigchunk_xy2[0] + BIGCHUNK_SIZE_CHUNKS, BIGCHUNK_SIZE_CHUNKS):
for y in range(bigchunk_xy1[1], bigchunk_xy2[1] + BIGCHUNK_SIZE_CHUNKS, BIGCHUNK_SIZE_CHUNKS):
yield (x, y)
def chunk_to_bigchunk(chunk_x, chunk_y):
bigchunk_x = (chunk_x // BIGCHUNK_SIZE_CHUNKS) * BIGCHUNK_SIZE_CHUNKS
bigchunk_y = (chunk_y // BIGCHUNK_SIZE_CHUNKS) * BIGCHUNK_SIZE_CHUNKS
log.loud('Converted chunk %s, %s to bigchunk %s, %s', chunk_x, chunk_y, bigchunk_x, bigchunk_y)
return (bigchunk_x, bigchunk_y)
def chunk_range_to_bigchunk_range(chunk_xy1, chunk_xy2):
bigchunk_range = (chunk_to_bigchunk(*chunk_xy1), chunk_to_bigchunk(*chunk_xy2))
return bigchunk_range
def chunk_to_pixel(chunk_x, chunk_y):
pixel_x = chunk_x * CHUNK_SIZE_PIX - ORIGIN_OFFSET_X
pixel_y = chunk_y * CHUNK_SIZE_PIX - ORIGIN_OFFSET_Y
log.loud('Converted chunk %s, %s to pixel %s, %s', chunk_x, chunk_y, pixel_x, pixel_y)
return (pixel_x, pixel_y)
def chunk_range_to_pixel_range(chunk_xy1, chunk_xy2):
pixel_range = (chunk_to_pixel(*chunk_xy1), chunk_to_pixel(*chunk_xy2))
return pixel_range
def pixel_to_chunk(pixel_x, pixel_y):
chunk_x = (pixel_x + ORIGIN_OFFSET_X) // CHUNK_SIZE_PIX
chunk_y = (pixel_y + ORIGIN_OFFSET_Y) // CHUNK_SIZE_PIX
log.loud('Converted pixel %s, %s to chunk %s, %s', pixel_x, pixel_y, chunk_x, chunk_y)
return (chunk_x, chunk_y)
def pixel_range_to_chunk_range(pixel_xy1, pixel_xy2):
chunk_range = (pixel_to_chunk(*pixel_xy1), pixel_to_chunk(*pixel_xy2))
log.loud('Converted pixel range %s, %s to chunk range %s, %s', pixel_xy1, pixel_xy2, *chunk_range)
return chunk_range
def pixel_to_bigchunk(pixel_x, pixel_y):
bigchunk_x = ((pixel_x + ORIGIN_OFFSET_X) // BIGCHUNK_SIZE_PIX) * BIGCHUNK_SIZE_CHUNKS
bigchunk_y = ((pixel_y + ORIGIN_OFFSET_Y) // BIGCHUNK_SIZE_PIX) * BIGCHUNK_SIZE_CHUNKS
log.loud('Converted pixel %s, %s to bigchunk %s, %s', pixel_x, pixel_y, bigchunk_x, bigchunk_y)
return (bigchunk_x, bigchunk_y)
def pixel_range_to_bigchunk_range(pixel_xy1, pixel_xy2):
bigchunk_range = (pixel_to_bigchunk(*pixel_xy1), pixel_to_bigchunk(*pixel_xy2))
log.loud('Converted pixel range %s, %s to bigchunk range %s, %s', pixel_xy1, pixel_xy2, *bigchunk_range)
return bigchunk_range
def split_bigchunk(bigchunk_x, bigchunk_y, bigchunk_data):
'''
Chunks are downloaded from the site as a "bigchunk" which is just 15x15
chunks stitched together.
The chunks are arranged left to right, top to bottom.
For example, the byte stream:
000011112222333344445555666677778888
represents the bitmap:
001122
001122
334455
334455
667788
667788
'''
chunks = []
chunk_count = int(len(bigchunk_data) / CHUNK_SIZE_BYTES)
for chunk_index in range(chunk_count):
chunk_x = (chunk_index % BIGCHUNK_SIZE_CHUNKS) + bigchunk_x
chunk_y = (chunk_index // BIGCHUNK_SIZE_CHUNKS) + bigchunk_y
start_index = chunk_index * CHUNK_SIZE_BYTES
end_index = start_index + CHUNK_SIZE_BYTES
chunk_data = bigchunk_data[start_index:end_index]
chunk = (chunk_x, chunk_y, chunk_data)
chunks.append(chunk)
return chunks
# IMAGE FUNCTIONS
################################################################################
def chunk_to_rgb(chunk_data):
'''
Convert the data chunk into RGB tuples.
PixelCanvas chunks are strings of bytes where every byte represents two
horizontal pixels. Each pixel is 4 bits since there are 16 colors.
Chunks are 32x64 bytes for a total of 64x64 pixels.
'''
# Each byte actually represents two horizontal pixels. 8F is actually 8, F.
# So create a generator that takes in the bytes and yields the pixel bits.
pixels = (
pixel
for byte in chunk_data
for pixel in (byte >> 4, byte & 0xf)
)
matrix = [None for x in range(len(chunk_data) * 2)]
for (index, pixel) in enumerate(pixels):
px = index % CHUNK_SIZE_PIX
py = index // CHUNK_SIZE_PIX
matrix[(py * CHUNK_SIZE_PIX) + px] = COLOR_MAP[pixel]
return matrix
def rgb_to_image(matrix):
matrix = bytes([color for pixel in matrix for color in pixel])
i = PIL.Image.frombytes(mode='RGB', size=(CHUNK_SIZE_PIX, CHUNK_SIZE_PIX), data=matrix)
return i
def chunk_to_image(chunk_data, scale=1):
image = rgb_to_image(chunk_to_rgb(chunk_data))
if scale is not None and scale != 1:
new_size = (int(image.size[0] * scale), int(image.size[1] * scale))
image = image.resize(new_size, resample=PIL.Image.NEAREST)
return image
def chunks_to_image(chunks, scale=1):
'''
Combine all of the given chunks into a single image.
'''
log.debug('Creating image from %s chunks', len(chunks))
min_x = min(chunk[0] for chunk in chunks)
max_x = max(chunk[0] for chunk in chunks)
min_y = min(chunk[1] for chunk in chunks)
max_y = max(chunk[1] for chunk in chunks)
span_x = max_x - min_x + 1
span_y = max_y - min_y + 1
img_width = int(span_x * CHUNK_SIZE_PIX * scale)
img_height = int(span_y * CHUNK_SIZE_PIX * scale)
img = PIL.Image.new(mode='RGB', size=(img_width, img_height), color=WHITE)
for (chunk_x, chunk_y, chunk_data) in chunks:
paste_x = int((chunk_x - min_x) * CHUNK_SIZE_PIX * scale)
paste_y = int((chunk_y - min_y) * CHUNK_SIZE_PIX * scale)
chunk_image = chunk_to_image(chunk_data, scale)
img.paste(chunk_image, (paste_x, paste_y))
return img
def crop_image(image, pixel_xy1, pixel_xy2):
'''
Because the images are rendered on a chunk basis, they are probably larger
than the exact area that you want. Use this function to crop the image to
the exact coordinates.
pixel_xy1 and pixel_xy2 are the world coordinates that you used to get this
image in the first place, not coordinates within this picture.
'''
img_width = pixel_xy2[0] - pixel_xy1[0] + 1
img_height = pixel_xy2[1] - pixel_xy1[1] + 1
basis_xy = chunk_to_pixel(*pixel_to_chunk(*pixel_xy1))
xy1 = (pixel_xy1[0] - (basis_xy[0]), pixel_xy1[1] - (basis_xy[1]))
xy2 = (xy1[0] + img_width, xy1[1] + img_height)
bbox = (xy1[0], xy1[1], xy2[0], xy2[1])
log.debug('Cropping image down to %s', bbox)
image = image.crop(bbox)
return image
# COMMAND LINE
################################################################################
from voussoirkit import betterhelp
def parse_coordinate_string(coordinates):
'''
Convert the given '~100.~100--100.100' to ((-100, -100), (100, 100)).
'''
coordinates = coordinates.strip()
if '--' in coordinates:
(xy1, xy2) = coordinates.split('--', 1)
else:
# If you are only updating a single chunk.
xy1 = coordinates
xy2 = coordinates
def split_xy(xy):
xy = xy.replace('~', '-')
xy = xy.replace(',', '.')
(x, y) = xy.split('.')
return (int(x), int(y))
(xy1, xy2) = (split_xy(xy1), split_xy(xy2))
log.debug('Parsed coordinates %s into %s %s', coordinates, xy1, xy2)
return (xy1, xy2)
def overview_argparse(args):
cur.execute('SELECT x, y, updated_at FROM chunks GROUP BY x, y ORDER BY updated_at DESC')
chunks = cur.fetchall()
min_x = min(chunk[0] for chunk in chunks)
max_x = max(chunk[0] for chunk in chunks)
min_y = min(chunk[1] for chunk in chunks)
max_y = max(chunk[1] for chunk in chunks)
width = max_x - min_x + 1
height = max_y - min_y + 1
x_offset = abs(min(min_x, 0))
y_offset = abs(min(min_y, 0))
matrix = [[' ' for x in range(width)] for y in range(height)]
for (x, y, updated_at) in chunks:
x += x_offset
y += y_offset
matrix[y][x] = '.'
for (x, y, updated_at) in chunks:
if (x % 15 == 0) and (y % 15 == 0):
text = f'{x},{y}'
x += x_offset
y += y_offset
for c in text:
matrix[y][x] = c
x+=1
for row in matrix:
for column in row:
print(column, end='')
print()
def render_argparse(args):
coordinates = parse_coordinate_string(args.coordinates)
if args.is_chunks:
chunk_range = coordinates
coordinates = chunk_range_to_pixel_range(*coordinates)
else:
chunk_range = pixel_range_to_chunk_range(*coordinates)
chunks = [get_chunk(*chunk_xy) for chunk_xy in chunk_range_iterator(*chunk_range)]
image = chunks_to_image(chunks, scale=args.scale)
if args.scale == 1 and not args.is_chunks:
image = crop_image(image, *coordinates)
if args.do_show:
image.show()
else:
((p1x, p1y), (p2x, p2y)) = coordinates
scale_s = f'_{args.scale}' if args.scale != 1 else ''
filename = f'{p1x}.{p1y}--{p2x}.{p2y}{scale_s}.png'
image.save(filename)
log.debug('Wrote %s', filename)
def update_argparse(args):
coordinates = parse_coordinate_string(args.coordinates)
if args.is_chunks:
bigchunk_range = chunk_range_to_bigchunk_range(*coordinates)
else:
bigchunk_range = pixel_range_to_bigchunk_range(*coordinates)
chunks = download_bigchunk_range(*bigchunk_range, shuffle=args.shuffle, threads=args.threads)
try:
insert_chunks(chunks)
except KeyboardInterrupt:
db_commit()
@vlogging.main_decorator
def main(argv):
parser = argparse.ArgumentParser(
description='''
This tool is run from the command line, where you provide the coordinates you
want to download and render.
The format for coordinates is `UPPERLEFT--LOWERRIGHT`. The format for each
of those pieces is `X.Y`.
Sometimes, argparse gets confused by negative coordinates because it thinks
you're trying to provide another argument. If this happens, use a tilde `~`
as the negative sign instead.
Remember, because this is an image, up and left are negative; down and right
are positive.
''',
)
subparsers = parser.add_subparsers()
################################################################################################
p_update = subparsers.add_parser(
'update',
description='''
Download chunks into the database.
''',
)
p_update.examples = [
'0.0--100.100',
'~100.~100--100.100',
'~1200.300--~900.600',
]
p_update.add_argument(
'coordinates',
)
p_update.add_argument(
'--chunks',
dest='is_chunks',
action='store_true',
help='''
The coordinates which you provided are chunk coordinates instead of
pixel coordinates.
''',
)
p_update.add_argument(
'--shuffle',
action='store_true',
help='''
Download chunks in a random order instead of from corner to corner.
''',
)
p_update.add_argument(
'--threads',
type=int,
default=1,
help='''
Use X threads to download bigchunks.
''',
)
p_update.set_defaults(func=update_argparse)
################################################################################################
p_render = subparsers.add_parser(
'render',
description='''
Export an image as PNG.
''',
)
p_render.examples = [
'0.0--100.100',
'~100.~100--100.100 --scale 2',
'~1200.300--~900.600 --show',
]
p_render.add_argument(
'coordinates',
)
p_render.add_argument(
'--chunks',
dest='is_chunks',
action='store_true',
help='''
The coordinates which you provided are chunk coordinates instead of
pixel coordinates.
''',
)
p_render.add_argument(
'--show',
dest='do_show',
action='store_true',
help='''
Render the image at a different scale.
For best results, use powers of 2 like 0.5, 0.25, etc.
This will disable the autocropping.
''',
)
p_render.add_argument(
'--scale',
dest='scale',
type=float,
default=1,
help='''
Instead of saving the image, display it on the screen.
pillow.readthedocs.io/en/stable/reference/Image.html#PIL.Image.Image.show
''',
)
p_render.set_defaults(func=render_argparse)
################################################################################################
p_overview = subparsers.add_parser(
'overview',
help='''
Draw an ascii map representing the owned chunks.
''',
)
p_overview.set_defaults(func=overview_argparse)
return betterhelp.go(parser, argv)
if __name__ == '__main__':
raise SystemExit(main(sys.argv[1:]))
Reference in New Issue View Git Blame Copy Permalink