259 lines
9.5 KiB
Python
259 lines
9.5 KiB
Python
|
import os
|
||
|
|
import UniTAP
|
||
|
|
import time
|
||
|
|
import math
|
||
|
|
import threading
|
||
|
|
|
||
|
|
#
|
||
|
|
# Config.
|
||
|
|
#
|
||
|
|
|
||
|
|
BULK_SIZE_MB = 100 # Mb
|
||
|
|
FOLDER_TO_GENERATE = "./BulkData"
|
||
|
|
|
||
|
|
tps3 = True
|
||
|
|
tps4 = True
|
||
|
|
fec = True
|
||
|
|
|
||
|
|
lanes = [
|
||
|
|
1,
|
||
|
|
2,
|
||
|
|
4
|
||
|
|
]
|
||
|
|
|
||
|
|
# If you want to use UCD-400, please, set 'dp_14_mode = True' and 'enable_dp_21_mode = False'
|
||
|
|
# If you want to use UCD-500 in DP 1.4 mode, please, set 'dp_14_mode = True' and 'enable_dp_21_mode = False'
|
||
|
|
# If you want to use UCD-500 in DP 2.1 mode, please, set 'dp_14_mode = False' and 'enable_dp_21_mode = True'
|
||
|
|
dp_14_mode = False
|
||
|
|
bit_rates_14 = [
|
||
|
|
1.62,
|
||
|
|
2.7,
|
||
|
|
5.4,
|
||
|
|
6.75,
|
||
|
|
8.1
|
||
|
|
]
|
||
|
|
|
||
|
|
enable_dp_21_mode = True
|
||
|
|
bit_rates_21 = [
|
||
|
|
10,
|
||
|
|
13.5,
|
||
|
|
20
|
||
|
|
]
|
||
|
|
|
||
|
|
mst = True
|
||
|
|
streams = [
|
||
|
|
0,
|
||
|
|
1,
|
||
|
|
2,
|
||
|
|
3
|
||
|
|
]
|
||
|
|
|
||
|
|
#
|
||
|
|
# Trigger config.
|
||
|
|
#
|
||
|
|
|
||
|
|
trigger_positions = [
|
||
|
|
UniTAP.bulk.TriggerPosition.TP_Start,
|
||
|
|
UniTAP.bulk.TriggerPosition.TP_25,
|
||
|
|
UniTAP.bulk.TriggerPosition.TP_50,
|
||
|
|
UniTAP.bulk.TriggerPosition.TP_75,
|
||
|
|
UniTAP.bulk.TriggerPosition.TP_End
|
||
|
|
]
|
||
|
|
|
||
|
|
# If you do not want to use trigger config, please, set 'None' to variable
|
||
|
|
# Use other types of trigger config. Full list located in 'TriggerType'
|
||
|
|
trigger_config = UniTAP.bulk.TriggerType.U13()
|
||
|
|
trigger_config.position = UniTAP.bulk.TriggerTypeEnum.SourceTypePosition.InitialLT
|
||
|
|
|
||
|
|
|
||
|
|
lUniTAP = UniTAP.TsiLib()
|
||
|
|
|
||
|
|
#
|
||
|
|
# For opening device, please, put serial number of the device as 8 symbol str or put index of device.
|
||
|
|
#
|
||
|
|
# serial_number = "2150C474"
|
||
|
|
# dev = lUniTAP.open(serial_number)
|
||
|
|
|
||
|
|
# If you do not want to write device serial number, you can write index position of device
|
||
|
|
index_position = 0
|
||
|
|
dev = lUniTAP.open(index_position)
|
||
|
|
|
||
|
|
# Need to select type of device: UCD-400 or UCD-500
|
||
|
|
# role = dev.select_role(UniTAP.dev.UCD400.DPSourceDPSink)
|
||
|
|
role = dev.select_role(UniTAP.dev.UCD500.DPSourceDPSink)
|
||
|
|
|
||
|
|
timing_manager = role.dptx.pg.timing_manager
|
||
|
|
|
||
|
|
timings = [
|
||
|
|
timing_manager.get_cta(16),
|
||
|
|
timing_manager.get_cta(90),
|
||
|
|
timing_manager.get_cta(97),
|
||
|
|
timing_manager.get_cta(126),
|
||
|
|
timing_manager.get_cta(196),
|
||
|
|
]
|
||
|
|
|
||
|
|
color_formats = [
|
||
|
|
UniTAP.ColorInfo.ColorFormat.CF_RGB,
|
||
|
|
UniTAP.ColorInfo.ColorFormat.CF_YCbCr_444,
|
||
|
|
UniTAP.ColorInfo.ColorFormat.CF_YCbCr_422,
|
||
|
|
UniTAP.ColorInfo.ColorFormat.CF_YCbCr_420,
|
||
|
|
UniTAP.ColorInfo.ColorFormat.CF_Y_ONLY,
|
||
|
|
UniTAP.ColorInfo.ColorFormat.CF_RAW
|
||
|
|
]
|
||
|
|
|
||
|
|
colorimetry_list = [
|
||
|
|
UniTAP.ColorInfo.Colorimetry.CM_sRGB,
|
||
|
|
UniTAP.ColorInfo.Colorimetry.CM_ITUR_BT601,
|
||
|
|
UniTAP.ColorInfo.Colorimetry.CM_ITUR_BT709
|
||
|
|
]
|
||
|
|
|
||
|
|
# Note: BPC 7 and 14 will be work only with CF_Y_ONLY and CF_RAW
|
||
|
|
bpc_list = [
|
||
|
|
6,
|
||
|
|
7,
|
||
|
|
8,
|
||
|
|
10,
|
||
|
|
12,
|
||
|
|
14,
|
||
|
|
16
|
||
|
|
]
|
||
|
|
|
||
|
|
|
||
|
|
def link_training(dev_role):
|
||
|
|
time.sleep(1)
|
||
|
|
dev_role.dptx.link.start_link_training()
|
||
|
|
|
||
|
|
|
||
|
|
def bool_str_format(value: bool) -> str:
|
||
|
|
return "ON" if value else "OFF"
|
||
|
|
|
||
|
|
|
||
|
|
def prepare_directory_name(_timing, _color_mode, _lane, _rate, _stream, tr_pos, tr_config=None) -> str:
|
||
|
|
tr_conf_str = type(tr_config).__name__ if tr_config is not None else "-"
|
||
|
|
auto_generated_folder = f"cap_[TG_Pos_{tr_pos.name}][TG_Conf-{tr_conf_str}][{'DP1.4' if dp_14_mode else 'DP2.1'}]" \
|
||
|
|
f"[MST-{bool_str_format(mst)}][FEC-{bool_str_format(fec)}][stream-{_stream}]" \
|
||
|
|
f"[TPS3-{bool_str_format(tps3)}][TPS4-{bool_str_format(tps4)}]" \
|
||
|
|
f"[{_timing.hactive}x{_timing.vactive}@" \
|
||
|
|
f"{math.ceil(_timing.frame_rate / 1000)}_{_color_mode.color_format.name}_{_color_mode.bpc}bpc" \
|
||
|
|
f"-{_color_mode.colorimetry.name}]"
|
||
|
|
directory_name = os.path.join(FOLDER_TO_GENERATE, auto_generated_folder)
|
||
|
|
|
||
|
|
return directory_name
|
||
|
|
|
||
|
|
|
||
|
|
def do_bulk_capture(dev_role, _timing, _color_mode, _lane, _rate, _stream):
|
||
|
|
# bulk_size - size of data in megabytes
|
||
|
|
# trigger_position - available position of trigger. Variants: TP_Start, TP_25, TP_50, TP_75, TP_End
|
||
|
|
# trigger_config - one of the variant of active trigger.
|
||
|
|
# assume_scrambler - just flag of assume scrambler
|
||
|
|
|
||
|
|
for tr_pos in trigger_positions:
|
||
|
|
|
||
|
|
thread = None
|
||
|
|
if trigger_config is not None:
|
||
|
|
thread = threading.Thread(target=link_training, args=(role,))
|
||
|
|
thread.start()
|
||
|
|
|
||
|
|
dev_role.dprx.bulk_capturer.start(bulk_size=BULK_SIZE_MB, trigger_position=tr_pos, trigger_config=trigger_config,
|
||
|
|
assume_scrambler=False, gpio=False)
|
||
|
|
dev_role.dprx.bulk_capturer.stop()
|
||
|
|
|
||
|
|
if trigger_config is not None and thread is not None:
|
||
|
|
thread.join()
|
||
|
|
thread.stop_thread = False
|
||
|
|
|
||
|
|
# If you want to get captured data, please, use 'capture_result'. It is the object of class 'ResultBulkObject'.
|
||
|
|
# For saving data in file, please, use 'save_to_bin_file'. You need to set 'directory_name' - name of directory,
|
||
|
|
# where you want to save files. If directory have already existed, is will be clean up.
|
||
|
|
# If directory does not exist, it will be created.
|
||
|
|
result = dev_role.dprx.bulk_capturer.capture_result
|
||
|
|
|
||
|
|
directory_name = prepare_directory_name(_timing, _color_mode, _lane, _rate, _stream, tr_pos, trigger_config)
|
||
|
|
result.save_to_bin_file(directory_name=directory_name)
|
||
|
|
|
||
|
|
|
||
|
|
for lane in lanes:
|
||
|
|
if dp_14_mode:
|
||
|
|
rates = bit_rates_14
|
||
|
|
else:
|
||
|
|
rates = bit_rates_21
|
||
|
|
for timing in timings:
|
||
|
|
for rate in rates:
|
||
|
|
for stream in streams:
|
||
|
|
for color_format in color_formats:
|
||
|
|
for bpc in bpc_list:
|
||
|
|
for colorimetry in colorimetry_list:
|
||
|
|
|
||
|
|
# Configure Sink Link caps
|
||
|
|
caps = UniTAP.LinkCapabilities()
|
||
|
|
caps.max_lane = lane
|
||
|
|
if dp_14_mode:
|
||
|
|
caps.bit_rate = rate
|
||
|
|
else:
|
||
|
|
caps.dp_128_132_bitrates = [rate]
|
||
|
|
caps.fec = fec
|
||
|
|
caps.tps3 = tps3
|
||
|
|
caps.tps4 = tps4
|
||
|
|
caps.mst = mst
|
||
|
|
role.dprx.link.capabilities.set(caps)
|
||
|
|
role.dprx.link.hpd_pulse()
|
||
|
|
time.sleep(0.5)
|
||
|
|
|
||
|
|
# Configure Source Link
|
||
|
|
if dp_14_mode:
|
||
|
|
config = UniTAP.LinkConfig.DP8b10b()
|
||
|
|
config.mst = mst
|
||
|
|
else:
|
||
|
|
config = UniTAP.LinkConfig.DP128b132b()
|
||
|
|
|
||
|
|
config.lane_count = lane
|
||
|
|
config.bit_rate = rate
|
||
|
|
|
||
|
|
if enable_dp_21_mode:
|
||
|
|
config.force_dp_128_132 = True
|
||
|
|
config.try_dp_128_132 = True
|
||
|
|
else:
|
||
|
|
config.force_dp_128_132 = False
|
||
|
|
config.try_dp_128_132 = False
|
||
|
|
|
||
|
|
role.dptx.link.config.set(config)
|
||
|
|
role.dptx.link.start_link_training()
|
||
|
|
time.sleep(0.5)
|
||
|
|
|
||
|
|
color_mode = UniTAP.ColorInfo()
|
||
|
|
color_mode.color_format = color_format
|
||
|
|
color_mode.bpc = bpc
|
||
|
|
color_mode.colorimetry = colorimetry
|
||
|
|
if color_format in [UniTAP.ColorInfo.ColorFormat.CF_RGB,
|
||
|
|
UniTAP.ColorInfo.ColorFormat.CF_RAW,
|
||
|
|
UniTAP.ColorInfo.ColorFormat.CF_Y_ONLY]:
|
||
|
|
color_mode.dynamic_range = UniTAP.ColorInfo.DynamicRange.DR_VESA
|
||
|
|
else:
|
||
|
|
color_mode.dynamic_range = UniTAP.ColorInfo.DynamicRange.DR_CTA
|
||
|
|
|
||
|
|
video_mode = UniTAP.VideoMode(timing=timing, color_info=color_mode)
|
||
|
|
if stream < role.dptx.pg.max_stream_count:
|
||
|
|
role.dptx.pg[stream].set_vm(vm=video_mode)
|
||
|
|
role.dptx.pg[stream].set_pattern(pattern=UniTAP.VideoPattern.ColorBars)
|
||
|
|
role.dptx.pg[stream].apply()
|
||
|
|
res_app = role.dptx.pg[stream].status().error
|
||
|
|
print(f"Stream {stream} - Apply {res_app.__str__()}")
|
||
|
|
if res_app == res_app.OK:
|
||
|
|
print("Start do bulk capturing")
|
||
|
|
try:
|
||
|
|
do_bulk_capture(role, timing, color_mode, lane, rate, stream)
|
||
|
|
print("Bulk capture completed successfully")
|
||
|
|
except BaseException as e:
|
||
|
|
print(f'Error during bulk capturing. Error: {e}')
|
||
|
|
else:
|
||
|
|
print('Cannot start to do bulk capturing, because PG does not apply settings.')
|
||
|
|
|
||
|
|
#
|
||
|
|
# Since the 3.5 version, TsiLib and TSIDevice objects have the option to be closed earlier.
|
||
|
|
# TSIDevice can be closed with the TsiLib method close(). TsiLib can be closed with cleanup().
|
||
|
|
# Clean up will close all opened devices and block ability to open any devices
|
||
|
|
# with same TsiLib object.
|
||
|
|
#
|
||
|
|
lUniTAP.close(dev)
|
||
|
|
|
||
|
|
lUniTAP.cleanup()
|