111 lines
4.1 KiB
Python
Executable File
111 lines
4.1 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 subprocess
|
|
import sys
|
|
import re
|
|
from pathlib import Path
|
|
import threading
|
|
import time
|
|
|
|
|
|
def monitor_and_output_progress(proc):
|
|
"""Monitor subprocess output and emit simple progress markers."""
|
|
|
|
last_progress = 0
|
|
last_emit_time = 0
|
|
MIN_EMIT_INTERVAL = 0.5 # Minimum time between progress emissions
|
|
|
|
# Track if we've seen explicit progress markers
|
|
has_explicit_progress = False
|
|
|
|
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
|
|
]
|
|
|
|
for pattern, extractor in stage_patterns:
|
|
match = re.search(pattern, line)
|
|
if match:
|
|
try:
|
|
progress = extractor(match)
|
|
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
|
|
# Emit simple progress marker
|
|
print(f"\nPROGRESS: {progress:.0f}", flush=True)
|
|
break
|
|
except:
|
|
pass
|
|
|
|
# Ensure we reach 100% if successful
|
|
if proc.wait() == 0 and last_progress < 100:
|
|
print(f"\nPROGRESS: 100", flush=True)
|
|
|
|
|
|
def main():
|
|
"""Run the converter with progress monitoring."""
|
|
# Pass all arguments to the actual converter
|
|
proc = 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,
|
|
)
|
|
|
|
monitor_and_output_progress(proc)
|
|
return proc.returncode
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|