/share/MONSDA/configs
+ os.path.abspath(
+ os.path.join(install_share, "MONSDA", "configs", TEMPLATE_FILE)
+ ),
+ # generic venv/conda prefix share path
+ os.path.abspath(
+ os.path.join(sys.prefix, "share", "MONSDA", "configs", TEMPLATE_FILE)
+ ),
+ # last-resort: run directory configs
+ os.path.abspath(os.path.join(os.getcwd(), "configs", TEMPLATE_FILE)),
+ ]
+
+ # de-duplicate while preserving order and dropping empty values
+ deduped: List[str] = []
+ for c in candidates:
+ c = (c or "").strip()
+ if c and c not in deduped:
+ deduped.append(c)
+ return deduped
+
+
+def resolve_template_path() -> str:
+ for path in _template_candidates():
+ if os.path.isfile(path):
+ return path
+ tried = "\n - ".join(_template_candidates())
+ raise FileNotFoundError(
+ "Could not find template_base_commented.json. Tried:\n - " + tried
+ )
def load_template() -> Dict[str, Any]:
- with open(TEMPLATE_PATH, "r") as f:
- return json.load(f)
+ return load_configfile(resolve_template_path())
-def strip_comments(d):
+def strip_comments(d: Any) -> Any:
if isinstance(d, dict):
return {k: strip_comments(v) for k, v in d.items() if k != "comment"}
- elif isinstance(d, list):
+ if isinstance(d, list):
return [strip_comments(x) for x in d]
+ return d
+
+
+def set_in_path(root: Dict[str, Any], path: List[str], value: Any) -> None:
+ node = root
+ for key in path[:-1]:
+ if key not in node or not isinstance(node[key], dict):
+ node[key] = {}
+ node = node[key]
+ node[path[-1]] = value
+
+
+def get_condition_paths_from_settings(settings: Dict[str, Any]) -> List[List[str]]:
+ out: List[List[str]] = []
+
+ def _walk(node: Any, path: List[str]) -> None:
+ if not isinstance(node, dict):
+ return
+ if "SAMPLES" in node and isinstance(node.get("SAMPLES"), list):
+ out.append(path.copy())
+ return
+ for k, v in node.items():
+ if isinstance(v, dict):
+ _walk(v, path + [k])
+
+ _walk(settings, [])
+ return out
+
+
+def build_workflow_block(
+ workflow_name: str,
+ workflow_template: Dict[str, Any],
+ condition_paths: List[List[str]],
+ selected_tools: List[str],
+) -> Dict[str, Any]:
+ block: Dict[str, Any] = {}
+
+ all_tools = workflow_template.get("TOOLS", {})
+ if not selected_tools:
+ selected_tools = list(all_tools.keys())
+
+ if all_tools:
+ block["TOOLS"] = {k: all_tools[k] for k in selected_tools if k in all_tools}
+
+ # carry workflow-level defaults if present
+ for passthrough in ["FEATURES", "CUTOFFS", "COMPARABLE", "EXCLUDE"]:
+ if passthrough in workflow_template:
+ block[passthrough] = copy.deepcopy(workflow_template[passthrough])
+
+ # populate per-condition tool settings like CLI configurator does
+ for cond_path in condition_paths:
+ for tool in selected_tools:
+ tool_def = workflow_template.get(tool)
+ if not isinstance(tool_def, dict):
+ continue
+ tool_def_no_comment = strip_comments(tool_def)
+ if not tool_def_no_comment:
+ continue
+ set_in_path(block, cond_path + [tool], copy.deepcopy(tool_def_no_comment))
+
+ return block
+
+
+def build_config(req: BuildConfigRequest) -> Dict[str, Any]:
+ template = strip_comments(load_template())
+
+ if req.settings is not None:
+ settings = req.settings
+ elif req.samplesheet_path:
+ samplesheet_path = os.path.abspath(req.samplesheet_path)
+ if not os.path.isfile(samplesheet_path):
+ raise HTTPException(
+ status_code=400,
+ detail=f"Samplesheet does not exist: {samplesheet_path}",
+ )
+ settings = samplesheet_to_settings(samplesheet_path)
else:
- return d
+ raise HTTPException(
+ status_code=400,
+ detail="Provide either settings JSON or samplesheet_path.",
+ )
- config_name: str
- config: Dict[str, Any]
- output_dir: str
+ condition_paths = get_condition_paths_from_settings(settings)
+ if not condition_paths:
+ raise HTTPException(
+ status_code=400,
+ detail="No condition leaves with SAMPLES found in SETTINGS.",
+ )
+
+ workflows = req.workflows or []
+ if not workflows:
+ raise HTTPException(
+ status_code=400,
+ detail="At least one workflow must be selected.",
+ )
+
+ invalid = [w for w in workflows if w not in template or w in NONE_WORKFLOW_KEYS]
+ if invalid:
+ raise HTTPException(
+ status_code=400,
+ detail=f"Unknown workflow(s): {', '.join(invalid)}",
+ )
+
+ final_config: Dict[str, Any] = {
+ "WORKFLOWS": ", ".join(workflows),
+ "BINS": template.get("BINS", ""),
+ "MAXTHREADS": str(req.maxthreads),
+ "VERSION": __version__,
+ "SETTINGS": settings,
+ }
+
+ for wf in workflows:
+ wf_template = template.get(wf, {})
+ final_config[wf] = build_workflow_block(
+ wf,
+ wf_template,
+ condition_paths,
+ req.tools.get(wf, []),
+ )
+
+ return final_config
@app.get("/template", response_model=Dict[str, Any])
-def get_template():
- """Get the config template (with comments)."""
- return load_template()
+def get_template() -> Dict[str, Any]:
+ try:
+ return load_template()
+ except Exception as e:
+ raise HTTPException(status_code=500, detail=f"Failed to load template: {e}")
@app.get("/template/fields", response_model=Dict[str, Any])
-def get_template_fields():
- """Get the config template (without comments)."""
- return strip_comments(load_template())
+def get_template_fields() -> Dict[str, Any]:
+ try:
+ return strip_comments(load_template())
+ except Exception as e:
+ raise HTTPException(
+ status_code=500, detail=f"Failed to load template fields: {e}"
+ )
+
+
+def _normalize_path(path: str) -> str:
+ if path:
+ return os.path.abspath(os.path.expanduser(path.strip()))
+ return os.getcwd()
+
+
+@app.get("/fs/roots", response_model=Dict[str, Any])
+def fs_roots() -> Dict[str, Any]:
+ cwd = os.getcwd()
+ home = os.path.expanduser("~")
+ roots = []
+ for p in [cwd, home, os.path.sep]:
+ if p and p not in roots and os.path.isdir(p):
+ roots.append(p)
+ return {"roots": roots}
+
+
+@app.get("/fs/list", response_model=Dict[str, Any])
+def fs_list(path: str = "", mode: str = "dirs") -> Dict[str, Any]:
+ mode = (mode or "dirs").lower()
+ if mode not in {"dirs", "all", "samplesheets"}:
+ raise HTTPException(
+ status_code=400, detail="mode must be dirs, all, or samplesheets"
+ )
+
+ current = _normalize_path(path)
+ if not os.path.isdir(current):
+ raise HTTPException(status_code=400, detail=f"Not a directory: {current}")
+
+ try:
+ entries = list(os.scandir(current))
+ except PermissionError:
+ raise HTTPException(status_code=403, detail=f"Permission denied: {current}")
+
+ dirs = []
+ files = []
+ for e in sorted(entries, key=lambda x: x.name.lower()):
+ if e.name in {".", ".."}:
+ continue
+ if e.is_dir(follow_symlinks=True):
+ dirs.append({"name": e.name, "path": e.path})
+ continue
+ if not e.is_file(follow_symlinks=True):
+ continue
+ if mode in {"all", "samplesheets"}:
+ files.append({"name": e.name, "path": e.path})
+
+ parent = (
+ os.path.dirname(current.rstrip(os.sep))
+ if current != os.path.sep
+ else os.path.sep
+ )
+ return {
+ "current": current,
+ "parent": parent,
+ "dirs": dirs,
+ "files": files,
+ "mode": mode,
+ }
+
+
+@app.post("/samplesheet/parse", response_model=Dict[str, Any])
+def parse_samplesheet(req: SamplesheetRequest) -> Dict[str, Any]:
+ path = os.path.abspath(req.samplesheet_path.strip())
+ if not os.path.isfile(path):
+ raise HTTPException(status_code=400, detail=f"Samplesheet not found: {path}")
+ settings = samplesheet_to_settings(path)
+ return {
+ "samplesheet": path,
+ "settings": settings,
+ "conditions": [
+ "/".join(p) for p in get_condition_paths_from_settings(settings)
+ ],
+ }
+
+@app.post("/config/preview", response_model=Dict[str, Any])
+def preview_config(req: BuildConfigRequest) -> Dict[str, Any]:
+ return {"config": build_config(req)}
-@app.post("/generate_config")
-def generate_config(req: ConfigRequest):
- """Generate a config file from user input. User specifies output_dir."""
+
+@app.post("/config/save", response_model=Dict[str, Any])
+def save_config(req: SaveConfigRequest) -> Dict[str, Any]:
config_name = req.config_name.strip()
- output_dir = os.path.abspath(req.output_dir.strip())
if not config_name or any(c in config_name for c in "/\\"):
raise HTTPException(status_code=400, detail="Invalid config name.")
- if not output_dir or not os.path.isdir(output_dir):
- raise HTTPException(status_code=400, detail="Invalid output directory.")
- config_path = os.path.join(output_dir, f"config_{config_name}.json")
- with open(config_path, "w") as f:
- json.dump(req.config, f, indent=4)
- return {"message": "Config generated.", "path": config_path}
-
-
-# Download config by full path (for security, restrict to files under allowed parent dir)
-@app.get("/download_config/")
-def download_config(config_path: str):
- config_path = os.path.abspath(config_path)
- if not os.path.exists(config_path):
- raise HTTPException(status_code=404, detail="Config not found.")
- if not config_path.endswith(".json"):
- raise HTTPException(status_code=400, detail="Only .json files allowed.")
- return FileResponse(config_path, filename=os.path.basename(config_path))
-
-
-class DirRequest(BaseModel):
- config: Dict[str, Any]
- project_dir: str
+ output_dir = os.path.abspath(req.output_dir.strip())
+ if not os.path.isdir(output_dir):
+ raise HTTPException(status_code=400, detail="Output directory does not exist.")
-def safe_makedirs(path):
- os.makedirs(path, exist_ok=True)
+ path = os.path.join(output_dir, f"config_{config_name}.json")
+ with open(path, "w") as fh:
+ json.dump(req.config, fh, indent=4)
+ return {"message": "Config written.", "path": path}
-def create_project_structure(config: Dict[str, Any], project_dir: str):
- # Example: create folders for workflows, logs, counts, etc.
- safe_makedirs(project_dir)
- for wf in config.get("WORKFLOWS", "").split(","):
- wf = wf.strip()
- if wf:
- safe_makedirs(os.path.join(project_dir, wf))
- safe_makedirs(os.path.join(project_dir, "LOGS"))
- safe_makedirs(os.path.join(project_dir, "COUNTS"))
- # Add more as needed
+@app.post("/project/create", response_model=Dict[str, Any])
+def create_project(req: ProjectRequest) -> Dict[str, Any]:
+ base_dir = os.path.abspath(req.project_dir.strip())
+ project_name = req.project_name.strip() or "monsda"
-@app.post("/generate_project_dir")
-def generate_project_dir(req: DirRequest):
- project_dir = os.path.abspath(req.project_dir)
- if not project_dir.startswith(os.getcwd()):
- raise HTTPException(
- status_code=400, detail="Project dir must be inside working directory."
- )
- if os.path.exists(project_dir) and os.listdir(project_dir):
- raise HTTPException(
- status_code=400, detail="Project dir already exists and is not empty."
- )
- create_project_structure(req.config, project_dir)
- return {"message": "Project directory structure created.", "path": project_dir}
+ # Project name becomes a subdirectory under the chosen path
+ project_dir = os.path.join(base_dir, project_name)
+ os.makedirs(project_dir, exist_ok=True)
+ # FASTQ directly under project dir (no extra project_name level)
+ fastq_dir = os.path.join(project_dir, "FASTQ")
+ gen_dir = os.path.join(project_dir, "GENOMES")
+ os.makedirs(fastq_dir, exist_ok=True)
+ os.makedirs(gen_dir, exist_ok=True)
+
+ settings = req.settings or (req.config.get("SETTINGS") if req.config else None)
+ linked_files: List[str] = []
+ warnings: List[str] = []
+
+ # Build lookup: condition_path -> ConditionFiles
+ cond_lookup: Dict[str, ConditionFiles] = {}
+ for cf in req.condition_files:
+ cond_lookup[cf.condition] = cf
+
+ if settings and isinstance(settings, dict):
+ condition_paths = get_condition_paths_from_settings(settings)
+ for cond_path in condition_paths:
+ cond_key = "/".join(cond_path)
+ cond_info = cond_lookup.get(cond_key)
+
+ # Create condition subdirectories under FASTQ
+ cond_dir = os.path.join(fastq_dir, *cond_path)
+ os.makedirs(cond_dir, exist_ok=True)
+
+ # Walk to the leaf node to find SAMPLES
+ node = settings
+ for key in cond_path:
+ node = node.get(key, {})
+
+ samples = node.get("SAMPLES", [])
+
+ # Link FASTQ files: prefer explicit file list, fall back to directory+sample matching
+ if cond_info and cond_info.fastq_files:
+ # Explicit file selection mode - link exactly the files the user chose
+ for fpath in cond_info.fastq_files:
+ src_file = os.path.abspath(fpath)
+ if not os.path.isfile(src_file):
+ warnings.append(f"FASTQ file not found: {fpath}")
+ continue
+ dst = os.path.join(cond_dir, os.path.basename(src_file))
+ if not os.path.exists(dst):
+ os.symlink(os.path.realpath(src_file), dst)
+ linked_files.append(dst)
+ # Handle paired-end mate
+ if cond_info.sequencing == "paired":
+ basename = os.path.basename(src_file)
+ if "_R1" in basename:
+ mate_name = basename.replace("_R1", "_R2")
+ elif "_1." in basename:
+ mate_name = basename.replace("_1.", "_2.")
+ else:
+ mate_name = None
+ if mate_name:
+ mate_src = os.path.join(
+ os.path.dirname(src_file), mate_name
+ )
+ mate_dst = os.path.join(cond_dir, mate_name)
+ if os.path.isfile(mate_src) and not os.path.exists(
+ mate_dst
+ ):
+ os.symlink(os.path.realpath(mate_src), mate_dst)
+ linked_files.append(mate_dst)
+
+ elif cond_info and cond_info.fastq_dir and isinstance(samples, list):
+ # Directory mode - find files matching sample names from samplesheet
+ src_dir = os.path.abspath(cond_info.fastq_dir)
+ if os.path.isdir(src_dir):
+ for sample in samples:
+ # Find matching files (sample name may or may not have extension)
+ matched = _find_fastq_files(src_dir, sample)
+ if not matched:
+ warnings.append(
+ f"No FASTQ file found for sample '{sample}' in {src_dir}"
+ )
+ continue
+ for src_file in matched:
+ dst = os.path.join(cond_dir, os.path.basename(src_file))
+ if not os.path.exists(dst):
+ os.symlink(os.path.realpath(src_file), dst)
+ linked_files.append(dst)
+ # Handle paired-end mate
+ if cond_info.sequencing == "paired":
+ basename = os.path.basename(src_file)
+ if "_R1" in basename:
+ mate_name = basename.replace("_R1", "_R2")
+ elif "_1." in basename:
+ mate_name = basename.replace("_1.", "_2.")
+ elif "_R2" in basename:
+ mate_name = basename.replace("_R2", "_R1")
+ elif "_2." in basename:
+ mate_name = basename.replace("_2.", "_1.")
+ else:
+ mate_name = None
+ if mate_name:
+ mate_src = os.path.join(src_dir, mate_name)
+ mate_dst = os.path.join(cond_dir, mate_name)
+ if os.path.isfile(mate_src) and not os.path.exists(
+ mate_dst
+ ):
+ os.symlink(os.path.realpath(mate_src), mate_dst)
+ linked_files.append(mate_dst)
+ else:
+ warnings.append(
+ f"FASTQ directory not found for condition '{cond_key}': {src_dir}"
+ )
+
+ # Link genome files: REFERENCE, DECOY, GTF, GFF into GENOMES/
+ ref_path = (cond_info.reference if cond_info else "") or node.get(
+ "REFERENCE", ""
+ )
+ if ref_path and os.path.isfile(ref_path):
+ dst = os.path.join(gen_dir, os.path.basename(ref_path))
+ if not os.path.exists(dst):
+ os.symlink(os.path.realpath(ref_path), dst)
+ linked_files.append(dst)
+ # Update settings path to relative
+ rel = os.path.relpath(dst, start=project_dir)
+ node["REFERENCE"] = rel
+ elif ref_path:
+ warnings.append(f"REFERENCE not found: {ref_path}")
+
+ decoy_path = (cond_info.decoy if cond_info else "") or node.get("DECOY", "")
+ if decoy_path and os.path.isfile(decoy_path):
+ dst = os.path.join(gen_dir, os.path.basename(decoy_path))
+ if not os.path.exists(dst):
+ os.symlink(os.path.realpath(decoy_path), dst)
+ linked_files.append(dst)
+ rel = os.path.relpath(dst, start=project_dir)
+ node["DECOY"] = rel
+
+ gtf_path = (cond_info.gtf if cond_info else "") or ""
+ anno = node.get("ANNOTATION", {})
+ if isinstance(anno, dict):
+ gtf_path = gtf_path or anno.get("GTF", "")
+ if gtf_path and os.path.isfile(gtf_path):
+ dst = os.path.join(gen_dir, os.path.basename(gtf_path))
+ if not os.path.exists(dst):
+ os.symlink(os.path.realpath(gtf_path), dst)
+ linked_files.append(dst)
+ rel = os.path.relpath(dst, start=project_dir)
+ if isinstance(anno, dict):
+ anno["GTF"] = rel
+ node["ANNOTATION"] = anno
+ elif gtf_path:
+ warnings.append(f"GTF not found: {gtf_path}")
+
+ gff_path = (cond_info.gff if cond_info else "") or ""
+ if isinstance(anno, dict):
+ gff_path = gff_path or anno.get("GFF", "")
+ if gff_path and os.path.isfile(gff_path):
+ dst = os.path.join(gen_dir, os.path.basename(gff_path))
+ if not os.path.exists(dst):
+ os.symlink(os.path.realpath(gff_path), dst)
+ linked_files.append(dst)
+ rel = os.path.relpath(dst, start=project_dir)
+ if isinstance(anno, dict):
+ anno["GFF"] = rel
+ node["ANNOTATION"] = anno
+ elif gff_path:
+ warnings.append(f"GFF not found: {gff_path}")
+
+ # Write config if provided (with updated relative paths in SETTINGS)
+ config_path = ""
+ if req.config:
+ if settings:
+ req.config["SETTINGS"] = settings
+ config_file = f"config_{project_name}.json"
+ config_path = os.path.join(project_dir, config_file)
+ with open(config_path, "w") as fh:
+ json.dump(req.config, fh, indent=4)
+
+ return {
+ "message": "Project created.",
+ "path": project_dir,
+ "config_path": config_path,
+ "linked_files": len(linked_files),
+ "warnings": warnings,
+ }
-from fastapi.responses import HTMLResponse
+
+def _find_fastq_files(src_dir: str, sample_name: str) -> List[str]:
+ """Find FASTQ files in src_dir matching a sample name.
+
+ Matches: exact filename, or sample_name*.fastq.gz / .fq.gz / .fastq / .fq
+ Only returns R1 (or unpaired) to avoid double-linking.
+ """
+ import glob as _glob
+
+ candidates: List[str] = []
+ # Try exact match first
+ exact = os.path.join(src_dir, sample_name)
+ if os.path.isfile(exact):
+ return [exact]
+
+ # Try common FASTQ extensions
+ for ext in [".fastq.gz", ".fq.gz", ".fastq", ".fq"]:
+ path = os.path.join(src_dir, sample_name + ext)
+ if os.path.isfile(path):
+ candidates.append(path)
+ # Try with _R1 suffix
+ path_r1 = os.path.join(src_dir, sample_name + "_R1" + ext)
+ if os.path.isfile(path_r1):
+ candidates.append(path_r1)
+ # Try with _1 suffix
+ path_1 = os.path.join(src_dir, sample_name + "_1" + ext)
+ if os.path.isfile(path_1):
+ candidates.append(path_1)
+
+ if candidates:
+ return candidates
+
+ # Glob fallback: any file starting with sample_name
+ pattern = os.path.join(src_dir, sample_name + "*")
+ matches = _glob.glob(pattern)
+ # Filter to only fastq-like files and only R1/unpaired
+ for m in sorted(matches):
+ if os.path.isfile(m):
+ lower = m.lower()
+ if any(lower.endswith(e) for e in [".fastq.gz", ".fq.gz", ".fastq", ".fq"]):
+ # Skip R2/_2 to avoid double-linking
+ base = os.path.basename(m)
+ if "_R2" in base or "_2." in base:
+ continue
+ candidates.append(m)
+
+ return candidates
@app.get("/", response_class=HTMLResponse)
-def root():
+def root() -> str:
return """
-
+
-
- MONSDA Configurator
-
+
+ MONSDA Web Configurator
+
+
+
- MONSDA Configurator
-
-
Step 1: Edit Configuration
-
-
-
-
-
-
Step 2: Generate Config File
-
-
-
-
-
-
-
-
-
Step 3: Create Project Directory
-
-
-
-
-
-
+ }
+ walkSettings(settings, []);
+
+ if (!conditions.length) {
+ setStatus('projectStatus', 'No conditions with SAMPLES found in settings.', false);
+ return;
+ }
+
+ const wizard = document.getElementById('conditionWizard');
+ wizard.innerHTML = `${conditions.length} condition(s) found. For each, select FASTQ files by directory (auto-matched by sample name) or pick individual files.
`;
+
+ conditions.forEach((c, idx) => {
+ const id = `cond_${idx}`;
+ const hasSamples = c.samples && c.samples.length > 0;
+ const samplesStr = hasSamples
+ ? c.samples.slice(0, 5).join(', ') + (c.samples.length > 5 ? ` (+${c.samples.length - 5} more)` : '')
+ : '(no samples defined)';
+ const card = document.createElement('div');
+ card.className = 'cond-card';
+ card.dataset.condition = c.path;
+ card.innerHTML = `
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+