"""
Author: Ismael Seidel
Affiliation: Embedded Computing Lab (ECL), Federal University of Santa Catarina (UFSC)
Description:
This module generates rate-distortion plots, reports, and analysis tables
for light field encoding experiments, visualizing codec performance across metrics.
"""
import json
import os
import sys
from hashlib import md5
from pathlib import Path
from typing import Any, Dict, List
# Add the parent directory to the Python path
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from lfc_toolkit.src.codecwrappers.codec_wrapper import CodecWrapper
from lfc_toolkit.src.configuration.configuration_reader import (
ConfigurationReader, read_config_from_argv)
from lfc_toolkit.src.ctc.lightfield_factory import LightFieldFactory
from lfc_toolkit.src.data_handlers.formatters import (
get_formatted_filename_for_lf, get_formatted_filename_for_rd_report)
from lfc_toolkit.src.data_handlers.lightfield import LightField
from lfc_toolkit.src.quality.bd_wrapper import BDWrapper
from lfc_toolkit.src.quality.compound_metrics import CompoundMetrics
from lfc_toolkit.src.quality.rd_plot import (RDCurveFromVMAFLikeReports,
RDCurveMatplotlibView,
RDPlotMatplotlib,
process_target_rate)
[docs]
def generate_rd_plots_for_lf_and_metric(
metric: str,
lightfield: LightField,
rd_plot_config: List[Dict[str, Any]],
configuration_reader: ConfigurationReader,
) -> None:
"""Generate rate-distortion plot for a specific light field and quality metric.
Creates matplotlib-based R-D curves for all configured codecs at specified
target bitrates, including anchor codec detection and plot customization.
:param metric: Quality metric name to plot
:type metric: str
:param lightfield: Light field object for plotting
:type lightfield: LightField
:param rd_plot_config: Plot configuration with output path and options
:type rd_plot_config: List[Dict[str, Any]]
:param configuration_reader: Configuration reader with codec and quality settings
:type configuration_reader: ConfigurationReader
:return: None
:rtype: None
"""
output_path = rd_plot_config["path"]
target_rates = configuration_reader.lightfield_configurations[lightfield.name][
"target-rates"
]
codecs = configuration_reader["codecs-to-run"]
sample_codecs = configuration_reader["sample-codecs-to-run"]
# Combined object
codecs.update(sample_codecs)
rd_curves = []
anchor_name = rd_plot_config.get("anchor", None)
anchor_curve = None
unit, used_tool = configuration_reader.get_used_quality_unit_and_tool(metric=metric)
used_tool_details = configuration_reader["quality"]["wrappers"][used_tool]
used_tool_short_name = used_tool_details["short_name"]
used_tool_raw_type = used_tool_details["raw_type"]
decoded_lightfield_yuv = LightFieldFactory.get_raw_lightfield(
configuration=configuration_reader,
lightfield_name=lightfield.name,
raw_type=used_tool_raw_type,
)
for codec_name, codec_info in codecs.items():
print(f"Generating rate-{metric} curve for {codec_name}")
encoded_lightfield = LightFieldFactory.get_raw_lightfield(
configuration_reader,
lightfield_name=lightfield.name,
raw_type=codec_info["raw_type"].lower(),
)
rd_prefences = codec_info["rd_preferences"]
path = f"{codec_info['results']}/metrics_reports/{used_tool_short_name}" if codec_name not in sample_codecs else ""
encoded_path = f"{codec_info['results']}/{decoded_lightfield_yuv.name}/encoded" if codec_name not in sample_codecs else ""
encoded_extension=f"{codec_info['encoded_extension']}" if codec_name not in sample_codecs else ""
rd_curve = RDCurveMatplotlibView(
configuration=configuration_reader,
rd_curve=RDCurveFromVMAFLikeReports(
path=path,
tool_short_name=used_tool_short_name,
configuration=configuration_reader,
encoded_path=encoded_path,
encoded_extension=encoded_extension,
lightfield=decoded_lightfield_yuv,
encoded_lightfield=encoded_lightfield,
target_rates=target_rates,
distortion_name=metric,
distortion_unit=unit,
codec_name=codec_name,
title=rd_prefences["title"],
),
color=rd_prefences["color"],
marker=rd_prefences["marker"],
)
if anchor_name == codec_name:
anchor_curve = rd_curve
rd_curves.append(rd_curve)
plot = RDPlotMatplotlib(
configuration=configuration_reader,
rd_plot_config=rd_plot_config,
lightfield=lightfield,
)
file_extension = rd_plot_config.get("format", "pdf")
filename = get_formatted_filename_for_lf(
output_path,
lightfield=decoded_lightfield_yuv,
file_extension=file_extension,
extra_end=metric,
)
plot.plot(
rd_curves=rd_curves,
anchor=anchor_curve,
title=f"{decoded_lightfield_yuv.name}",
bpp_logscale=rd_plot_config.get("bpp_logscale", True),
target_bpps=target_rates,
interpolation=rd_plot_config.get("interpolation", None),
filename=filename,
)
[docs]
def generate_rd_plots_for_lf(
lightfield,
rd_plot_config: List[Dict[str, Any]],
configuration_reader: ConfigurationReader,
) -> None:
"""Generate all rate-distortion plots for a specific light field.
Creates R-D plots for all metrics configured in rd_plot_config.
:param lightfield: Light field object
:type lightfield: LightField
:param rd_plot_config: Plot configuration with metrics and output options
:type rd_plot_config: List[Dict[str, Any]]
:param configuration_reader: Configuration reader
:type configuration_reader: ConfigurationReader
:return: None
:rtype: None
"""
output_path = rd_plot_config["path"]
os.makedirs(output_path, exist_ok=True)
metrics = rd_plot_config["metrics"]
for metric in metrics:
generate_rd_plots_for_lf_and_metric(
metric=metric,
lightfield=lightfield,
rd_plot_config=rd_plot_config,
configuration_reader=configuration_reader,
)
[docs]
def create_json(
lightfield: LightField,
rd_plot_config: List[Dict[str, Any]],
configuration_reader: ConfigurationReader,
) -> None:
"""Generate JSON report with rate-distortion data for all codecs and metrics.
Creates detailed JSON files containing R-D values, metadata, and codec information
organized by target bitrate and quality metric.
:param lightfield: Light field object
:type lightfield: LightField
:param rd_plot_config: Plot configuration
:type rd_plot_config: List[Dict[str, Any]]
:param configuration_reader: Configuration reader
:type configuration_reader: ConfigurationReader
:return: None
:rtype: None
"""
lightfield_name = lightfield.name
target_rates = configuration_reader.lightfield_configurations[lightfield_name][
"target-rates"
]
codecs = configuration_reader["codecs-to-run"]
quality_config = configuration_reader["quality"]
metrics = rd_plot_config["metrics"]
for codec_name, codec_info in codecs.items():
if codec_name in configuration_reader["sample-codecs-to-run"]:
return
print(f"Generating RD JSON for {codec_name}")
codec_json_data = {
"lightfield_name": lightfield_name,
"codec": codec_name,
"codec_title": codec_info["rd_preferences"]["title"],
"raw_md5": LightFieldFactory.get_raw_lightfield(
configuration=configuration_reader,
lightfield_name=lightfield_name,
raw_type=codec_info["raw_type"],
).get_md5(),
"metrics": {},
"results": {},
}
for metric in metrics:
if CompoundMetrics.is_weighted_metric(
metric=metric, configuration=configuration_reader
):
metric_name = quality_config["weighted-metrics"][metric]["name"]
if "metrics" in quality_config["weighted-metrics"][metric]:
quality_wrapper = {
"metrics": quality_config["weighted-metrics"][metric]["metrics"]
}
elif "weights" in quality_config["weighted-metrics"][metric]:
quality_wrapper = {
"weighted": quality_config["weighted-metrics"][metric][
"weights"
]
}
else:
quality_wrapper = None
label = quality_config["weighted-metrics"][metric]["label"]
tex_label = quality_config["weighted-metrics"][metric].get(
"tex_label", label
)
# adding bd-adjusted metrics to report
elif CompoundMetrics.is_bd_adjusted_metric(
metric=metric, configuration=configuration_reader
):
metric_name = quality_config["bd-adjusted-metrics"][metric]["name"]
quality_wrapper = {
"origin": quality_config["bd-adjusted-metrics"][metric]["origin"]
}
label = quality_config["bd-adjusted-metrics"][metric]["label"]
tex_label = quality_config["bd-adjusted-metrics"][metric].get(
"tex_label", label
)
else:
metric_name = quality_config["metrics"][metric]["name"]
quality_wrapper = quality_config["metrics"][metric]["quality-wrapper"]
label = quality_config["metrics"][metric]["label"]
tex_label = quality_config["metrics"][metric].get("tex_label", label)
codec_json_data["metrics"][metric] = {
"name": metric_name,
"label": label,
"tex_label": tex_label,
"quality_wrapper": quality_wrapper,
"unit": None,
}
ctc_enc = CodecWrapper(
codec=codec_name,
codec_path=None,
results_path=codec_info["results"],
encoded_extension=codec_info["encoded_extension"],
repository=None,
)
for rate in target_rates:
rate_data = {"rate": None, "md5_of_encoded": None}
for metric in metrics:
unit, used_tool = configuration_reader.get_used_quality_unit_and_tool(
metric=metric
)
used_tool_details = quality_config["wrappers"][used_tool]
used_tool_short_name = used_tool_details["short_name"]
used_tool_raw_type = used_tool_details["raw_type"]
codec_json_data["metrics"][metric]["unit"] = unit
try:
result = process_target_rate(
bpp=rate,
path=f"{codec_info['results']}/metrics_reports/{used_tool_short_name}",
tool_short_name=used_tool_short_name,
configuration=configuration_reader,
encoded_path=f"{codec_info['results']}/{lightfield_name}/encoded",
encoded_extension=codec_info["encoded_extension"],
lightfield=LightFieldFactory.get_raw_lightfield(
configuration=configuration_reader,
lightfield_name=lightfield_name,
raw_type=used_tool_raw_type,
),
encoded_lightfield=LightFieldFactory.get_raw_lightfield(
configuration=configuration_reader,
lightfield_name=lightfield_name,
raw_type=codec_info["raw_type"],
),
distortion_name=metric,
ctc_enc=ctc_enc,
)
rate_data["rate"] = result["bpp"]
rate_data[metric] = result.asdict(exclude=("bpp",))
except Exception as e:
print(f"Warning: rate {rate} for metric {metric} not found.")
encoded_file_path = Path(
get_formatted_filename_for_lf(
path=f"{codec_info['results']}/{lightfield_name}/encoded",
lightfield=LightFieldFactory.get_raw_lightfield(
configuration=configuration_reader,
lightfield_name=lightfield_name,
raw_type=codec_info["raw_type"],
),
file_extension=codec_info["encoded_extension"],
bpp=rate,
)
)
if encoded_file_path.exists():
with open(encoded_file_path, "rb") as f:
rate_data["md5_of_encoded"] = md5(f.read()).hexdigest()
else:
print(f"WARNING: Encoded file not found: {encoded_file_path}")
rate_data["md5_of_encoded"] = None
codec_json_data["results"][rate] = rate_data
output_path = Path(configuration_reader["logs"]["rd_results_path"])
output_path.mkdir(parents=True, exist_ok=True)
json_filename = output_path / get_formatted_filename_for_rd_report(lightfield_name, codec_name)
with open(json_filename, "w") as json_file:
json.dump(codec_json_data, json_file, indent=2)
[docs]
def create_latex_tables(
lightfield: LightField,
rd_plot_config: List[Dict[str, Any]],
configuration_reader: ConfigurationReader,
) -> None:
"""Generate LaTeX tables with rate-distortion results for documentation.
Creates formatted LaTeX tables with R-D data for all codecs and metrics,
suitable for inclusion in research papers and technical documents.
:param lightfield: Light field object
:type lightfield: LightField
:param rd_plot_config: Plot configuration
:type rd_plot_config: List[Dict[str, Any]]
:param configuration_reader: Configuration reader
:type configuration_reader: ConfigurationReader
:return: None
:rtype: None
"""
output_path = Path(configuration_reader["logs"]["rd_results_path"])
codecs = configuration_reader["codecs-to-run"]
metrics = rd_plot_config["metrics"]
lightfield_name = lightfield.name
for codec_name, codec_info in codecs.items():
json_filename = output_path / get_formatted_filename_for_rd_report(lightfield_name, codec_name)
if not json_filename.exists():
print(f"WARNING: RD JSON not found: {json_filename}")
continue
with open(json_filename, "r") as f:
data = json.load(f)
target_bpps = list(data["results"].keys())
# LaTeX layout
caption = "{title} results for \\textbf{{{lightfield_name}}}.".format(
title=codec_info["rd_preferences"]["title"],
lightfield_name=lightfield_name.replace("_", "\\_"),
)
label = f"tab.{codec_name}.{lightfield_name.lower()}"
latex = []
latex.append("\\begin{table}[h]")
latex.append(f"\\caption{{{caption}}}")
latex.append(f"\\label{{{label}}}")
latex.append("\\vspace*{-3mm}")
latex.append("\\begin{center}")
latex.append(f" \\begin{{tabular}}{{l|{'c|' * (len(target_bpps) - 1)}c}}")
latex.append(" \\toprule")
# Target bpps
header = (
" \\textbf{Target bpps} & "
+ " & ".join([f"\\textbf{{{float(bpp):g}}}" for bpp in target_bpps])
+ " \\\\ \\midrule"
)
latex.append(header)
# Obtained bpps
obtained_bpps = [
(
f"{data['results'][bpp]['rate']:.4f}"
if data["results"][bpp]["rate"] is not None
else ""
)
for bpp in target_bpps
]
obtained_line = (
" \\textbf{Obtained bpps} & " + " & ".join(obtained_bpps) + " \\\\"
)
latex.append(obtained_line)
latex.append("\\midrule %\\hline")
# Complete with the quality metrics
for i, metric in enumerate(metrics):
metric_info = data["metrics"][metric]
metric_tex_label = metric_info["tex_label"]
unit = metric_info.get("unit", "")
if unit:
metric_tex_label += f" ({unit})"
row_values = []
for bpp in target_bpps:
value = data["results"][bpp].get(metric, {}).get("mean", "")
if value is not None and value != "":
row_values.append(f"{value:.2f}")
else:
row_values.append("")
row = f"{metric_tex_label} & " + " & ".join(row_values) + " \\\\"
latex.append(row)
if i < len(metrics) - 1:
if i % 2 == 0:
latex.append("\\hline")
else:
latex.append("\\midrule")
latex.append("\\bottomrule")
latex.append("\\end{tabular}")
latex.append("\\end{center}")
latex.append("\\end{table}")
latex_full = "\n".join(latex)
tex_dir = output_path / "LaTeX_tables" / codec_name
os.makedirs(tex_dir, exist_ok=True)
tex_filename = tex_dir / f"{lightfield_name}_{codec_name}_rd_table.tex"
with open(tex_filename, "w") as f:
f.write(latex_full)
print(f"LaTeX table saved to {tex_filename}")
[docs]
def main(configuration: ConfigurationReader = None) -> None:
"""Generate rate-distortion plots and reports for all light fields.
Orchestrates complete R-D visualization pipeline:
1. Generates R-D plots for all configured metrics
2. Creates JSON reports with detailed R-D data
3. Generates LaTeX tables for documentation
4. Computes Bjøntegaard-Delta (BD) rate analysis if configured
Usage: python generate_rd_plots.py <configuration.json>
:param configuration: Configuration reader (read from argv if None)
:type configuration: ConfigurationReader
:return: None
:rtype: None
"""
if not configuration:
configuration = read_config_from_argv()
print("Using the paths from", configuration)
lightfields = [
LightFieldFactory.get_lightfield(
configuration=configuration.lightfield_configurations[name]
)
for name in configuration.lightfield_names
]
rd_plot_configs = configuration["rd_plots"]
configs_size = len(rd_plot_configs)
for i, rd_plot_config in enumerate(rd_plot_configs, start=1):
for lightfield in lightfields:
print(f"({lightfield.name} LF for plot configuration {i}/{configs_size})")
generate_rd_plots_for_lf(
lightfield=lightfield,
rd_plot_config=rd_plot_config,
configuration_reader=configuration,
)
# save rd_values in a json file and create latex tables
generate_rd_reports = rd_plot_config.get("generate_rd_reports", True)
if generate_rd_reports:
create_json(
lightfield=lightfield,
rd_plot_config=rd_plot_config,
configuration_reader=configuration,
)
create_latex_tables(
lightfield=lightfield,
rd_plot_config=rd_plot_config,
configuration_reader=configuration,
)
# Create BD-rate tables in selected formats (LaTeX, HTML, CSV, ...)
if rd_plot_config.get("interpolation", None):
BDWrapper.create_bd_rate_tables(configuration_reader=configuration)
# for codec_name, codec_info in configuration["codecs"].items():
if __name__ == "__main__":
main()