Python API reference

def to_puml(
    cwl_document: Mapping[str, Process],
    diagram_type: DiagramType,
    output_stream: TextIO,
    workflow_id: str = "main",
) -> None:
    """Convert a CWL, given its document model, to a PlantUML diagram.

    Args:
        cwl_document: The Processes object model representing the CWL document.
        diagram_type: The PlantUML diagram type to render.
        output_stream: The output stream for the PlantUML diagram.
        workflow_id: The identifier of the workflow to render.
    """
    template = _jinja_environment.get_template(f"{diagram_type.name.lower()}.puml")

    output_stream.write(
        template.render(
            version=_get_version(),
            timestamp=datetime.fromtimestamp(time.time()).isoformat(timespec="milliseconds"),
            workflows=cwl_document.values(),
            workflow_id=workflow_id,
            index=cwl_document,
        )
    )

def get_ogc_inputs(process: Process) -> str:
    """Serialize the process inputs as indented OGC API Processes JSON."""
    return _dump_json(BaseCWLtypes2OGCConverter(process).get_inputs())

def get_ogc_outputs(process: Process) -> str:
    """Serialize the process outputs as indented OGC API Processes JSON."""
    return _dump_json(BaseCWLtypes2OGCConverter(process).get_outputs())

@transpiler_plugin(
    name="cwl2puml",
    description="Converts a CWL, given its document model, to PlantUML diagram(s).",
    options_model=Cwl2PumlOptions,
)
def cwl2puml(context: TranspilerContext, options: Cwl2PumlOptions) -> None:
    """Converts a CWL, given its document model, to PlantUML diagram(s)."""

    try:
        for workflow in context.get_processes_by_type(
            Workflow, [context.process_id] if context.process_id else None
        ):
            for diagram_type in options.diagrams:
                logger.info(f"Converting to {diagram_type.name.lower()} PlantUML diagram...")
                out = StringIO()
                to_puml(
                    cwl_document=context.document,
                    workflow_id=workflow.id,
                    diagram_type=diagram_type,
                    output_stream=out,
                )

                target = Path(options.output, workflow.id, f"{diagram_type.name.lower()}.puml")
                target.parent.mkdir(parents=True, exist_ok=True)

                clear_output = out.getvalue()
                logger.info(
                    f"Saving PlantUML {diagram_type.name.lower()} diagram to {target.absolute()}..."
                )

                with target.open("w") as f:
                    f.write(clear_output)

                logger.success(
                    f"PlantUML {diagram_type.name.lower()} diagram successfully dumped to {target.absolute()}!"
                )

                if options.convert_image:
                    image_format: str = options.image_format.name.lower()

                    logger.info(
                        f"Converting PlantUML {diagram_type.name.lower()} diagram to '{image_format}'..."
                    )

                    encoded = deflate_and_encode(clear_output)
                    diagram_url = f"https://{options.puml_server}/plantuml/{image_format}/{encoded}"
                    response = requests.get(diagram_url, timeout=30)
                    if HTTPStatus.OK.value == response.status_code:
                        target = Path(
                            target.parent,
                            f"{diagram_type.name.lower()}.{image_format}",
                        )
                        logger.info(
                            f"Saving PlantUML {diagram_type.name.lower()} {image_format} image to {target.absolute()}..."
                        )

                        with target.open("wb") as f:
                            f.write(response.content)

                        logger.success(
                            f"PlantUML {diagram_type.name.lower()} {image_format} image successfully dumped to {target.absolute()}!"
                        )
                    else:
                        raise PluginExecutionError(
                            f"Impossible to render {diagram_type.name.lower()} {image_format} image",
                            f"{options.puml_server} server replied: {response.status_code} {response.reason}",
                            f"Deflated and encoded PlantUML Diagram: {encoded}",
                        )
    except Exception as e:
        raise PluginExecutionError(
            f"An error occurred when serializing to {options.output.absolute()}, see nested exception"
        ) from e