Plugin and API reference¶

class CWL2WebGLOptions(BaseModel):
    """Options accepted by the CWL to WebGL plugin."""

    model_config = ConfigDict(extra="forbid")

    output: Path = Field(default=Path("workflow.html"), description="Output HTML file")
    overwrite: bool = Field(default=False, description="Replace an existing output file")

class Projection:
    """Per-invocation traversal state, containing references to the existing DOM."""

    def __init__(self, context: TranspilerContext) -> None:
        self.context = context
        self.views: dict[str, dict[str, Any]] = {}
        self.resolved: dict[str, TranspilerContext] = {}
        self.active: set[str] = set()

    def run(
        self, step: WorkflowStep, context: TranspilerContext
    ) -> tuple[Process, TranspilerContext]:
        """Resolve a step process and its owning context.

        Raises:
            PluginExecutionError: If the external resolver fails.
            PluginFailureError: If the reference is ambiguous or unresolved.
        """
        if not isinstance(step.run, str):
            return step.run, context
        process = find_process(context, step.run)
        if process is not None:
            return process, context
        # cwl-utils normally supplies absolute URIs; normalized DOMs may retain fragments.
        location = urljoin(str(context.source), step.run)
        process = find_process(context, location)
        if process is not None:
            return process, context
        if location not in self.resolved:
            try:
                self.resolved[location] = context.resolver.resolve(location)
            except Exception as exc:
                raise PluginExecutionError(
                    f"Cannot resolve run {step.run!r} for {step.id}"
                ) from exc
        other = self.resolved[location]
        return self._resolved_process(other, location), other

    def _resolved_process(self, other: TranspilerContext, location: str) -> Process:
        """Select the referenced process from a resolved document."""
        process = find_process(other, location)
        if process is None and other.process_id:
            process = find_process(other, other.process_id)
        if process is None and not urldefrag(location)[1] and len(other.document) == 1:
            process = next(iter(other.processes))
        if process is None:
            raise PluginFailureError(f"Resolved source does not identify run {location!r}")
        return process

    def view(self, workflow: Workflow, context: TranspilerContext) -> str:
        """Build a workflow view, rejecting recursive workflow references."""
        key = str(context.source) + "|" + workflow.id
        if key in self.active:
            raise PluginFailureError(f"Recursive workflow reference: {workflow.id}")
        if key in self.views:
            return key
        self.active.add(key)
        sources: dict[str, tuple[str, str]] = {}
        nodes = self._nodes(workflow, context, sources)
        edges = self._edges(workflow, sources)
        self.views[key] = {
            "id": workflow.id,
            "label": workflow.label or short(workflow.id),
            "details": details(workflow, ("id", "label", "doc", "requirements", "hints")),
            "nodes": nodes,
            "edges": edges,
        }
        self.active.remove(key)
        return key

    def _nodes(
        self,
        workflow: Any,
        context: TranspilerContext,
        sources: dict[str, tuple[str, str]],
    ) -> list[dict[str, Any]]:
        """Build presentation nodes and register their source ports in place."""
        nodes: list[dict[str, Any]] = []

        def node(identifier: str, kind: str, obj: Any, info: dict[str, Any]) -> dict[str, Any]:
            result = {
                "id": node_id("step" if kind == "workflow" else kind, identifier),
                "cwlId": identifier,
                "kind": kind,
                "label": getattr(obj, "label", None) or short(identifier),
                "details": info,
            }
            if any(n["id"] == result["id"] for n in nodes):
                raise PluginFailureError(f"Duplicate node ID: {identifier}")
            nodes.append(result)
            return result

        for port in workflow.inputs:
            node(port.id, "input", port, details(port, PORT_FIELDS))
            sources[port.id] = (node_id("input", port.id), port.id)
        for step in workflow.steps:
            process, owner = self.run(step, context)
            info = details(step, STEP_FIELDS)
            info["run"] = process.id
            info["bindings"] = [details(port, BINDING_FIELDS) for port in step.in_]
            info["inputs"] = [details(port, PORT_FIELDS) for port in process.inputs]
            info["outputs"] = [details(port, PORT_FIELDS) for port in process.outputs]
            info["process"] = details(
                process,
                (
                    "class_",
                    "label",
                    "doc",
                    "requirements",
                    "hints",
                    "baseCommand",
                    "arguments",
                    "expression",
                ),
            )
            result = node(
                step.id,
                "workflow" if isinstance(process, Workflow) else "step",
                step,
                info,
            )
            if isinstance(process, Workflow):
                result["child"] = self.view(process, owner)
            self._register_outputs(step, sources)
        for port in workflow.outputs:
            node(
                port.id,
                "output",
                port,
                details(port, (*PORT_FIELDS, "outputSource", "linkMerge", "pickValue")),
            )

        return nodes

    @staticmethod
    def _register_outputs(step: WorkflowStep, sources: dict[str, tuple[str, str]]) -> None:
        """Register step outputs, rejecting duplicate source ports."""
        for output in step.out:
            identifier = output if isinstance(output, str) else output.id
            # cwl-loader's normalized DOM has out=["result"], source="step/result".
            if "/" not in identifier and "#" not in identifier:
                identifier = step.id + "/" + identifier
            if identifier in sources:
                raise PluginFailureError(f"Duplicate source port: {identifier}")
            sources[identifier] = (node_id("step", step.id), identifier)

    @staticmethod
    def _edge(
        reference: str,
        target: str,
        port: str,
        binding: dict[str, Any],
        sources: dict[str, tuple[str, str]],
        workflow_id: str,
    ) -> dict[str, Any]:
        """Create an edge after checking that its source exists."""
        if reference not in sources:
            raise PluginFailureError(f"Unknown source {reference!r} in workflow {workflow_id}")
        source, source_port = sources[reference]
        return {
            "source": source,
            "target": target,
            "sourcePort": source_port,
            "targetPort": port,
            "binding": binding,
        }

    def _edges(self, workflow: Any, sources: dict[str, tuple[str, str]]) -> list[dict[str, Any]]:
        """Connect workflow ports to their validated source nodes."""
        edges: list[dict[str, Any]] = []

        for step in workflow.steps:
            for port in step.in_:
                edges.extend(
                    self._edge(
                        reference,
                        node_id("step", step.id),
                        port.id,
                        details(port, BINDING_FIELDS),
                        sources,
                        workflow.id,
                    )
                    for reference in many(port.source)
                )
        for port in workflow.outputs:
            edges.extend(
                self._edge(
                    reference,
                    node_id("output", port.id),
                    port.id,
                    details(port, ("linkMerge", "pickValue")),
                    sources,
                    workflow.id,
                )
                for reference in many(port.outputSource)
            )
        return edges

    def build(self, workflow_id: str | None = None) -> dict[str, Any]:
        requested = workflow_id or self.context.process_id
        if requested:
            selected = find_process(self.context, requested)
            if selected is None:
                # Convenient short selection is allowed only when unique, never for run resolution.
                matches = [
                    p for p in self.context.processes if short(p.id) == requested.lstrip("#")
                ]
                selected = matches[0] if len(matches) == 1 else None
            if selected is None or not isinstance(selected, Workflow):
                raise PluginFailureError(f"Select an existing Workflow ID; got {requested!r}")
            workflows = [selected]
        else:
            workflows = [
                process for process in self.context.processes if isinstance(process, Workflow)
            ]
        if not workflows:
            raise PluginFailureError("The source contains no Workflow processes")
        roots = [self.view(workflow, self.context) for workflow in workflows]
        return {"version": 1, "roots": roots, "views": self.views}