Updated DTutils and folder simultarion (logs, reception errors, etc)

This commit is contained in:
Emilio Martinez 2023-06-26 20:05:50 -03:00
parent acd5afa624
commit 25921298c3
2 changed files with 70 additions and 43 deletions

View File

@ -22,6 +22,9 @@ from scipy import signal
from PIL import Image from PIL import Image
from utils.DTutils import TMDS_encoding_original, TMDS_serial from utils.DTutils import TMDS_encoding_original, TMDS_serial
import sys import sys
import logging
from utils import utils_logger
from datetime import datetime
#%% #%%
@ -47,7 +50,7 @@ def image_transmition_simulation(I, blanking=False):
return I_TMDS_Tx, I_TMDS.shape return I_TMDS_Tx, I_TMDS.shape
def image_capture_simulation(I_Tx, h_total, v_total, N_harmonic, noise_std=0, def image_capture_simulation(I_Tx, h_total, v_total, N_harmonic, noise_std=0,
fps=60): fps=60, freq_error=0, phase_error=0):
# Compute pixelrate and bitrate # Compute pixelrate and bitrate
px_rate = h_total*v_total*fps px_rate = h_total*v_total*fps
@ -75,13 +78,14 @@ def image_capture_simulation(I_Tx, h_total, v_total, N_harmonic, noise_std=0,
# AM modulation frequency according to pixel harmonic # AM modulation frequency according to pixel harmonic
harm = N_harmonic*px_rate harm = N_harmonic*px_rate
# Harmonic oscilator # Harmonic oscilator (including frequency and phase error)
baseband_exponential = np.exp(2j*np.pi*harm*t_continuous) baseband_exponential = np.exp(2j*np.pi*(harm+freq_error)*t_continuous + 1j*phase_error)
usrp_rate = 50e6 usrp_rate = 50e6
usrp_BW = usrp_rate/2
# AM modulation and SDR sampling # AM modulation and SDR sampling
I_Rx = signal.resample_poly(I_Tx_noisy*baseband_exponential,up=int(usrp_rate), down=sample_rate) I_Rx = signal.resample_poly(I_Tx_noisy*baseband_exponential,up=int(usrp_BW), down=sample_rate)
# Reshape signal to the image size # Reshape signal to the image size
I_capture = signal.resample(I_Rx, h_total*v_total).reshape(v_total,h_total) I_capture = signal.resample(I_Rx, h_total*v_total).reshape(v_total,h_total)
@ -97,58 +101,81 @@ def save_simulation_image(I,path_and_name):
I_real = np.real(I) I_real = np.real(I)
I_imag = np.imag(I) I_imag = np.imag(I)
realmax = I_real.max() I_save[:,:,0], I_save[:,:,1] = I_real, I_imag
realmin = I_real.min() min_value, max_value = np.min(I_save[:,:,:2]), np.max(I_save[:,:,:2])
imagmax = I_imag.max() I_save[:,:,0] = 255*(I_real-min_value)/(max_value-min_value)
imagmin = I_imag.min() I_save[:,:,1] = 255*(I_imag-min_value)/(max_value-min_value)
# Stretch contrast on every channel
I_save[:,:,0] = 255*(I_real-realmin)/(realmax-realmin)
I_save[:,:,1] = 255*(I_imag-imagmin)/(imagmax-imagmin)
im = Image.fromarray(I_save.astype('uint8')) im = Image.fromarray(I_save.astype('uint8'))
im.save(path_and_name) im.save(path_and_name)
def main(): def main():
logs_dir = './logfiles/'
# Create logs directory if not exist
if not os.path.exists(logs_dir):
os.mkdir(logs_dir)
logger_name = 'simulations_'+datetime.now().strftime("%d-%m-%Y_%H:%M:%S")
utils_logger.logger_info(logger_name, logs_dir+logger_name+'.log')
logger = logging.getLogger(logger_name)
# Get foldername argument # Get foldername argument
foldername = sys.argv[-1] foldername = sys.argv[-1]
message = f'Tempest capture simulation for image folder {foldername}\n'
logger.info(message)
# Get images and subfolders names # Get images and subfolders names
images = get_images_names_from_folder(foldername) images = get_images_names_from_folder(foldername)
simulations_folder = foldername+'/simulations/' # Create simulation directory if not exist at the folder path
simulations_path = foldername+'/simulations/'
if not os.path.exists(simulations_path):
os.mkdir(simulations_path)
message = f'Created simulation directory at {simulations_path}\n'
logger.info(message)
os.mkdir(simulations_folder) # Possible noise std values
# noise_stds = np.array([ 0, 5, 10, 15, 20, 25])
for image in images:
# timestamp for simulation starting # timestamp for simulation starting
t1_image = time.time() t1_image = time.time()
for image in images:
# Read image # Read image
image_path = foldername+'/'+image image_path = foldername+'/'+image
I = imread(image_path) I = imread(image_path)
# TMDS coding and bit serialization
I_Tx, resolution = image_transmition_simulation(I)
v_res, h_res, _ = resolution
# Choose random pixelrate harmonic number # Choose random pixelrate harmonic number
N_harmonic = np.random.randint(1,10) N_harmonic = np.random.randint(1,10)
message = f'Initiate simulation for image {image} with {N_harmonic} pixel harmonic frequency'
logger.info(message)
# TMDS coding and bit serialization
I_Tx, resolution = image_transmition_simulation(I, blanking=True)
v_res, h_res, _ = resolution
I_capture = image_capture_simulation(I_Tx, h_res, v_res, N_harmonic) I_capture = image_capture_simulation(I_Tx, h_res, v_res, N_harmonic)
path = simulations_folder+image path = simulations_path+image
save_simulation_image(I_capture,path) save_simulation_image(I_capture,path)
# timestamp for simulation ending # timestamp for simulation ending
t2_image = time.time() t2_image = time.time()
t_images = t2_images-t1_images t_images = t2_image-t1_image
message = 'Processing time: {:.2f}'.format(t_images)+'s\n'
logger.info(message)
print('\nTiempo total de las '+str(len(images))+' simulaciones:','{:.2f}'.format(t_images)+'s\n')
if __name__ == "__main__": if __name__ == "__main__":
main() main()

View File

@ -110,33 +110,33 @@ def TMDS_pixel_rare (pix):
""" """
# Convert 8-bit pixel to binary list D # Convert 8-bit pixel to binary list D
D = uint8_to_binarray(pix) d = uint8_to_binarray(pix)
# Initialize output q # Initialize output q
qm = [D[0]] Qm = [d[0]]
# 1's unbalanced condition at current pixel # 1's unbalanced condition at current pixel
N1_D = np.sum(D) N1_D = np.sum(d)
if N1_D>4 or (N1_D==4 and not(D[0])): if N1_D>4 or (N1_D==4 and not(d[0])):
# XNOR of consecutive bits # XNOR of consecutive bits
for k in range(1,8): for k in range(1,8):
qm.append( not(qm[k-1] ^ D[k]) ) Qm.append( not(Qm[k-1] ^ d[k]) )
qm.append(0) Qm.append(0)
else: else:
# XOR of consecutive bits # XOR of consecutive bits
for k in range(1,8): for k in range(1,8):
qm.append( qm[k-1] ^ D[k] ) Qm.append( Qm[k-1] ^ d[k] )
qm.append(1) Qm.append(1)
qm.append(np.random.choice([0,1])) Qm.append(np.random.choice([0,1]))
# Return the TMDS coded pixel as uint and 0's y 1's balance # Return the TMDS coded pixel as uint and 0's y 1's balance
return binarray_to_uint(qm) return binarray_to_uint(Qm)
@jit(nopython=True) @jit(nopython=True)
def TMDS_pixel_numba(pix:uint8, cnt:int8)->tuple: def TMDS_pixel_numba(pix:uint8, cnt:int8)->tuple:
@ -293,11 +293,11 @@ def TMDS_encoding_original (I, blanking = False):
if blanking: if blanking:
# Get blanking resolution for input image # Get blanking resolution for input image
v = (v_in==1080)*1125 + (v_in==720)*750 + (v_in==600)*628 + (v_in==480)*525 v = (v_in==1080)*1125 + (v_in==900)*1000 + (v_in==720)*750 + (v_in==600)*628 + (v_in==480)*525
h = (h_in==1920)*2200 + (h_in==1280)*1650 + (h_in==800)*1056 + (h_in==640)*800 h = (h_in==1920)*2200 + (h_in==1600)*1800 + (h_in==1280)*1650 + (h_in==800)*1056 + (h_in==640)*800
vdiff = v - v_in v_diff = v - v_in
hdiff = h - h_in h_diff = h - h_in
# Create image with blanking and change type to uint16 # Create image with blanking and change type to uint16
# Assuming the blanking corresponds to 10bit number [0, 0, 1, 0, 1, 0, 1, 0, 1, 1] (LSB first) # Assuming the blanking corresponds to 10bit number [0, 0, 1, 0, 1, 0, 1, 0, 1, 1] (LSB first)