adctk.builder.Builder Class Reference

ADC Toolkit message and subsection construction class. More...

Inheritance diagram for adctk.builder.Builder:

Public Member Functions
	__init__ (self)
adc_types.Field	get_value (self, str path_full)
	get the existing named nested field in the builder.
	get_value_string (self, str path)
	get the existing named nested scalar string value.
	get_value_int64 (self, str path)
	get the existing named nested scalar int64-equivalent value.
	get_value_uint64 (self, str path)
	get the existing named nested scalar uint64-equivalent value.
	__json__ (self)
	add_header_section (self, str application_name)
	Create or replace the 'header' section using application_name.
	add_host_section (self, osinfo=False, ramsize=False, env=False, cpu=False, gpu=False, numa=False, all_hs=False)
	Add the 'host' section, with optional subgroups of properties.
	add_app_data_section (self, 'Builder' app_data)
	Create the "app_data" section with data defined by the application writer.
	add_model_data_section (self, 'Builder' model_data)
	populate application run-time physics (re)configuration or result to "model_data" section.
	add_code_section (self, str tag, 'Builder' version, 'Builder|None' code_details=None)
	Add the 'code' section, with user-specified version subsection and optional code_details section describing things which do not normally vary with invocation.
	add_code_configuration_section (self, 'Builder' build_details)
	add_exit_data_section (self, 'int|str' return_code, str status, 'Builder|None' status_details=None)
	populate "exit_data" section with return code and status string & user provided details.
	add_memory_usage_section (self)
	Populate "memory_usage" section with current host /proc/meminfo data in the style of free(1).
'Builder|None'	get_section (self, str name)
	get the existing named section.
'list[str]'	get_section_names (self)
	Get the names of sections.
	add_mpi_section (self, name, communicator, bool rank=False, bool size=False, bool mpi_name=False, bool hostlist=False, bool rank_host=False, bool ver=False, bool lib_ver=False, bool mpiall=False, bool local=False)
	Unimplemented: python with mpi is not yet supported.
	add_gitlab_ci_section (self)
	Add gitlab_ci environment variable dictionary.
	add_workflow_section (self, str|None wfid=None, str|None parent=None, list[str]|str|None children=None)
	add data from adc_wfid_ environment variables.
	add_workflow_children (self, collections.abc.Iterable[str] child_uuids)
	Add list of child uuids to "adc_workflow" section after add_workflow_section has been called.
	add_slurm_section (self, collections.abc.Iterable[str]|None output_vars=None)
	Add slurm output environment variable dictionary elements.
	add_section (self, str name, 'Builder|collections.abc.Mapping[str,...]' section)
	Add or replace a user-defined, named section.
str	serialize (self)
	get json string of the builder.
	add (self, str name, any value, value_type=None)
	add named item and its datatype information to the builder.
	add_path (self, name, str path)
	Add string file path.
	add_json_string (self, str name, str js)
	Add stringified json.
	add_yaml_string (self, str name, str yaml)
	Add stringified yaml.
	add_xml_string (self, str name, str xml)
	Add stringified xml.
	add_number_string (self, str name, decimal.Decimal dec)
	Add stringified number of arbitrary precision.
	add_epoch (self, name, int epoch)
	add unix epoch seconds (gettimeofday)
	add_mime (self, str name, str mime_type, str encoding, str file_name, str data)
	add a string encoded object (e.g.
	add_array_typed_string (self, name, adc_types.ScalarType st, collections.abc.Iterable[str] strings)
	Add string group tagged with scalar type.

Static Public Member Functions
'Builder'	BasicBuilder (str app_name)
	DEPRECATED.

Static Public Attributes
	API_VERSION = adc_version.AdcVersion()._dict()

Protected Member Functions
	_set_attr (self, name, value)
	_get_attr (self, name)
	_find_path (self, obj, path)
dict	_flatten (self, depth=0)
	Get a dictionary tree-copy constructed from keys and sections.
adc_types.KeyType	_kind (self, str name)

Detailed Description

ADC Toolkit message and subsection construction class.

Use of the Builder ensures that adctk messages:

• meet the schema definitions for commonly defined sections and the required header.
• capture the correct data typing information for application-defined fields. Presently, some of the functions only work as intended on Linux platforms, as they examine /proc.

Constructor & Destructor Documentation

__init__()

adctk.builder.Builder.__init__

(

self

)

Reimplemented in adctk.builder.Memory, adctk.builder.Host, and adctk.builder.Header.

    def __init__(self):
        self._sections: dict[Builder]  = {}
        # dict of key/values defined by add directly on this builder
        self._d: dict[dict] = {}
 

Member Function Documentation

__json__()

adctk.builder.Builder.__json__

(

self

)

    def __json__(self):
        print("CALLED __json__?")
        return self._flatten()
 

_find_path()

adctk.builder.Builder._find_path (   self,   obj,   path  )

protected

look up value in nested string-named structures using /-delimited path

    def _find_path(self, obj, path):
        """look up value in nested string-named structures using /-delimited path"""
        parts = path.strip("/").split("/")
        
        for p in parts:
            if p in obj:
                obj = obj[p]
            else:
                return None
        return obj
 

_flatten()

dict adctk.builder.Builder._flatten (   self,   depth = 0  )

protected

Get a dictionary tree-copy constructed from keys and sections.

Changes made to the returned dictionary are not reflected in the Builder instance.

    def _flatten(self, depth=0) -> dict:
        """! @brief Get a dictionary tree-copy constructed from keys and sections.
 
         Changes made to the returned dictionary are not reflected in
         the Builder instance.
        """
        d = copy.deepcopy(self._d)
        for name,section in self._sections.items():
            d[name] = section._flatten(depth=depth+1)
        return d
 

_get_attr()

adctk.builder.Builder._get_attr (   self,   name  )

protected

    def _get_attr(self, name):
        if name in self._d:
            return self._d[name]
        return None
 

_kind()

adc_types.KeyType adctk.builder.Builder._kind (   self, str  name  )

protected

 return the type that name resolves to in this builder.

    def _kind(self, name: str) -> adc_types.KeyType:
        """ return the type that name resolves to in this builder."""
        if name in self._sections:
            
            return adc_types.KeyType.k_section
        if name in self._d:
            
            return adc_types.KeyType.k_value
        
        return adc_types.KeyType.k_none
 

_set_attr()

adctk.builder.Builder._set_attr (   self,   name,   value  )

protected

    def _set_attr(self, name, value):
        self._d[name] = value
 

add()

adctk.builder.Builder.add	(		self,
		str	name,
		any	value,
			value_type = None
	)

add named item and its datatype information to the builder.

Supported value types are

• built-ins: int, float, complex, bool, str
• standard: decimal, numpy.ndarray (as 1-d list and dimensions if originally multidimensional)
• character types: Python often wraps non-python applications; where the developer knows a python string represents a single character, the optional value_type parameter can be passed to disambiguate. Supported value_type options are (signed)"char",(unsigned)"char8", (signed)"char16",(signed)"char32"
• lower-precision types: Python often wraps non-python applications; where the developer knows a python number represents something smaller than 64-bit, the optional value_type parameter can be passed to disambiguate. Supported numeric value_type options are int8, int16,int32, uint8, uint16, uint32, f32. (signed)"char16",(signed)"char32"
• any value not explicitly supported which can be converted to a json string will be.

Any value not explicitly supported which cannot be converted to a json string will be silently ignored.

Notes:

• int is represented as type int64 if possible and number_str if too big.

    def add(self, name: str, value: any, value_type=None):
        """! @brief add named item and its datatype information to the builder.
 
         Supported value types are
         - built-ins: int, float, complex, bool, str
         - standard:  decimal, numpy.ndarray (as 1-d list and dimensions if
          originally multidimensional)
         - character types:  Python often wraps non-python applications; where
          the developer knows a python string represents a single character,
          the optional value_type parameter can be passed to disambiguate.
          Supported value_type options are (signed)"char",(unsigned)"char8",
          (signed)"char16",(signed)"char32"
         - lower-precision types:  Python often wraps non-python applications; where
          the developer knows a python number represents something smaller than
          64-bit, the optional value_type parameter can be passed to disambiguate.
          Supported numeric value_type options are int8, int16,int32, uint8,
          uint16, uint32,  f32.
          (signed)"char16",(signed)"char32"
         - any value not explicitly supported which can be converted to a
          json string will be.
 
         Any value not explicitly supported which cannot be converted to
         a json string will be silently ignored.
 
         Notes:
         - int is represented as type int64 if possible and number_str if too big.
    
        """
        if name is None or value is None:
            raise ValueError("add: name and value cannot be None")
        t = type(value)
        
        if isinstance(t, Builder):
            self.add_section(name, value)
            return
        if isinstance(value, bool):
            self._set_attr(name,
                    { "type": adc_types.scalar_to_json(adc_types.type_to_scalar(t)),
                        "value": value })
            return
        if isinstance(value, float):
            if not value_type:
                
                self._set_attr(name,
                        { "type": adc_types.scalar_to_json(adc_types.type_to_scalar(t, value)),
                            "value": value })
                return
            if value_type in ["f32", "f64"]:
                self._set_attr(name,
                        { "type": adc_types.scalar_to_json(adc_types.scalar_from_json(value_type)),
                            "value": value })
                return
            raise ValueError("add: value_type option for float value must be f32 or f64")
        if isinstance(value, str):
            if not value_type:
                
                self._set_attr(name,
                        { "type": adc_types.scalar_to_json(adc_types.type_to_scalar(t)),
                            "value": value })
                return
            if value_type in ["char", "char16", "char32", "char8"]:
                
                self._set_attr(name,
                        { "type": adc_types.scalar_to_json(adc_types.scalar_from_json(value_type)),
                            "value": value })
                return
            raise ValueError("add: value_type option for str value must be one of: char, char16, char32, char8")
        if t == complex:
            self._set_attr(name,
                    { "type": adc_types.scalar_to_json(adc_types.type_to_scalar(t, value)),
                        "value": [value.real, value.imag] })
            return
        if t == int:
            if not value_type:
                jtname = adc_types.scalar_to_json(adc_types.type_to_scalar(int, value))
                if adc_types.fits_int64(value):
                    self._set_attr(name, { "type": jtname, "value": value })
                else:
                    self._set_attr(name, { "type": jtname, "value": str(value) })
                return
            if value_type in ["int16", "int32", "int8", "uint8", "uint16", "uint32", "int64"]:
                
                self._set_attr(name,
                        { "type": adc_types.scalar_to_json(adc_types.scalar_from_json(value_type)),
                            "value": value })
                return
            raise ValueError("add: value_type option for int value must be one of: int16, int32, int8, uint8, uint16, uint32, int64")
        if isinstance(value, decimal.Decimal):
            jtname = adc_types.scalar_to_json(adc_types.type_to_scalar(t))
            self._set_attr(name, { "type": jtname, "value": str(value) })
            return
        if type(value).__module__ == numpy.__name__:
            st = adc_types.scalar_from_numpy(value.dtype)
            if st == adc_types.ScalarType.cp_none:
                raise ValueError(f"add: unknown scalar type for {value.dtype}")
            jtype = adc_types.scalar_to_json(st)
            if isinstance(value, numpy.ndarray):
                if value.flags['F_CONTIGUOUS']:
                    rowcol = 'F'
                else:
                    rowcol = 'C'
                shape = list( value.shape )
                list_value = adc_types.get_list_format(value, st)
                self._set_attr(name, {"type": "array_"+jtype,
                    "value": list_value, "container_type": "ndarray", "order": rowcol,
                    "shape": shape})
                return
            match value:
                case numpy.float64() | numpy.float32() | numpy.float16():
                    self._set_attr(name, {"type": jtype, "value": \
                        float(numpy.format_float_scientific(value, unique=True)) , "BUG":"TRUE"} )
                case numpy.int64() | numpy.int32() | numpy.int16() | numpy.int8() | \
                     numpy.uint32() | numpy.uint16() | numpy.uint8() |numpy.bool():
                    self._set_attr(name, {"type": jtype, "value": int(value) } )
                case numpy.bool():
                    self._set_attr(name, {"type": jtype, "value": bool(value) } )
                case numpy.str_() | numpy.bytes_():
                    self._set_attr(name, {"type": jtype, "value": value.tostring() } )
                    # fixme numpy str_, bytes_ conversion may need work.
                case numpy.uint64():
                    self._set_attr(name, {"type": jtype, "value": value.tostring() })
                case numpy.complex64() | numpy.complex128():
                    self._set_attr(name, {"type": jtype, "value": [ \
                         float(numpy.format_float_scientific(value.real, unique=True)), \
                         float(numpy.format_float_scientific(value.imag, unique=True))] })
                case numpy.complex192() | numpy.complex256():
                    self._set_attr(name, {"type": jtype, "value": [ \
                        numpy.format_float_scientific(value.real, unique=True), \
                        numpy.format_float_scientific(value.imag, unique=True)] } )
                case _:
                    msg = f"add: item {name} contains unserializable numpy type." + \
                            f"({type(value)})" + \
                            "  consider add() on a type conversion to work around this."
                    print(dir(type(value).__module__ ))
                    print(dir( numpy.__name__))
                    print(dir(value))
                    raise ValueError(msg)
            return # from all matched setattr
        if isinstance(value, list|collections.abc.Set|tuple):
            depth = 0
            maxdepth=[0]
            it = adc_types.get_common_scalar(value, depth, maxdepth)
            pname = value.__class__.__name__
            if it != adc_types.ScalarType.cp_none:
                list_value = adc_types.get_list_format(value, it)
                if maxdepth[0] == 1:
                    cname = "vector"
                    pname = value.__class__.__name__
                    jt = "array_" + adc_types.scalar_to_json(it)
                    self._set_attr(name, {"type": jt,
                        "value": list_value, "container_type": cname, "python": pname})
                    return
                # this isn't documented in json spec; fixme-spec
                cname = "nested_list"
                major = "row" # fixme: use in setattr call? need test cases
                self._set_attr(name, {"type": adc_types.scalar_to_json(it),
                    "value": list_value, "container_type": cname, "major": major})
                return
            # if we had a 'warn-untyped-items' mode, it would warn here.
            try:
                tmp = json.dumps(value)
                self._set_attr(name,
                        {"type": adc_types.scalar_to_json(adc_types.ScalarType.cp_json),
                        "value": tmp, "python": pname})
                return
            except TypeError as orig_exception:
                msg = f"add: item {name} contains unserializable types." + \
                        " Use add() on individual elements to work around this."
                raise ValueError(msg) from orig_exception
        elif isinstance(value, array.array):
            it = adc_types.typecode_to_scalar(value)
            if it != adc_types.ScalarType.cp_none:
                list_value = value.tolist()
                if it == adc_types.ScalarType.cp_number_str:
                    list_value = [str(v) for v in list_value]
                cname = "vector"
                pname = f"array.array('{value.typecode}')"
                jt = "array_" + adc_types.scalar_to_json(it)
                self._set_attr(name, {"type": jt,
                    "value": list_value, "container_type": cname, "python": pname})
                return
            msg = f"add: array.array {name} contains unserializable types. how?!"
            raise ValueError(msg)
        else:
            # if we had a 'warn-untyped-items' mode, it would warn here.
            
            pname = value.__class__.__name__
            try:
                tmp = json.dumps(value)
                self._set_attr(name,
                        {"type": adc_types.scalar_to_json(
                            adc_types.ScalarType.cp_json),
                        "value": value, "python": pname})
                return
            except TypeError as orig_exception:
                msg = f"add: item {name} contains unserializable types." + \
                        " Use add() on individual elements to work around this."
                raise ValueError(msg) from orig_exception
 

add_app_data_section()

adctk.builder.Builder.add_app_data_section	(		self,
		'Builder'	app_data
	)

Create the "app_data" section with data defined by the application writer.

It is recommended that any relationship identifiers to previous jobs or higher level environments goes in app_data, unless supported by add_workflow_section, add_slurm_section or other normalized sections.

Example: app_data = Factory.get_builder(); app_data.add("saw_id", getenv("SAW_WORKFLOW_ID"); builder.add_app_data_section(app_data)

    def add_app_data_section(self, app_data: 'Builder'):
        """! Create the "app_data" section with data defined by the application writer.
 
            It is recommended that any relationship identifiers to previous jobs or higher
            level environments goes in app_data, unless supported by add_workflow_section,
            add_slurm_section or other normalized sections.
 
            Example:
             app_data = Factory.get_builder();
             app_data.add("saw_id", getenv("SAW_WORKFLOW_ID");
             builder.add_app_data_section(app_data)
        """
 
        if isinstance(app_data, Builder):
            self.add_section("app_data", app_data)
        else:
            raise TypeError("argument app_data must be of type adctk.Builder")
 

add_array_typed_string()

adctk.builder.Builder.add_array_typed_string	(		self,
			name,
		adc_types.ScalarType	st,
		collections.abc.Iterable[str]	strings
	)

Add string group tagged with scalar type.

Parameters

st	the ScalarType
strings	a one-dimensional string group
name	the field name.

Exceptions

ValueError

if strings cannot be converted.

            strings: collections.abc.Iterable[str]):
        """! Add string group tagged with scalar type.
 
        @param st the ScalarType
        @param strings a one-dimensional string group
        @param name the field name.
        @exception ValueError if strings cannot be converted.
        """
        jtname = adc_types.scalar_to_json(adc_types.type_to_scalar(st))
        maxdepth = [0]
        jt = adc_types.get_common_scalar(strings, 0, maxdepth)
        if not adc_types.is_string_type(jt):
            raise ValueError("add_array_typed_string: strings argument must be strings only")
        if maxdepth[0] > 1:
            raise ValueError("add_array_typed_string: strings argument must be one-dimensional")
        data = adc_types.get_list_format(strings)
        self._set_attr(name, {"type": "array_"+jtname, "value": data,
            "container_type": "vector", "python": strings.__class__.__name__})
 

add_code_configuration_section()

adctk.builder.Builder.add_code_configuration_section	(		self,
		'Builder'	build_details
	)

 @brief Populate build/install "configuration" information such as options enabled

    @param build_details option list.

    def add_code_configuration_section(self, build_details: 'Builder'):
        """ @brief Populate build/install "configuration" information such as options enabled
 
            @param build_details option list.
        """
        if build_details:
            if isinstance(build_details, Builder):
                self.add_section("code_configuration", build_details)
            else:
                raise TypeError("argument build_details must be from type adctk.Builder")
        else:
            raise TypeError("argument build_details must not be None")
 

add_code_section()

adctk.builder.Builder.add_code_section	(		self,
		str	tag,
		'Builder'	version,
		'Builder|None'	code_details = None
	)

Add the 'code' section, with user-specified version subsection and optional code_details section describing things which do not normally vary with invocation.

Automatic contents include tag, base program, full path, loaded modules and shared libraries. Shared libraries detected may vary with invocation if LD_LIBRARY_PATH or related environment settings change.

                        code_details: 'Builder|None' = None):
        """! Add the 'code' section, with user-specified version subsection and
        optional code_details section describing things which do not normally
        vary with invocation.
 
        Automatic contents include tag, base program, full path, loaded modules
        and shared libraries.
        Shared libraries detected may vary with invocation if LD_LIBRARY_PATH
        or related environment settings change.
        """
        fullpath = sys.executable
        basename = os.path.basename(fullpath)
        libs = get_libs()
        mods = get_modules()
        if code_details:
            if isinstance(code_details, Builder):
                d = code_details._flatten()
                c = { "name": tag,
                        "program": basename,
                        "path": fullpath,
                        "version": version._flatten(),
                        "libs": libs,
                        "modules": mods,
                        "details": d }
            else:
                raise TypeError("argument status_details must be from type adctk.Builder")
        else:
            c = { "name": tag,
                    "program": basename,
                    "path": fullpath,
                    "version": version._flatten(),
                    "libs": libs}
        self.add_section("code", c)
 

add_epoch()

adctk.builder.Builder.add_epoch	(		self,
			name,
		int	epoch
	)

add unix epoch seconds (gettimeofday)

    def add_epoch(self, name, epoch: int):
        """! add unix epoch seconds (gettimeofday)"""
        self._set_attr(name, { "type": adc_types.scalar_to_json(
            adc_types.ScalarType.cp_epoch),
            "value": epoch })
 

add_exit_data_section()

adctk.builder.Builder.add_exit_data_section	(		self,
		'int|str'	return_code,
		str	status,
		'Builder|None'	status_details = None
	)

populate "exit_data" section with return code and status string & user provided details.

Parameters

return_code	numeric value of exit status
status	one-line summary of the execution
status_details	any final or summary results of interest to application users or developers.

                            status: str, status_details: 'Builder|None' = None):
        """! @brief populate "exit_data" section with return code and status string
             & user provided details.
 
             @param return_code numeric value of exit status
             @param status one-line summary of the execution
             @param status_details any final or summary results of interest to application
             users or developers.
        """
        if status_details:
            if isinstance(status_details, Builder):
                d = status_details._flatten()
                ed = { "return_code": str(return_code), "status": status, "details": d }
            else:
                raise TypeError("argument status_details must be from type adctk.Builder")
        else:
            ed = { "return_code": str(return_code), "status": status }
        self.add_section("exit_data", ed)
 

add_gitlab_ci_section()

adctk.builder.Builder.add_gitlab_ci_section

(

self

)

Add gitlab_ci environment variable dictionary.

The section added is named "gitlab_ci". The variables collected from env() are:

• ci_runner_id
• ci_runner_version
• ci_project_id
• ci_project_name
• ci_server_fqdn
• ci_server_version
• ci_job_id
• ci_job_started_at
• ci_pipeline_id
• ci_pipeline_source
• ci_commit_sha
• gitlab_user_login

Where the values are strings from the corresponding env() values.

    def add_gitlab_ci_section(self):
        """! @brief Add gitlab_ci environment variable dictionary.
 
            The section added is named "gitlab_ci".
            The variables collected from env() are:
            - ci_runner_id
            - ci_runner_version
            - ci_project_id
            - ci_project_name
            - ci_server_fqdn
            - ci_server_version
            - ci_job_id
            - ci_job_started_at
            - ci_pipeline_id
            - ci_pipeline_source
            - ci_commit_sha
            - gitlab_user_login
 
            Where the values are strings from the corresponding env() values.
        """
        gitlab_ci_names = [
            "CI_RUNNER_ID",
            "CI_RUNNER_VERSION",
            "CI_PROJECT_ID",
            "CI_PROJECT_NAME",
            "CI_SERVER_FQDN",
            "CI_SERVER_VERSION",
            "CI_JOB_ID",
            "CI_JOB_STARTED_AT",
            "CI_PIPELINE_ID",
            "CI_PIPELINE_SOURCE",
            "CI_COMMIT_SHA",
            "GITLAB_USER_LOGIN"
        ]
        b = Builder()
        for i in gitlab_ci_names:
            v = os.getenv(i)
            if v:
                b._set_attr(i.lower(), v)
            else:
                b._set_attr(i.lower(), "")
        self.add_section("gitlab_ci", b)
 

add_header_section()

adctk.builder.Builder.add_header_section	(		self,
		str	application_name
	)

Create or replace the 'header' section using application_name.

To reuse a Builder object for a new message, call add_header_section on it again before sending it to publish again.

    def add_header_section(self, application_name: str):
        """! @brief Create or replace the 'header' section using application_name.
 
        To reuse a Builder object for a new message, call add_header_section on it again
        before sending it to publish again."""
        header = Header(application_name)
        self.add_section("header", header)
 

add_host_section()

adctk.builder.Builder.add_host_section	(		self,
			osinfo = False,
			ramsize = False,
			env = False,
			cpu = False,
			gpu = False,
			numa = False,
			all_hs = False
	)

Add the 'host' section, with optional subgroups of properties.

builder.add_host_section(os=True, ramsize=True, env=True, cpu=True, gpu=True, numa=True) is equivalent to builder.add_host_section(all_hs=True)

                        env=False, cpu=False, gpu=False, numa=False, all_hs=False):
        """! Add the 'host' section, with optional subgroups of properties.
 
        builder.add_host_section(os=True, ramsize=True, env=True, cpu=True, gpu=True, numa=True)
        is equivalent to builder.add_host_section(all_hs=True)
        """
        if all_hs:
            osinfo=True
            ramsize=True
            env=True
            cpu=True
            gpu=True
            numa=True
        host = Host(osinfo, ramsize, env, cpu, gpu, numa)
        self.add_section("host", host)
 

add_json_string()

adctk.builder.Builder.add_json_string	(		self,
		str	name,
		str	js
	)

Add stringified json.

    def add_json_string(self, name: str, js: str ):
        """! Add stringified json"""
        if js is None or isinstance(js, str):
            self._set_attr(name,
                    { "type": adc_types.scalar_to_json(
                        adc_types.ScalarType.cp_json_str),
                    "value": js })
        else:
            raise ValueError("add_json_string: json must be a string")
 

add_memory_usage_section()

adctk.builder.Builder.add_memory_usage_section

(

self

)

Populate "memory_usage" section with current host /proc/meminfo data in the style of free(1).

Values included are: mem_total mem_used mem_free mem_shared mem_buffers mem_cache mem_available swap_total swap_used swap_free

    def add_memory_usage_section(self):
        """! @brief Populate "memory_usage" section with current host /proc/meminfo data
            in the style of free(1).
 
            Values included are:
            mem_total mem_used mem_free mem_shared
            mem_buffers mem_cache mem_available
            swap_total swap_used swap_free
        """
        memory = Memory()
        self.add_section("memory_usage", memory)
 

add_mime()

adctk.builder.Builder.add_mime	(		self,
		str	name,
		str	mime_type,
		str	encoding,
		str	file_name,
		str	data
	)

add a string encoded object (e.g.

image)

Parameters

name	the name in the message of this field
mime_type	e.g. "image/png"
encoding	e.g. "base64"
file_name	e.g. "origin.png"
data	character array of the encoded object.

    def add_mime(self, name:str, mime_type:str, encoding:str, file_name:str, data: str):
        """! @brief add a string encoded object (e.g. image)
 
            @param name the name in the message of this field
            @param mime_type e.g. "image/png"
            @param encoding e.g. "base64"
            @param file_name e.g. "origin.png"
            @param data character array of the encoded object.
        """
        if not name or not mime_type or not encoding or not file_name or not data:
            return
        self._set_attr(name, { "type": adc_types.scalar_to_json(
            adc_types.ScalarType.cp_mime),
            "mimetype": mime_type,
            "encoding": encoding,
            "filename": file_name,
            "value": data })
 
 

add_model_data_section()

adctk.builder.Builder.add_model_data_section	(		self,
		'Builder'	model_data
	)

populate application run-time physics (re)configuration or result to "model_data" section.

For example initial or changes in mesh or particle decomp go here.

    def add_model_data_section(self, model_data: 'Builder'):
        """!  @brief populate application run-time physics (re)configuration or
            result to "model_data" section.
            For example initial or changes in mesh or particle decomp go here.
        """
        if isinstance(model_data, Builder):
            self.add_section("model_data", model_data)
        else:
            raise TypeError("argument model_data must be from type adctk.Builder")
 

add_mpi_section()

adctk.builder.Builder.add_mpi_section	(		self,
			name,
			communicator,
		bool	rank = False,
		bool	size = False,
		bool	mpi_name = False,
		bool	hostlist = False,
		bool	rank_host = False,
		bool	ver = False,
		bool	lib_ver = False,
		bool	mpiall = False,
		bool	local = False
	)

Unimplemented: python with mpi is not yet supported.

                        mpiall:bool = False, local:bool = False):
        """! Unimplemented: python with mpi is not yet supported."""
        if mpiall:
            local = True
            hostlist = True
            rank_host = True
        if local:
            rank = True
            size = True
            mpi_name = True
            ver = True
            lib_ver = True
        b = Builder()
        # fixme: shall we assume add_mpi_section items from mpi4py communicator?
        # check that communicator is the type from mpi4py.
        # default to 1-rank assumptions
        if rank:
            b._set_attr("mpi_rank", 0)
        if size:
            b._set_attr("mpi_size", 1)
        if mpi_name:
            b._set_attr("mpi_name", "unimplemented")
        if ver:
            # major.minor
            b._set_attr("mpi_version", "unimplemented")
        if lib_ver:
            # vmajor.vminor.vrelease...
            b._set_attr("mpi_library_version", "unimplemented")
        if rank_host:
            b._set_attr("mpi_rank_host", [])
        if hostlist:
            b._set_attr("mpi_hostlist", [])
        self.add_section("mpi_comm_" + name, b)
 

add_number_string()

adctk.builder.Builder.add_number_string	(		self,
		str	name,
		decimal.Decimal	dec
	)

Add stringified number of arbitrary precision.

    def add_number_string(self, name: str, dec: decimal.Decimal ):
        """! Add stringified number of arbitrary precision"""
        decstr = str(dec)
        self._set_attr(name,
                    {"type": adc_types.scalar_to_json(adc_types.ScalarType.cp_number_str),
                    "value": decstr })
 

add_path()

adctk.builder.Builder.add_path	(		self,
			name,
		str	path
	)

Add string file path.

    def add_path(self, name, path:str ):
        """! Add string file path."""
        if path is None or isinstance(path, str):
            self._set_attr(name,
                    { "type": adc_types.scalar_to_json(adc_types.ScalarType.cp_path),
                    "value": path })
        else:
            raise ValueError("add_path: path must be a string")
 

add_section()

adctk.builder.Builder.add_section	(		self,
		str	name,
		'Builder|collections.abc.Mapping[str,...]'	section
	)

Add or replace a user-defined, named section.

Using a string-keyed dictionary instead of a Builder is allowed but is not recommended for numeric or path or markup data because this omits variable type information needed for correct extraction in languages other than python. An exception occurs replacing a key/value pair with a section is attempted.

                section: 'Builder|collections.abc.Mapping[str,...]'):
        """! @brief Add or replace a user-defined, named section.
 
        Using a string-keyed dictionary instead of a Builder is allowed but is
        not recommended for numeric or path or markup data because
        this omits variable type information needed for correct extraction in
        languages other than python.
        An exception occurs replacing a key/value pair with a section is attempted.
        """
        if name in self._d:
            raise TypeError("attempting to replace a key/value with a section. Not Allowed. {}: {}".format(name, section))
        if isinstance(section, Builder) and isinstance(name, str):
            self._sections[name] = section
        elif isinstance(section, collections.abc.Mapping):
            b = Builder()
            for i,j in section.items():
                if isinstance(i, str):
                    if isinstance(j,Builder):
                        b.add_section(i, j)
                    else:
                        b._set_attr(i,j)
            self._sections[name] = b
        else:
            raise TypeError("arguments name,section must be of types str," +
                f"adctk.Builder,Mapping not {type(name)},{type(section)}")
 

add_slurm_section()

adctk.builder.Builder.add_slurm_section	(		self,
		collections.abc.Iterable[str]|None	output_vars = None
	)

Add slurm output environment variable dictionary elements.

The section added is named "slurm".

The variables collected from env() are:

• cluster: SLURM_CLUSTER_NAME
• job_id: SLURM_JOB_ID
• num_nodes: SLURM_JOB_NUM_NODES
• dependency: SLURM_JOB_DEPENDENCY

Where the values are strings from the corresponding env() values.

    def add_slurm_section(self, output_vars: collections.abc.Iterable[str]|None = None):
        """! @brief Add slurm output environment variable dictionary elements.
 
            The section added is named "slurm".
 
            The variables collected from env() are:
            - cluster: SLURM_CLUSTER_NAME
            - job_id: SLURM_JOB_ID
            - num_nodes: SLURM_JOB_NUM_NODES
            - dependency: SLURM_JOB_DEPENDENCY
 
            Where the values are strings from the corresponding env() values.
        """
        b = Builder()
        for n,e in [
            ( "cluster", "SLURM_CLUSTER_NAME"),
            ( "job_id", "SLURM_JOB_ID"),
            ( "num_nodes", "SLURM_JOB_NUM_NODES"),
            ( "dependency", "SLURM_JOB_DEPENDENCY") ]:
            v = os.getenv(e)
            if v:
                b._set_attr(n, v)
            else:
                b._set_attr(n, "")
        for e in output_vars:
            if isinstance(e,str):
                v = os.getenv(e)
                if v:
                    b._set_attr(e, v)
                else:
                    b._set_attr(e, "")
        self.add_section("slurm", b)
 

add_workflow_children()

adctk.builder.Builder.add_workflow_children	(		self,
		collections.abc.Iterable[str]	child_uuids
	)

Add list of child uuids to "adc_workflow" section after add_workflow_section has been called.

This call is optional.

This call may be repeated if necessary, incrementally building the child list. wfid_children: [ user defined list of ids ]

Where a workflow can track its immediate children, it may substantially improve downstream workflow analyses if the child items can be captured. The result appears in the resulting json as wfid_children: $ADC_WFID_CHILDREN

    def add_workflow_children(self, child_uuids: collections.abc.Iterable[str]):
        """! @brief Add list of child uuids to "adc_workflow" section
            after add_workflow_section has been called. This call is optional.
 
            This call may be repeated if necessary, incrementally building the child list.
            wfid_children: [ user defined list of ids ]
 
            Where a workflow can track its immediate children, it may substantially
            improve downstream workflow analyses if the child items can be captured.
            The result appears in the resulting json as
                wfid_children: $ADC_WFID_CHILDREN
        """
        wsec = self.get_section("adc_workflow")
        if wsec:
            cl = wsec._get_attr("wfid_children")
            if cl:
                if isinstance(cl, list):
                    for i in child_uuids:
                        if isinstance(i, str) and not i in cl:
                            cl.append(i)
                    wsec._set_attr("wfid_children", cl)
                else:
                    raise TypeError("adc_workflow.wfid_children found but isn't an array (?)")
            else:
                cl = []
                for i in child_uuids:
                    if isinstance(i, str) and not i in cl:
                        cl.append(i)
                wsec._set_attr("wfid_children", cl)
        else:
            raise APISequenceError("add_workflow_children called before add_workflow")
 

add_workflow_section()

adctk.builder.Builder.add_workflow_section	(		self,
		str|None	wfid = None,
		str|None	parent = None,
		list[str]|str|None	children = None
	)

add data from adc_wfid_ environment variables.

The section name is "adc_workflow".

The env variables collected are: wfid: ADC_WFID wfid_parent: $ADC_WFID_PARENT wfid_path: $ADC_WFID_PATH

The suggested format of an adc workflow identifier (wfid) is as: uuid -v1 -F STR run at the appropriate scope. For example, when starting numerous processes with mpi under slurm, in the sbatch script before launching anything else do: export ADC_WFID=$(uuid -v1 -F STR) and then make sure it gets propagated to all the processes via the launch mechamism. This ties all the messages from mpi ranks together in adc.

Where a workflow parent (such as an agent launching multiple slurm jobs) can, export ADC_WFID_PARENT=$(uuid -v1 -F STR) and then make sure this value is propagated to the slurm environments.

Where possible (requires coordination at all workflow levels) export ADC_WFID_PATH=(higher_level_wfid_path)/$ADC_WFID the entire task hierarchy identifier can be collected.

The choice of uuid version is up to the user.

                            children: list[str]|str|None =None):
        """! @brief add data from adc_wfid_ environment variables.
 
            The section name is "adc_workflow".
 
            The env variables collected are:
            wfid: ADC_WFID
            wfid_parent: $ADC_WFID_PARENT
            wfid_path: $ADC_WFID_PATH
 
            The suggested format of an adc workflow identifier (wfid) is as:
                uuid -v1 -F STR
            run at the appropriate scope.
            For example, when starting numerous processes with mpi under slurm,
            in the sbatch script before launching anything else do:
                export ADC_WFID=$(uuid -v1 -F STR)
            and then make sure it gets propagated to all the processes via the launch
            mechamism. This ties all the messages from mpi ranks together in adc.
 
            Where a workflow parent (such as an agent launching multiple slurm jobs)
            can,
                export ADC_WFID_PARENT=$(uuid -v1 -F STR)
            and then make sure this value is propagated to the slurm environments.
 
            Where possible (requires coordination at all workflow levels)
                export ADC_WFID_PATH=(higher_level_wfid_path)/$ADC_WFID
            the entire task hierarchy identifier can be collected.
 
            The choice of uuid version is up to the user.
        """
 
        b = Builder()
        for n,e in [
                ( "wfid", "ADC_WFID"),
                ( "wfid_parent", "ADC_WFID_PARENT"),
                ( "wfid_path", "ADC_WFID_PATH")
            ]:
            v = os.getenv(e)
            if v:
                b._set_attr(n, v)
            else:
                b._set_attr(n, "")
        cenv = os.getenv("ADC_WFID_CHILDREN")
        if cenv:
            echildren = list(filter(None,cenv.split(":")))
            b._set_attr("wfid_children", echildren)
        # now override from arguments anything found in the environment
        if children:
            if isinstance(children, list):
                clist = []
                for i in children:
                    if isinstance(i, str) and len(i):
                        clist.append(i)
                b._set_attr("wfid_children", echildren)
            if isinstance(children, str):
                echildren = list(filter(None, children.split(":")))
                b._set_attr("wfid_children", echildren)
        if parent and len(parent):
            b._set_attr("wfid_parent", parent)
        if wfid and len(wfid):
            b._set_attr("wfid", parent)
        self.add_section("adc_workflow", b)
 

add_xml_string()

adctk.builder.Builder.add_xml_string	(		self,
		str	name,
		str	xml
	)

Add stringified xml.

    def add_xml_string(self, name: str, xml: str ):
        """! Add stringified xml"""
        if xml is None or isinstance(xml, str):
            self._set_attr(name,
                    {"type": adc_types.scalar_to_json(adc_types.ScalarType.cp_xml_str),
                    "value": xml })
        else:
            raise ValueError("add_xml_string: xml must be a string")
 

add_yaml_string()

adctk.builder.Builder.add_yaml_string	(		self,
		str	name,
		str	yaml
	)

Add stringified yaml.

    def add_yaml_string(self, name: str, yaml: str ):
        """! Add stringified yaml"""
        if yaml is None or isinstance(yaml, str):
            self._set_attr(name,
                    { "type": adc_types.scalar_to_json(
                        adc_types.ScalarType.cp_yaml_str),
                    "value": yaml })
        else:
            raise ValueError("add_yaml_string: yaml must be a string")
 

BasicBuilder()

'Builder' adctk.builder.Builder.BasicBuilder ( str  app_name )

static

DEPRECATED.

    def BasicBuilder(app_name: str) -> 'Builder':
        """! DEPRECATED"""
        # pylint: disable=C0103
        b = Builder()
        b.add_header_section(app_name)
        return b
 

get_section()

'Builder|None' adctk.builder.Builder.get_section	(		self,
		str	name
	)

get the existing named section.

Returns

the section, or an empty pointer if it doesn't exist.

    def get_section(self, name: str) -> 'Builder|None':
        """! @brief get the existing named section.
 
            @return the section, or an empty pointer if it doesn't exist.
        """
        if name in self._sections:
            return self._sections[name]
        return None
 

get_section_names()

'list[str]' adctk.builder.Builder.get_section_names

(

self

)

Get the names of sections.

Returns

list of names, or an empty list.

    def get_section_names(self) -> 'list[str]':
        """! @brief Get the names of sections
 
             @return list of names, or an empty list.
         """
        return self._sections.keys()
 

get_value()

adc_types.Field adctk.builder.Builder.get_value	(		self,
		str	path_full
	)

get the existing named nested field in the builder.

Parameters

path_full

a simple json path such as /a/b/c which resolves to a value added via one of the add* functions.

Returns

the field description, with kt==k_none if not found. If the value was not set via this interface, k_none may be returned. Leading / is optional in path_full.

    def get_value(self, path_full: str) -> adc_types.Field:
        """! @brief get the existing named nested field in the builder.
 
            @param path_full a simple json path such as /a/b/c which resolves to
                   a value added via one of the add* functions.
            @return the field description, with kt==k_none if not found.
            If the value was not set via this interface, k_none may be returned.
            Leading / is optional in path_full.
 
        """
        field = adc_types.Field()
        path = path_full.strip('/')
        
        pos = path.find('/')
        if pos == -1:
            name = path
        else:
            name = path[:pos]
        
        kt = self._kind(name)
        
        child = path[pos:]
        jit = None
        match kt:
            case adc_types.KeyType.k_none:
                
                return field
            case adc_types.KeyType.k_section:
                
                return self._sections[name].get_value(child)
            case adc_types.KeyType.k_value:
                
                jit = self._d[name]
                if not jit:
                    
                    return field
        field.kt = kt
        if isinstance(jit, dict):
            
            v = None
            if "value" in jit:
                v = jit["value"]
            if "type" in jit and isinstance(jit["type"], str):
                type_name = jit["type"]
            else:
                return field # not defined by api
            if len(type_name):
                if "container_type" in jit:
                    c = jit["container_type"]
                else:
                    c = None
                st = adc_types.scalar_from_json(type_name)
                field.st = st
                field.data = v
                if c:
                    field.container = c
                
                return field
        if isinstance(jit, str):
            field.st = adc_types.ScalarType.cp_cstr
            field.data = jit
            
            return field
        if isinstance(jit, bool):
            field.st = adc_types.ScalarType.cp_bool
            field.data = jit
            
            return field
        if isinstance(jit, float):
            field.st = adc_types.ScalarType.cp_f64
            field.data = jit
            
            return field
        if isinstance(jit, int):
            if adc_types.fits_int64(jit):
                field.st = adc_types.ScalarType.cp_int64
                field.data = jit
                
                return field
            if adc_types.fits_uint64(jit):
                field.st = adc_types.ScalarType.cp_uint64
                field.data = jit
                
                return field
            field.st = adc_types.ScalarType.cp_number_str
            field.data = jit
            
            return field
        if isinstance(jit, decimal.Decimal):
            field.st = adc_types.type_to_scalar(type(jit), jit)
            field.data = jit
            
            return field
        field.data = jit
        field.st = adc_types.type_to_scalar(type(jit), jit)
        
        return field
 

get_value_int64()

adctk.builder.Builder.get_value_int64	(		self,
		str	path
	)

get the existing named nested scalar int64-equivalent value.

Parameters

path	a simple json path such as /a/b/c which resolves to a value added via one of the add* functions.

Returns

None if the path is not matched or int64_max if it is not int64 compatible.

    def get_value_int64(self, path: str):
        """! @brief get the existing named nested scalar int64-equivalent value.
 
        @param path a simple json path such as /a/b/c which resolves to
               a value added via one of the add* functions.
        @return None if the path is not matched or int64_max if it is
               not int64 compatible.
        """
        field = self.get_value(path)
        if not field or field.kt != adc_types.KeyType.k_value or len(field.data) != 1:
            return None
        i = numpy.iinfo(numpy.int64).max
        match field.st:
            case adc_types.ScalarType.cp_bool:
                if field.data:
                    return 1
                return 0
            case adc_types.ScalarType.cp_char32  | \
                adc_types.ScalarType.cp_char16  | \
                adc_types.ScalarType.cp_char:
                return ord(field.data)
            case adc_types.ScalarType.cp_int8  | \
                adc_types.ScalarType.cp_int16  | \
                adc_types.ScalarType.cp_int32  | \
                adc_types.ScalarType.cp_int64:
                return field.data
            case _:
                return i
 

get_value_string()

adctk.builder.Builder.get_value_string	(		self,
		str	path
	)

get the existing named nested scalar string value.

Parameters

path	a simple json path such as /a/b/c which resolves to a value added via one of the add* functions.

Returns

None if the path is not matched or is not a string of some sort.

    def get_value_string(self, path: str):
        """! @brief get the existing named nested scalar string value.
 
        @param path a simple json path such as /a/b/c which resolves to
               a value added via one of the add* functions.
        @return None if the path is not matched or is not a string of some sort.
        """
        field = self.get_value(path)
        if (not field or field.kt != adc_types.KeyType.k_value or
                (field.container and len(field.container))):
            return None
        match field.st:
            case adc_types.ScalarType.cp_cstr  | \
                adc_types.ScalarType.cp_json_str  | \
                adc_types.ScalarType.cp_yaml_str  | \
                adc_types.ScalarType.cp_xml_str  | \
                adc_types.ScalarType.cp_json  | \
                adc_types.ScalarType.cp_path  | \
                adc_types.ScalarType.cp_number_str:
                return field.data
            case _:
                return None
 

get_value_uint64()

adctk.builder.Builder.get_value_uint64	(		self,
		str	path
	)

get the existing named nested scalar uint64-equivalent value.

Parameters

path	a simple json path such as /a/b/c which resolves to a uint64-equivalent value added via one of the add* functions.

Returns

None if the path is not matched or uint64_max if it is not uint64 compatible.

    def get_value_uint64(self, path: str):
        """! @brief get the existing named nested scalar uint64-equivalent value.
 
        @param path a simple json path such as /a/b/c which resolves to
               a uint64-equivalent value added via one of the add* functions.
        @return None if the path is not matched or uint64_max if it is
               not uint64 compatible.
        """
        field = self.get_value(path)
        if not field or field.kt != adc_types.KeyType.k_value or len(field.data) != 1:
            return None
        i = numpy.iinfo(numpy.uint64).max
        match field.st:
            case adc_types.ScalarType.cp_bool:
                if field.data:
                    return 1
                return 0
            case adc_types.ScalarType.cp_char32  | \
                adc_types.ScalarType.cp_char16  | \
                adc_types.ScalarType.cp_char:
                return ord(field.data)
            case adc_types.ScalarType.cp_uint8  | \
                adc_types.ScalarType.cp_uint16  | \
                adc_types.ScalarType.cp_uint32  | \
                adc_types.ScalarType.cp_uint64:
                return field.data
            case _:
                return i
 

serialize()

str adctk.builder.Builder.serialize

(

self

)

get json string of the builder.

Items within builder must already be json compatible; custom encoders/decoders for json to handle numpy and complex correctly are not provided.

    def serialize(self) -> str:
        """! @brief get json string of the builder.
 
        Items within builder must already be json compatible; custom
        encoders/decoders for json to handle numpy and complex correctly
        are not provided.
        """
        p = self._flatten()
        # pprint.pprint(self)
        return json.dumps(p)
 

Member Data Documentation

API_VERSION

adctk.builder.Builder.API_VERSION = adc_version.AdcVersion()._dict()

static

---

The documentation for this class was generated from the following file:

• src/adctk/builder.py