diff --git a/Lib/profiling/sampling/binary_reader.py b/Lib/profiling/sampling/binary_reader.py
index 2087c0da7c43b6..ece79853c696f7 100644
--- a/Lib/profiling/sampling/binary_reader.py
+++ b/Lib/profiling/sampling/binary_reader.py
@@ -103,6 +103,28 @@ def get_stats(self):
return self._reader.get_stats()
+def _set_replay_stats(collector, info, sample_interval_usec):
+ """Pass recorded statistics to a collector using the live sampling API."""
+ duration_sec = info.get("duration_sec")
+ sample_rate = info.get("sample_rate")
+ mode = info.get("mode")
+ if (
+ duration_sec is not None
+ and sample_rate is not None
+ and hasattr(collector, "set_stats")
+ ):
+ collector.set_stats(
+ sample_interval_usec,
+ duration_sec,
+ sample_rate,
+ error_rate=info.get("error_rate"),
+ missed_samples=info.get("missed_samples"),
+ mode=mode,
+ )
+ elif hasattr(collector, "set_mode"):
+ collector.set_mode(mode)
+
+
def convert_binary_to_format(input_file, output_file, output_format,
sample_interval_usec=None, progress_callback=None):
"""Convert a binary profiling file to another format.
@@ -137,10 +159,7 @@ def convert_binary_to_format(input_file, output_file, output_format,
# Replay samples through collector
count = reader.replay_samples(collector, progress_callback)
- if hasattr(collector, "set_replay_stats"):
- collector.set_replay_stats(info)
- if hasattr(collector, "set_mode"):
- collector.set_mode(info.get("mode"))
+ _set_replay_stats(collector, info, interval)
# Export to target format
collector.export(output_file)
diff --git a/Lib/profiling/sampling/cli.py b/Lib/profiling/sampling/cli.py
index c5fdcd7331096b..e4928638568f3e 100644
--- a/Lib/profiling/sampling/cli.py
+++ b/Lib/profiling/sampling/cli.py
@@ -22,7 +22,7 @@
from .gecko_collector import GeckoCollector
from .jsonl_collector import JsonlCollector
from .binary_collector import BinaryCollector
-from .binary_reader import BinaryReader
+from .binary_reader import BinaryReader, _set_replay_stats
from .constants import (
MICROSECONDS_PER_SECOND,
PROFILING_MODE_ALL,
@@ -793,10 +793,7 @@ def progress_callback(current, total):
)
count = reader.replay_samples(collector, progress_callback)
- if hasattr(collector, "set_replay_stats"):
- collector.set_replay_stats(info)
- if hasattr(collector, "set_mode"):
- collector.set_mode(info.get("mode"))
+ _set_replay_stats(collector, info, interval)
print()
if args.format == "pstats":
diff --git a/Lib/profiling/sampling/heatmap_collector.py b/Lib/profiling/sampling/heatmap_collector.py
index 62f42755827f06..fb25e01ebfdbd9 100644
--- a/Lib/profiling/sampling/heatmap_collector.py
+++ b/Lib/profiling/sampling/heatmap_collector.py
@@ -752,6 +752,8 @@ def _generate_index_html(self, index_path: Path, file_stats: List[FileStats]):
missed_samples_class = "good"
# Populate template
+ duration_sec = self.stats.get('duration_sec')
+ sample_rate = self.stats.get('sample_rate')
replacements = {
"": f"",
"": f"",
@@ -759,8 +761,8 @@ def _generate_index_html(self, index_path: Path, file_stats: List[FileStats]):
"": f"{sys.version_info.major}.{sys.version_info.minor}",
"": f"{len(file_stats):n}",
"": f"{self._total_samples:n}",
- "": fmt(self.stats.get('duration_sec', 0)),
- "": fmt(self.stats.get('sample_rate', 0)),
+ "": f"{fmt(duration_sec)} sec" if duration_sec is not None else "N/A",
+ "": fmt(sample_rate) if sample_rate is not None else "N/A",
"": error_rate_str,
"": str(error_rate_width),
"": error_rate_class,
diff --git a/Lib/profiling/sampling/stack_collector.py b/Lib/profiling/sampling/stack_collector.py
index 60061b4a2ecf11..9a26ea6fbe54e5 100644
--- a/Lib/profiling/sampling/stack_collector.py
+++ b/Lib/profiling/sampling/stack_collector.py
@@ -152,20 +152,6 @@ def set_stats(self, sample_interval_usec, duration_sec, sample_rate,
"mode": mode
}
- def set_replay_stats(self, info):
- """Restore measured statistics stored in a binary profile."""
- duration_sec = info.get("duration_sec")
- sample_rate = info.get("sample_rate")
- if duration_sec is None or sample_rate is None:
- return
- self.set_stats(
- self.sample_interval_usec,
- duration_sec,
- sample_rate,
- error_rate=info.get("error_rate"),
- missed_samples=info.get("missed_samples"),
- mode=self.stats.get("mode"),
- )
def set_mode(self, mode):
self.stats["mode"] = mode
diff --git a/Lib/test/test_profiling/test_sampling_profiler/test_binary_format.py b/Lib/test/test_profiling/test_sampling_profiler/test_binary_format.py
index 66a9c68a61c47d..9a29b0050245c1 100644
--- a/Lib/test/test_profiling/test_sampling_profiler/test_binary_format.py
+++ b/Lib/test/test_profiling/test_sampling_profiler/test_binary_format.py
@@ -7,10 +7,11 @@
import struct
import tempfile
import unittest
+from types import SimpleNamespace
from unittest import mock
from collections import defaultdict
-from test.support import captured_stderr
+from test.support import captured_stderr, captured_stdout, run_with_locale
try:
import _remote_debugging
@@ -28,6 +29,7 @@
)
from profiling.sampling.binary_collector import BinaryCollector
from profiling.sampling.binary_reader import BinaryReader, convert_binary_to_format
+ from profiling.sampling.cli import _replay_with_reader
from profiling.sampling.constants import PROFILING_MODE_CPU
from profiling.sampling.gecko_collector import GeckoCollector
@@ -1937,6 +1939,48 @@ def test_original_stats_extension_remains_readable(self):
self.assertIsNone(info["missed_samples"])
+class TestBinaryReplayToHeatmap(unittest.TestCase):
+ @run_with_locale("LC_NUMERIC", "C")
+ def test_replay_with_and_without_persisted_stats(self):
+ frame = make_frame(__file__, 1, "hot_func")
+ for with_stats in (True, False):
+ with (
+ self.subTest(with_stats=with_stats),
+ tempfile.TemporaryDirectory() as tmp,
+ captured_stdout(),
+ ):
+ bin_path = os.path.join(tmp, "profile.bin")
+ with BinaryCollector(bin_path, 2000, compression="none") as collector:
+ collector.collect([
+ make_interpreter(0, [make_thread(1, [frame])])
+ ])
+ if with_stats:
+ collector.set_stats(
+ 2000, 2.5, 0.4, error_rate=2.0, missed_samples=1.5
+ )
+ args = SimpleNamespace(
+ format="heatmap", input_file=bin_path,
+ outfile=tmp, diff_baseline=None,
+ )
+ with BinaryReader(bin_path) as reader:
+ _replay_with_reader(args, reader)
+ content = pathlib.Path(tmp, "index.html").read_text(
+ encoding="utf-8"
+ )
+ if with_stats:
+ self.assertIn('2.5 sec', content)
+ self.assertIn('0.4', content)
+ self.assertIn('2.0%', content)
+ self.assertIn('1.5%', content)
+ else:
+ self.assertEqual(
+ content.count('N/A'), 2
+ )
+ self.assertEqual(
+ content.count('N/A'), 2
+ )
+
+
class TestBinaryReplayToJsonl(BinaryFormatTestBase):
"""Tests for binary -> JSONL replay via convert_binary_to_format."""
diff --git a/Lib/test/test_profiling/test_sampling_profiler/test_collectors.py b/Lib/test/test_profiling/test_sampling_profiler/test_collectors.py
index d4ba2df5044559..6f411e87a6e715 100644
--- a/Lib/test/test_profiling/test_sampling_profiler/test_collectors.py
+++ b/Lib/test/test_profiling/test_sampling_profiler/test_collectors.py
@@ -13,6 +13,7 @@
try:
import _remote_debugging # noqa: F401
from profiling.sampling import gecko_collector
+ from profiling.sampling.binary_reader import _set_replay_stats
from profiling.sampling.pstats_collector import PstatsCollector
from profiling.sampling.stack_collector import (
CollapsedStackCollector,
@@ -1218,12 +1219,12 @@ def test_flamegraph_collector_stats_accumulation(self):
def test_flamegraph_collector_restores_replay_stats(self):
"""Replay restores measured values from binary metadata."""
collector = FlamegraphCollector(1000)
- collector.set_replay_stats({
+ _set_replay_stats(collector, {
"duration_sec": 1.25,
"sample_rate": 4.0,
"error_rate": 2.5,
"missed_samples": 1.5,
- })
+ }, 1000)
self.assertEqual(collector.stats["duration_sec"], 1.25)
self.assertEqual(collector.stats["sample_rate"], 4.0)
@@ -1233,10 +1234,12 @@ def test_flamegraph_collector_restores_replay_stats(self):
def test_flamegraph_collector_leaves_legacy_replay_stats_unavailable(self):
collector = FlamegraphCollector(1000)
original_stats = collector.stats.copy()
- collector.set_replay_stats({
+ original_stats["mode"] = PROFILING_MODE_CPU
+ _set_replay_stats(collector, {
"duration_sec": None,
"sample_rate": None,
- })
+ "mode": PROFILING_MODE_CPU,
+ }, 1000)
self.assertEqual(collector.stats, original_stats)
def test_flamegraph_collector_per_thread_stats(self):
diff --git a/Misc/NEWS.d/next/Library/2026-10-31-18-25-26.gh-issue-161008.YotuUZ.rst b/Misc/NEWS.d/next/Library/2026-10-31-18-25-26.gh-issue-161008.YotuUZ.rst
new file mode 100644
index 00000000000000..5656ecb87df2d2
--- /dev/null
+++ b/Misc/NEWS.d/next/Library/2026-10-31-18-25-26.gh-issue-161008.YotuUZ.rst
@@ -0,0 +1,2 @@
+Fix zero duration and sample rate in heatmap replay output. Patch by Timofei
+Ivankov.