rocrate_action_recorder.core¶

Core functionality for recording CLI invocations in RO-Crate format.

Attributes¶

logger

Classes¶

`Program`	Container for program details.
`IOArgumentPath`	Container for the details of an input/output argument.
`IOArgumentPaths`	Container for all the input/output paths for a recording.

Functions¶

`detect_software_version`(→ str)	Detect software version from package name or executable or caller package.
`_detect_package_from_stack`(→ str \| None)	Detect the distribution name from the current call stack.
`_dectect_version_by_running`(→ str)	Try to detect version by running the program with --version flag.
`make_action_id`(→ str)	Create an action ID from command-line arguments.
`record`(→ pathlib.Path)	Record a CLI invocation in an RO-Crate.
`build_software_application`(...)	Build a SoftwareApplication object for the crate.
`add_software_application`(...)	Add or get a SoftwareApplication in the crate.
`add_agent`(→ rocrate.model.person.Person)	Add or get a Person agent in the crate.
`_get_mime_type`(→ str)	Detect MIME type from file path.
`get_relative_path`(→ pathlib.Path)	Get the relative path from root.
`add_file`(→ rocrate.model.File)	Add or update a File in the crate.
`add_files`(→ list[rocrate.model.File])	Add multiple files to the crate.
`add_dir`(→ rocrate.model.dataset.Dataset)	Add or get a Dataset (directory) in the crate.
`add_dirs`(→ list[rocrate.model.dataset.Dataset])	Add multiple directories to the crate.
`add_action`(→ None)	Add an action to the crate.
`_record_run`(→ pathlib.Path)	Internal function to record a run into an RO-Crate.
`_unique_by_id`(→ list[_unique_by_id.T])	Get unique entities based on their IDs.
`conform_to_process_run_crate_profile`(...)	Makes crate conform to Process Run Crate profile
`_update_crate`(→ rocrate.rocrate.ROCrate)	Internal function to update a crate with all necessary entities.
`playback`(→ str)	Extract and return recorded action commands sorted by execution time.

Module Contents¶

rocrate_action_recorder.core.logger¶

class rocrate_action_recorder.core.Program[source]¶

Container for program details.

name: str¶: Name of the program.

description: str¶: Description of the program.

subcommands: dict[str, Program]¶: Dictionary of subcommand names to Program instances. Only contains used by current action.

version: str | None = None¶: Version of the program. If None, will be detected automatically.

class rocrate_action_recorder.core.IOArgumentPath[source]¶

Container for the details of an input/output argument.

name: str¶: The name of the argument as known in the parser with – stripped.

path: pathlib.Path¶: The path value of the argument.

help: str¶: Help text associated with the argument.

class rocrate_action_recorder.core.IOArgumentPaths[source]¶

Container for all the input/output paths for a recording.

input_files: list[IOArgumentPath]¶: List of input file argument paths.

output_files: list[IOArgumentPath]¶: List of output file argument paths.

input_dirs: list[IOArgumentPath]¶: List of input directory argument paths.

output_dirs: list[IOArgumentPath]¶: List of output directory argument paths.

rocrate_action_recorder.core.detect_software_version(program_name: str) → str[source]¶

Detect software version from package name or executable or caller package.

Parameters:: program_name – Name of the program/package or path to executable.
Returns:: Version string if found, otherwise empty string.

rocrate_action_recorder.core._detect_package_from_stack() → str | None[source]¶

Detect the distribution name from the current call stack.

Walks stack frames and returns the first distribution name that provides the top-level package/module referenced in the frame. Returns None if not found.

rocrate_action_recorder.core._dectect_version_by_running(program_name: str) → str[source]¶

Try to detect version by running the program with –version flag.

Parameters:: program_name – Name of the program or path to executable.
Returns:: Version string if found, otherwise empty string.

rocrate_action_recorder.core.make_action_id(argv: list[str] | None = None) → str[source]¶

Create an action ID from command-line arguments.

Parameters:: argv – Command-line arguments. If None, uses sys.argv.
Returns:: Quoted and joined command-line string.

rocrate_action_recorder.core.record(program: Program, ioargs: IOArgumentPaths, start_time: datetime.datetime, crate_dir: pathlib.Path | None = None, argv: list[str] | None = None, end_time: datetime.datetime | None = None, current_user: str | None = None, dataset_license: str | None = None) → pathlib.Path[source]¶

Record a CLI invocation in an RO-Crate.

This is a low-level function, better use one of the adapter functions for specific argument parsing frameworks like record_argparse() or record_cyclopts().

Example

Example with single input and output file arguments:

from datetime import datetime, UTC
from pathlib import Path

from rocrate_action_recorder import (
    IOArgumentPath,
    IOArgumentPaths,
    Program,
    record,
)

crate_dir = Path()
input_path = crate_dir / "input.txt"
output_path = crate_dir / "output.txt"
input_path.write_text("Hello World")
argv = [
    "myscript",
    "--input",
    str(input_path),
    "--output",
    str(output_path),
]
start_time = datetime(2026, 1, 16, 12, 0, 0, tzinfo=UTC)
# Simulate the script's main operation
output_path.write_text(input_path.read_text().upper())
end_time = datetime(2026, 1, 16, 12, 0, 5, tzinfo=UTC)

crate_meta = record(
    program=Program(
        name="myscript", description="My test script", version="1.2.3"
    ),
    ioargs=IOArgumentPaths(
        input_files=[
            IOArgumentPath(name="input", path=input_path, help="Input file")
        ],
        output_files=[
            IOArgumentPath(name="output", path=output_path, help="Output file")
        ],
    ),
    argv=argv,
    current_user="tester",
    start_time=start_time,
    end_time=end_time,
    crate_dir=crate_dir,
    dataset_license="CC-BY-4.0",
)
# crate_meta == Path("ro-crate-metadata.json")

Parameters:

program – The program details.
ioargs – Which files/directories are involved in action.
start_time – The datetime when the action started.
crate_dir – Optional path to the RO-Crate directory. If None, uses current working directory.
argv – Optional list of command-line arguments. If None, uses sys.argv.
end_time – Optional datetime when the action ended. If None, uses current time.
current_user – Optional username of the user running the action. If None, attempts to determine it from the system.
dataset_license – Optional license string to set for the RO-Crate dataset. For example “CC-BY-4.0”.

Returns:

Path to the generated ro-crate-metadata.json file.

Raises:

ValueError – If the current user cannot be determined. If the specified paths are outside the crate root. If the software version cannot be determined based on the program name. If start_time and end_time have different timezone information. If start_time is after end_time.

rocrate_action_recorder.core.build_software_application(crate: rocrate.rocrate.ROCrate, program: Program) → rocrate.rocrate.SoftwareApplication[source]¶

Build a SoftwareApplication object for the crate.

Parameters:

crate – The ROCrate object.
program – The Program object with name and description.
software_version – The version string.

Returns:

A SoftwareApplication object.

rocrate_action_recorder.core.add_software_application(crate: rocrate.rocrate.ROCrate, program: Program) → rocrate.rocrate.SoftwareApplication[source]¶

Add or get a SoftwareApplication in the crate.

Parameters:

crate – The ROCrate object.
program – The Program object.
software_version – The version string.

Returns:

The SoftwareApplication object.

rocrate_action_recorder.core.add_agent(crate: rocrate.rocrate.ROCrate, current_user: str) → rocrate.model.person.Person[source]¶

Add or get a Person agent in the crate.

Parameters:

crate – The ROCrate object.
current_user – Username of the agent.

Returns:

The Person object.

rocrate_action_recorder.core._get_mime_type(path: pathlib.Path) → str[source]¶

Detect MIME type from file path.

Parameters:: path – Path to the file.
Returns:: MIME type string, defaults to ‘application/octet-stream’.

rocrate_action_recorder.core.get_relative_path(path: pathlib.Path, root: pathlib.Path) → pathlib.Path[source]¶

Get the relative path from root.

Parameters:

path – The absolute or relative path.
root – The root directory.

Returns:

The relative path from root.

Raises:

ValueError – If path is outside the root.

rocrate_action_recorder.core.add_file(crate: rocrate.rocrate.ROCrate, crate_root: pathlib.Path, ioarg: IOArgumentPath) → rocrate.model.File[source]¶

Add or update a File in the crate.

Parameters:

crate – The ROCrate object.
crate_root – The root directory of the crate.
ioarg – The IOArgument specifying the file.

Returns:

The File object.

rocrate_action_recorder.core.add_files(crate: rocrate.rocrate.ROCrate, crate_root: pathlib.Path, ioargs: list[IOArgumentPath]) → list[rocrate.model.File][source]¶

Add multiple files to the crate.

Parameters:

crate – The ROCrate object.
crate_root – The root directory of the crate.
ioargs – List of IOArgument objects.

Returns:

List of File objects.

rocrate_action_recorder.core.add_dir(crate: rocrate.rocrate.ROCrate, crate_root: pathlib.Path, ioarg: IOArgumentPath) → rocrate.model.dataset.Dataset[source]¶

Add or get a Dataset (directory) in the crate.

Parameters:

crate – The ROCrate object.
crate_root – The root directory of the crate.
ioarg – The IOArgument specifying the directory.

Returns:

The Dataset object.

rocrate_action_recorder.core.add_dirs(crate: rocrate.rocrate.ROCrate, crate_root: pathlib.Path, ioargs: list[IOArgumentPath]) → list[rocrate.model.dataset.Dataset][source]¶

Add multiple directories to the crate.

Parameters:

crate – The ROCrate object.
crate_root – The root directory of the crate.
ioargs – List of IOArgument objects.

Returns:

List of Dataset objects.

rocrate_action_recorder.core.add_action(crate: rocrate.rocrate.ROCrate, action_id: str, start_time: datetime.datetime, end_time: datetime.datetime, software: rocrate.rocrate.SoftwareApplication, all_inputs: list[rocrate.model.File | rocrate.model.dataset.Dataset], all_outputs: list[rocrate.model.File | rocrate.model.dataset.Dataset], agent: rocrate.model.person.Person) → None[source]¶

Add an action to the crate.

Parameters:

crate – The ROCrate object.
action_id – Unique identifier for the action.
start_time – When the action started.
end_time – When the action ended.
software – The instrument (SoftwareApplication) used.
all_inputs – List of input files/datasets.
all_outputs – List of output files/datasets.
agent – The Person who ran the action.

rocrate_action_recorder.core._record_run(crate_root: pathlib.Path, program: Program, ioargs: IOArgumentPaths, action_id: str, start_time: datetime.datetime, end_time: datetime.datetime, current_user: str, dataset_license: str | None = None) → pathlib.Path[source]¶

Internal function to record a run into an RO-Crate.

Parameters:

crate_root – Root directory of the crate.
program – The Program object.
ioargs – IOArgs with input/output files and directories.
action_id – Unique action identifier.
start_time – When the action started.
end_time – When the action ended.
current_user – Username of the user.
dataset_license – Optional license for the dataset.

Returns:

Path to the ro-crate-metadata.json file.

rocrate_action_recorder.core._unique_by_id(entities: list[_unique_by_id.T]) → list[_unique_by_id.T][source]¶

Get unique entities based on their IDs.

Parameters:: entities – List of entities.
Returns:: List of unique entities.

rocrate_action_recorder.core.conform_to_process_run_crate_profile(crate: rocrate.rocrate.ROCrate) → rocrate.model.creativework.CreativeWork[source]¶

Makes crate conform to Process Run Crate profile

See https://www.researchobject.org/workflow-run-crate/profiles/0.5/process_run_crate/

Parameters:: crate – The ROCrate object.
Returns:: The CreativeWork entity representing the Process Run Crate profile.

rocrate_action_recorder.core._update_crate(crate: rocrate.rocrate.ROCrate, crate_root: pathlib.Path, program: Program, ioargs: IOArgumentPaths, action_id: str, start_time: datetime.datetime, end_time: datetime.datetime, current_user: str, dataset_license: str | None) → rocrate.rocrate.ROCrate[source]¶

Internal function to update a crate with all necessary entities.

Parameters:

crate – The ROCrate object.
crate_root – Root directory.
program – The Program object.
ioargs – IOArgs with inputs/outputs.
action_id – Unique action identifier.
start_time – When the action started.
end_time – When the action ended.
current_user – Username.
dataset_license – Optional license.

Returns:

The updated ROCrate object.

rocrate_action_recorder.core.playback(crate_root: pathlib.Path) → str[source]¶

Extract and return recorded action commands sorted by execution time.

Parameters:: crate_root – Root directory of the RO-Crate.
Returns:: Newline-separated string of action command lines, sorted by endTime. Returns empty string if no actions are recorded.