Skip to content

Commit

Permalink
Python formatting tools (#8)
Browse files Browse the repository at this point in the history
* Formatting tools

* JPEG formatting

* PR fixes
  • Loading branch information
Xierumeng authored Mar 24, 2024
1 parent 1729650 commit 05b3446
Show file tree
Hide file tree
Showing 9 changed files with 353 additions and 90 deletions.
49 changes: 49 additions & 0 deletions .github/workflows/run-tests.yml
Original file line number Diff line number Diff line change
@@ -0,0 +1,49 @@
# This workflow will install Python dependencies and run tests with PyTest using Python 3.8
# For more information see: https://help.github.com/actions/language-and-framework-guides/using-python-with-github-actions

name: Run tests

on:
push:
branches: [main]
pull_request:
branches: [main]

jobs:
build:
runs-on: ubuntu-latest

steps:
# Checkout repository
- uses: actions/checkout@v3

# Set Python version
- name: Set up Python 3.8
uses: actions/setup-python@v4
with:
python-version: 3.8

# Set up submodules and submodule dependencies
# TODO: Uncomment when there are submodules
# - name: Set up submodule and submodule dependencies
# run: |
# git submodule update --init --recursive --remote
# pip install -r ./modules/common/requirements.txt

# Install project dependencies
- name: Install project dependencies
run: |
python -m pip install --upgrade pip
pip install -r requirements.txt
# Run linters and formatters
- name: Linters and formatters
run: |
black --check .
flake8 .
pylint .
# Run unit tests with PyTest
# TODO: Uncomment when there are unit tests to run
# - name: Run unit tests
# run: pytest -vv
7 changes: 5 additions & 2 deletions .gitignore
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
# IDE
# IDEs
.idea/
.vscode/

Expand All @@ -7,4 +7,7 @@ __pycache__/
venv/

# Logging
*log*
logs/

# Testing
test_images/
47 changes: 31 additions & 16 deletions avif_benchmark.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
as well as a .json with the test data and a .csv which provides a more human-friendly summary
of the data.
"""

import gc
import io
import json
Expand All @@ -21,7 +22,7 @@
FRAME_COUNT = 300 # Total number of frames
FRAME_TO_SAVE = 69 # This frame is good, it has both landing pads in it
INPUT_PATH = pathlib.Path("test_images", "Encode Test Dataset 2024")
OUTPUT_PATH = pathlib.Path(f"log_{int(time.time())}")
OUTPUT_PATH = pathlib.Path("logs", str(int(time.time())))
# All the quality settings to test (-1 should represent 'lossless',
# although it is only lossless in case of 444 subsampling)
QUALITY_SETTINGS = [0, 10, 20, 30, 40, 50, 60, 70, 80, 90, 100]
Expand All @@ -44,7 +45,7 @@
"Quality",
"Chroma",
"Min Time (ms)",
"Max Time (ms)",
"Max Time (ms)",
"Avg Time (ms)",
"Min Size (B)",
"Max Size (B)",
Expand All @@ -56,9 +57,11 @@
HEADER_LINE = ",".join(HEADERS) + "\n"


def update_min_max(min_value: "int | float",
max_value: "int | float",
current_value: "int | float",) -> "tuple[int, int] | tuple[float, float]":
def update_min_max(
min_value: "int | float",
max_value: "int | float",
current_value: "int | float",
) -> "tuple[int, int] | tuple[float, float]":
"""
Updates the min and max values for a measurement.
Expand All @@ -70,7 +73,7 @@ def update_min_max(min_value: "int | float",
Returns: (min_value, max_value)
min_value: new updated minimum recorded value
max_value: new updated maximum recorded value
The intended output is something like [int, int] or [float, float],
but it is not guaranteed because the inputs could be a combination of int and float.
eg. could also be tuple[float, int]
Expand All @@ -83,7 +86,10 @@ def update_min_max(min_value: "int | float",
return min_value, max_value


def run():
def main() -> int:
"""
Main function.
"""
pillow_heif.register_avif_opener(thumbnails=False)

OUTPUT_PATH.mkdir(parents=True, exist_ok=True)
Expand All @@ -102,14 +108,15 @@ def run():
MAX_SIZE_RATIO_COMPRESSED_TO_ORIGINAL: 0,
AVG_SIZE_RATIO_COMPRESSED_TO_ORIGINAL: 0,
FRAME_DATA: [],
} for chroma in CHROMA_SETTINGS
} for quality in QUALITY_SETTINGS
}
for chroma in CHROMA_SETTINGS
}
for quality in QUALITY_SETTINGS
}

test_begin = time.time()
print("Start time:", test_begin)


for quality in QUALITY_SETTINGS:
print(f"-----------------QUALITY = {quality}--------------------")
for chroma in CHROMA_SETTINGS:
Expand Down Expand Up @@ -145,9 +152,10 @@ def run():
# Save singular test results
time_ns = end - start
size_B = buffer.getbuffer().nbytes
compression_ratio = 100 * size_B / os.path.getsize(
original_size_B = os.path.getsize(
pathlib.Path(INPUT_PATH, f"{frame_index}.png"),
)
compression_ratio = 100 * size_B / original_size_B

min_time_ns, max_time_ns = update_min_max(min_time_ns, max_time_ns, time_ns)
min_size_B, max_size_B = update_min_max(min_size_B, max_size_B, size_B)
Expand All @@ -163,7 +171,7 @@ def run():
test_result = {
"time_ns": time_ns,
"size_B": size_B,
"size_ratio_compressed_to_original_%": compression_ratio
"size_ratio_compressed_to_original_%": compression_ratio,
}
current_result[FRAME_DATA].append(test_result)

Expand All @@ -185,20 +193,21 @@ def run():
current_result[AVG_SIZE_B] = total_size_B / FRAME_COUNT
current_result[MIN_SIZE_RATIO_COMPRESSED_TO_ORIGINAL] = min_compression_ratio
current_result[MAX_SIZE_RATIO_COMPRESSED_TO_ORIGINAL] = max_compression_ratio
current_result[AVG_SIZE_RATIO_COMPRESSED_TO_ORIGINAL] = \
current_result[AVG_SIZE_RATIO_COMPRESSED_TO_ORIGINAL] = (
total_compression_ratio / FRAME_COUNT
)
print(f"chroma {chroma} completed")

print("")
print("-------------------TEST COMPLETED------------------")
print("")

# Saving full results
with open(pathlib.Path(OUTPUT_PATH, "results.json"), 'w', encoding="utf-8") as file:
with open(pathlib.Path(OUTPUT_PATH, "results.json"), "w", encoding="utf-8") as file:
file.write(json.dumps(results, indent=2))

# Saving shortcut results without frame data (for more human readability)
with open(pathlib.Path(OUTPUT_PATH, "summary.csv"), 'w', encoding="utf-8") as file:
with open(pathlib.Path(OUTPUT_PATH, "summary.csv"), "w", encoding="utf-8") as file:
file.write(HEADER_LINE)
for quality in QUALITY_SETTINGS:
for chroma in CHROMA_SETTINGS:
Expand Down Expand Up @@ -229,6 +238,12 @@ def run():
"secs",
)

return 0


if __name__ == "__main__":
run()
result_main = main()
if result_main < 0:
print(f"ERROR: Status code: {result_main}")

print("Done!")
54 changes: 34 additions & 20 deletions heif_benchmark.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
as well as a .json with the test data and a .csv which provides a more human-friendly summary
of the data.
"""

import gc
import io
import json
Expand All @@ -21,7 +22,7 @@
FRAME_COUNT = 300 # Total number of frames
FRAME_TO_SAVE = 69 # This frame is good, it has both landing pads in it
INPUT_PATH = pathlib.Path("test_images", "Encode Test Dataset 2024")
OUTPUT_PATH = pathlib.Path(f"log_{int(time.time())}")
OUTPUT_PATH = pathlib.Path("logs", str(int(time.time())))

# All the quality settings to test (-1 should represent 'lossless',
# although it is only lossless in case of 444 subsampling)
Expand All @@ -45,7 +46,7 @@
"Quality",
"Chroma",
"Min Time (ms)",
"Max Time (ms)",
"Max Time (ms)",
"Avg Time (ms)",
"Min Size (B)",
"Max Size (B)",
Expand All @@ -57,9 +58,11 @@
HEADER_LINE = ",".join(HEADERS) + "\n"


def update_min_max(min_value: "int | float",
max_value: "int | float",
current_value: "int | float",) -> "tuple[int, int] | tuple[float, float]":
def update_min_max(
min_value: "int | float",
max_value: "int | float",
current_value: "int | float",
) -> "tuple[int, int] | tuple[float, float]":
"""
Updates the min and max values for a measurement.
Expand All @@ -71,7 +74,7 @@ def update_min_max(min_value: "int | float",
Returns: (min_value, max_value)
min_value: new updated minimum recorded value
max_value: new updated maximum recorded value
The intended output is something like [int, int] or [float, float],
but it is not guaranteed because the inputs could be a combination of int and float.
eg. could also be tuple[float, int]
Expand All @@ -84,7 +87,10 @@ def update_min_max(min_value: "int | float",
return min_value, max_value


def run():
def main() -> int:
"""
Main function.
"""
register_heif_opener(thumbnails=False)

OUTPUT_PATH.mkdir(parents=True, exist_ok=True)
Expand All @@ -103,14 +109,15 @@ def run():
MAX_SIZE_RATIO_COMPRESSED_TO_ORIGINAL: 0,
AVG_SIZE_RATIO_COMPRESSED_TO_ORIGINAL: 0,
FRAME_DATA: [],
} for chroma in CHROMA_SETTINGS
} for quality in QUALITY_SETTINGS
}
for chroma in CHROMA_SETTINGS
}
for quality in QUALITY_SETTINGS
}

test_begin = time.time()
print("Start time:", test_begin)


for quality in QUALITY_SETTINGS:
print(f"-----------------QUALITY = {quality}--------------------")
for chroma in CHROMA_SETTINGS:
Expand All @@ -123,8 +130,7 @@ def run():
min_compression_ratio = float("inf")
max_compression_ratio = 0
total_compression_ratio = 0
current_result = \
results[f"quality_{quality}"][f"chroma_{chroma}"]
current_result = results[f"quality_{quality}"][f"chroma_{chroma}"]
for frame_index in range(FRAME_COUNT):
img = Image.open(pathlib.Path(INPUT_PATH, f"{frame_index}.png"))
buffer = io.BytesIO()
Expand All @@ -144,9 +150,10 @@ def run():
# Save singular test results
time_ns = end - start
size_B = buffer.getbuffer().nbytes
compression_ratio = 100 * size_B / os.path.getsize(
original_size_B = os.path.getsize(
pathlib.Path(INPUT_PATH, f"{frame_index}.png"),
)
compression_ratio = 100 * size_B / original_size_B

min_time_ns, max_time_ns = update_min_max(min_time_ns, max_time_ns, time_ns)
min_size_B, max_size_B = update_min_max(min_size_B, max_size_B, size_B)
Expand All @@ -162,7 +169,7 @@ def run():
test_result = {
"time_ns": time_ns,
"size_B": size_B,
"size_ratio_compressed_to_original_%": compression_ratio
"size_ratio_compressed_to_original_%": compression_ratio,
}
current_result[FRAME_DATA].append(test_result)

Expand All @@ -184,25 +191,25 @@ def run():
current_result[AVG_SIZE_B] = total_size_B / FRAME_COUNT
current_result[MIN_SIZE_RATIO_COMPRESSED_TO_ORIGINAL] = min_compression_ratio
current_result[MAX_SIZE_RATIO_COMPRESSED_TO_ORIGINAL] = max_compression_ratio
current_result[AVG_SIZE_RATIO_COMPRESSED_TO_ORIGINAL] = \
current_result[AVG_SIZE_RATIO_COMPRESSED_TO_ORIGINAL] = (
total_compression_ratio / FRAME_COUNT
)
print(f"chroma {chroma} complete")

print("")
print("-------------------TEST COMPLETED------------------")
print("")

# Saving full results
with open(pathlib.Path(OUTPUT_PATH, "results.json"), 'w', encoding="utf-8") as file:
with open(pathlib.Path(OUTPUT_PATH, "results.json"), "w", encoding="utf-8") as file:
file.write(json.dumps(results, indent=2))

# Saving shortcut results without frame data (for more human readability)
with open(pathlib.Path(OUTPUT_PATH, "summary.csv"), 'w', encoding="utf-8") as file:
with open(pathlib.Path(OUTPUT_PATH, "summary.csv"), "w", encoding="utf-8") as file:
file.write(HEADER_LINE)
for quality in QUALITY_SETTINGS:
for chroma in CHROMA_SETTINGS:
current_result = \
results[f"quality_{quality}"][f"chroma_{chroma}"]
current_result = results[f"quality_{quality}"][f"chroma_{chroma}"]
line_stats = [
str(quality),
str(chroma),
Expand All @@ -229,5 +236,12 @@ def run():
"secs",
)

return 0


if __name__ == "__main__":
run()
result_main = main()
if result_main < 0:
print(f"ERROR: Status code: {result_main}")

print("Done!")
Loading

0 comments on commit 05b3446

Please sign in to comment.