"""
Author: Leonardo de Sousa Marques
Affiliation: Embedded Computing Lab (ECL), Federal University of Santa Catarina (UFSC)
Description:
This module combines two or more R-D curves loaded from pre-computed RD report
JSON files and generates a single comparative R-D plot per extra_rd_curves entry.
Each entry in ``extra_rd_curves`` declares a list of ``rd_files``, each identified
by an ``id`` and pointing to a report JSON file with its own ``rd_preferences``
(title, color, marker). An optional ``anchor`` field names the ``id`` of the
curve to be used as BD-rate reference.
The module reuses the existing RDCurveMatplotlibView / RDPlotMatplotlib
infrastructure from rd_plot without modifying it.
"""
import json
import sys
from pathlib import Path
from typing import Dict, List, Optional, Tuple
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.quality.rd_curve import RDCurve
from lfc_toolkit.src.quality.rd_plot import RDCurveMatplotlibView, RDPlotMatplotlib
def _read_json_report(filename: str) -> Optional[Dict]:
"""Read a JSON RD report file, returning None on any failure.
:param filename: Absolute or relative path to the JSON report
:return: Parsed report dictionary or None
"""
path = Path(filename)
if not path.exists():
print(f"Warning: report file not found: {filename}")
return None
try:
with open(path, "r") as f:
return json.load(f)
except Exception as e:
print(f"Error reading report {filename}: {e}")
return None
def _load_rd_curve_view_from_report(
report_data: Dict,
rd_preferences: Dict,
metric: str,
configuration: ConfigurationReader,
) -> Optional[RDCurveMatplotlibView]:
"""Build an RDCurveMatplotlibView from a loaded RD report dictionary.
The report is expected to follow the structure produced by
``create_json_from_bundle`` / ``create_json``:
.. code-block:: json
{
"results": {
"0.150": {"rate": 0.148, "<metric>": {"mean": 35.2}},
...
}
}
:param report_data: Parsed RD report dictionary
:param rd_preferences: Display preferences with keys ``title``, ``color``, ``marker``
:param metric: Quality metric name to extract (e.g. ``"psnr_y"``)
:param configuration: Configuration reader (used for unit look-up)
:return: Populated RDCurveMatplotlibView or None if fewer than 2 valid points
"""
results = report_data.get("results", {})
if not results:
print("Warning: report has no 'results' section.")
return None
points: List[Tuple[float, float]] = []
for _key, entry in results.items():
bpp = entry.get("rate")
metric_block = entry.get(metric)
if bpp is None or not isinstance(metric_block, dict):
continue
value = metric_block.get("mean")
if value is not None:
points.append((bpp, value))
if len(points) < 2:
print(f"Warning: fewer than 2 valid points for metric '{metric}' — skipping curve.")
return None
points.sort(key=lambda p: p[0])
bpps, values = zip(*points)
unit, _ = configuration.get_used_quality_unit_and_tool(metric=metric)
rd_curve = RDCurve(
rates=list(bpps),
rate_unit="bpp",
distortions=list(values),
distortion_name=metric,
distortion_unit=unit,
codec_name=rd_preferences.get("title", metric),
title=rd_preferences.get("title", metric),
)
return RDCurveMatplotlibView(
configuration=configuration,
rd_curve=rd_curve,
color=rd_preferences.get("color", "blue"),
marker=rd_preferences.get("marker", "o"),
)
def _collect_metrics_from_entry(rd_files_cfg: List[Dict]) -> List[str]:
"""Collect all metric names present across all report files in an entry.
:param rd_files_cfg: List of rd_file config dicts (each with ``filename``)
:return: Sorted list of unique metric names found in the reports
"""
metrics: set = set()
for rd_file in rd_files_cfg:
data = _read_json_report(rd_file.get("filename", ""))
if data is None:
continue
results = data.get("results", {})
for entry in results.values():
for key, val in entry.items():
if isinstance(val, dict) and "mean" in val:
metrics.add(key)
return sorted(metrics)
def _build_curve_views_for_metric(
rd_files_cfg: List[Dict],
metric: str,
configuration: ConfigurationReader,
) -> Tuple[List[RDCurveMatplotlibView], Dict[str, RDCurveMatplotlibView]]:
"""Load and build all RD curve views for a given metric.
:param rd_files_cfg: List of rd_file config dicts with ``id``, ``filename``,
and ``rd_preferences``
:param metric: Quality metric name
:param configuration: Configuration reader
:return: Tuple of (ordered list of views, mapping from id to view)
"""
views: List[RDCurveMatplotlibView] = []
id_to_view: Dict[str, RDCurveMatplotlibView] = {}
for rd_file in rd_files_cfg:
file_id = rd_file.get("id", "")
filename = rd_file.get("filename", "")
rd_preferences = rd_file.get("rd_preferences", {})
data = _read_json_report(filename)
if data is None:
continue
view = _load_rd_curve_view_from_report(
report_data=data,
rd_preferences=rd_preferences,
metric=metric,
configuration=configuration,
)
if view is None:
continue
views.append(view)
id_to_view[file_id] = view
return views, id_to_view
[docs]
def generate_combined_rd_plots(
configuration: ConfigurationReader,
results_path: Path,
) -> None:
"""Generate combined R-D plots for all entries declared in ``extra_rd_curves``.
Each entry in ``configuration["rd_plots_from_bundle"]["extra_rd_curves"]``
produces one plot file per metric found across its report files. If an
``anchor`` id is specified, BD-rate values are computed relative to that curve.
:param configuration: Configuration reader containing the full config
:param results_path: Root directory where plots will be saved
"""
bundle_cfg = configuration["rd_plots_from_bundle"]
extra_entries: List[Dict] = bundle_cfg.get("extra_rd_curves", [])
if not extra_entries:
print("No extra_rd_curves entries found — nothing to do.")
return
rd_plots_path = results_path / "rd_plots" / "extra_rd_curves"
rd_plots_path.mkdir(parents=True, exist_ok=True)
rd_plot_configs = configuration["rd_plots"]
rd_plot_config = rd_plot_configs[0] if rd_plot_configs else {}
for entry in extra_entries:
output_filename_stem: str = entry.get("filename", "combined")
plot_title: str = entry.get("title", output_filename_stem)
rd_files_cfg: List[Dict] = entry.get("rd_files", [])
anchor_id: Optional[str] = entry.get("anchor")
if not rd_files_cfg:
print(f"Warning: entry '{output_filename_stem}' has no rd_files — skipping.")
continue
metrics = _collect_metrics_from_entry(rd_files_cfg)
if not metrics:
print(f"Warning: no metrics found for entry '{output_filename_stem}' — skipping.")
continue
print(f"\nProcessing entry: {output_filename_stem} (metrics: {metrics})")
# Use the first report's lightfield_name to retrieve a LightField object,
# falling back gracefully when the name is not in configuration.
first_data = _read_json_report(rd_files_cfg[0].get("filename", ""))
lf_name: str = (first_data or {}).get("lightfield_name", "")
if lf_name and lf_name not in configuration.lightfield_names:
configuration.lightfield_configurations[lf_name] = (
configuration.get_lightfield_configuration(lf_name)
)
lightfield = None
if lf_name:
try:
lightfield = LightFieldFactory.get_lightfield(
configuration=configuration.lightfield_configurations[lf_name]
)
except Exception as e:
print(f"Warning: could not load lightfield '{lf_name}': {e}")
for metric in metrics:
print(f" Metric: {metric}")
views, id_to_view = _build_curve_views_for_metric(
rd_files_cfg=rd_files_cfg,
metric=metric,
configuration=configuration,
)
if len(views) < 1:
print(f" No valid curves for metric '{metric}' — skipping.")
continue
anchor_view: Optional[RDCurveMatplotlibView] = (
id_to_view.get(anchor_id) if anchor_id else None
)
if anchor_id and anchor_view is None:
print(f" Warning: anchor id '{anchor_id}' not found among loaded curves.")
target_bpps = sorted(
{bpp for v in views for bpp in v.rd_curve.rate}
)
file_extension = rd_plot_config.get("format", "pdf")
bpp_logscale = rd_plot_config.get("bpp_logscale", True)
out_file = rd_plots_path / f"{output_filename_stem}_{metric}.{file_extension}"
plot = RDPlotMatplotlib(
configuration=configuration,
rd_plot_config=rd_plot_config,
lightfield=lightfield,
)
plot.plot(
rd_curves=views,
anchor=anchor_view,
title=plot_title,
bpp_logscale=bpp_logscale,
target_bpps=target_bpps,
interpolation=rd_plot_config.get("interpolation", None),
filename=str(out_file),
generate_bd_report=False,
)
print(f" Saved: {out_file}")
[docs]
def main() -> None:
"""Generate combined R-D plots from extra_rd_curves declared in configuration.
Usage: python combine_curves.py <configuration.json>
:raises SystemExit: If configuration file is not provided or paths are invalid
"""
if len(sys.argv) < 2:
print("Usage: combine_curves.py <configuration.json>")
sys.exit(1)
configuration = read_config_from_argv()
bundle_cfg = configuration["rd_plots_from_bundle"]
results_path = Path(bundle_cfg.get("results_path", ""))
if not results_path:
print("Error: 'results_path' not set in rd_plots_from_bundle configuration.")
sys.exit(1)
generate_combined_rd_plots(
configuration=configuration,
results_path=results_path,
)
print("\nDone.")
if __name__ == "__main__":
main()