Mpall -

def _save_summary_text(self): """Save summary to text file.""" with open(self.args.output_summary, 'w') as f: f.write("MPALL EXECUTION SUMMARY\n") f.write("=" * 50 + "\n") for r in self.results: f.write(f"Task r.task_id: 'SUCCESS' if r.success else 'FAIL'\n") if not r.success: f.write(f" Error: r.stderr[:200]\n") def main(): parser = argparse.ArgumentParser( description="mpall - Run commands across multiple parallel processes", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: mpall -w 2 -c echo hello -r {} Replace placeholders mpall -c "echo name is age" -r name=alice,age=30 -r name=bob,age=25 Use replacement file mpall -c "process_file.py input output" -f replacements.txt -w 4 With retries and timeout mpall -c "curl url" -r url=http://example.com -t 30 --retries 3 Save results mpall -c "test.sh param" -r param=1 -o results.json --summary summary.txt """ )

# Test 2: With retries (simulate failure) ./mpall.py -c "sh -c 'exit code'" -r code=0 -r code=1 --retries 2 def _save_summary_text(self): """Save summary to text file

parser.add_argument( "-o", "--output-json", help="Save detailed results to JSON file" ) age=30 -r name=bob

def parse_replacements(self) -> List[Dict[str, str]]: """Parse replacement arguments into list of parameter dictionaries.""" replacements = [] if self.args.replace_file: with open(self.args.replace_file, 'r') as f: for line in f: line = line.strip() if not line or line.startswith('#'): continue parts = line.split() if len(parts) < 2: self.logger.warning(f"Skipping invalid line: line") continue # Format: key1=val1 key2=val2 ... replacement = {} for part in parts: if '=' in part: k, v = part.split('=', 1) replacement[k] = v if replacement: replacements.append(replacement) elif self.args.replace: # Format: key1=val1,key2=val2 or multiple -r flags for rep in self.args.replace: rep_dict = {} for pair in rep.split(','): if '=' in pair: k, v = pair.split('=', 1) rep_dict[k] = v replacements.append(rep_dict) else: # Single run with no replacements replacements.append({}) return replacements v = part.split('='

def _save_results_json(self): """Save detailed results to JSON file.""" data = "timestamp": datetime.now().isoformat(), "command": self.args.command, "total_tasks": len(self.results), "successful": sum(1 for r in self.results if r.success), "failed": sum(1 for r in self.results if not r.success), "results": [ "task_id": r.task_id, "args": dict(r.args), "success": r.success, "stdout": r.stdout, "stderr": r.stderr, "exit_code": r.exit_code, "duration": r.duration, "retries": r.retries for r in self.results ] with open(self.args.output_json, 'w') as f: json.dump(data, f, indent=2) self.logger.info(f"Results saved to self.args.output_json")

def _print_summary(self, total_tasks: int, total_duration: float): """Print execution summary.""" succeeded = sum(1 for r in self.results if r.success) failed = total_tasks - succeeded self.logger.info("=" * 50) self.logger.info("SUMMARY") self.logger.info(f"Total tasks: total_tasks") self.logger.info(f"Succeeded: succeeded") self.logger.info(f"Failed: failed") self.logger.info(f"Total duration: total_duration:.2fs") if failed > 0: self.logger.info("\nFailed tasks:") for r in self.results: if not r.success: self.logger.info(f" Task r.task_id: r.stderr[:200]")

parser.add_argument( "--retries", type=int, default=0, help="Number of retries on failure (default: 0)" )