"""
Author: Ismael Seidel (ismael.seidel@ufsc.br)
Affiliation: Embedded Computing Lab (ECL), Federal University of Santa Catarina (UFSC)
Description:
This module contains the `JPLMWrapper` class, which extends the `CodecWrapper` class to provide
specific encoding and decoding functionality for the JPLM codec.
"""
# from typing import override
import json
import subprocess
from pathlib import Path
from typing import Any, Dict, List, Optional, Union
from lfc_toolkit.src.data_handlers.formatters import (
get_bpp_str, get_formatted_filename_for_lf)
from lfc_toolkit.src.data_handlers.lightfield import (
EncodedLightField, RAW_BT709_FR_PGX_LightField_Data, RAWLightFieldData)
from .codec_wrapper import CodecWrapper
# Extract data from the log file created for each repetition
[docs]
def get_data_from_preffix(file_lines: List[str], preffix: str) -> str:
"""Extracts the value from a log line that starts with the given prefix.
:param file_lines: List of lines from the log file
:type file_lines: List[str]
:param preffix: Prefix string to search for at the start of lines
:type preffix: str
:return: The value after the prefix, or empty string if not found
:rtype: str
"""
for line in file_lines:
if line.startswith(preffix):
return line.removeprefix(preffix).strip()
return ""
JPLM_ENCODER_COMMAND_TEMPLATE = """
{jplm_bin}/jpl-encoder-bin
--show-runtime-statistics
--show-error-estimate
--config {bpp_configuration_file}
--input {input}
--output {output}
--output_json {output_json}
--number-of-threads {number_of_threads}
{extra_params}
"""
JPLM_DECODER_COMMAND_TEMPLATE = """
{jplm_bin}/jpl-decoder-bin
--show-runtime-statistics
--number-of-threads {number_of_threads}
--input {input}
--output {output}
"""
[docs]
class JPLMWrapper(CodecWrapper):
[docs]
def __init__(
self,
codec: str,
codec_path: Union[str, Path],
results_path: Union[str, Path],
encoded_extension: str,
repository: str,
clear_log: bool = False,
repetitions: int = 1,
force_encoding: bool = False,
num_cores: int = None,
number_of_threads: int = 1,
json_configuration_path: Optional[Union[str, Path]] = None,
extra_params: Optional[Any] = None,
) -> None:
"""Initializes the JPLMWrapper instance with the given parameters.
:param codec: Name of the codec
:type codec: str
:param codec_path: Path to the codec binary
:type codec_path: Union[str, Path]
:param results_path: Path to store results
:type results_path: Union[str, Path]
:param encoded_extension: File extension for encoded files
:type encoded_extension: str
:param repository: Repository object for managing codec source
:type repository: str
:param clear_log: Whether to clear logs after execution, defaults to False
:type clear_log: bool, optional
:param repetitions: Number of repetitions for encoding/decoding, defaults to 1
:type repetitions: int, optional
:param force_encoding: Whether to force encoding even if results exist, defaults to False
:type force_encoding: bool, optional
:param num_cores: Number of cores to use for encoding/decoding, defaults to None
:type num_cores: int, optional
:param number_of_threads: Number of threads to use, defaults to 1
:type number_of_threads: int, optional
:param json_configuration_path: Path to the JSON configuration file, defaults to None
:type json_configuration_path: Union[str, Path], optional
:param extra_params: Additional parameters for the codec, defaults to None
:type extra_params: optional
"""
super().__init__(
codec=codec,
codec_path=codec_path,
results_path=results_path,
clear_log=clear_log,
repetitions=repetitions,
encoded_extension=encoded_extension,
force_encoding=force_encoding,
num_cores=num_cores,
repository=repository,
)
if not json_configuration_path:
json_configuration_path = Path(codec_path) / "cfg" / "part2" / "4DTransformMode"
self.json_configuration_path = Path(json_configuration_path)
self.number_of_threads = number_of_threads
self.extra_params = extra_params
[docs]
def create_required_paths(self, raw_lightfield: RAWLightFieldData) -> None:
"""Creates the required directories for storing encoded and log data.
:param raw_lightfield: The raw light field data
:type raw_lightfield: RAWLightFieldData
:return: None
:rtype: None
"""
self.get_encoded_path(raw_lightfield).mkdir(parents=True, exist_ok=True)
self.get_logs_path(raw_lightfield).mkdir(parents=True, exist_ok=True)
[docs]
def encode_lightfield_for_target_bpps(self, raw_lightfield: RAWLightFieldData, bpps: List[float]) -> List[EncodedLightField]:
"""Encodes the light field for the specified target bits per pixel (bpp) values.
:param raw_lightfield: The raw light field data
:type raw_lightfield: RAWLightFieldData
:param bpps: List of target bpp values
:type bpps: List[float]
:return: List of encoded light fields
:rtype: List[EncodedLightField]
"""
self.create_required_paths(raw_lightfield)
lf_name = raw_lightfield.name
self.results.setdefault(
lf_name, {
"raw_lightfield": raw_lightfield,
"target_bpps": {}
}
)
encoded_lfs = []
for bpp in bpps:
bpp_configuration = self.json_configuration_path / lf_name / f"{lf_name}_{get_bpp_str(bpp)}.json"
with open(bpp_configuration, "r") as file:
configuration_data: dict = json.load(file)
encoded_filename = get_formatted_filename_for_lf(
path=self.get_encoded_path(raw_lightfield=raw_lightfield),
lightfield=raw_lightfield,
bpp=bpp,
file_extension=self.encoded_extension,
)
log_filename = get_formatted_filename_for_lf(
path=self.get_logs_path(raw_lightfield=raw_lightfield),
lightfield=raw_lightfield,
bpp=bpp,
file_extension="txt",
)
all_logs_exist = all(
log_filename.with_stem(f"{log_filename.stem}_rep_{rep+1}").is_file()
for rep in range(self.repetitions)
)
json_log_filename = get_formatted_filename_for_lf(
path=self.get_logs_path(raw_lightfield=raw_lightfield),
lightfield=raw_lightfield,
bpp=bpp,
file_extension="json",
)
encoded_lightfield = EncodedLightField(
raw_lightfield.copy(),
encoded_path=encoded_filename,
target_bitrate=bpp,
actual_bitrate=0,
encoder_name="JPLM",
)
self.results[lf_name]["target_bpps"].setdefault(bpp, {
"encoded": {
"encoded_lf": encoded_lightfield,
"log_filename": log_filename,
"add_to_log": {"lambda": configuration_data["lambda"]}
},
"profile_encoder": None,
"profile_decoder": None
})
if (not self.force_encoding and encoded_filename.is_file() and all_logs_exist):
print(f"Encoded file {encoded_filename} and log {log_filename} are valid. Skipping...")
_, obtained_bpp = self.compute_bytes_and_bpp(
encoded_filename=encoded_filename,
raw_lightfield=raw_lightfield
)
encoded_lightfield.actual_bitrate = obtained_bpp
encoded_lfs.append(encoded_lightfield)
continue
command = JPLM_ENCODER_COMMAND_TEMPLATE.format(
jplm_bin=(self.codec_path / "bin"),
input=raw_lightfield.raw_path,
output=encoded_filename,
output_json=json_log_filename,
number_of_threads=self.number_of_threads,
bpp_configuration_file=bpp_configuration,
extra_params=self.extra_params
).split()
print(f"Encoding for bpp: {bpp}")
execution_measurements = self.execute_command(
bpp=bpp,
lightfield=encoded_lightfield,
command=command,
log_filename=log_filename
)
_, obtained_bpp = self.compute_bytes_and_bpp(
encoded_filename=encoded_filename,
raw_lightfield=raw_lightfield
)
encoded_lightfield.actual_bitrate = obtained_bpp
self.results[lf_name]["target_bpps"][bpp]["profile_encoder"] = {
"encoder": execution_measurements
}
encoded_lfs.append(encoded_lightfield)
return encoded_lfs
[docs]
def decode(self, encoded_lightfield: EncodedLightField) -> RAWLightFieldData:
"""Decodes an encoded light field using the JPLM decoder.
:param encoded_lightfield: The encoded light field to decode
:type encoded_lightfield: EncodedLightField
:return: Decoded raw light field data in PGX format
:rtype: RAWLightFieldData
"""
lightfield = encoded_lightfield.copy()
lf_name = lightfield.name
bpp = encoded_lightfield.target_bitrate
self.results.setdefault(lf_name, {"target_bpps": dict()})
self.results[lf_name]["target_bpps"].setdefault(bpp, dict())
output_decoded_raw_lf = RAW_BT709_FR_PGX_LightField_Data(
lightfield=lightfield,
pgx_path=f"{self.get_decoded_path(encoded_lightfield)}/pgx",
bpp_for_naming=bpp,
)
output_decoded_raw_lf.raw_path.mkdir(parents=True, exist_ok=True)
log_filename = get_formatted_filename_for_lf(
path=self.get_logs_path(encoded_lightfield),
lightfield=encoded_lightfield,
bpp=bpp,
file_extension="txt"
)
command = JPLM_DECODER_COMMAND_TEMPLATE.format(
jplm_bin=(self.codec_path / "bin"),
input=encoded_lightfield.encoded_path,
output=output_decoded_raw_lf.raw_path,
number_of_threads=self.number_of_threads
).split()
execution_measurements = self.execute_command(
bpp=bpp,
lightfield=output_decoded_raw_lf,
command=command,
log_filename=log_filename
)
self.results[lf_name]["target_bpps"][bpp]["decoded"] = output_decoded_raw_lf
self.results[lf_name]["target_bpps"][bpp]["profile_decoder"] = {
"decoder": execution_measurements
}
return output_decoded_raw_lf