import os import time from random import randint from ast import literal_eval import numpy as np import cv2 as cv import pymemcache from avdb.db import DBConnection, DBOps LIMIT=100 LEVEL=randint(3,7) SEARCH_TILE_OFFSET=-2 class ImagePyramidState: def __init__(self, levels=8): self._connection = DBConnection('testdb') self._memcache = pymemcache.Client("127.0.0.1:11211") self.image_width = 1280 self.image_height = 960 self.levels = levels self.colorspace = 'lab' self.state_map = {} self.img_map = {} self.src_img_id, self.xlimit, self.ylimit, _ = self.select_initial_tile(src_img_id=None,level=LEVEL, inception=True) filename = self.get_imagefile_token_from_id(self.src_img_id) self.image_filetoken = os.path.basename(filename).removesuffix(".jpeg") print(self.image_filetoken) self.initialize_image_map() self.initialize_state_map() def initialize_state_map(self): """ Initialize state arrays with 0s (0 = original/no change) """ for lvl in range(1,8): self.state_map[lvl] = np.zeros( ( # int(self.image_height/(2**lvl)), # int(self.image_width/(2**lvl)), int(self.xlimit), int(self.ylimit), 3 ), dtype=np.uint16 ) def initialize_image_map(self): """ Initialize image arrays with downsampled images """ for lvl in range(1,8): imgpath = f"/home/kfl/dev/git/avdb-py-v1/img/{self.image_filetoken}/{self.image_filetoken}.{self.colorspace}.pyr_{lvl}_display.jpeg" img = cv.imread(imgpath) print(f"imgpath:{imgpath} (level:) SHAPE: {img.shape}") self.img_map[lvl] = img def match_tile(self, src_img_id, src_tilex, src_tiley, search_tile, level=3, limit=100): """ pyrdata_id, img_id, level, tilex_a, tiley_a, x_a, y_a, x0_a, y0_a, w_a, h_a, tilex_b, tiley_b, x_b, y_b, x0_b, y0_b, w_b, h_b, hash, tile_data -- and tilex_a = {src_tilex} -- and tiley_a = {src_tiley} """ search_query = f""" select *, (tile_data <-> '{search_tile}') from pyramiddata_2tiles where level = {level} and img_id <> {src_img_id} order by tile_data <-> '{search_tile}' asc limit {limit}; """ psqldata = self._memcache.get(f"{str(search_tile).replace(', ', ',')}_{src_img_id}_{level}_{limit}") if not psqldata: with self._connection.get_cursor() as cursor: cursor.execute(search_query, (search_tile, level, src_img_id, LIMIT)) rows = cursor.fetchall() self._memcache.set(f"{str(search_tile).replace(', ', ',')}_{src_img_id}_{level}_{limit}", rows) return rows else: print ("Loaded data from memcached") return tuple(literal_eval(psqldata.decode("utf-8"))) def get_imagefile_token_from_id(self, src_img_id): """ """ if src_img_id is not None: search_q = f""" select * from avdb_image where img_id = {src_img_id}; """ # print(search_q) with self._connection.get_cursor() as cursor: cursor.execute(search_q, src_img_id) row = cursor.fetchone() if row is None: raise ValueError("No image found with the given img_id.") return row[2] def select_initial_tile(self, src_img_id=None, level=LEVEL, src_tilex=None, src_tiley=None, inception=False): """ """ if src_img_id is None: src_img_id = randint(1, 5) # 845) if src_tilex is None: src_tilex = randint(0, (1280 // (2**level)) - 5) if src_tiley is None: src_tiley = randint(0, (960 // (2**level)) - 5) search_query = f""" select * from pyramiddata_2tiles where level = {level} and img_id = {src_img_id} and tilex_a = {src_tilex} and tiley_a = {src_tiley} limit {LIMIT}; """ # print(search_query) with self._connection.get_cursor() as cursor: cursor.execute(search_query, (level, src_img_id, src_tilex, src_tiley)) row = cursor.fetchone() if inception: return src_img_id, (1280 // (2**level)) - 1, (960 // (2**level)) - 1, row return src_img_id, src_tilex, src_tiley, row def run_generation(self): """ """ # choose random tile to update level = LEVEL generation_counter = 0 # grid = [] # for y in range(0, self.ylimit - 6, 3): # for x in range(0, self.xlimit - 6, 3): # grid += [(y,x)] # print(f"limits: {(self.ylimit, self.xlimit)}") # for (y,x) in grid: while True: lvl = level # _, tilex, tiley, init_tile = self.select_initial_tile(src_img_id=self.src_img_id, level=lvl, src_tilex=x, src_tiley=y, inception=False) _, x, y, init_tile = self.select_initial_tile(src_img_id=self.src_img_id, level=lvl, src_tilex=None, src_tiley=None, inception=False) if init_tile is not None: # print(f"INIT: {init_tile}") # find similar tiles to swap in found_tiles = self.match_tile( src_img_id=self.src_img_id, src_tilex=x, src_tiley=y, search_tile=init_tile[-1], level=lvl ) # print(found_tiles[:3]) next_tile = found_tiles[randint(0, len(found_tiles)-1)] # update img + state map sz = 2**lvl # print(next_tile) next_tile_ = list(map(int, next_tile[SEARCH_TILE_OFFSET].replace('[','').replace(']','').split(','))) # print(f"next_tile_: {next_tile_}") # print(f"{next_tile[1]}, {init_tile[3]}, {init_tile[4]}") self.img_map[lvl][y*sz:(y+1)*sz,x*sz:(x+1)*sz] = [ next_tile_[0], next_tile_[1], next_tile_[2] ] if generation_counter % 100 == 0: shape = self.state_map[lvl].shape print(shape) for x in range(shape[0]): for y in range(shape[1]): st = self.state_map[lvl][x,y] if st[0] > 0 or st[1] > 0.0 or st[2] > 0.0: print( f"x: {x}, y: {y} st: {st}" ) self.state_map[lvl][x, y] = [next_tile[1], next_tile[3], next_tile[4]] cv.namedWindow("Image", cv.WINDOW_NORMAL) cv.resizeWindow("Image", 1280, 960) cv.imshow("Image", cv.cvtColor(self.img_map[lvl], cv.COLOR_Lab2RGB)) time.sleep(0.2) cv.waitKey(1) generation_counter += 1 if __name__ == '__main__': ips = ImagePyramidState() for i in range(100): ips.run_generation()