Documentation

ThermoDBHub

Bases: ThermoLink

ThermoDBHub class used to manage thermodynamic databases.

Source code in pyThermoLinkDB/docs/thermodbhub.py

class ThermoDBHub(ThermoLink):
    """ThermoDBHub class used to manage thermodynamic databases."""
    # vars
    _thermodb = {}
    _thermodb_rule = {}
    _hub = {}

    def __init__(self):
        # init super class
        super().__init__()

    @property
    def thermodb(self):
        return self._thermodb

    @property
    def thermodb_rule(self):
        return self._thermodb_rule

    @property
    def hub(self):
        return self._hub

    def items(self):
        '''
        Gets all items in thermodb link

        Returns
        -------
        list
            list of items in thermodb link
        '''
        try:
            return list(self._thermodb.keys())
        except Exception as e:
            raise Exception('Getting components failed!, ', e)

    def config_thermodb_rule(
        self,
        rule: Union[str, Path, Dict[str, Dict[str, Dict[str, str]]]],
        names: Optional[List[str]] = None,
        disp: bool = False
    ) -> bool | str:
        '''
        Configs thermodb rule defined for each component

        Parameters
        ----------
        rule: str | Path | dict
            config file path or content, file can be a `yml`, `md`, or `txt` file
            or a string content in the same format
        names: List[str]
            name of the record, optional, default is None
        disp: bool
            display the config log

        Returns
        -------
        log_info : list
            log info of the config

        Notes
        -----
        rule is a `yml file` format as:

        ```yaml
        EtOH:
            DATA:
                Pc: Pc1
                Tc: Tc2
                AcFa: AcFa3
            EQUATIONS:
                VAPOR-PRESSURE: VaPr
        MeOH:
            DATA:
                Pc: Pc
                Tc: Tc
                AcFa: AcFa
            EQUATIONS:
                VAPOR-PRESSURE: VaPr
                HEAT-CAPACITY: Cp_IG
        ```
        '''
        try:
            # log info
            log_info = ['Logging thermodb rule...']

            # SECTION: check rule
            if not rule:
                # log warning
                _log = 'No configuration file provided!'
                if disp:
                    print(_log)
                log_info.append(_log)

                # set
                log_info = '\n'.join(log_info)
                # return
                return log_info

            # SECTION: check file
            # check the rule is a file or content
            if isinstance(rule, str):
                # check if rule is a file
                if os.path.isfile(rule):
                    # ! load thermodb file
                    _ref = thermodb_file_loader(rule)
                else:
                    # ! parse thermodb content
                    _ref = thermodb_parser(rule)
            elif isinstance(rule, Path):
                # check if rule is a file
                if rule.is_file():
                    # ! load thermodb file
                    _ref = thermodb_file_loader(rule)
                else:
                    # ! parse thermodb content
                    _ref = thermodb_parser(rule.read_text())
            elif isinstance(rule, dict):
                # check if rule is a dict
                if rule:
                    # set _ref
                    _ref = rule
                else:
                    # log warning
                    _log = 'thermodb rule file is empty!'
                    if disp:
                        print(_log)
                    log_info.append(_log)

                    # set
                    log_info = '\n'.join(log_info)
                    # return
                    return log_info
            else:
                raise TypeError(
                    'rule should be a string or a file path!')

            # SECTION: analyze the reference
            # NOTE: check if _ref is None
            if _ref is None:
                _log = 'thermodb rule file is empty!'
                # log warning
                if disp:
                    print(_log)
                log_info.append(_log)

                # set
                log_info = '\n'.join(log_info)
                # return
                return log_info

            # NOTE: check config mode
            if names is None:
                # check name exists
                if _ref.keys():
                    # looping through
                    for key in _ref.keys():
                        # get record
                        record_thermodb_rule = _ref[key]

                        # check key exists
                        if (
                            key in self._thermodb.keys()
                        ):
                            # add
                            self._thermodb_rule[key].update(
                                record_thermodb_rule)

                            # check disp
                            _log = f'{key} thermodb rule successfully set.'
                            if disp:
                                print(_log)
                            log_info.append(_log)
                        else:
                            _log = f'{key} not found, no thermodb provided!'
                            # log warning
                            if disp:
                                print(_log)
                            log_info.append(_log)
                else:
                    # empty thermodb rule
                    _log = 'No thermodb rule found!'
                    # log warning
                    if disp:
                        print(_log)
                    log_info.append(_log)

                    # set
                    log_info = '\n'.join(log_info)

                    # return
                    return log_info
            else:
                # SECTION
                # check keys
                if not _ref.keys():
                    _log = 'No thermodb rule found!'
                    # log warning
                    if disp:
                        print(_log)
                    log_info.append(_log)

                    # set
                    log_info = '\n'.join(log_info)
                    # return
                    return log_info

                # looping through names
                for name in names:
                    # NOTE: check name exists
                    if name in _ref.keys():
                        # get record
                        record_thermodb_rule = _ref[name]

                        # check name exists
                        if name in self._thermodb.keys():
                            # looping through
                            self._thermodb_rule[name].update(
                                record_thermodb_rule)

                            # check disp
                            _log = f'{name} thermodb rule successfully registered.'
                            if disp:
                                print(_log)
                            log_info.append(_log)
                        else:
                            _log = f'{name} not found, no thermodb provided!'
                            if disp:
                                print(_log)
                            log_info.append(_log)
                    else:
                        _log = f'{name} not found, no thermodb provided!'
                        if disp:
                            print(_log)
                        log_info.append(_log)

            # NOTE: convert log_info to string
            if isinstance(log_info, list):
                log_info = '\n'.join(log_info)

            return log_info

        except Exception as e:
            raise Exception('Configuration failed!, ', e)

    def add_thermodb_rule(
        self,
        item: str,
        rules: dict
    ) -> bool | str:
        '''
        Adds or update a thermodb rule for a thermodb rule item.

        Parameters
        ----------
        item: str
            name of the record
        rules: dict
            thermodb rule dict for all components

        Returns
        -------
        bool
            True if success, False otherwise

        Notes
        -----
        - A dictionary file with the following format:

        ```python
        # example of thermodb rule
        thermodb_rule_CO2 = {
            'DATA': {
                'Pc': 'Pc1',
                'Tc': 'Tc1',
                'AcFa': 'AcFa1'
            },
            'EQUATIONS': {
                'vapor-pressure': 'VaPr1',
                'heat-capacity': 'Cp_IG1'
            }
        }

        # add thermodb rule for CO2
        thub1.add_thermodb_rule('CO2', thermodb_rule_CO2)
        ```
        '''
        try:
            # logger
            log_info = ['Logging thermodb rule...']

            # log res
            def log_res(x): return "\n".join(x)

            # SECTION: check item exists
            if item not in self._thermodb_rule.keys():
                # add item
                self._thermodb_rule[item] = {}
                # log
                log_ = f"{item} thermodb successfully set."
                # log warning
                log_info.append(log_)

            # SECTION: check rules
            if 'DATA' not in self._thermodb_rule[item].keys():
                # add DATA
                self._thermodb_rule[item]['DATA'] = {}

                # log
                log_ = f"{item} DATA successfully set."
                # log warning
                log_info.append(log_)

            if 'EQUATIONS' not in self._thermodb_rule[item].keys():
                # add EQUATIONS
                self._thermodb_rule[item]['EQUATIONS'] = {}

                # log
                log_ = f"{item} EQUATIONS successfully set."
                # log warning
                log_info.append(log_)

            # SECTION: check item exist
            if item not in self._thermodb_rule.keys():
                # log warning
                log_ = f"{item} is not in thermodb rule!"
                # log warning
                log_info.append(log_)

                # res
                return log_res(log_info)

            # LINK: add DATA
            if 'DATA' in rules.keys():
                # data
                data_ = rules['DATA']
                # check data
                if not isinstance(data_, dict):
                    raise Exception('DATA should be a dictionary!')

                # check data exist
                if not data_.keys():
                    raise Exception('DATA is empty!')

                # looping through data
                for k, v in data_.items():
                    # initialize
                    if k not in self._thermodb_rule[item].keys():
                        # add data (new record)
                        self._thermodb_rule[item]['DATA'][k] = {}

                    # add data (update record)
                    self._thermodb_rule[item]['DATA'][k] = v

                    # log
                    log_ = f"{item} {k} successfully set."
                    # log warning
                    log_info.append(log_)

            # LINK: add EQUATIONS
            if 'EQUATIONS' in rules.keys():
                # equations
                equations_ = rules['EQUATIONS']
                # check equations
                if not isinstance(equations_, dict):
                    raise Exception('EQUATIONS should be a dictionary!')

                # check equations exist
                if not equations_.keys():
                    raise Exception('EQUATIONS is empty!')

                # looping through equations
                for k, v in equations_.items():
                    # initialize
                    if k not in self._thermodb_rule[item].keys():
                        # add equations (new record)
                        self._thermodb_rule[item]['EQUATIONS'][k] = {}

                    # add equations (update record)
                    self._thermodb_rule[item]['EQUATIONS'][k] = v
                    # log
                    log_ = f"{item} {k} successfully set."
                    # log warning
                    log_info.append(log_)

            # NOTE: check if log_info is empty
            if len(log_info) == 1:
                # log warning
                log_ = "No thermodb rule found!"
                # log warning
                log_info.append(log_)

            # res
            return log_res(log_info)

        except Exception as e:
            raise Exception('Adding new rule failed!, ', e)

    def delete_thermodb_rule(
        self,
        name: str
    ) -> bool:
        '''
        Deletes an item from thermodb rule

        Parameters
        ----------
        name: str
            name of the record to be deleted

        Returns
        -------
        bool
            True if success, False otherwise
        '''
        try:
            # check key exist
            if name not in self._thermodb_rule.keys():
                # log warning
                print(f"{name} is not in thermodb rule!")
                return False

            # del
            self._thermodb_rule[name] = {}
            # res
            return True
        except Exception as e:
            raise Exception('Deleting rule failed!, ', e)

    def add_thermodb(
        self,
        name: str,
        data: CompBuilder,
        rules: Optional[Dict[str, Dict[str, str]]] = None
    ) -> bool:
        '''
        Adds new thermodb such as: CO2_thermodb

        Parameters
        ----------
        name: str
            name of the record
        data: CompBuilder
            data of the record
        rules: dict, optional
            thermodb rule for the record, default is None

        Returns
        -------
        True : bool
            success
        '''
        try:
            # NOTE: data
            if not isinstance(data, CompBuilder):
                logger.error('data should be a pyThermoDB.CompBuilder object!')
                return False

            # set thermodb
            self._thermodb[name] = data

            # NOTE: create thermodb rule
            self._thermodb_rule[name] = {}

            # check rules
            if rules is not None:
                # check rules is a dict
                if not isinstance(rules, dict):
                    logger.error('rules should be a dictionary!')
                    raise TypeError('rules should be a dictionary!')

                # check rules exist
                if not rules.keys():
                    logger.error('rules is empty!')
                    raise ValueError('rules is empty!')

                # add thermodb rule
                self.add_thermodb_rule(name, rules)
            # res
            return True
        except Exception as e:
            logger.error(f'Adding new record failed!, {e}')
            return False

    def update_thermodb(
        self,
        name,
        data: CompBuilder
    ) -> bool:
        '''
        Updates existing record

        Parameters
        ----------
        name: str
            name of the record
        data: pyThermoDB.CompBuilder
            data of the record

        Returns
        -------
        bool:
            True if success, False otherwise
        '''
        try:
            self._thermodb[name] = data
            # res
            return True
        except Exception as e:
            raise Exception('Updating record failed!, ', e)

    def delete_thermodb(self, name: str) -> bool:
        '''
        Deletes existing record

        Parameters
        ----------
        name: str
            name of the record

        Returns
        -------
        bool
            True if success, False otherwise
        '''
        try:
            del self._thermodb[name]
            # delete thermodb rule
            del self._thermodb_rule[name]
            # res
            return True
        except Exception as e:
            raise Exception('Deleting record failed!, ', e)

    def info_thermodb(self, name) -> dict:
        '''
        Gets info of thermodb

        Parameters
        ----------
        name: str
            name of the record

        Returns
        -------
        res : CompBuilder
            thermodb check
        '''
        try:
            # check key exist
            if name not in self._thermodb.keys():
                # raise
                raise Exception(f"{name} is not in thermodb!")

            # res
            res = self._thermodb[name].check()
            return res
        except Exception as e:
            raise Exception('Getting info of record failed!, ', e)

    def build(self):
        '''
        Builds `datasource` and `equationsource` for each component registered in thermodb

        Parameters
        ----------
        None

        Returns
        -------
        datasource : dict
            datasource including component data such as Tc, Pc, etc.
        equationsource : dict
            equationsource including component equations such as VAPOR-PRESSURE, etc.

        Notes
        -----
        - A dictionary contains data and equations

        ```python
        # CO2 data
        dt1_ = datasource['CO2']['Pc']
        print(type(dt1_))
        print(dt1_)

        # MeOH data
        dt2_ = datasource['MeOH']['Tc']
        print(type(dt2_))
        print(dt2_)

        # NRTL data
        dt3_ = datasource['NRTL']['alpha_i_j']
        print(type(dt3_))
        print(dt3_.ij("Alpha_methanol_ethanol"))

        # CO2 equation
        eq1_ = equationsource['CO2']['VaPr']
        print(type(eq1_))
        print(eq1_)
        print(eq1_.args)
        print(eq1_.cal(T=298.15))

        # nrtl equation
        eq2_ = equationsource['NRTL']['tau_i_j']
        print(type(eq2_))
        print(eq2_)
        print(eq2_.args)
        print(eq2_.cal(T=298.15))
        ```
        '''
        try:
            # components
            components = list(self._thermodb.keys())
            # datasource
            datasource = self._set_datasource(
                self._thermodb,
                self._thermodb_rule,
                components
            )
            # equationsource
            equationsource = self._set_equationsource(
                self._thermodb,
                self._thermodb_rule,
                components
            )

            # reset
            self._hub = {}
            # update hub

            # for each component
            for component in components:
                _data = datasource[component]
                _eq = equationsource[component]
                # save
                self._hub[component] = {**_data, **_eq}
            # res
            return datasource, equationsource
        except Exception as e:
            raise Exception('Building data/equation source failed!, ', e)

    def build_model_source(self) -> ModelSource:
        '''
        Builds model source containing data source and equation source for multiple components

        Parameters
        ----------
        None

        Returns
        -------
        ModelSource
            ModelSource containing data source and equation source for multiple components

        Notes
        -----
        - This function is a wrapper of `build` function and returns a `ModelSource` object
        '''
        try:
            # build
            datasource, equationsource = self.build()
            # res
            return ModelSource(
                data_source=datasource,
                equation_source=equationsource
            )
        except Exception as e:
            raise Exception('Building ModelSource failed!, ', e)

    def clean(self):
        '''
        Cleans thermodb including thermodb, thermodb_rule, and hub

        Parameters
        ----------
        None

        Returns
        -------
        None
        '''
        try:
            # reset
            self._thermodb = {}
            # reset
            self._thermodb_rule = {}
            # reset
            self._hub = {}

            # res
            return True
        except Exception as e:
            raise Exception('Cleaning data/equation source failed!, ', e)

    def check(self):
        '''
        Checks data/equation source

        Parameters
        ----------
        None

        Returns
        -------
        hub : dict
            hub data
        '''
        try:
            # check summary
            return generate_summary(self._hub)
        except Exception as e:
            raise Exception('Checking data/equation source failed!, ', e)

add_thermodb(name, data, rules=None)

Adds new thermodb such as: CO2_thermodb

Parameters

name: str name of the record data: CompBuilder data of the record rules: dict, optional thermodb rule for the record, default is None

Returns

True : bool success

Source code in pyThermoLinkDB/docs/thermodbhub.py

def add_thermodb(
    self,
    name: str,
    data: CompBuilder,
    rules: Optional[Dict[str, Dict[str, str]]] = None
) -> bool:
    '''
    Adds new thermodb such as: CO2_thermodb

    Parameters
    ----------
    name: str
        name of the record
    data: CompBuilder
        data of the record
    rules: dict, optional
        thermodb rule for the record, default is None

    Returns
    -------
    True : bool
        success
    '''
    try:
        # NOTE: data
        if not isinstance(data, CompBuilder):
            logger.error('data should be a pyThermoDB.CompBuilder object!')
            return False

        # set thermodb
        self._thermodb[name] = data

        # NOTE: create thermodb rule
        self._thermodb_rule[name] = {}

        # check rules
        if rules is not None:
            # check rules is a dict
            if not isinstance(rules, dict):
                logger.error('rules should be a dictionary!')
                raise TypeError('rules should be a dictionary!')

            # check rules exist
            if not rules.keys():
                logger.error('rules is empty!')
                raise ValueError('rules is empty!')

            # add thermodb rule
            self.add_thermodb_rule(name, rules)
        # res
        return True
    except Exception as e:
        logger.error(f'Adding new record failed!, {e}')
        return False

add_thermodb_rule(item, rules)

Adds or update a thermodb rule for a thermodb rule item.

Parameters

item: str name of the record rules: dict thermodb rule dict for all components

Returns

bool True if success, False otherwise

Notes

• A dictionary file with the following format:

# example of thermodb rule
thermodb_rule_CO2 = {
    'DATA': {
        'Pc': 'Pc1',
        'Tc': 'Tc1',
        'AcFa': 'AcFa1'
    },
    'EQUATIONS': {
        'vapor-pressure': 'VaPr1',
        'heat-capacity': 'Cp_IG1'
    }
}

# add thermodb rule for CO2
thub1.add_thermodb_rule('CO2', thermodb_rule_CO2)

Source code in pyThermoLinkDB/docs/thermodbhub.py

def add_thermodb_rule(
    self,
    item: str,
    rules: dict
) -> bool | str:
    '''
    Adds or update a thermodb rule for a thermodb rule item.

    Parameters
    ----------
    item: str
        name of the record
    rules: dict
        thermodb rule dict for all components

    Returns
    -------
    bool
        True if success, False otherwise

    Notes
    -----
    - A dictionary file with the following format:

    ```python
    # example of thermodb rule
    thermodb_rule_CO2 = {
        'DATA': {
            'Pc': 'Pc1',
            'Tc': 'Tc1',
            'AcFa': 'AcFa1'
        },
        'EQUATIONS': {
            'vapor-pressure': 'VaPr1',
            'heat-capacity': 'Cp_IG1'
        }
    }

    # add thermodb rule for CO2
    thub1.add_thermodb_rule('CO2', thermodb_rule_CO2)
    ```
    '''
    try:
        # logger
        log_info = ['Logging thermodb rule...']

        # log res
        def log_res(x): return "\n".join(x)

        # SECTION: check item exists
        if item not in self._thermodb_rule.keys():
            # add item
            self._thermodb_rule[item] = {}
            # log
            log_ = f"{item} thermodb successfully set."
            # log warning
            log_info.append(log_)

        # SECTION: check rules
        if 'DATA' not in self._thermodb_rule[item].keys():
            # add DATA
            self._thermodb_rule[item]['DATA'] = {}

            # log
            log_ = f"{item} DATA successfully set."
            # log warning
            log_info.append(log_)

        if 'EQUATIONS' not in self._thermodb_rule[item].keys():
            # add EQUATIONS
            self._thermodb_rule[item]['EQUATIONS'] = {}

            # log
            log_ = f"{item} EQUATIONS successfully set."
            # log warning
            log_info.append(log_)

        # SECTION: check item exist
        if item not in self._thermodb_rule.keys():
            # log warning
            log_ = f"{item} is not in thermodb rule!"
            # log warning
            log_info.append(log_)

            # res
            return log_res(log_info)

        # LINK: add DATA
        if 'DATA' in rules.keys():
            # data
            data_ = rules['DATA']
            # check data
            if not isinstance(data_, dict):
                raise Exception('DATA should be a dictionary!')

            # check data exist
            if not data_.keys():
                raise Exception('DATA is empty!')

            # looping through data
            for k, v in data_.items():
                # initialize
                if k not in self._thermodb_rule[item].keys():
                    # add data (new record)
                    self._thermodb_rule[item]['DATA'][k] = {}

                # add data (update record)
                self._thermodb_rule[item]['DATA'][k] = v

                # log
                log_ = f"{item} {k} successfully set."
                # log warning
                log_info.append(log_)

        # LINK: add EQUATIONS
        if 'EQUATIONS' in rules.keys():
            # equations
            equations_ = rules['EQUATIONS']
            # check equations
            if not isinstance(equations_, dict):
                raise Exception('EQUATIONS should be a dictionary!')

            # check equations exist
            if not equations_.keys():
                raise Exception('EQUATIONS is empty!')

            # looping through equations
            for k, v in equations_.items():
                # initialize
                if k not in self._thermodb_rule[item].keys():
                    # add equations (new record)
                    self._thermodb_rule[item]['EQUATIONS'][k] = {}

                # add equations (update record)
                self._thermodb_rule[item]['EQUATIONS'][k] = v
                # log
                log_ = f"{item} {k} successfully set."
                # log warning
                log_info.append(log_)

        # NOTE: check if log_info is empty
        if len(log_info) == 1:
            # log warning
            log_ = "No thermodb rule found!"
            # log warning
            log_info.append(log_)

        # res
        return log_res(log_info)

    except Exception as e:
        raise Exception('Adding new rule failed!, ', e)

build()

Builds datasource and equationsource for each component registered in thermodb

Parameters

None

Returns

datasource : dict datasource including component data such as Tc, Pc, etc. equationsource : dict equationsource including component equations such as VAPOR-PRESSURE, etc.

Notes

• A dictionary contains data and equations

# CO2 data
dt1_ = datasource['CO2']['Pc']
print(type(dt1_))
print(dt1_)

# MeOH data
dt2_ = datasource['MeOH']['Tc']
print(type(dt2_))
print(dt2_)

# NRTL data
dt3_ = datasource['NRTL']['alpha_i_j']
print(type(dt3_))
print(dt3_.ij("Alpha_methanol_ethanol"))

# CO2 equation
eq1_ = equationsource['CO2']['VaPr']
print(type(eq1_))
print(eq1_)
print(eq1_.args)
print(eq1_.cal(T=298.15))

# nrtl equation
eq2_ = equationsource['NRTL']['tau_i_j']
print(type(eq2_))
print(eq2_)
print(eq2_.args)
print(eq2_.cal(T=298.15))

Source code in pyThermoLinkDB/docs/thermodbhub.py

def build(self):
    '''
    Builds `datasource` and `equationsource` for each component registered in thermodb

    Parameters
    ----------
    None

    Returns
    -------
    datasource : dict
        datasource including component data such as Tc, Pc, etc.
    equationsource : dict
        equationsource including component equations such as VAPOR-PRESSURE, etc.

    Notes
    -----
    - A dictionary contains data and equations

    ```python
    # CO2 data
    dt1_ = datasource['CO2']['Pc']
    print(type(dt1_))
    print(dt1_)

    # MeOH data
    dt2_ = datasource['MeOH']['Tc']
    print(type(dt2_))
    print(dt2_)

    # NRTL data
    dt3_ = datasource['NRTL']['alpha_i_j']
    print(type(dt3_))
    print(dt3_.ij("Alpha_methanol_ethanol"))

    # CO2 equation
    eq1_ = equationsource['CO2']['VaPr']
    print(type(eq1_))
    print(eq1_)
    print(eq1_.args)
    print(eq1_.cal(T=298.15))

    # nrtl equation
    eq2_ = equationsource['NRTL']['tau_i_j']
    print(type(eq2_))
    print(eq2_)
    print(eq2_.args)
    print(eq2_.cal(T=298.15))
    ```
    '''
    try:
        # components
        components = list(self._thermodb.keys())
        # datasource
        datasource = self._set_datasource(
            self._thermodb,
            self._thermodb_rule,
            components
        )
        # equationsource
        equationsource = self._set_equationsource(
            self._thermodb,
            self._thermodb_rule,
            components
        )

        # reset
        self._hub = {}
        # update hub

        # for each component
        for component in components:
            _data = datasource[component]
            _eq = equationsource[component]
            # save
            self._hub[component] = {**_data, **_eq}
        # res
        return datasource, equationsource
    except Exception as e:
        raise Exception('Building data/equation source failed!, ', e)

build_model_source()

Builds model source containing data source and equation source for multiple components

Parameters

None

Returns

ModelSource ModelSource containing data source and equation source for multiple components

Notes

• This function is a wrapper of build function and returns a ModelSource object

Source code in pyThermoLinkDB/docs/thermodbhub.py

def build_model_source(self) -> ModelSource:
    '''
    Builds model source containing data source and equation source for multiple components

    Parameters
    ----------
    None

    Returns
    -------
    ModelSource
        ModelSource containing data source and equation source for multiple components

    Notes
    -----
    - This function is a wrapper of `build` function and returns a `ModelSource` object
    '''
    try:
        # build
        datasource, equationsource = self.build()
        # res
        return ModelSource(
            data_source=datasource,
            equation_source=equationsource
        )
    except Exception as e:
        raise Exception('Building ModelSource failed!, ', e)

check()

Checks data/equation source

Parameters

None

Returns

hub : dict hub data

Source code in pyThermoLinkDB/docs/thermodbhub.py

def check(self):
    '''
    Checks data/equation source

    Parameters
    ----------
    None

    Returns
    -------
    hub : dict
        hub data
    '''
    try:
        # check summary
        return generate_summary(self._hub)
    except Exception as e:
        raise Exception('Checking data/equation source failed!, ', e)

clean()

Cleans thermodb including thermodb, thermodb_rule, and hub

Parameters

None

Returns

None

Source code in pyThermoLinkDB/docs/thermodbhub.py

def clean(self):
    '''
    Cleans thermodb including thermodb, thermodb_rule, and hub

    Parameters
    ----------
    None

    Returns
    -------
    None
    '''
    try:
        # reset
        self._thermodb = {}
        # reset
        self._thermodb_rule = {}
        # reset
        self._hub = {}

        # res
        return True
    except Exception as e:
        raise Exception('Cleaning data/equation source failed!, ', e)

config_thermodb_rule(rule, names=None, disp=False)

Configs thermodb rule defined for each component

Parameters

rule: str | Path | dict config file path or content, file can be a yml, md, or txt file or a string content in the same format names: List[str] name of the record, optional, default is None disp: bool display the config log

Returns

log_info : list log info of the config

Notes

rule is a yml file format as:

EtOH:
    DATA:
        Pc: Pc1
        Tc: Tc2
        AcFa: AcFa3
    EQUATIONS:
        VAPOR-PRESSURE: VaPr
MeOH:
    DATA:
        Pc: Pc
        Tc: Tc
        AcFa: AcFa
    EQUATIONS:
        VAPOR-PRESSURE: VaPr
        HEAT-CAPACITY: Cp_IG

Source code in pyThermoLinkDB/docs/thermodbhub.py

def config_thermodb_rule(
    self,
    rule: Union[str, Path, Dict[str, Dict[str, Dict[str, str]]]],
    names: Optional[List[str]] = None,
    disp: bool = False
) -> bool | str:
    '''
    Configs thermodb rule defined for each component

    Parameters
    ----------
    rule: str | Path | dict
        config file path or content, file can be a `yml`, `md`, or `txt` file
        or a string content in the same format
    names: List[str]
        name of the record, optional, default is None
    disp: bool
        display the config log

    Returns
    -------
    log_info : list
        log info of the config

    Notes
    -----
    rule is a `yml file` format as:

    ```yaml
    EtOH:
        DATA:
            Pc: Pc1
            Tc: Tc2
            AcFa: AcFa3
        EQUATIONS:
            VAPOR-PRESSURE: VaPr
    MeOH:
        DATA:
            Pc: Pc
            Tc: Tc
            AcFa: AcFa
        EQUATIONS:
            VAPOR-PRESSURE: VaPr
            HEAT-CAPACITY: Cp_IG
    ```
    '''
    try:
        # log info
        log_info = ['Logging thermodb rule...']

        # SECTION: check rule
        if not rule:
            # log warning
            _log = 'No configuration file provided!'
            if disp:
                print(_log)
            log_info.append(_log)

            # set
            log_info = '\n'.join(log_info)
            # return
            return log_info

        # SECTION: check file
        # check the rule is a file or content
        if isinstance(rule, str):
            # check if rule is a file
            if os.path.isfile(rule):
                # ! load thermodb file
                _ref = thermodb_file_loader(rule)
            else:
                # ! parse thermodb content
                _ref = thermodb_parser(rule)
        elif isinstance(rule, Path):
            # check if rule is a file
            if rule.is_file():
                # ! load thermodb file
                _ref = thermodb_file_loader(rule)
            else:
                # ! parse thermodb content
                _ref = thermodb_parser(rule.read_text())
        elif isinstance(rule, dict):
            # check if rule is a dict
            if rule:
                # set _ref
                _ref = rule
            else:
                # log warning
                _log = 'thermodb rule file is empty!'
                if disp:
                    print(_log)
                log_info.append(_log)

                # set
                log_info = '\n'.join(log_info)
                # return
                return log_info
        else:
            raise TypeError(
                'rule should be a string or a file path!')

        # SECTION: analyze the reference
        # NOTE: check if _ref is None
        if _ref is None:
            _log = 'thermodb rule file is empty!'
            # log warning
            if disp:
                print(_log)
            log_info.append(_log)

            # set
            log_info = '\n'.join(log_info)
            # return
            return log_info

        # NOTE: check config mode
        if names is None:
            # check name exists
            if _ref.keys():
                # looping through
                for key in _ref.keys():
                    # get record
                    record_thermodb_rule = _ref[key]

                    # check key exists
                    if (
                        key in self._thermodb.keys()
                    ):
                        # add
                        self._thermodb_rule[key].update(
                            record_thermodb_rule)

                        # check disp
                        _log = f'{key} thermodb rule successfully set.'
                        if disp:
                            print(_log)
                        log_info.append(_log)
                    else:
                        _log = f'{key} not found, no thermodb provided!'
                        # log warning
                        if disp:
                            print(_log)
                        log_info.append(_log)
            else:
                # empty thermodb rule
                _log = 'No thermodb rule found!'
                # log warning
                if disp:
                    print(_log)
                log_info.append(_log)

                # set
                log_info = '\n'.join(log_info)

                # return
                return log_info
        else:
            # SECTION
            # check keys
            if not _ref.keys():
                _log = 'No thermodb rule found!'
                # log warning
                if disp:
                    print(_log)
                log_info.append(_log)

                # set
                log_info = '\n'.join(log_info)
                # return
                return log_info

            # looping through names
            for name in names:
                # NOTE: check name exists
                if name in _ref.keys():
                    # get record
                    record_thermodb_rule = _ref[name]

                    # check name exists
                    if name in self._thermodb.keys():
                        # looping through
                        self._thermodb_rule[name].update(
                            record_thermodb_rule)

                        # check disp
                        _log = f'{name} thermodb rule successfully registered.'
                        if disp:
                            print(_log)
                        log_info.append(_log)
                    else:
                        _log = f'{name} not found, no thermodb provided!'
                        if disp:
                            print(_log)
                        log_info.append(_log)
                else:
                    _log = f'{name} not found, no thermodb provided!'
                    if disp:
                        print(_log)
                    log_info.append(_log)

        # NOTE: convert log_info to string
        if isinstance(log_info, list):
            log_info = '\n'.join(log_info)

        return log_info

    except Exception as e:
        raise Exception('Configuration failed!, ', e)

delete_thermodb(name)

Deletes existing record

Parameters

name: str name of the record

Returns

bool True if success, False otherwise

Source code in pyThermoLinkDB/docs/thermodbhub.py

def delete_thermodb(self, name: str) -> bool:
    '''
    Deletes existing record

    Parameters
    ----------
    name: str
        name of the record

    Returns
    -------
    bool
        True if success, False otherwise
    '''
    try:
        del self._thermodb[name]
        # delete thermodb rule
        del self._thermodb_rule[name]
        # res
        return True
    except Exception as e:
        raise Exception('Deleting record failed!, ', e)

delete_thermodb_rule(name)

Deletes an item from thermodb rule

Parameters

name: str name of the record to be deleted

Returns

bool True if success, False otherwise

Source code in pyThermoLinkDB/docs/thermodbhub.py

def delete_thermodb_rule(
    self,
    name: str
) -> bool:
    '''
    Deletes an item from thermodb rule

    Parameters
    ----------
    name: str
        name of the record to be deleted

    Returns
    -------
    bool
        True if success, False otherwise
    '''
    try:
        # check key exist
        if name not in self._thermodb_rule.keys():
            # log warning
            print(f"{name} is not in thermodb rule!")
            return False

        # del
        self._thermodb_rule[name] = {}
        # res
        return True
    except Exception as e:
        raise Exception('Deleting rule failed!, ', e)

info_thermodb(name)

Gets info of thermodb

Parameters

name: str name of the record

Returns

res : CompBuilder thermodb check

Source code in pyThermoLinkDB/docs/thermodbhub.py

def info_thermodb(self, name) -> dict:
    '''
    Gets info of thermodb

    Parameters
    ----------
    name: str
        name of the record

    Returns
    -------
    res : CompBuilder
        thermodb check
    '''
    try:
        # check key exist
        if name not in self._thermodb.keys():
            # raise
            raise Exception(f"{name} is not in thermodb!")

        # res
        res = self._thermodb[name].check()
        return res
    except Exception as e:
        raise Exception('Getting info of record failed!, ', e)

items()

Gets all items in thermodb link

Returns

list list of items in thermodb link

Source code in pyThermoLinkDB/docs/thermodbhub.py

def items(self):
    '''
    Gets all items in thermodb link

    Returns
    -------
    list
        list of items in thermodb link
    '''
    try:
        return list(self._thermodb.keys())
    except Exception as e:
        raise Exception('Getting components failed!, ', e)

update_thermodb(name, data)

Updates existing record

Parameters

name: str name of the record data: pyThermoDB.CompBuilder data of the record

Returns

bool: True if success, False otherwise

Source code in pyThermoLinkDB/docs/thermodbhub.py

def update_thermodb(
    self,
    name,
    data: CompBuilder
) -> bool:
    '''
    Updates existing record

    Parameters
    ----------
    name: str
        name of the record
    data: pyThermoDB.CompBuilder
        data of the record

    Returns
    -------
    bool:
        True if success, False otherwise
    '''
    try:
        self._thermodb[name] = data
        # res
        return True
    except Exception as e:
        raise Exception('Updating record failed!, ', e)

ThermoLink

Source code in pyThermoLinkDB/docs/thermolink.py

class ThermoLink:

    def __init__(self):
        # load reference
        pass

    def _set_datasource(
        self,
        thermodb: dict,
        thermodb_rule: dict,
        components: list
    ) -> dict:
        '''
        Sets a datasource

        Parameters
        ----------
        thermodb : dict
            thermodb
        thermodb_rule : dict
            thermodb rule
        components : list
            list of components

        Returns
        -------
        datasource : dict
            datasource
        '''
        try:
            # datasource
            datasource = {}

            # looping through each component
            for component in components:

                # NOTE: check if component is in thermodb
                if component in thermodb:
                    # init datasource for component
                    datasource[component] = {}

                    # ! saved component data
                    # check properties -> data
                    data = list(
                        thermodb[component].check_properties().keys()
                    )

                    # check
                    if len(data) != 0:

                        # NOTE: looping through each data source (GENERAL)
                        for src in data:
                            # REVIEW: get data
                            src_ = thermodb[component].select(src)

                            # init
                            df_src = None

                            # check
                            if isinstance(src_, TableData):
                                # NOTE: set
                                df_src = src_.data_structure()

                                # ? take all columns (header)
                                header = df_src['COLUMNS'].tolist()

                                # ? take all symbols
                                symbols = df_src['SYMBOL'].tolist()
                            elif isinstance(src_, TableMatrixData):
                                # NOTE: set
                                matrix_symbol_ = src_.matrix_symbol

                                # take all symbols
                                if matrix_symbol_ is None:
                                    raise Exception(
                                        'Matrix symbol is None, ',
                                        component
                                    )

                                # ? get all symbols
                                symbols = matrix_symbol_

                                # ? take all columns (header)
                                header = []
                            else:
                                # log
                                logger.warning(
                                    f'Unknown data type {type(src_)} for component {component}'
                                )
                                symbols = []
                                header = []

                            # looping through item data
                            # SECTION: looking through each symbol
                            for symbol in symbols:
                                # check
                                if symbol is not None and symbol != 'None':
                                    # ! symbol
                                    symbol = str(symbol).strip()

                                    # ? get property value
                                    # check if TableData or TableMatrixData
                                    _val = src_.get_property(symbol) if \
                                        isinstance(src_, TableData) else src_

                                    # NOTE: check symbol rename is required
                                    if component in thermodb_rule.keys():
                                        # get thermodb rule
                                        _rules = thermodb_rule[component].get(
                                            'DATA',
                                            None
                                        )

                                        # check
                                        if _rules:
                                            # set
                                            if symbol in _rules.keys():
                                                # ! rename
                                                symbol = _rules[symbol]

                                    # LINK: update
                                    datasource[component][symbol] = _val

                            # SECTION: looking through each header
                            if df_src is not None:
                                # looping through each header
                                for header_ in header:
                                    # check
                                    if header_ is not None and header_ != 'None':
                                        # ! header
                                        header_ = str(header_).strip()

                                        # ! header index
                                        header_index = df_src['COLUMNS'].tolist().\
                                            index(header_)

                                        # ? find symbol
                                        symbol_ = df_src['SYMBOL'].tolist()[
                                            header_index
                                        ]

                                        # ! check symbol if None and '' and 'None'
                                        if (
                                            symbol_ is None or
                                            symbol_ == '' or
                                            symbol_ == 'None'
                                        ):
                                            continue

                                        # NOTE: check already set
                                        # if symbol_ in datasource[component].keys():
                                        #     # skip
                                        #     continue

                                        _val = src_.get_property(symbol_) if \
                                            isinstance(src_, TableData) else src_

                                        # NOTE: check symbol rename is required
                                        if component in thermodb_rule.keys():
                                            # get thermodb rule
                                            _rules = thermodb_rule[component].get(
                                                'DATA',
                                                None
                                            )

                                            # check
                                            if _rules:
                                                # set
                                                if header_ in _rules.keys():
                                                    # ! rename
                                                    symbol_ = _rules[header_]

                                        # LINK: update
                                        datasource[component][symbol_] = _val
                    else:
                        # no data registered
                        raise Exception(
                            'No data registered in thermodb, ', component)
            # res
            return datasource
        except Exception as e:
            raise Exception('Building datasource failed!, ', e)

    def _set_equationsource(
        self,
        thermodb: dict,
        thermodb_rule: dict,
        components: list
    ) -> dict:
        '''
        Sets equation source

        Parameters
        ----------
        thermodb : dict
            thermodb
        thermodb_rule : dict
            thermodb rule
        components : list
            list of components

        Returns
        -------
        datasource : dict
            datasource
        '''
        try:
            # datasource
            datasource = {}
            for component in components:
                if component in thermodb:
                    # set
                    datasource[component] = {}

                    # SECTION: setting based on key functions
                    # component registered data/equations
                    eq_data = list(
                        thermodb[component].check_functions().keys()
                    )

                    # check
                    if len(eq_data) != 0:
                        # looping through each equation data
                        for eq in eq_data:
                            # >> equation identifier
                            equation_identifier = str(eq).strip()
                            # ! initial symbol set
                            symbol = equation_identifier

                            # NOTE: select equation
                            _val = thermodb[component].select(eq)
                            # check
                            if not isinstance(_val, TableEquation):
                                logger.warning(
                                    f'Unknown equation type {type(_val)} for component {component}'
                                )
                                continue

                            # NOTE: get original symbol
                            original_symbols: list[str] = list(
                                _val.return_symbols.keys()
                            )

                            # NOTE: update equation symbol with rules
                            # check if component is in thermodb_rule
                            if component in thermodb_rule.keys():
                                # get thermodb rule
                                _rules = thermodb_rule[component].get(
                                    'EQUATIONS',
                                    None
                                )

                                # NOTE: verify _rules keys to set with equation identifier
                                if _rules:
                                    # ! keys (equation identifiers)
                                    keys_ = _rules.keys()
                                    # set
                                    if equation_identifier in keys_:
                                        # rename: use the symbol from rules
                                        symbol = _rules[equation_identifier]
                            # else:
                            #     # ! use original symbols
                            #     if len(original_symbols) > 0:
                            #         symbol = original_symbols[0]

                            # LINK: update
                            datasource[component][symbol] = _val
            # res
            return datasource
        except Exception as e:
            raise Exception('Building datasource failed!, ', e)