113 lines
3.8 KiB
Python
Executable File
113 lines
3.8 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Wrapper for convert_rdf_to_hf_dataset.py that adds simple progress output
|
|
for subprocess monitoring in parallel processing.
|
|
|
|
Progress Scale:
|
|
0-10%: Initialization and format detection
|
|
10-50%: Parsing RDF file to graph
|
|
50-90%: Converting graph to dataset format
|
|
90-100%: Final preparation and saving
|
|
"""
|
|
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from scripts.common import print_progress
|
|
|
|
|
|
def monitor_and_output_progress(proc: subprocess.Popen[Any]) -> None:
|
|
"""Monitor subprocess output and emit simple progress markers."""
|
|
|
|
last_progress = 0.0
|
|
last_emit_time = 0.0
|
|
min_emit_interval = 0.5 # Minimum time between progress emissions
|
|
|
|
# Track if we've seen explicit progress markers
|
|
has_explicit_progress = False
|
|
|
|
assert proc.stdout is not None
|
|
for line in iter(proc.stdout.readline, ""):
|
|
if not line:
|
|
break
|
|
|
|
# Pass through the original output
|
|
print(line, end="", flush=True)
|
|
|
|
# Look for explicit PROGRESS markers first
|
|
progress_match = re.search(r"PROGRESS:\s*(\d+)", line)
|
|
if progress_match:
|
|
progress = float(progress_match.group(1))
|
|
has_explicit_progress = True
|
|
current_time = time.time()
|
|
|
|
# Only emit if progress increased and enough time has passed
|
|
if (
|
|
progress > last_progress
|
|
and (current_time - last_emit_time) > min_emit_interval
|
|
):
|
|
last_progress = progress
|
|
last_emit_time = current_time
|
|
# Progress already in the output; don't double-emit
|
|
continue
|
|
|
|
# If no explicit progress, try to infer from stage markers
|
|
if not has_explicit_progress:
|
|
stage_patterns = [
|
|
# Phase detection with consistent progress mapping
|
|
(r"Parsing RDF file:", lambda m: 5), # Starting
|
|
(r"Phase 1:", lambda m: 10), # Starting reading
|
|
(r"Reading file", lambda m: 15), # Reading phase
|
|
(r"✓ Read (\d+\.?\d*) MB", lambda m: 25), # File read complete
|
|
(r"Phase 2:", lambda m: 30), # Starting parsing
|
|
(r"Parsing RDF", lambda m: 35), # Parsing phase
|
|
(r"Successfully loaded (\d+)", lambda m: 45), # Loaded triples
|
|
(r"✓ Parsed", lambda m: 50), # Parse complete
|
|
(r"Preparing graph", lambda m: 55), # Prep phase
|
|
(r"Ready to convert", lambda m: 60), # Ready phase
|
|
(r"Converting triples", lambda m: 65), # Converting phase
|
|
(r"Creating HuggingFace", lambda m: 90), # Creating dataset
|
|
(r"Building dataset", lambda m: 92), # Building dataset
|
|
(r"Dataset built", lambda m: 95), # Built
|
|
(r"Dataset saved|Successfully", lambda m: 100), # Complete
|
|
]
|
|
|
|
last_progress = print_progress(
|
|
last_emit_time,
|
|
last_progress,
|
|
line,
|
|
min_emit_interval,
|
|
stage_patterns,
|
|
)
|
|
|
|
# Ensure we reach 100% if successful
|
|
if proc.wait() == 0 and last_progress < 100:
|
|
print("\nPROGRESS: 100", flush=True)
|
|
|
|
|
|
def main() -> int:
|
|
"""Run the converter with progress monitoring."""
|
|
# Pass all arguments to the actual converter
|
|
with subprocess.Popen(
|
|
[
|
|
sys.executable,
|
|
"-u",
|
|
str(Path(__file__).parent / "convert_rdf_to_hf_dataset.py"),
|
|
*sys.argv[1:],
|
|
],
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
text=True,
|
|
bufsize=1,
|
|
) as proc:
|
|
monitor_and_output_progress(proc)
|
|
return proc.returncode
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|