"""
Author: Ismael Seidel
Affiliation: Embedded Computing Lab (ECL), Federal University of Santa Catarina (UFSC)
Description:
This module handles quality metrics computation, utilizing the tools defined
in the quality.json file (e. g, VMAF, COMPARE) for each metric.
"""
import importlib
import os
import sys
from pathlib import Path
from typing import Any, Dict, List, Union
# 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.configuration.configuration_reader import (
ConfigurationReader,
read_config_from_argv,
)
from lfc_toolkit.src.ctc.lightfield_factory import LightFieldFactory
[docs]
def get_wrapper_instances(
configuration: ConfigurationReader, path: Union[str | Path]
) -> dict:
"""
Get wrapper instances for quality metrics
:param configuration: Configuration file
:type configuration: ConfigurationReader
:param path: Path in which the reports will be saved
:type path: Union[str | Path]
:return: Dictionary in which the key is the metric name
and the value is the quality wrapper
:rtype: dict
"""
# Create a list of all metrics needed to execute the experiment
listed_metrics = set()
try:
listed_metrics.update(configuration["quality_heatmaps"]["metrics"])
except KeyError:
print("No metrics found for quality heatmaps generation in configuration.")
try:
rd_plot_config = configuration["rd_plots"]
except KeyError:
rd_plot_config = []
print("No metrics found for RD plots generation in configuration.")
for rd_plot_configs in rd_plot_config:
listed_metrics.update(rd_plot_configs["metrics"])
# Load quality configuration from quality.json and additional files added to configuraton
quality_configuration = configuration["quality"]
# Get all metrics types (simple, bd_adjusted and weighted)
metrics = quality_configuration["metrics"]
bd_adjusted_metrics = quality_configuration["bd-adjusted-metrics"]
weighted_metrics = quality_configuration["weighted-metrics"]
if not listed_metrics:
listed_metrics.update(metrics)
listed_metrics.update(bd_adjusted_metrics)
listed_metrics.update(weighted_metrics)
print("Computing all quality metrics.")
# Create wrappers set
wrappers = set()
for metric in listed_metrics:
if metric in metrics:
quality_wrapper = metrics[metric]["quality-wrapper"]
wrappers.add(quality_wrapper)
elif metric in bd_adjusted_metrics:
origin = bd_adjusted_metrics[metric]["origin"]
quality_wrapper = metrics[origin]["quality-wrapper"]
wrappers.add(quality_wrapper)
elif metric in weighted_metrics:
# For weighted_metrics, we will consider either the weights (int values) or the used metrics for calculation.
weighted_metric = weighted_metrics[metric]
metrics_to_process = weighted_metric.get(
"metrics", weighted_metric.get("weights", {}).keys()
)
for m in metrics_to_process:
quality_wrapper = metrics[m]["quality-wrapper"]
wrappers.add(quality_wrapper)
else:
raise ValueError(f"Metric {metric} not found in quality file.")
# Load wrapper configuration from quality object
wrapper_configurations = quality_configuration["wrappers"]
# Create dictionary (to be returned) contatining pair (metric_name, quality_wrapper)
wrapper_instances = dict()
for wrapper in wrappers:
wrapper_configuration = wrapper_configurations[wrapper]
wrapper_name = wrapper_configuration["wrapper_name"]
wrapper_file = wrapper_configuration["wrapper_file"]
try:
quality_wrapper_module = importlib.import_module(
f"lfc_toolkit.src.quality.{wrapper_file}"
)
except ModuleNotFoundError:
try:
quality_wrapper_module = importlib.import_module(
f"lfc_toolkit.examples.quality.{wrapper_file}"
)
except ModuleNotFoundError:
raise ModuleNotFoundError(
"Error. You must provide a valid wrapper_file name in the quality configuration."
)
QualityToolWrapperClass = getattr(quality_wrapper_module, wrapper_name)
short_name = wrapper_configuration["short_name"]
kwargs = wrapper_configuration.get("kwargs", dict())
quality_tool_wrapper_instance = QualityToolWrapperClass(
reports_path=Path(path) / "metrics_reports" / short_name,
**kwargs,
)
wrapper_instances[short_name] = quality_tool_wrapper_instance
return wrapper_instances
[docs]
def get_metrics_bpps(
configuration: ConfigurationReader,
path: Union[str | Path],
codec_raw_type: str,
lightfield_name: str,
wrapper_instances: Dict[str, Any],
) -> List[Any]:
"""
Calculate quality metric for each target bitrate (bpp)
:param configuration: Configuration file
:type configuration: ConfigurationReader
:param path: Path to decoded light fields
:type path: Union[str | Path]
:param codec_raw_type: Raw light field file type (e. g, PGX, YUV)
:type codec_raw_type: str
:param lightfield_name: Light field name (e. g, Bikes, Danger_de_Mort)
:type lightfield_name: str
:param wrapper_instances: Wrapper instances dictionary with each metric
:type wrapper_instances: dict
:return: Metrics list result
:rtype: list
"""
# Create results list (to be returned)
results = list()
# Get original light field from factory
original_lightfield = LightFieldFactory.get_raw_lightfield(
configuration=configuration,
lightfield_name=lightfield_name,
raw_type=codec_raw_type,
)
# Load target bitarates
bpps = configuration.lightfield_configurations[original_lightfield.name][
"target-rates"
]
for bpp in bpps:
decoded_lightfield = LightFieldFactory.get_raw_lightfield(
configuration=configuration,
lightfield_name=lightfield_name,
raw_type=codec_raw_type,
decoded_base_path=path,
bpp_for_naming=bpp,
)
print(f"Computing quality for bpp {bpp}")
for name, quality_tool_wrapper in wrapper_instances.items():
print(f"Using tool {name}...")
if Path(decoded_lightfield.raw_path).exists():
result = quality_tool_wrapper.get_metrics(
original_lightfield,
decoded_lightfield,
remove_converted_lfs_after_using=True,
)
if result:
results.append(result)
else:
print(
f"File {decoded_lightfield.raw_path} does not exist. Skipping quality computation..."
)
return results
[docs]
def main(configuration: ConfigurationReader = None):
if not configuration:
configuration = read_config_from_argv()
# Load codecs being executed as object
codecs = configuration["codecs-to-run"]
sample_codecs = configuration["sample-codecs-to-run"]
# Check if there are codecs to run that are not example (sample) codecs.
if (len(codecs) == 0 and len(sample_codecs) > 0):
print("Skipping quality metrics computation. Will use sampled values.")
return
for codec, codec_configuration in codecs.items():
results_path = codec_configuration["results"]
codec_raw_type = codec_configuration["raw_type"]
wrapper_instances = get_wrapper_instances(
configuration=configuration, path=results_path
)
for name in configuration.lightfield_names:
print(
"Computing quality metrics",
f"for {codec} ({name} LF)",
)
get_metrics_bpps(
configuration=configuration,
path=results_path,
codec_raw_type=codec_raw_type,
lightfield_name=name,
wrapper_instances=wrapper_instances,
)
if __name__ == "__main__":
main()