"""
Author: Ismael Seidel
Affiliation: Embedded Computing Lab (ECL), Federal University of Santa Catarina (UFSC)
Description:
This module encodes the JPEG Pleno LF Dataset using the JPEG Pleno Model (JPLM).
This implementation is for JPEG Pleno CE 16.
"""
import importlib
import os
import sys
from pathlib import Path
from typing import Dict, List, Optional
# 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.codecrepos.git_repository_cloner_and_builder import (
GitRepositoryClonerAndBuilder,
)
# from codecwrappers import JPLMWrapper
from lfc_toolkit.src.configuration.configuration_reader import (
ConfigurationReader,
read_config_from_argv,
)
from lfc_toolkit.src.converters.lightfield_converter import LightfieldConverter
from lfc_toolkit.src.ctc.lightfield_factory import LightFieldFactory
[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 ""
[docs]
def clone_and_build_repo(
codec_configuration: Dict, global_configuration: Dict, fulldocker: Optional[bool] = None
) -> Optional[GitRepositoryClonerAndBuilder]:
"""Clones and builds the codec repository if configured.
If `repository` is present in codec_configuration, clones and builds it using
GitRepositoryClonerAndBuilder and returns the created object. Otherwise returns None.
:param codec_configuration: Codec-specific configuration
:type codec_configuration: Dict
:param global_configuration: Global configuration
:type global_configuration: Dict
:return: GitRepositoryClonerAndBuilder instance or None
:rtype: Optional[GitRepositoryClonerAndBuilder]
"""
if "repository" not in codec_configuration:
return None
if Path(codec_configuration["path"]).exists() and not codec_configuration.get(
"force_building", True
):
print("Repository already exists. Forcing build is not enabled.")
return None
repo = GitRepositoryClonerAndBuilder(
repository_address=codec_configuration["repository"],
local_path=codec_configuration["path"],
tag=codec_configuration.get("tag", "master"),
build_tool=codec_configuration.get("build_tool", "cmake"),
build_path=codec_configuration.get("build_path", "build"),
cmake_path=codec_configuration.get("cmake_path", ".."),
build_options=codec_configuration.get("build_options", []),
use_docker_to_build=not fulldocker if fulldocker is not None else global_configuration["codecs"].get(
"use-docker-for-build", False
),
)
return repo
[docs]
def create_codec_wrapper_instance(
codec: str,
codec_configuration: Dict,
repository_obj: Optional[object] = None,
) -> object:
"""Imports the wrapper module/class and creates an instance configured with codec_configuration.
:param codec: Codec name
:type codec: str
:param codec_configuration: Codec configuration
:type codec_configuration: Dict
:param repository_obj: Optional repository object from clone_and_build_repo
:type repository_obj: Optional[object]
:return: Configured codec wrapper instance
:rtype: object
"""
wrapper_name: str = codec_configuration["wrapper_name"]
wrapper_file: str = codec_configuration["wrapper_file"]
try:
codec_wrapper_module = importlib.import_module(
f"lfc_toolkit.src.codecwrappers.{wrapper_file}"
)
except ModuleNotFoundError:
try:
codec_wrapper_module = importlib.import_module(wrapper_file)
except ModuleNotFoundError:
raise ModuleNotFoundError(
"Error. You must provide a valid wrapper_file name in the codec wrapper configuration."
)
CodecWrapperClass = getattr(codec_wrapper_module, wrapper_name)
path = codec_configuration["path"]
results_path = codec_configuration["results"]
force_encoding = codec_configuration.get("force_encoding", True)
encoded_extension = codec_configuration["encoded_extension"]
kwargs = codec_configuration.get("kwargs", {})
clear_log = codec_configuration.get("clear_log", True)
repetitions = codec_configuration.get("repetitions", 1)
codec_wrapper_instance = CodecWrapperClass(
codec=codec,
codec_path=f"{path}",
results_path=results_path,
clear_log=clear_log,
repetitions=repetitions,
force_encoding=force_encoding,
num_cores=codec_configuration.get("num_cores", None),
encoded_extension=encoded_extension,
repository=repository_obj,
**kwargs,
)
return codec_wrapper_instance
[docs]
def process_codec_using_wrapper(
codec_wrapper_instance: object,
configuration: ConfigurationReader,
codec_configuration: Dict,
) -> None:
"""Encodes and decodes all lightfields using the codec wrapper.
:param codec_wrapper_instance: Prepared codec wrapper instance
:type codec_wrapper_instance: object
:param configuration: Configuration reader
:type configuration: ConfigurationReader
:param codec_configuration: Codec-specific configuration
:type codec_configuration: Dict
:return: None
:rtype: None
"""
raw_type = codec_configuration["raw_type"]
decoded_conversions = codec_configuration.get("decoded_conversions", list())
keep_decoded = codec_configuration.get("keep_decoded", [raw_type])
if raw_type not in keep_decoded:
print("Warning: currently not keeping the decoded raw format is not supported")
# Prepare raw lightfields
raw_lfs = [
LightFieldFactory.get_raw_lightfield(
configuration=configuration, lightfield_name=name, raw_type=raw_type
)
for name in configuration.lightfield_names
]
for raw_lf in raw_lfs:
print(
f"Encoding {raw_lf.name} with {codec_wrapper_instance.codec} from {raw_type}..."
)
encoded_lfs = codec_wrapper_instance.encode_lightfield_for_target_bpps(
raw_lf,
bpps=configuration.lightfield_configurations[raw_lf.name]["target-rates"],
)
print(
f"Finished encoding {raw_lf.name} with {codec_wrapper_instance.codec} from {raw_type}."
)
if codec_configuration.get("decode", True):
for encoded_lf in encoded_lfs:
print(
f"Decoding {encoded_lf.name} with {codec_wrapper_instance.codec} for {encoded_lf.target_bitrate} bpp..."
)
decoded_lf = codec_wrapper_instance.decode(encoded_lf)
print("Decoded.")
for decoded_conversion_target_type in decoded_conversions:
print(
f"Converting decoded from {raw_type} to {decoded_conversion_target_type}..."
)
remove_this_conversion = (
decoded_conversion_target_type not in keep_decoded
)
converted = LightfieldConverter.convert(
source=decoded_lf,
destination_type=decoded_conversion_target_type,
remove_after_using=remove_this_conversion,
)
codec_wrapper_instance.add_decoded_conversion(converted)
[docs]
def main(configuration: ConfigurationReader = None, fulldocker: Optional[bool] = None) -> None:
"""Encode and decode light fields using configured codecs.
Orchestrates the complete encoding/decoding pipeline:
1. Clones and builds codec repositories if configured
2. Initializes codec wrappers
3. Encodes all light fields at target bitrates
4. Decodes encoded files and converts to configured formats
Usage: python run_codecs.py <configuration.json>
:param configuration: Configuration reader (read from argv if None)
:type configuration: ConfigurationReader
:param fulldocker: Whether to use full Docker for building and running codecs
:type fulldocker: bool
:return: None
:rtype: None
"""
if not configuration:
configuration = read_config_from_argv()
codecs = configuration["codecs-to-run"]
sample_codecs = configuration["sample-codecs-to-run"]
if len(codecs) == 0 and len(sample_codecs) > 0:
print("Skipping codecs execution. Will use sampled values.")
return
for codec, codec_configuration in codecs.items():
print(f"Executing codec {codec}")
# Clone/build repo if needed (returns repo object or None)
repo = clone_and_build_repo(
codec_configuration=codec_configuration,
global_configuration=configuration,
fulldocker=fulldocker,
)
# Create wrapper instance
codec_wrapper_instance = create_codec_wrapper_instance(
codec=codec,
codec_configuration=codec_configuration,
repository_obj=repo,
)
# Process encoding/decoding
process_codec_using_wrapper(
codec_wrapper_instance, configuration, codec_configuration
)
# Create execution log if configured
results_path = codec_configuration.get("results")
output_log_filename = (
f'{results_path}/{codec_configuration.get("log","execution_log.json")}'
)
try:
codec_wrapper_instance.create_execution_log(
output_log_filename=output_log_filename
)
except Exception:
# Keep behavior lenient: some wrappers may not implement logging the same way
pass
if __name__ == "__main__":
main()