| import subprocess |
| import tempfile |
| from pathlib import Path |
| from typing import List, Optional |
|
|
| |
| |
|
|
| from mcp.server.fastmcp import FastMCP |
|
|
| SERVER_NAME = 'local_bedops' |
| mcp = FastMCP(SERVER_NAME) |
|
|
| @mcp.tool() |
| def bedops_complement( |
| files: List[Path], |
| chop_to_limits: bool = False, |
| chrom: Optional[str] = None, |
| ec: bool = False, |
| header: bool = False, |
| range_str: Optional[str] = None, |
| output_file: Optional[Path] = None, |
| ) -> dict: |
| """ |
| Computes the complement of one or more BED files. |
| |
| This corresponds to the `bedops -c` or `bedops --complement` operation. |
| It finds the regions within chromosome boundaries that are not covered by |
| any intervals in the input file(s). |
| |
| Args: |
| files: A list of one or more input BED/Starch files. Must be sorted. |
| chop_to_limits: If True, chop complementary regions to chromosome limits |
| defined by the first input file (-L flag). |
| chrom: Process data for the given chromosome only. |
| ec: Error check input files (slower). |
| header: Accept headers (VCF, GFF, SAM, BED, WIG) in any input file. |
| range_str: Pad coordinates. Use 'L:R' format (e.g., '-10:20') or a |
| single value 'S' for symmetric padding (e.g., '100'). |
| output_file: Optional path to save the output. If not provided, |
| output is returned as a string in the result dictionary. |
| |
| Returns: |
| A dictionary containing the command executed, stdout, stderr, and a |
| list of output files generated. |
| """ |
| if not files: |
| raise ValueError("At least one input file must be provided for the complement operation.") |
| for file_path in files: |
| if not file_path.exists(): |
| raise FileNotFoundError(f"Input file not found: {file_path}") |
|
|
| cmd = ["bedops"] |
|
|
| if chrom: |
| cmd.extend(["--chrom", chrom]) |
| if ec: |
| cmd.append("--ec") |
| if header: |
| cmd.append("--header") |
| if range_str: |
| cmd.extend(["--range", range_str]) |
|
|
| cmd.append("--complement") |
| if chop_to_limits: |
| cmd.append("-L") |
|
|
| cmd.extend([str(p) for p in files]) |
|
|
| try: |
| if output_file: |
| with open(output_file, "w") as f: |
| result = subprocess.run( |
| cmd, check=True, text=True, stdout=f, stderr=subprocess.PIPE |
| ) |
| stdout_capture = "" |
| output_files_list = [str(output_file)] |
| else: |
| result = subprocess.run( |
| cmd, check=True, text=True, capture_output=True |
| ) |
| stdout_capture = result.stdout |
| output_files_list = [] |
|
|
| return { |
| "command_executed": " ".join(cmd), |
| "stdout": stdout_capture, |
| "stderr": result.stderr, |
| "output_files": output_files_list, |
| } |
| except subprocess.CalledProcessError as e: |
| raise RuntimeError( |
| f"bedops complement failed with exit code {e.returncode}\n" |
| f"Stderr: {e.stderr}\n" |
| f"Stdout: {e.stdout}\n" |
| f"Command: {' '.join(cmd)}" |
| ) from e |
|
|
| @mcp.tool() |
| def bedops_difference( |
| files: List[Path], |
| chrom: Optional[str] = None, |
| ec: bool = False, |
| header: bool = False, |
| range_str: Optional[str] = None, |
| output_file: Optional[Path] = None, |
| ) -> dict: |
| """ |
| Computes the difference between a reference BED file and one or more other BED files. |
| |
| This corresponds to the `bedops -d` or `bedops --difference` operation. |
| It returns regions from the first (reference) file that do not overlap |
| with any regions in the subsequent files. |
| |
| Args: |
| files: A list of two or more input BED/Starch files. The first file is |
| the reference. All files must be sorted. |
| chrom: Process data for the given chromosome only. |
| ec: Error check input files (slower). |
| header: Accept headers (VCF, GFF, SAM, BED, WIG) in any input file. |
| range_str: Pad coordinates. Use 'L:R' format (e.g., '-10:20') or a |
| single value 'S' for symmetric padding (e.g., '100'). |
| output_file: Optional path to save the output. If not provided, |
| output is returned as a string in the result dictionary. |
| |
| Returns: |
| A dictionary containing the command executed, stdout, stderr, and a |
| list of output files generated. |
| """ |
| if len(files) < 2: |
| raise ValueError("At least two input files (a reference and one other) must be provided for the difference operation.") |
| for file_path in files: |
| if not file_path.exists(): |
| raise FileNotFoundError(f"Input file not found: {file_path}") |
|
|
| cmd = ["bedops"] |
|
|
| if chrom: |
| cmd.extend(["--chrom", chrom]) |
| if ec: |
| cmd.append("--ec") |
| if header: |
| cmd.append("--header") |
| if range_str: |
| cmd.extend(["--range", range_str]) |
|
|
| cmd.append("--difference") |
| cmd.extend([str(p) for p in files]) |
|
|
| try: |
| if output_file: |
| with open(output_file, "w") as f: |
| result = subprocess.run( |
| cmd, check=True, text=True, stdout=f, stderr=subprocess.PIPE |
| ) |
| stdout_capture = "" |
| output_files_list = [str(output_file)] |
| else: |
| result = subprocess.run( |
| cmd, check=True, text=True, capture_output=True |
| ) |
| stdout_capture = result.stdout |
| output_files_list = [] |
|
|
| return { |
| "command_executed": " ".join(cmd), |
| "stdout": stdout_capture, |
| "stderr": result.stderr, |
| "output_files": output_files_list, |
| } |
| except subprocess.CalledProcessError as e: |
| raise RuntimeError( |
| f"bedops difference failed with exit code {e.returncode}\n" |
| f"Stderr: {e.stderr}\n" |
| f"Stdout: {e.stdout}\n" |
| f"Command: {' '.join(cmd)}" |
| ) from e |
|
|
| @mcp.tool() |
| def bedops_element_of( |
| files: List[Path], |
| overlap_criterion: str = "100%", |
| chrom: Optional[str] = None, |
| ec: bool = False, |
| header: bool = False, |
| range_str: Optional[str] = None, |
| output_file: Optional[Path] = None, |
| ) -> dict: |
| """ |
| Finds elements in the reference file that overlap other files by a specified amount. |
| |
| This corresponds to the `bedops -e` or `bedops --element-of` operation. |
| It returns elements from the first (reference) file that overlap elements |
| in any of the other files by at least the specified amount. |
| |
| Args: |
| files: A list of two or more input BED/Starch files. The first file is |
| the reference. All files must be sorted. |
| overlap_criterion: The required overlap, as base pairs (e.g., '1') or |
| percentage (e.g., '50%'). Defaults to '100%'. |
| chrom: Process data for the given chromosome only. |
| ec: Error check input files (slower). |
| header: Accept headers (VCF, GFF, SAM, BED, WIG) in any input file. |
| range_str: Pad coordinates. Use 'L:R' or 'S' format. The first (reference) |
| file is NOT padded with this operation. |
| output_file: Optional path to save the output. If not provided, |
| output is returned as a string in the result dictionary. |
| |
| Returns: |
| A dictionary containing the command executed, stdout, stderr, and a |
| list of output files generated. |
| """ |
| if len(files) < 2: |
| raise ValueError("At least two input files (a reference and one other) must be provided for the element-of operation.") |
| for file_path in files: |
| if not file_path.exists(): |
| raise FileNotFoundError(f"Input file not found: {file_path}") |
|
|
| cmd = ["bedops"] |
|
|
| if chrom: |
| cmd.extend(["--chrom", chrom]) |
| if ec: |
| cmd.append("--ec") |
| if header: |
| cmd.append("--header") |
| if range_str: |
| cmd.extend(["--range", range_str]) |
|
|
| cmd.extend(["--element-of", overlap_criterion]) |
| cmd.extend([str(p) for p in files]) |
|
|
| try: |
| if output_file: |
| with open(output_file, "w") as f: |
| result = subprocess.run( |
| cmd, check=True, text=True, stdout=f, stderr=subprocess.PIPE |
| ) |
| stdout_capture = "" |
| output_files_list = [str(output_file)] |
| else: |
| result = subprocess.run( |
| cmd, check=True, text=True, capture_output=True |
| ) |
| stdout_capture = result.stdout |
| output_files_list = [] |
|
|
| return { |
| "command_executed": " ".join(cmd), |
| "stdout": stdout_capture, |
| "stderr": result.stderr, |
| "output_files": output_files_list, |
| } |
| except subprocess.CalledProcessError as e: |
| raise RuntimeError( |
| f"bedops element-of failed with exit code {e.returncode}\n" |
| f"Stderr: {e.stderr}\n" |
| f"Stdout: {e.stdout}\n" |
| f"Command: {' '.join(cmd)}" |
| ) from e |
|
|
| @mcp.tool() |
| def bedops_intersect( |
| files: List[Path], |
| chrom: Optional[str] = None, |
| ec: bool = False, |
| header: bool = False, |
| range_str: Optional[str] = None, |
| output_file: Optional[Path] = None, |
| ) -> dict: |
| """ |
| Computes the intersection of two or more BED files. |
| |
| This corresponds to the `bedops -i` or `bedops --intersect` operation. |
| It returns regions that are common to all input files. |
| |
| Args: |
| files: A list of two or more input BED/Starch files. All files must be sorted. |
| chrom: Process data for the given chromosome only. |
| ec: Error check input files (slower). |
| header: Accept headers (VCF, GFF, SAM, BED, WIG) in any input file. |
| range_str: Pad coordinates. Use 'L:R' format (e.g., '-10:20') or a |
| single value 'S' for symmetric padding (e.g., '100'). |
| output_file: Optional path to save the output. If not provided, |
| output is returned as a string in the result dictionary. |
| |
| Returns: |
| A dictionary containing the command executed, stdout, stderr, and a |
| list of output files generated. |
| """ |
| if len(files) < 2: |
| raise ValueError("At least two input files must be provided for the intersect operation.") |
| for file_path in files: |
| if not file_path.exists(): |
| raise FileNotFoundError(f"Input file not found: {file_path}") |
|
|
| cmd = ["bedops"] |
|
|
| if chrom: |
| cmd.extend(["--chrom", chrom]) |
| if ec: |
| cmd.append("--ec") |
| if header: |
| cmd.append("--header") |
| if range_str: |
| cmd.extend(["--range", range_str]) |
|
|
| cmd.append("--intersect") |
| cmd.extend([str(p) for p in files]) |
|
|
| try: |
| if output_file: |
| with open(output_file, "w") as f: |
| result = subprocess.run( |
| cmd, check=True, text=True, stdout=f, stderr=subprocess.PIPE |
| ) |
| stdout_capture = "" |
| output_files_list = [str(output_file)] |
| else: |
| result = subprocess.run( |
| cmd, check=True, text=True, capture_output=True |
| ) |
| stdout_capture = result.stdout |
| output_files_list = [] |
|
|
| return { |
| "command_executed": " ".join(cmd), |
| "stdout": stdout_capture, |
| "stderr": result.stderr, |
| "output_files": output_files_list, |
| } |
| except subprocess.CalledProcessError as e: |
| raise RuntimeError( |
| f"bedops intersect failed with exit code {e.returncode}\n" |
| f"Stderr: {e.stderr}\n" |
| f"Stdout: {e.stdout}\n" |
| f"Command: {' '.join(cmd)}" |
| ) from e |
|
|
| @mcp.tool() |
| def bedops_merge( |
| files: List[Path], |
| chrom: Optional[str] = None, |
| ec: bool = False, |
| header: bool = False, |
| range_str: Optional[str] = None, |
| output_file: Optional[Path] = None, |
| ) -> dict: |
| """ |
| Merges overlapping regions from one or more BED files. |
| |
| This corresponds to the `bedops -m` or `bedops --merge` operation. |
| It combines overlapping or adjacent intervals into a single, larger interval. |
| |
| Args: |
| files: A list of one or more input BED/Starch files. Must be sorted. |
| chrom: Process data for the given chromosome only. |
| ec: Error check input files (slower). |
| header: Accept headers (VCF, GFF, SAM, BED, WIG) in any input file. |
| range_str: Pad coordinates. Use 'L:R' format (e.g., '-10:20') or a |
| single value 'S' for symmetric padding (e.g., '100'). |
| output_file: Optional path to save the output. If not provided, |
| output is returned as a string in the result dictionary. |
| |
| Returns: |
| A dictionary containing the command executed, stdout, stderr, and a |
| list of output files generated. |
| """ |
| if not files: |
| raise ValueError("At least one input file must be provided for the merge operation.") |
| for file_path in files: |
| if not file_path.exists(): |
| raise FileNotFoundError(f"Input file not found: {file_path}") |
|
|
| cmd = ["bedops"] |
|
|
| if chrom: |
| cmd.extend(["--chrom", chrom]) |
| if ec: |
| cmd.append("--ec") |
| if header: |
| cmd.append("--header") |
| if range_str: |
| cmd.extend(["--range", range_str]) |
|
|
| cmd.append("--merge") |
| cmd.extend([str(p) for p in files]) |
|
|
| try: |
| if output_file: |
| with open(output_file, "w") as f: |
| result = subprocess.run( |
| cmd, check=True, text=True, stdout=f, stderr=subprocess.PIPE |
| ) |
| stdout_capture = "" |
| output_files_list = [str(output_file)] |
| else: |
| result = subprocess.run( |
| cmd, check=True, text=True, capture_output=True |
| ) |
| stdout_capture = result.stdout |
| output_files_list = [] |
|
|
| return { |
| "command_executed": " ".join(cmd), |
| "stdout": stdout_capture, |
| "stderr": result.stderr, |
| "output_files": output_files_list, |
| } |
| except subprocess.CalledProcessError as e: |
| raise RuntimeError( |
| f"bedops merge failed with exit code {e.returncode}\n" |
| f"Stderr: {e.stderr}\n" |
| f"Stdout: {e.stdout}\n" |
| f"Command: {' '.join(cmd)}" |
| ) from e |
|
|
| @mcp.tool() |
| def bedops_not_element_of( |
| files: List[Path], |
| overlap_criterion: str = "100%", |
| chrom: Optional[str] = None, |
| ec: bool = False, |
| header: bool = False, |
| range_str: Optional[str] = None, |
| output_file: Optional[Path] = None, |
| ) -> dict: |
| """ |
| Finds elements in the reference file that DO NOT overlap other files by a specified amount. |
| |
| This corresponds to the `bedops -n` or `bedops --not-element-of` operation. |
| It is the inverse of the `element-of` operation. |
| |
| Args: |
| files: A list of two or more input BED/Starch files. The first file is |
| the reference. All files must be sorted. |
| overlap_criterion: The required overlap, as base pairs (e.g., '1') or |
| percentage (e.g., '50%'). Defaults to '100%'. |
| chrom: Process data for the given chromosome only. |
| ec: Error check input files (slower). |
| header: Accept headers (VCF, GFF, SAM, BED, WIG) in any input file. |
| range_str: Pad coordinates. Use 'L:R' or 'S' format. The first (reference) |
| file is NOT padded with this operation. |
| output_file: Optional path to save the output. If not provided, |
| output is returned as a string in the result dictionary. |
| |
| Returns: |
| A dictionary containing the command executed, stdout, stderr, and a |
| list of output files generated. |
| """ |
| if len(files) < 2: |
| raise ValueError("At least two input files (a reference and one other) must be provided for the not-element-of operation.") |
| for file_path in files: |
| if not file_path.exists(): |
| raise FileNotFoundError(f"Input file not found: {file_path}") |
|
|
| cmd = ["bedops"] |
|
|
| if chrom: |
| cmd.extend(["--chrom", chrom]) |
| if ec: |
| cmd.append("--ec") |
| if header: |
| cmd.append("--header") |
| if range_str: |
| cmd.extend(["--range", range_str]) |
|
|
| cmd.extend(["--not-element-of", overlap_criterion]) |
| cmd.extend([str(p) for p in files]) |
|
|
| try: |
| if output_file: |
| with open(output_file, "w") as f: |
| result = subprocess.run( |
| cmd, check=True, text=True, stdout=f, stderr=subprocess.PIPE |
| ) |
| stdout_capture = "" |
| output_files_list = [str(output_file)] |
| else: |
| result = subprocess.run( |
| cmd, check=True, text=True, capture_output=True |
| ) |
| stdout_capture = result.stdout |
| output_files_list = [] |
|
|
| return { |
| "command_executed": " ".join(cmd), |
| "stdout": stdout_capture, |
| "stderr": result.stderr, |
| "output_files": output_files_list, |
| } |
| except subprocess.CalledProcessError as e: |
| raise RuntimeError( |
| f"bedops not-element-of failed with exit code {e.returncode}\n" |
| f"Stderr: {e.stderr}\n" |
| f"Stdout: {e.stdout}\n" |
| f"Command: {' '.join(cmd)}" |
| ) from e |
|
|
| @mcp.tool() |
| def bedops_partition( |
| files: List[Path], |
| chrom: Optional[str] = None, |
| ec: bool = False, |
| header: bool = False, |
| range_str: Optional[str] = None, |
| output_file: Optional[Path] = None, |
| ) -> dict: |
| """ |
| Partitions the input BED file(s) into disjoint segments. |
| |
| This corresponds to the `bedops -p` or `bedops --partition` operation. |
| It breaks the input regions into non-overlapping segments, reporting each |
| new segment and which input files it came from. |
| |
| Args: |
| files: A list of one or more input BED/Starch files. Must be sorted. |
| chrom: Process data for the given chromosome only. |
| ec: Error check input files (slower). |
| header: Accept headers (VCF, GFF, SAM, BED, WIG) in any input file. |
| range_str: Pad coordinates. Use 'L:R' format (e.g., '-10:20') or a |
| single value 'S' for symmetric padding (e.g., '100'). |
| output_file: Optional path to save the output. If not provided, |
| output is returned as a string in the result dictionary. |
| |
| Returns: |
| A dictionary containing the command executed, stdout, stderr, and a |
| list of output files generated. |
| """ |
| if not files: |
| raise ValueError("At least one input file must be provided for the partition operation.") |
| for file_path in files: |
| if not file_path.exists(): |
| raise FileNotFoundError(f"Input file not found: {file_path}") |
|
|
| cmd = ["bedops"] |
|
|
| if chrom: |
| cmd.extend(["--chrom", chrom]) |
| if ec: |
| cmd.append("--ec") |
| if header: |
| cmd.append("--header") |
| if range_str: |
| cmd.extend(["--range", range_str]) |
|
|
| cmd.append("--partition") |
| cmd.extend([str(p) for p in files]) |
|
|
| try: |
| if output_file: |
| with open(output_file, "w") as f: |
| result = subprocess.run( |
| cmd, check=True, text=True, stdout=f, stderr=subprocess.PIPE |
| ) |
| stdout_capture = "" |
| output_files_list = [str(output_file)] |
| else: |
| result = subprocess.run( |
| cmd, check=True, text=True, capture_output=True |
| ) |
| stdout_capture = result.stdout |
| output_files_list = [] |
|
|
| return { |
| "command_executed": " ".join(cmd), |
| "stdout": stdout_capture, |
| "stderr": result.stderr, |
| "output_files": output_files_list, |
| } |
| except subprocess.CalledProcessError as e: |
| raise RuntimeError( |
| f"bedops partition failed with exit code {e.returncode}\n" |
| f"Stderr: {e.stderr}\n" |
| f"Stdout: {e.stdout}\n" |
| f"Command: {' '.join(cmd)}" |
| ) from e |
|
|
| @mcp.tool() |
| def bedops_symmdiff( |
| files: List[Path], |
| chrom: Optional[str] = None, |
| ec: bool = False, |
| header: bool = False, |
| range_str: Optional[str] = None, |
| output_file: Optional[Path] = None, |
| ) -> dict: |
| """ |
| Computes the symmetric difference of two or more BED files. |
| |
| This corresponds to the `bedops -s` or `bedops --symmdiff` operation. |
| It returns regions that are unique to any of the input files (i.e., not |
| present in their intersection). |
| |
| Args: |
| files: A list of two or more input BED/Starch files. All files must be sorted. |
| chrom: Process data for the given chromosome only. |
| ec: Error check input files (slower). |
| header: Accept headers (VCF, GFF, SAM, BED, WIG) in any input file. |
| range_str: Pad coordinates. Use 'L:R' format (e.g., '-10:20') or a |
| single value 'S' for symmetric padding (e.g., '100'). |
| output_file: Optional path to save the output. If not provided, |
| output is returned as a string in the result dictionary. |
| |
| Returns: |
| A dictionary containing the command executed, stdout, stderr, and a |
| list of output files generated. |
| """ |
| if len(files) < 2: |
| raise ValueError("At least two input files must be provided for the symmetric difference operation.") |
| for file_path in files: |
| if not file_path.exists(): |
| raise FileNotFoundError(f"Input file not found: {file_path}") |
|
|
| cmd = ["bedops"] |
|
|
| if chrom: |
| cmd.extend(["--chrom", chrom]) |
| if ec: |
| cmd.append("--ec") |
| if header: |
| cmd.append("--header") |
| if range_str: |
| cmd.extend(["--range", range_str]) |
|
|
| cmd.append("--symmdiff") |
| cmd.extend([str(p) for p in files]) |
|
|
| try: |
| if output_file: |
| with open(output_file, "w") as f: |
| result = subprocess.run( |
| cmd, check=True, text=True, stdout=f, stderr=subprocess.PIPE |
| ) |
| stdout_capture = "" |
| output_files_list = [str(output_file)] |
| else: |
| result = subprocess.run( |
| cmd, check=True, text=True, capture_output=True |
| ) |
| stdout_capture = result.stdout |
| output_files_list = [] |
|
|
| return { |
| "command_executed": " ".join(cmd), |
| "stdout": stdout_capture, |
| "stderr": result.stderr, |
| "output_files": output_files_list, |
| } |
| except subprocess.CalledProcessError as e: |
| raise RuntimeError( |
| f"bedops symmdiff failed with exit code {e.returncode}\n" |
| f"Stderr: {e.stderr}\n" |
| f"Stdout: {e.stdout}\n" |
| f"Command: {' '.join(cmd)}" |
| ) from e |
|
|
| @mcp.tool() |
| def bedops_everything( |
| files: List[Path], |
| chrom: Optional[str] = None, |
| ec: bool = False, |
| header: bool = False, |
| range_str: Optional[str] = None, |
| output_file: Optional[Path] = None, |
| ) -> dict: |
| """ |
| Returns the union of all elements from input files without merging. |
| |
| This corresponds to the `bedops -u` or `bedops --everything` operation. |
| It effectively concatenates the input files while maintaining sort order |
| and preserving all original columns. |
| |
| Args: |
| files: A list of one or more input BED/Starch files. Must be sorted. |
| chrom: Process data for the given chromosome only. |
| ec: Error check input files (slower). |
| header: Accept headers (VCF, GFF, SAM, BED, WIG) in any input file. |
| range_str: Pad coordinates. Use 'L:R' format (e.g., '-10:20') or a |
| single value 'S' for symmetric padding (e.g., '100'). |
| output_file: Optional path to save the output. If not provided, |
| output is returned as a string in the result dictionary. |
| |
| Returns: |
| A dictionary containing the command executed, stdout, stderr, and a |
| list of output files generated. |
| """ |
| if not files: |
| raise ValueError("At least one input file must be provided for the everything operation.") |
| for file_path in files: |
| if not file_path.exists(): |
| raise FileNotFoundError(f"Input file not found: {file_path}") |
|
|
| cmd = ["bedops"] |
|
|
| if chrom: |
| cmd.extend(["--chrom", chrom]) |
| if ec: |
| cmd.append("--ec") |
| if header: |
| cmd.append("--header") |
| if range_str: |
| cmd.extend(["--range", range_str]) |
|
|
| cmd.append("--everything") |
| cmd.extend([str(p) for p in files]) |
|
|
| try: |
| if output_file: |
| with open(output_file, "w") as f: |
| result = subprocess.run( |
| cmd, check=True, text=True, stdout=f, stderr=subprocess.PIPE |
| ) |
| stdout_capture = "" |
| output_files_list = [str(output_file)] |
| else: |
| result = subprocess.run( |
| cmd, check=True, text=True, capture_output=True |
| ) |
| stdout_capture = result.stdout |
| output_files_list = [] |
|
|
| return { |
| "command_executed": " ".join(cmd), |
| "stdout": stdout_capture, |
| "stderr": result.stderr, |
| "output_files": output_files_list, |
| } |
| except subprocess.CalledProcessError as e: |
| raise RuntimeError( |
| f"bedops everything failed with exit code {e.returncode}\n" |
| f"Stderr: {e.stderr}\n" |
| f"Stdout: {e.stdout}\n" |
| f"Command: {' '.join(cmd)}" |
| ) from e |
|
|
| @mcp.tool() |
| def bedops_chop( |
| files: List[Path], |
| bp: int = 1, |
| stagger: Optional[int] = None, |
| exclusive_chop: bool = False, |
| chrom: Optional[str] = None, |
| ec: bool = False, |
| header: bool = False, |
| range_str: Optional[str] = None, |
| output_file: Optional[Path] = None, |
| ) -> dict: |
| """ |
| Chops elements into fixed-size, potentially staggered sub-elements. |
| |
| This corresponds to the `bedops -w` or `bedops --chop` operation. |
| |
| Args: |
| files: A list of one or more input BED/Starch files. Must be sorted. |
| bp: The size in base pairs of each chopped element. Defaults to 1. |
| stagger: The stagger distance in nucleotides. If not set, no staggering is done. |
| exclusive_chop: If True, removes single-base elements that can result |
| from chopping (-x flag). |
| chrom: Process data for the given chromosome only. |
| ec: Error check input files (slower). |
| header: Accept headers (VCF, GFF, SAM, BED, WIG) in any input file. |
| range_str: Pad coordinates. Use 'L:R' format (e.g., '-10:20') or a |
| single value 'S' for symmetric padding (e.g., '100'). |
| output_file: Optional path to save the output. If not provided, |
| output is returned as a string in the result dictionary. |
| |
| Returns: |
| A dictionary containing the command executed, stdout, stderr, and a |
| list of output files generated. |
| """ |
| if not files: |
| raise ValueError("At least one input file must be provided for the chop operation.") |
| for file_path in files: |
| if not file_path.exists(): |
| raise FileNotFoundError(f"Input file not found: {file_path}") |
|
|
| cmd = ["bedops"] |
|
|
| if chrom: |
| cmd.extend(["--chrom", chrom]) |
| if ec: |
| cmd.append("--ec") |
| if header: |
| cmd.append("--header") |
| if range_str: |
| cmd.extend(["--range", range_str]) |
|
|
| cmd.extend(["--chop", str(bp)]) |
| if stagger is not None: |
| cmd.extend(["--stagger", str(stagger)]) |
| if exclusive_chop: |
| cmd.append("-x") |
|
|
| cmd.extend([str(p) for p in files]) |
|
|
| try: |
| if output_file: |
| with open(output_file, "w") as f: |
| result = subprocess.run( |
| cmd, check=True, text=True, stdout=f, stderr=subprocess.PIPE |
| ) |
| stdout_capture = "" |
| output_files_list = [str(output_file)] |
| else: |
| result = subprocess.run( |
| cmd, check=True, text=True, capture_output=True |
| ) |
| stdout_capture = result.stdout |
| output_files_list = [] |
|
|
| return { |
| "command_executed": " ".join(cmd), |
| "stdout": stdout_capture, |
| "stderr": result.stderr, |
| "output_files": output_files_list, |
| } |
| except subprocess.CalledProcessError as e: |
| raise RuntimeError( |
| f"bedops chop failed with exit code {e.returncode}\n" |
| f"Stderr: {e.stderr}\n" |
| f"Stdout: {e.stdout}\n" |
| f"Command: {' '.join(cmd)}" |
| ) from e |
|
|
| if __name__ == "__main__": |
| mcp.run(transport="stdio") |
|
|
