QCSC Prefect API Reference

This page is generated from Python docstrings in the core qcsc-prefect packages. Native Qiskit primitives are documented separately in Native Qiskit API.

Core Models

qcsc_prefect_core.models.execution_profile.ExecutionProfile dataclass

ExecutionProfile(
    command_key: str,
    num_nodes: int,
    mpiprocs: int | None = None,
    ompthreads: int | None = None,
    walltime: str | None = None,
    launcher: Launcher = "single",
    mpi_options: list[str] = list(),
    modules: list[str] = list(),
    pre_commands: list[str] = list(),
    environments: dict[str, str] = dict(),
    arguments: list[str] = list(),
)

Minimal execution profile for Miyabi PBS template.

NOTE: - This is intentionally "template-driven" MVP. - Later you can evolve this into common+overrides and canonicalization.

Prefect Blocks

qcsc_prefect_blocks.common.blocks.CommandBlock

Bases: Block

Command definition (what to run). Common for all HPC systems.

qcsc_prefect_blocks.common.blocks.ExecutionProfileBlock

Bases: Block

Execution baseline (how to run). Common for all HPC systems.

qcsc_prefect_blocks.common.blocks.HPCProfileBlock

Bases: Block

Execution target profile for local and supported HPC runtimes.

Block-Based Execution

qcsc_prefect_executor.from_blocks.SubmissionTarget dataclass

SubmissionTarget(
    hpc_target: str, queue_name: str, project: str
)

Execution routing information resolved from Prefect blocks.

Attributes:

Name	Type	Description
hpc_target	str	Runtime target name, such as "local", "miyabi", "fugaku", or "slurm".
queue_name	str	Queue, partition, or resource-group name selected for the execution profile's resource class. Empty for local execution.
project	str	Project, group, or account name selected for the resource class. Empty for local execution and scheduler targets that do not require an account.

qcsc_prefect_executor.from_blocks.resolve_hpc_target async

resolve_hpc_target(*, hpc_profile_block_name: str) -> str

Load an HPCProfileBlock and return its execution target name.

Parameters:

Name	Type	Description	Default
hpc_profile_block_name	str	Prefect block document name for qcsc_prefect_blocks.common.blocks.HPCProfileBlock.	required

Returns:

Type	Description
str	The configured hpc_target value, for example "local",
str	"miyabi", "fugaku", or "slurm".

qcsc_prefect_executor.from_blocks.resolve_submission_target async

resolve_submission_target(
    *,
    hpc_profile_block_name: str,
    execution_profile_block_name: str,
) -> SubmissionTarget

Resolve scheduler routing from block names without submitting a job.

This helper is useful when a flow needs to inspect the target queue or project before it creates scheduler-specific filenames or logs. It loads the HPCProfileBlock and ExecutionProfileBlock and chooses CPU or GPU queue/project fields from the execution profile's resource_class.

Parameters:

Name	Type	Description	Default
hpc_profile_block_name	str	Prefect block document name for target-specific scheduler settings.	required
execution_profile_block_name	str	Prefect block document name for scheduler-independent execution settings.	required

Returns:

Type	Description
SubmissionTarget	Resolved scheduler target, queue/partition/resource group, and
SubmissionTarget	project/account values.

qcsc_prefect_executor.from_blocks.build_scheduler_script_filename

build_scheduler_script_filename(
    script_stem: str, hpc_target: str
) -> str

Build a scheduler-specific script filename from a logical stem.

Existing scheduler suffixes are replaced, while names without a known scheduler suffix receive the target suffix appended. For example, "batch" becomes "batch.pbs" for Miyabi and "batch.slurm" for Slurm; "batch.pbs" becomes "batch.pjm" for Fugaku.

Parameters:

Name	Type	Description	Default
script_stem	str	Logical script name or existing scheduler script filename.	required
hpc_target	str	Scheduler target name.	required

Returns:

Type	Description
str	Script filename with the suffix required by the scheduler target.

Raises:

Type	Description
NotImplementedError	If hpc_target is not supported.

qcsc_prefect_executor.from_blocks.resolve_scheduler_script_filename async

resolve_scheduler_script_filename(
    *, script_stem: str, hpc_profile_block_name: str
) -> str

Resolve scheduler target from blocks and return a matching filename.

Parameters:

Name	Type	Description	Default
script_stem	str	Logical script name or existing scheduler script filename.	required
hpc_profile_block_name	str	Prefect block document name used to determine the scheduler target.	required

Returns:

Type	Description
str	Scheduler-specific script filename.

qcsc_prefect_executor.from_blocks.run_job_from_blocks async

run_job_from_blocks(
    *,
    command_block_name: str,
    execution_profile_block_name: str,
    hpc_profile_block_name: str,
    work_dir: Path,
    script_filename: str | None = None,
    user_args: list[str] | None = None,
    watch_poll_interval: float = 10.0,
    timeout_seconds: float | None = None,
    metrics_artifact_key: str = "hpc-job-metrics",
    fugaku_job_name: str | None = None,
    execution_profile_overrides: dict[str, Any]
    | None = None,
) -> Any

Resolve Prefect blocks and execute a job on the configured target.

This is the main block-driven entrypoint for workflow authors. It loads the command, execution profile, and HPC profile blocks; converts them into the internal runtime models; and dispatches to local execution or the Miyabi, Fugaku, or Slurm executor.

Parameters:

Name	Type	Description	Default
command_block_name	str	Prefect block document name for the command to run.	required
execution_profile_block_name	str	Prefect block document name describing resources, launcher, environment, and default execution behavior.	required
hpc_profile_block_name	str	Prefect block document name describing the execution target and executable mapping, plus scheduler routing fields when applicable.	required
work_dir	Path	Working directory for the process or scheduler job.	required
script_filename	str | None	Logical or scheduler-specific script filename. The suffix is normalized for scheduler targets. It is ignored for local execution and may be omitted.	None
user_args	list[str] | None	Optional extra command-line arguments appended after the command block's default arguments.	None
watch_poll_interval	float	Seconds to wait between scheduler status polls.	10.0
timeout_seconds	float | None	Optional maximum wait time for terminal job status.	None
metrics_artifact_key	str	Prefect artifact key used for job metrics.	'hpc-job-metrics'
fugaku_job_name	str | None	Optional Fugaku PJM job name. When omitted, a safe name is derived from the command name.	None
execution_profile_overrides	dict[str, Any] | None	Optional runtime overrides for selected execution profile fields, such as num_nodes or walltime.	None

Returns:

Type	Description
Any	A target-specific result object: LocalRunResult,
Any	MiyabiRunResult, FugakuRunResult, or SlurmRunResult.

Raises:

Type	Description
ValueError	If the command and execution profile blocks refer to different command names, if a required project/group is missing, if local execution receives modules or pre_commands, or if unsupported execution profile override keys are provided.
KeyError	If the command's executable key is missing from the HPC profile's executable map.
NotImplementedError	If the resolved hpc_target is unsupported.

Bulk HPC Execution

qcsc_prefect_executor.bulk.BulkJobStatus

Bases: str, Enum

Normalized lifecycle status for one logical bulk job.

is_terminal property

is_terminal: bool

Return whether this status is a terminal state.

is_active property

is_active: bool

Return whether this status represents an accepted scheduler job.

is_submit_candidate property

is_submit_candidate: bool

Return whether this status may be submitted in a future refill.

qcsc_prefect_executor.bulk.BulkJobSpec dataclass

BulkJobSpec(
    job_key: str,
    work_dir: Path,
    command_args: dict[str, Any] = dict(),
    wave_id: str | None = None,
    target_id: str | None = None,
    stage_id: str | None = None,
    priority: int = 0,
    expected_outputs: list[Path] = list(),
    max_submit_attempts: int = 5,
    execution_profile_block: str | None = None,
    hpc_profile_block: str | None = None,
)

Desired bulk job registration payload.

job_key is the stable idempotency key and must be unique within one registry.

qcsc_prefect_executor.bulk.BulkJobRecord dataclass

BulkJobRecord(
    job_key: str,
    wave_id: str | None,
    target_id: str | None,
    status: BulkJobStatus,
    work_dir: Path,
    scheduler_job_id: str | None,
    submit_attempts: int,
    monitor_attempts: int,
    command_args: dict[str, Any],
    expected_outputs: list[Path],
    created_at: str,
    updated_at: str,
    submitted_at: str | None,
    started_at: str | None,
    finished_at: str | None,
    last_error: str | None,
    priority: int = 0,
    max_submit_attempts: int = 5,
    stage_id: str | None = None,
    submit_mode: str = "single",
    bulk_group_key: str | None = None,
    bulk_parent_job_id: str | None = None,
    bulk_index: int | None = None,
    scheduler_subjob_id: str | None = None,
    execution_profile_block: str | None = None,
    hpc_profile_block: str | None = None,
)

Persisted state for one logical bulk job.

qcsc_prefect_executor.bulk.SubmittedJob dataclass

SubmittedJob(
    job_key: str,
    scheduler_job_id: str,
    status: BulkJobStatus,
    work_dir: Path,
)

Scheduler identity returned after a bulk job is accepted.

qcsc_prefect_executor.bulk.BulkRunResult dataclass

BulkRunResult(
    total_jobs: int,
    status_counts: dict[str, int],
    succeeded: int,
    failed: int,
    cancelled: int,
    submit_deferred: int,
    unknown: int,
    registry_path: Path,
    failed_jobs: list[str],
)

Summary returned by a future bulk run orchestration API.

qcsc_prefect_executor.bulk.BulkTickResult dataclass

BulkTickResult(
    submitted: list[SubmittedJob],
    monitored: dict[str, BulkJobStatus],
    status_counts: dict[str, int],
    registry_path: Path,
)

Summary returned by one non-blocking bulk runner tick.

qcsc_prefect_executor.bulk.BulkJobRegistry

BulkJobRegistry(path: str | Path)

Persistent SQLite registry for bulk HPC job state.

upsert_jobs

upsert_jobs(jobs: list[BulkJobSpec]) -> None

Register jobs idempotently without resetting existing progress.

qcsc_prefect_executor.bulk.GlobalFugakuBulkRunner dataclass

GlobalFugakuBulkRunner(
    command_block: str,
    execution_profile_block: str,
    hpc_profile_block: str,
    registry_path: Path,
    queue_probe: QueueProbe | None = None,
    max_active_jobs: int = 1000,
    safety_margin: int = 20,
    initial_submit_count: int | None = None,
    max_submit_per_refill: int = 100,
    target_active_jobs: int | None = None,
    no_check_directory: bool = False,
    submit_workers: int = 8,
)

Non-blocking queue-aware runner for staged Fugaku workflows.

The runner uses the existing single-submit path. Native PJM bulk submission remains an explicit experimental mode elsewhere and is not used here. Within one tick, selected pending jobs are submitted concurrently up to submit_workers without increasing the queue-aware batch size.

register_jobs

register_jobs(jobs: list[BulkJobSpec]) -> None

Register logical jobs idempotently and skip completed outputs.

tick async

tick() -> BulkTickResult

Run one monitor/refill cycle without waiting for terminal completion.

all_submitted

all_submitted(stage_id: str) -> bool

Return whether a stage has no pending or deferred logical jobs.

status_counts

status_counts(
    stage_id: str | None = None,
) -> dict[str, int]

Return status counts for all jobs or one stage.

qcsc_prefect_core.queue.QueueCapacity dataclass

QueueCapacity(
    max_active_jobs: int,
    current_active_jobs: int,
    available_slots: int,
    raw_output: str | None = None,
)

Current scheduler queue capacity estimate.

Attributes:

Name	Type	Description
max_active_jobs	int	Maximum active jobs allowed by the scheduler policy or local workflow policy.
current_active_jobs	int	Number of currently active scheduler jobs.
available_slots	int	Number of jobs that may be submitted now.
raw_output	str | None	Optional raw scheduler output used to estimate capacity.

qcsc_prefect_core.queue.QueueAwareSubmitGate

QueueAwareSubmitGate(
    *,
    queue_probe: QueueProbe,
    max_active_jobs: int = 1000,
    safety_margin: int = 20,
    max_submit_per_refill: int = 100,
)

Compute conservative submit allowance from a queue capacity probe.

allowed_submit_count

allowed_submit_count() -> int

Return how many new jobs may be submitted in this refill cycle.

Queue probing failures are treated conservatively and return zero.

qcsc_prefect_executor.bulk.QueueFullError

Bases: SubmitError

Raised when a scheduler rejects submission because queue capacity is full.

qcsc_prefect_executor.bulk.TemporarySubmitError

Bases: SubmitError

Raised when a scheduler submission failure should be retried later.

qcsc_prefect_executor.bulk.DuplicateJobKeyError

Bases: SubmitError

Raised when a bulk job key already refers to a submitted or terminal job.

qcsc_prefect_executor.from_blocks.submit_job_from_blocks async

submit_job_from_blocks(
    *,
    work_dir: Path,
    job_key: str,
    command_block: str | None = None,
    execution_profile_block: str | None = None,
    hpc_profile_block: str | None = None,
    command_args: dict[str, Any] | None = None,
    registry: BulkJobRegistry | None = None,
    command_block_name: str | None = None,
    execution_profile_block_name: str | None = None,
    hpc_profile_block_name: str | None = None,
    fugaku_no_check_directory: bool = False,
) -> SubmittedJob

Submit one block-defined HPC job without waiting for completion.

Queue-full and retryable scheduler failures are recorded as SUBMIT_DEFERRED when a registry is provided, then raised so a future refill loop can stop submitting more jobs in the current cycle. Set fugaku_no_check_directory to opt into pjsub --no-check-directory for Fugaku submissions only.

qcsc_prefect_executor.from_blocks.monitor_jobs_many async

monitor_jobs_many(
    *,
    scheduler_job_ids: list[str],
    hpc_profile_block: str | None = None,
    registry: BulkJobRegistry | None = None,
    hpc_profile_block_name: str | None = None,
) -> dict[str, BulkJobStatus]

Monitor many scheduler jobs with one aggregated scheduler query per target.

qcsc_prefect_executor.from_blocks.run_jobs_from_blocks_bulk async

run_jobs_from_blocks_bulk(
    *,
    jobs: list[BulkJobSpec],
    command_block: str,
    execution_profile_block: str,
    hpc_profile_block: str,
    registry_path: Path,
    queue_probe: QueueProbe | None = None,
    max_active_jobs: int = 1000,
    safety_margin: int = 20,
    max_submit_per_refill: int = 100,
    submit_mode: Literal[
        "single", "native_bulk"
    ] = "single",
    initial_submit_count: int | None = None,
    max_bulk_group_size: int = 100,
    target_active_jobs: int | None = None,
    poll_interval_seconds: int = 60,
    refill_interval_seconds: int = 60,
    stop_on_first_failure: bool = False,
    fugaku_no_check_directory: bool = False,
) -> BulkRunResult

Run many block-defined HPC jobs through one queue-aware bulk loop.

This API submits and monitors scheduler jobs from a shared pending pool. It does not create one Prefect task per scheduler job, and wave identifiers on BulkJobSpec remain registry metadata for downstream workflow readiness checks rather than submit units. The default submit_mode="single" keeps using one scheduler submit per logical job. Fugaku native bulk submission is an explicit opt-in path via submit_mode="native_bulk". Set fugaku_no_check_directory to opt into pjsub --no-check-directory for Fugaku submissions only.

Miyabi API

qcsc_prefect_adapters.miyabi.builder.MiyabiJobRequest dataclass

MiyabiJobRequest(
    queue_name: str, project: str, executable: str
)

Target-specific request fields required to build a Miyabi PBS job.

Attributes:

Name	Type	Description
queue_name	str	Miyabi PBS queue name, for example "regular-c".
project	str	Miyabi group/project name passed to the PBS template.
executable	str	Absolute or scheduler-visible command path to execute.

qcsc_prefect_adapters.miyabi.builder.to_miyabi_template_kwargs

to_miyabi_template_kwargs(
    *, exec_profile: ExecutionProfile, req: MiyabiJobRequest
) -> dict

Build template variables for the Miyabi PBS script.

Parameters:

Name	Type	Description	Default
exec_profile	ExecutionProfile	Scheduler-independent execution profile.	required
req	MiyabiJobRequest	Miyabi-specific scheduler request fields.	required

Returns:

Type	Description
dict	A dictionary that can be passed to the Miyabi Jinja template.

qcsc_prefect_adapters.miyabi.builder.render_script

render_script(
    *,
    work_dir: Path,
    exec_profile: ExecutionProfile,
    req: MiyabiJobRequest,
) -> str

Render a Miyabi job script text from the configured Jinja template.

.. note:: The template file is configured by module constant _TEMPLATE and is expected to be a .j2 file.

Parameters:

Name	Type	Description	Default
work_dir	Path	Working directory injected into the template.	required
exec_profile	ExecutionProfile	Scheduler-independent execution profile.	required
req	MiyabiJobRequest	Miyabi-specific scheduler request fields.	required

Returns:

Type	Description
str	Rendered PBS script text.

qcsc_prefect_adapters.miyabi.builder.write_script_file

write_script_file(
    *, work_dir: Path, filename: str, text: str
) -> Path

Write a rendered job script into the work directory.

.. note:: This function is expected to be called inside qcsc_prefect_executor.miyabi.run.run_miyabi_job. Workflow authors normally do not need to call it directly.

.. note:: The text argument is expected to come from render_script, which renders the .j2 template specified by _TEMPLATE.

Parameters:

Name	Type	Description	Default
work_dir	Path	Base working directory where the script file is created.	required
filename	str	Script file name (for example batch.pbs).	required
text	str	Rendered script text.	required

Returns:

Type	Description
Path	Absolute path to the created job script file.

qcsc_prefect_adapters.miyabi.runtime.SubmitResult dataclass

SubmitResult(job_id: str, raw_output: str)

Submission result returned after PBS accepts a batch script.

Attributes:

Name	Type	Description
job_id	str	PBS job id parsed from qsub stdout.
raw_output	str	Raw, stripped stdout emitted by qsub.

qcsc_prefect_adapters.miyabi.runtime.MiyabiPBSRuntime

Async runtime wrapper for Miyabi PBS scheduler commands.

The runtime maps to the core PBS commands used on Miyabi: qsub for submission, qstat -fH for completed-job status, and qdel for cancellation. Workflow code usually calls qcsc_prefect_executor.miyabi.run.run_miyabi_job or qcsc_prefect_executor.from_blocks.run_job_from_blocks instead.

submit async

submit(
    script_path: Path, *, cwd: Path | None = None
) -> SubmitResult

Submit a PBS script with qsub.

Parameters:

Name	Type	Description	Default
script_path	Path	Path to the PBS script file.	required
cwd	Path | None	Optional working directory for qsub execution.	None

Returns:

Type	Description
SubmitResult	Parsed submission result including job id and raw output.

Raises:

Type	Description
SubmitError	If submission fails or job id cannot be parsed.

wait_final_status async

wait_final_status(
    job_id: str,
    *,
    watch_poll_interval: float = 10.0,
    timeout_seconds: float | None = None,
) -> dict[str, Any]

Wait until PBS reports the job in the finished-job list.

qstat -fH exposes completed jobs on Miyabi. This method polls that view, parses PBS key = value records, and preserves continuation lines in the returned payload.

Parameters:

Name	Type	Description	Default
job_id	str	PBS job id to watch.	required
watch_poll_interval	float	Seconds to wait between qstat calls.	10.0
timeout_seconds	float | None	Optional maximum wait time.	None

Returns:

Type	Description
dict[str, Any]	Parsed final PBS status dictionary.

Raises:

Type	Description
WaitTimeout	If timeout_seconds elapses before final status is observed.
RuntimeError	If an underlying qstat command fails.

cancel async

cancel(job_id: str) -> None

Cancel a PBS job using qdel.

Parameters:

Name	Type	Description	Default
job_id	str	Target PBS job id.	required

Raises:

Type	Description
CancelError	If cancellation fails.

qcsc_prefect_executor.miyabi.run.MiyabiRunResult dataclass

MiyabiRunResult(
    job_id: str,
    exit_status: int,
    job_status: dict[str, Any],
)

Normalized result returned by run_miyabi_job.

Attributes:

Name	Type	Description
job_id	str	PBS job id returned by qsub.
exit_status	int	Integer PBS Exit_status value.
job_status	dict[str, Any]	Parsed final PBS status dictionary from qstat -fH.

qcsc_prefect_executor.miyabi.run.run_miyabi_job async

run_miyabi_job(
    *,
    work_dir: Path,
    script_filename: str,
    exec_profile: ExecutionProfile,
    req: MiyabiJobRequest,
    watch_poll_interval: float = 10.0,
    timeout_seconds: float | None = None,
    metrics_artifact_key: str = "miyabi-job-metrics",
) -> MiyabiRunResult

Execute a Miyabi job end-to-end from runtime models.

.. note:: This function is the high-level executor entrypoint. It internally renders a script, submits it, waits for final status, captures logs, and publishes a metrics artifact.

Parameters:

Name	Type	Description	Default
work_dir	Path	Working directory where scripts and job outputs are written.	required
script_filename	str	Job script filename to create in work_dir.	required
exec_profile	ExecutionProfile	Scheduler-independent execution profile.	required
req	MiyabiJobRequest	Miyabi-specific scheduler request fields.	required
watch_poll_interval	float	Poll interval in seconds for job status checks.	10.0
timeout_seconds	float | None	Optional timeout for waiting final status.	None
metrics_artifact_key	str	Prefect artifact key for job metrics table.	'miyabi-job-metrics'

Returns:

Type	Description
MiyabiRunResult	MiyabiRunResult containing job id, exit status, and final
MiyabiRunResult	scheduler status payload.

qcsc_prefect_executor.miyabi.from_blocks.run_miyabi_job_from_blocks async

run_miyabi_job_from_blocks(
    *,
    command_block_name: str,
    execution_profile_block_name: str,
    hpc_profile_block_name: str,
    work_dir: Path,
    script_filename: str,
    user_args: list[str] | None = None,
    watch_poll_interval: float = 10.0,
    timeout_seconds: float | None = None,
    metrics_artifact_key: str = "miyabi-job-metrics",
) -> MiyabiRunResult

Backward-compatible wrapper around run_job_from_blocks.

Fugaku API

qcsc_prefect_adapters.fugaku.builder.FugakuJobRequest dataclass

FugakuJobRequest(
    queue_name: str,
    project: str,
    executable: str,
    job_name: str = "prefect_job",
    gfscache: str | None = None,
    mpi_options_for_pjm: list[str] | None = None,
    spack_modules: list[str] | None = None,
    pjm_resources: list[str] | None = None,
)

Target-specific request fields required to build a Fugaku PJM job.

Attributes:

Name	Type	Description
queue_name	str	Fugaku resource group name.
project	str	Fugaku group name passed to PJM.
executable	str	Absolute or scheduler-visible command path to execute.
job_name	str	PJM job name used for scheduler display and output filenames.
gfscache	str | None	Optional PJM_LLIO_GFSCACHE path.
mpi_options_for_pjm	list[str] | None	Optional MPI options emitted as PJM directives.
spack_modules	list[str] | None	Optional Spack modules loaded by the generated script.
pjm_resources	list[str] | None	Additional raw #PJM -L resource directives.

qcsc_prefect_adapters.fugaku.builder.to_fugaku_template_kwargs

to_fugaku_template_kwargs(
    *,
    work_dir: Path,
    exec_profile: ExecutionProfile,
    req: FugakuJobRequest,
    script_basename: str = "batch.pjm",
) -> dict

Build template variables for the Fugaku PJM script.

Parameters:

Name	Type	Description	Default
work_dir	Path	Base working directory used to derive output paths.	required
exec_profile	ExecutionProfile	Scheduler-independent execution profile.	required
req	FugakuJobRequest	Fugaku-specific scheduler request fields.	required
script_basename	str	Script basename used for output/stat filenames.	'batch.pjm'

Returns:

Type	Description
dict	A dictionary that can be passed to the Fugaku Jinja template.

qcsc_prefect_adapters.fugaku.builder.render_script

render_script(
    *,
    work_dir: Path,
    exec_profile: ExecutionProfile,
    req: FugakuJobRequest,
    script_basename: str = "batch.pjm",
) -> str

Render a Fugaku job script text from the configured Jinja template.

.. note:: The template file is configured by module constant _TEMPLATE and is expected to be a .j2 file.

Parameters:

Name	Type	Description	Default
work_dir	Path	Base working directory used in template variables.	required
exec_profile	ExecutionProfile	Scheduler-independent execution profile.	required
req	FugakuJobRequest	Fugaku-specific scheduler request fields.	required
script_basename	str	Script basename used for output/stat filenames.	'batch.pjm'

Returns:

Type	Description
str	Rendered PJM script text.

qcsc_prefect_adapters.fugaku.builder.write_script_file

write_script_file(
    *, work_dir: Path, filename: str, text: str
) -> Path

Write a rendered job script into the work directory.

.. note:: This function is expected to be called inside qcsc_prefect_executor.fugaku.run.run_fugaku_job. Workflow authors normally do not need to call it directly.

.. note:: The text argument is expected to come from render_script, which renders the .j2 template specified by _TEMPLATE.

Parameters:

Name	Type	Description	Default
work_dir	Path	Base working directory where the script file is created.	required
filename	str	Script file name (for example batch.pjm).	required
text	str	Rendered script text.	required

Returns:

Type	Description
Path	Absolute path to the created job script file.

qcsc_prefect_adapters.fugaku.runtime.SubmitResult dataclass

SubmitResult(job_id: str, raw_output: str)

Submission result returned after PJM accepts a batch script.

Attributes:

Name	Type	Description
job_id	str	PJM job id parsed from pjsub stdout.
raw_output	str	Raw, stripped stdout emitted by pjsub.

qcsc_prefect_adapters.fugaku.runtime.FugakuPJMRuntime

FugakuPJMRuntime(*, no_check_directory: bool = False)

Async runtime wrapper for Fugaku PJM scheduler commands.

The runtime maps to the core PJM commands used on Fugaku: pjsub for submission, pjstat for status polling, and pjdel for cancellation. Workflow code usually calls qcsc_prefect_executor.fugaku.run.run_fugaku_job or qcsc_prefect_executor.from_blocks.run_job_from_blocks instead.

Create a Fugaku PJM runtime.

Parameters:

Name	Type	Description	Default
no_check_directory	bool	When true, pass --no-check-directory to pjsub submissions. This is intentionally opt-in because it skips PJM's data-area check for the submit working directory.	False

submit async

submit(
    script_path: Path, *, cwd: Path | None = None
) -> SubmitResult

Submit a PJM script with pjsub.

Parameters:

Name	Type	Description	Default
script_path	Path	Path to the PJM script file.	required
cwd	Path | None	Optional working directory for pjsub execution.	None

Returns:

Type	Description
SubmitResult	Parsed submission result including job id and raw output.

Raises:

Type	Description
SubmitError	If submission fails or job id cannot be parsed.

submit_bulk async

submit_bulk(
    script_path: Path,
    bulk_count: int,
    *,
    cwd: Path | None = None,
) -> str

Submit a PJM native bulk script and return the parent job id.

Parameters:

Name	Type	Description	Default
script_path	Path	Path to the PJM script file.	required
bulk_count	int	Number of native bulk subjobs. PJM subjob parameters are submitted as 0-{bulk_count - 1}.	required
cwd	Path | None	Optional working directory for pjsub execution.	None

Returns:

Type	Description
str	Parent PJM job id parsed from pjsub stdout.

Raises:

Type	Description
ValueError	If bulk_count is not positive.
SubmitError	If submission fails or the parent job id cannot be parsed.

wait_final_status async

wait_final_status(
    job_id: str,
    *,
    watch_poll_interval: float = 10.0,
    timeout_seconds: float | None = None,
) -> dict[str, Any]

Poll PJM status until a terminal state is reached.

This method first checks the active job view with pjstat -v and falls back to the historical view with pjstat -v -H when needed. The job is considered terminal when PJM reports EXT or CCL.

Parameters:

Name	Type	Description	Default
job_id	str	Target PJM job id.	required
watch_poll_interval	float	Poll interval in seconds.	10.0
timeout_seconds	float | None	Optional timeout for waiting terminal status.	None

Returns:

Type	Description
dict[str, Any]	Parsed final pjstat row.

Raises:

Type	Description
WaitTimeout	If timeout is exceeded.
RuntimeError	If an underlying pjstat command fails.

cancel async

cancel(job_id: str) -> None

Cancel a PJM job using pjdel.

Parameters:

Name	Type	Description	Default
job_id	str	Target PJM job id.	required

Raises:

Type	Description
CancelError	If cancellation fails.

qcsc_prefect_executor.fugaku.run.FugakuRunResult dataclass

FugakuRunResult(
    job_id: str,
    exit_status: int,
    state: str,
    job_status: dict[str, Any],
)

Normalized result returned by run_fugaku_job.

Attributes:

Name	Type	Description
job_id	str	PJM job id returned by pjsub.
exit_status	int	Integer exit status derived from PJM EC when present.
state	str	Final PJM state, such as "EXT" or "CCL".
job_status	dict[str, Any]	Parsed final PJM status dictionary from pjstat.

qcsc_prefect_executor.fugaku.run.run_fugaku_job async

run_fugaku_job(
    *,
    work_dir: Path,
    script_filename: str,
    exec_profile: ExecutionProfile,
    req: FugakuJobRequest,
    watch_poll_interval: float = 10.0,
    timeout_seconds: float | None = None,
    metrics_artifact_key: str = "fugaku-job-metrics",
) -> FugakuRunResult

Execute a Fugaku job end-to-end from runtime models.

.. note:: This function is the high-level executor entrypoint. It internally renders a script, submits it, waits for final status, captures logs, parses stats, and publishes a metrics artifact.

Parameters:

Name	Type	Description	Default
work_dir	Path	Working directory where scripts and job outputs are written.	required
script_filename	str	Job script filename to create in work_dir.	required
exec_profile	ExecutionProfile	Scheduler-independent execution profile.	required
req	FugakuJobRequest	Fugaku-specific scheduler request fields.	required
watch_poll_interval	float	Poll interval in seconds for job status checks.	10.0
timeout_seconds	float | None	Optional timeout for waiting final status.	None
metrics_artifact_key	str	Prefect artifact key for job metrics table.	'fugaku-job-metrics'

Returns:

Type	Description
FugakuRunResult	FugakuRunResult containing job id, exit status, state, and
FugakuRunResult	final scheduler status payload.

Slurm API

qcsc_prefect_adapters.slurm.builder.SlurmJobRequest dataclass

SlurmJobRequest(
    partition: str,
    executable: str,
    account: str | None = None,
    qpu: str | None = None,
    memory: str | None = None,
    ntasks: int | None = None,
)

Target-specific request fields required to build a Slurm batch job.

Attributes:

Name	Type	Description
partition	str	Slurm partition name passed to #SBATCH --partition.
executable	str	Absolute or scheduler-visible command path to execute.
account	str | None	Optional Slurm account passed to #SBATCH --account.
qpu	str | None	Optional QPU resource selector emitted by the Slurm template.
memory	str | None	Optional memory request passed to #SBATCH --mem.
ntasks	int | None	Optional task count passed to #SBATCH --ntasks.

qcsc_prefect_adapters.slurm.builder.to_slurm_template_kwargs

to_slurm_template_kwargs(
    *, exec_profile: ExecutionProfile, req: SlurmJobRequest
) -> dict

Build template variables for the Slurm job script.

Parameters:

Name	Type	Description	Default
exec_profile	ExecutionProfile	Scheduler-independent execution profile.	required
req	SlurmJobRequest	Slurm-specific scheduler request fields.	required

Returns:

Type	Description
dict	A dictionary that can be passed to the Slurm Jinja template.

qcsc_prefect_adapters.slurm.builder.render_script

render_script(
    *,
    work_dir: Path,
    exec_profile: ExecutionProfile,
    req: SlurmJobRequest,
) -> str

Render Slurm job script text from the configured Jinja template.

Parameters:

Name	Type	Description	Default
work_dir	Path	Working directory injected into the template.	required
exec_profile	ExecutionProfile	Scheduler-independent execution profile.	required
req	SlurmJobRequest	Slurm-specific scheduler request fields.	required

Returns:

Type	Description
str	Rendered Slurm script text.

qcsc_prefect_adapters.slurm.builder.write_script_file

write_script_file(
    *, work_dir: Path, filename: str, text: str
) -> Path

Write a rendered Slurm script into the work directory.

Parameters:

Name	Type	Description	Default
work_dir	Path	Base working directory where the script file is created.	required
filename	str	Script file name, for example batch.slurm.	required
text	str	Rendered script text.	required

Returns:

Type	Description
Path	Absolute path to the created job script file.

qcsc_prefect_adapters.slurm.runtime.SubmitResult dataclass

SubmitResult(job_id: str, raw_output: str)

Submission result returned after a scheduler accepts a batch script.

Attributes:

Name	Type	Description
job_id	str	Scheduler job id parsed from the submission command output.
raw_output	str	Raw, stripped stdout emitted by the submission command.

qcsc_prefect_adapters.slurm.runtime.SlurmRuntime

Async runtime wrapper for Slurm scheduler commands.

This class is the low-level boundary around sbatch, sacct, and scancel. Workflow code usually calls qcsc_prefect_executor.slurm.run.run_slurm_job or qcsc_prefect_executor.from_blocks.run_job_from_blocks instead.

submit async

submit(
    script_path: Path, *, cwd: Path | None = None
) -> SubmitResult

Submit a Slurm batch script with sbatch --parsable.

Parameters:

Name	Type	Description	Default
script_path	Path	Path to the generated Slurm script.	required
cwd	Path | None	Optional working directory for sbatch.	None

Returns:

Type	Description
SubmitResult	Parsed submission payload containing the Slurm job id.

Raises:

Type	Description
SubmitError	If sbatch fails or the job id cannot be parsed.

wait_final_status async

wait_final_status(
    job_id: str,
    *,
    watch_poll_interval: float = 10.0,
    timeout_seconds: float | None = None,
) -> dict[str, Any]

Poll sacct until the job reaches a terminal state.

The returned dictionary is normalized to the fields requested from sacct: JobID, State, ExitCode, Elapsed, AllocCPUS, and NodeList.

Parameters:

Name	Type	Description	Default
job_id	str	Slurm job id to watch.	required
watch_poll_interval	float	Seconds to wait between sacct calls.	10.0
timeout_seconds	float | None	Optional maximum wait time.	None

Returns:

Type	Description
dict[str, Any]	Parsed terminal sacct row for the batch job.

Raises:

Type	Description
WaitTimeout	If timeout_seconds elapses before a terminal state.
RuntimeError	If an underlying sacct command fails.

cancel async

cancel(job_id: str) -> None

Cancel a Slurm job using scancel.

Parameters:

Name	Type	Description	Default
job_id	str	Target Slurm job id.	required

Raises:

Type	Description
CancelError	If scancel exits with an error.

qcsc_prefect_executor.slurm.run.SlurmRunResult dataclass

SlurmRunResult(
    job_id: str,
    exit_status: int,
    state: str,
    job_status: dict[str, Any],
)

Normalized result returned by run_slurm_job.

Attributes:

Name	Type	Description
job_id	str	Slurm job id returned by sbatch.
exit_status	int	Integer process exit code parsed from Slurm ExitCode.
state	str	Final Slurm job state.
job_status	dict[str, Any]	Parsed terminal sacct status dictionary.

qcsc_prefect_executor.slurm.run.run_slurm_job async

run_slurm_job(
    *,
    work_dir: Path,
    script_filename: str,
    exec_profile: ExecutionProfile,
    req: SlurmJobRequest,
    watch_poll_interval: float = 10.0,
    timeout_seconds: float | None = None,
    metrics_artifact_key: str = "slurm-job-metrics",
) -> SlurmRunResult

Execute a Slurm job end-to-end from runtime models.

This high-level executor renders the Slurm script, submits it with sbatch, waits for terminal sacct status, captures stdout/stderr files, and publishes a Prefect table artifact with scheduler metrics.

Parameters:

Name	Type	Description	Default
work_dir	Path	Working directory where scripts and job outputs are written.	required
script_filename	str	Job script filename to create in work_dir.	required
exec_profile	ExecutionProfile	Scheduler-independent execution profile.	required
req	SlurmJobRequest	Slurm-specific scheduler request fields.	required
watch_poll_interval	float	Poll interval in seconds for job status checks.	10.0
timeout_seconds	float | None	Optional timeout for waiting final status.	None
metrics_artifact_key	str	Prefect artifact key for job metrics table.	'slurm-job-metrics'

Returns:

Type	Description
SlurmRunResult	SlurmRunResult containing job id, exit status, state, and
SlurmRunResult	final scheduler status payload.

qcsc_prefect_executor.slurm.from_blocks.run_slurm_job_from_blocks async

run_slurm_job_from_blocks(
    *,
    command_block_name: str,
    execution_profile_block_name: str,
    hpc_profile_block_name: str,
    work_dir: Path,
    script_filename: str,
    user_args: list[str] | None = None,
    watch_poll_interval: float = 10.0,
    timeout_seconds: float | None = None,
    metrics_artifact_key: str = "slurm-job-metrics",
) -> SlurmRunResult

Backward-compatible wrapper around run_job_from_blocks for Slurm.

DICE API

qcsc_prefect_dice.solver_job.DiceSHCISolverJob

Bases: Block

Prefect block facade for DICE execution through qcsc-prefect blocks.

run async

run(
    ci_strings: tuple[ndarray, ndarray],
    one_body_tensor: ndarray,
    two_body_tensor: ndarray,
    norb: int,
    nelec: tuple[int, int],
    spin_sq: float | None = None,
) -> SCIResult

Run DICE and return parsed SHCI outputs.

qcsc_prefect_dice.block_utils.register_dice_block_types

register_dice_block_types() -> None

Register common block schemas used by DICE integrations.

qcsc_prefect_dice.block_utils.create_dice_blocks

create_dice_blocks(
    *,
    hpc_target: Literal["local", "miyabi", "fugaku"],
    root_dir: str,
    dice_executable: str,
    project: str = "",
    queue: str = "",
    command_block_name: str = "cmd-dice-solver",
    execution_profile_block_name: str | None = None,
    hpc_profile_block_name: str | None = None,
    solver_block_name: str = "dice-solver",
    command_name: str = "dice",
    executable_key: str = "dice_solver",
    profile_name: str = "dice-mpi",
    launcher: Literal[
        "single", "mpirun", "mpiexec", "mpiexec.hydra"
    ]
    | None = None,
    num_nodes: int = 1,
    mpiprocs: int = 4,
    ompthreads: int | None = None,
    walltime: str = "01:00:00",
    resource_class: Literal["cpu", "gpu"] = "cpu",
    modules: list[str] | None = None,
    mpi_options: list[str] | None = None,
    pre_commands: list[str] | None = None,
    environments: dict[str, str] | None = None,
    script_filename: str | None = None,
    metrics_artifact_key: str | None = None,
    select_cutoff: float = 0.0005,
    davidson_tol: float = 1e-05,
    energy_tol: float = 1e-10,
    max_iter: int = 10,
    return_sci_state: bool = True,
    gfscache: str | None = None,
    spack_modules: list[str] | None = None,
    mpi_options_for_pjm: list[str] | None = None,
    pjm_resources: list[str] | None = None,
) -> dict[str, str]

Create the standard block set required to run DICE on qcsc-prefect.

This helper keeps a typed, explicit API on purpose. Algorithm-specific wrappers can translate TOML or CLI config dictionaries into these arguments without leaking config-shape coupling into the shared package.

qcsc_prefect_dice.io_utils.make_job_work_dir

make_job_work_dir(base_work_dir: Path) -> Path

Create a unique work directory for one DICE execution.

Parameters:

Name	Type	Description	Default
base_work_dir	Path	Parent directory under which per-job directories are created.	required

Returns:

Type	Description
Path	Newly created job directory named with a UTC timestamp and short random
Path	suffix.

Raises:

Type	Description
FileExistsError	If the generated unique directory already exists.
OSError	If the directory cannot be created.

qcsc_prefect_dice.io_utils.prep_dice_input_files

prep_dice_input_files(
    *,
    work_dir: Path,
    ci_strings: tuple[ndarray, ndarray],
    one_body_tensor: ndarray,
    two_body_tensor: ndarray,
    norb: int,
    nelec: tuple[int, int],
    spin_sq: float | None,
    select_cutoff: float,
    davidson_tol: float,
    energy_tol: float,
    max_iter: int,
) -> None

Prepare all input files required by the DICE SHCI executable.

The helper writes fcidump.txt from molecular integrals, renders input.dat from solver controls, and serializes alpha/beta determinant seeds into AlphaDets.bin and BetaDets.bin using the DICE binary format expected by the bundled template.

Parameters:

Name	Type	Description	Default
work_dir	Path	Directory where DICE input files are written.	required
ci_strings	tuple[ndarray, ndarray]	Alpha and beta determinant bitstrings used as the initial CI space.	required
one_body_tensor	ndarray	One-electron integral tensor passed to PySCF's FCIDUMP writer.	required
two_body_tensor	ndarray	Two-electron integral tensor passed to PySCF's FCIDUMP writer.	required
norb	int	Number of spatial orbitals.	required
nelec	tuple[int, int]	Number of alpha and beta electrons.	required
spin_sq	float | None	Optional target spin-squared value for input.dat.	required
select_cutoff	float	DICE selection cutoff.	required
davidson_tol	float	Davidson solver convergence tolerance.	required
energy_tol	float	SCI energy convergence tolerance.	required
max_iter	int	Maximum number of HCI iterations.	required

Raises:

Type	Description
OSError	If any input file cannot be written.
OverflowError	If a determinant bitstring cannot be encoded into the fixed-width DICE determinant representation.

qcsc_prefect_dice.io_utils.read_dice_output_files

read_dice_output_files(
    *,
    work_dir: Path,
    norb: int,
    nelec: tuple[int, int],
    return_sci_state: bool,
) -> SCIResult

Read DICE output files and reconstruct an SCIResult.

The helper reads orbital occupancies from spin1RDM.0.0.txt and the variational energy from shci.e. When return_sci_state is true, it also parses dets.bin and reconstructs a qiskit_addon_sqd.fermion.SCIState compatible with the SQD addon.

Parameters:

Name	Type	Description	Default
work_dir	Path	Directory containing DICE output files.	required
norb	int	Number of spatial orbitals.	required
nelec	tuple[int, int]	Number of alpha and beta electrons.	required
return_sci_state	bool	Whether to parse dets.bin and include the SCI state in the returned result.	required

Returns:

Type	Description
SCIResult	Parsed SCI result containing energy, optional SCI state, and alpha/beta
SCIResult	orbital occupancies.

Raises:

Type	Description
FileNotFoundError	If a required DICE output file is missing.
ValueError	If text output cannot be parsed into numeric arrays.
error	If binary output is truncated or malformed.