Derivation Client

PyDerivationClient ¶

PyDerivationClient(derivation_instance_base_url: str, query_endpoint: str, update_endpoint: str, kg_user: str = None, kg_password: str = None)

This is a wrapper class for the java class uk.ac.cam.cares.jps.base.derivation.DerivationClient. Only the methods that commonly used are wrapped here. For other methods, one may access them via self.derivation_client.nameOfJavaMethod(args).

Attributes:

Name	Type	Description
`derivation_client`		The Java DerivationClient object
`jpsBaseLib_view`		The JVM module view for accessing Java classes

Methods:

Name	Description
`createDerivation`	Create a normal derivation markup.
`createSyncDerivationForNewInfo`	Create a synchronous derivation for new info to be computed by the agent.
`createSyncDerivationForNewInfoWithHttpUrl`	Create a synchronous derivation for new info to be computed by the agent with the agent URL provided.
`bulkCreateDerivations`	Create multiple normal derivations in one go.
`createDerivationWithTimeSeries`	Create a time series derivation markup.
`bulkCreateDerivationsWithTimeSeries`	Create multiple time series derivations in one go.
`createAsyncDerivation`	Create an asynchronous derivation markup.
`createAsyncDerivationFromDerivation`	Create an asynchronous derivation markup from an existing derivation.
`bulkCreateAsyncDerivations`	Create multiple asynchronous derivations in one go.
`createAsyncDerivationForNewInfo`	Create an asynchronous derivation markup for new information.
`bulkCreateAsyncDerivationsForNewInfo`	Create multiple asynchronous derivations for new information in one go.
`validateDerivations`	Validate all derivations and their inputs are attached with timestamps, also no circular dependencies.
`createOntoAgentInstance`	Register an OntoAgent instance in the triple store.
`addTimeInstance`	Add a time instance to each entity in the list of entities.
`addTimeInstanceCurrentTimestamp`	Add time instance with current timestamp to the given entities.
`updateTimestamp`	Update the timestamp of the entity to the current time.
`updateTimestamps`	Update the timestamp of all entities in the list.
`dropTimestampsOf`	Drop the timestamp of the given entities.
`getDerivations`	Get the derivations of the given agent.
`getDerivationsOf`	Get the derivations of the given entities.
`unifiedUpdateDerivation`	Unified update derivation method.
`updatePureSyncDerivation`	Update a single synchronous derivation.
`updatePureSyncDerivations`	Update a list of synchronous derivations.
`updatePureSyncDerivationsInParallel`	Update a list of synchronous derivations in parallel.
`updateAllSyncDerivations`	Update all synchronous derivations within the knowledge graph.
`updateMixedAsyncDerivation`	Update a directed acyclic graph of a-/sync derivations.

Parameters:

Name	Type	Description	Default
`derivation_instance_base_url`	`str`	base url of the derivation instance to be created by the client	required
`query_endpoint`	`str`	query endpoint of the knowledge graph	required
`update_endpoint`	`str`	update endpoint of the knowledge graph	required
`kg_user`	`str`	username for the knowledge graph endpoint. Defaults to None.	`None`
`kg_password`	`str`	password for the knowledge graph endpoint. Defaults to None.	`None`

Source code in JPS_BASE_LIB/python_wrapper/twa/kg_operations/derivation_client.py

def __init__(
    self,
    derivation_instance_base_url: str,
    query_endpoint: str,
    update_endpoint: str,
    kg_user:str=None,
    kg_password:str=None
) -> None:
    """
    Initialise the derivation client

    Args:
        derivation_instance_base_url (str): base url of the derivation instance to be created by the client
        query_endpoint (str): query endpoint of the knowledge graph
        update_endpoint (str): update endpoint of the knowledge graph
        kg_user (str, optional): username for the knowledge graph endpoint. Defaults to None.
        kg_password (str, optional): password for the knowledge graph endpoint. Defaults to None.
    """

    # create a JVM module view and use it to import the required java classes
    self.jpsBaseLib_view = jpsBaseLibGW.createModuleView()
    jpsBaseLibGW.importPackages(self.jpsBaseLib_view,"uk.ac.cam.cares.jps.base.query.*")
    jpsBaseLibGW.importPackages(self.jpsBaseLib_view,"uk.ac.cam.cares.jps.base.derivation.*")

    # create store_client
    if kg_user is None:
        store_client = self.jpsBaseLib_view.RemoteStoreClient(query_endpoint, update_endpoint)
    else:
        store_client = self.jpsBaseLib_view.RemoteStoreClient(
            query_endpoint, update_endpoint, kg_user, kg_password)

    # create derivation_client
    self.derivation_client = self.jpsBaseLib_view.DerivationClient(store_client, derivation_instance_base_url)

createDerivation ¶

createDerivation(entities: List[str], agentIRI: str, inputs: List[str]) -> str

Create a normal derivation markup.

Parameters:

Name	Type	Description	Default
`entities`	`List[str]`	List of entities that belongsTo the derivation	required
`agentIRI`	`str`	List of agents that the derivation isDerivedUsing	required
`inputs`	`List[str]`	List of inputs that the derivation isDerivedFrom	required

Returns:

Name	Type	Description
`str`	`str`	IRI of the created derivation

Source code in JPS_BASE_LIB/python_wrapper/twa/kg_operations/derivation_client.py

def createDerivation(
    self,
    entities: List[str],
    agentIRI: str,
    inputs: List[str]
) -> str:
    """
    Create a normal derivation markup.

    Args:
        entities (List[str]): List of entities that belongsTo the derivation
        agentIRI (str): List of agents that the derivation isDerivedUsing
        inputs (List[str]): List of inputs that the derivation isDerivedFrom

    Returns:
        str: IRI of the created derivation
    """
    return self.derivation_client.createDerivation(entities, agentIRI, inputs)

createSyncDerivationForNewInfo ¶

createSyncDerivationForNewInfo(agentIRI: str, inputsIRI: List[str], derivationType: str) -> Derivation

Create a sync derivation for new info to be computed by the agent.

Parameters:

Name	Type	Description	Default
`agentIRI`	`str`	IRI of the agent that the derivation isDerivedUsing	required
`inputsIRI`	`List[str]`	List of inputs that the derivation isDerivedFrom	required
`derivationType`	`str`	IRI of the synchronous derivation type to be created	required

Returns:

Name	Type	Description
`Derivation`	`Derivation`	Object of the created derivation

Source code in JPS_BASE_LIB/python_wrapper/twa/kg_operations/derivation_client.py

def createSyncDerivationForNewInfo(
    self,
    agentIRI: str,
    inputsIRI: List[str],
    derivationType: str
) -> Derivation:
    """
    Create a sync derivation for new info to be computed by the agent.

    Args:
        agentIRI (str): IRI of the agent that the derivation isDerivedUsing
        inputsIRI (List[str]): List of inputs that the derivation isDerivedFrom
        derivationType (str): IRI of the synchronous derivation type to be created

    Returns:
        Derivation: Object of the created derivation
    """
    derivation_java = self.derivation_client.createSyncDerivationForNewInfo(agentIRI, inputsIRI, derivationType)
    return Derivation(derivation_java=derivation_java)

createSyncDerivationForNewInfoWithHttpUrl ¶

createSyncDerivationForNewInfoWithHttpUrl(agentIRI: str, agentURL: str, inputsIRI: List[str], derivationType: str) -> Derivation

Create a sync derivation for new info to be computed by the agent with the agent url provided.

Parameters:

Name	Type	Description	Default
`agentIRI`	`str`	IRI of the agent that the derivation isDerivedUsing	required
`agentURL`	`str`	HTTP URL of the agent that the derivation isDerivedUsing	required
`inputsIRI`	`List[str]`	List of inputs that the derivation isDerivedFrom	required
`derivationType`	`str`	IRI of the synchronous derivation type to be created	required

Returns:

Name	Type	Description
`Derivation`	`Derivation`	Object of the created derivation

Source code in JPS_BASE_LIB/python_wrapper/twa/kg_operations/derivation_client.py

def createSyncDerivationForNewInfoWithHttpUrl(
    self,
    agentIRI: str,
    agentURL: str,
    inputsIRI: List[str],
    derivationType: str
) -> Derivation:
    """
    Create a sync derivation for new info to be computed by the agent with the agent url provided.

    Args:
        agentIRI (str): IRI of the agent that the derivation isDerivedUsing
        agentURL (str): HTTP URL of the agent that the derivation isDerivedUsing
        inputsIRI (List[str]): List of inputs that the derivation isDerivedFrom
        derivationType (str): IRI of the synchronous derivation type to be created

    Returns:
        Derivation: Object of the created derivation
    """
    derivation_java = self.derivation_client.createSyncDerivationForNewInfo(agentIRI, agentURL, inputsIRI, derivationType)
    return Derivation(derivation_java=derivation_java)

bulkCreateDerivations ¶

bulkCreateDerivations(entitiesList: List[List[str]], agentIRIList: List[str], inputsList: List[List[str]]) -> List[str]

Create multiple normal derivations in one go.

Parameters:

Name	Type	Description	Default
`entitiesList`	`List[List[str]]`	List of list of entities that belongsTo the derivations	required
`agentIRIList`	`List[str]`	List of agents that the derivations isDerivedUsing	required
`inputsList`	`List[List[str]]`	List of list of inputs that the derivations isDerivedFrom	required

Returns:

Type	Description
`List[str]`	List[str]: List of IRIs of the created derivations

Source code in JPS_BASE_LIB/python_wrapper/twa/kg_operations/derivation_client.py

def bulkCreateDerivations(
    self,
    entitiesList: List[List[str]],
    agentIRIList: List[str],
    inputsList: List[List[str]]
) -> List[str]:
    """
    Create multiple normal derivations in one go.

    Args:
        entitiesList (List[List[str]]): List of list of entities that belongsTo the derivations
        agentIRIList (List[str]): List of agents that the derivations isDerivedUsing
        inputsList (List[List[str]]): List of list of inputs that the derivations isDerivedFrom

    Returns:
        List[str]: List of IRIs of the created derivations
    """
    return self.derivation_client.bulkCreateDerivations(entitiesList, agentIRIList, inputsList)

createDerivationWithTimeSeries ¶

createDerivationWithTimeSeries(entities: List[str], agentIRI: str, inputs: List[str]) -> str

Create a time series derivation markup.

Parameters:

Name	Type	Description	Default
`entities`	`List[str]`	List of entities that belongsTo the derivation	required
`agentIRI`	`str`	IRI of the agent that the derivation isDerivedUsing	required
`inputs`	`List[str]`	List of inputs that the derivation isDerivedFrom	required

Returns:

Name	Type	Description
`str`	`str`	IRI of the created time series derivation

Source code in JPS_BASE_LIB/python_wrapper/twa/kg_operations/derivation_client.py

def createDerivationWithTimeSeries(
    self,
    entities: List[str],
    agentIRI: str,
    inputs: List[str]
) -> str:
    """
    Create a time series derivation markup.

    Args:
        entities (List[str]): List of entities that belongsTo the derivation
        agentIRI (str): IRI of the agent that the derivation isDerivedUsing
        inputs (List[str]): List of inputs that the derivation isDerivedFrom

    Returns:
        str: IRI of the created time series derivation
    """
    return self.derivation_client.createDerivationWithTimeSeries(entities, agentIRI, inputs)

bulkCreateDerivationsWithTimeSeries ¶

bulkCreateDerivationsWithTimeSeries(entitiesList: List[List[str]], agentIRIList: List[str], inputsList: List[List[str]]) -> List[str]

Create multiple time series derivations in one go.

Parameters:

Name	Type	Description	Default
`entitiesList`	`List[List[str]]`	List of list of entities that belongsTo the derivations	required
`agentIRIList`	`List[str]`	List of agents that the derivations isDerivedUsing	required
`inputsList`	`List[List[str]]`	List of list of inputs that the derivations isDerivedFrom	required

Returns:

Type	Description
`List[str]`	List[str]: List of IRIs of the created time series derivations

Source code in JPS_BASE_LIB/python_wrapper/twa/kg_operations/derivation_client.py

def bulkCreateDerivationsWithTimeSeries(
    self,
    entitiesList: List[List[str]],
    agentIRIList: List[str],
    inputsList: List[List[str]]
) -> List[str]:
    """
    Create multiple time series derivations in one go.

    Args:
        entitiesList (List[List[str]]): List of list of entities that belongsTo the derivations
        agentIRIList (List[str]): List of agents that the derivations isDerivedUsing
        inputsList (List[List[str]]): List of list of inputs that the derivations isDerivedFrom

    Returns:
        List[str]: List of IRIs of the created time series derivations
    """
    return self.derivation_client.bulkCreateDerivationsWithTimeSeries(entitiesList, agentIRIList, inputsList)

createAsyncDerivation ¶

createAsyncDerivation(entities: List[str], agentIRI: str, inputs: List[str], forUpdate: bool) -> str

Create an asynchronous derivation markup. If forUpdate is True, the derivation will be marked as "Requested" with a timestamp of 0. Otherwise, the derivation will be marked without status but with a current timestamp.

Parameters:

Name	Type	Description	Default
`entities`	`List[str]`	List of entities that belongsTo the derivation	required
`agentIRI`	`str`	IRI of the agent that the derivation isDerivedUsing	required
`inputs`	`List[str]`	List of inputs that the derivation isDerivedFrom	required
`forUpdate`	`bool`	Boolean flag to indicate if the derivation markup is for update	required

Returns:

Name	Type	Description
`str`	`str`	IRI of the created asynchronous derivation

Source code in JPS_BASE_LIB/python_wrapper/twa/kg_operations/derivation_client.py

def createAsyncDerivation(
    self,
    entities: List[str],
    agentIRI: str,
    inputs: List[str],
    forUpdate: bool
) -> str:
    """
    Create an asynchronous derivation markup. If `forUpdate` is True, the derivation will be marked as "Requested"
    with a timestamp of 0. Otherwise, the derivation will be marked without status but with a current timestamp.

    Args:
        entities (List[str]): List of entities that belongsTo the derivation
        agentIRI (str): IRI of the agent that the derivation isDerivedUsing
        inputs (List[str]): List of inputs that the derivation isDerivedFrom
        forUpdate (bool): Boolean flag to indicate if the derivation markup is for update

    Returns:
        str: IRI of the created asynchronous derivation
    """
    return self.derivation_client.createAsyncDerivation(entities, agentIRI, inputs, forUpdate)

createAsyncDerivationFromDerivation ¶

createAsyncDerivationFromDerivation(entities: List[str], agentIRI: str, derivation: str, forUpdate: bool) -> str

Create an asynchronous derivation markup from an existing derivation. If forUpdate is True, the derivation will be marked as "Requested" with a timestamp of 0. Otherwise, the derivation will be marked without status but with a current timestamp.

Parameters:

Name	Type	Description	Default
`entities`	`List[str]`	List of entities that belongsTo the derivation	required
`agentIRI`	`str`	IRI of the agent that the derivation isDerivedUsing	required
`derivation`	`str`	IRI of an existing derivation whose outputs the new derivation isDerivedFrom	required
`forUpdate`	`bool`	Boolean flag to indicate if the derivation markup is for update	required

Returns:

Name	Type	Description
`str`	`str`	IRI of the created asynchronous derivation

Source code in JPS_BASE_LIB/python_wrapper/twa/kg_operations/derivation_client.py

def createAsyncDerivationFromDerivation(
    self,
    entities: List[str],
    agentIRI: str,
    derivation: str,
    forUpdate: bool
) -> str:
    """
    Create an asynchronous derivation markup from an existing derivation. If `forUpdate` is True, the derivation
    will be marked as "Requested" with a timestamp of 0. Otherwise, the derivation will be marked without status but
    with a current timestamp.

    Args:
        entities (List[str]): List of entities that belongsTo the derivation
        agentIRI (str): IRI of the agent that the derivation isDerivedUsing
        derivation (str): IRI of an existing derivation whose outputs the new derivation isDerivedFrom
        forUpdate (bool): Boolean flag to indicate if the derivation markup is for update

    Returns:
        str: IRI of the created asynchronous derivation
    """
    return self.derivation_client.createAsyncDerivation(entities, agentIRI, derivation, forUpdate)

bulkCreateAsyncDerivations ¶

bulkCreateAsyncDerivations(entitiesList: List[List[str]], agentIRIList: List[str], inputsList: List[List[str]], forUpdateFlagList: List[bool]) -> List[str]

Create multiple asynchronous derivations in one go. If the flag in forUpdateFlagList is True, the corresponding derivation will be marked as "Requested" with a timestamp of 0. Otherwise, the derivation will be marked without status but with a current timestamp.

Parameters:

Name	Type	Description	Default
`entitiesList`	`List[List[str]]`	List of list of entities that belongsTo the derivations	required
`agentIRIList`	`List[str]`	List of agents that the derivations isDerivedUsing	required
`inputsList`	`List[List[str]]`	List of list of inputs that the derivations isDerivedFrom	required
`forUpdateFlagList`	`List[bool]`	List of boolean flags to indicate if the derivation markup is for update	required

Returns:

Type	Description
`List[str]`	List[str]: List of IRIs of the created asynchronous derivations

Source code in JPS_BASE_LIB/python_wrapper/twa/kg_operations/derivation_client.py

def bulkCreateAsyncDerivations(
    self,
    entitiesList: List[List[str]],
    agentIRIList: List[str],
    inputsList: List[List[str]],
    forUpdateFlagList: List[bool]
) -> List[str]:
    """
    Create multiple asynchronous derivations in one go. If the flag in `forUpdateFlagList` is True, the corresponding
    derivation will be marked as "Requested" with a timestamp of 0. Otherwise, the derivation will be marked without
    status but with a current timestamp.

    Args:
        entitiesList (List[List[str]]): List of list of entities that belongsTo the derivations
        agentIRIList (List[str]): List of agents that the derivations isDerivedUsing
        inputsList (List[List[str]]): List of list of inputs that the derivations isDerivedFrom
        forUpdateFlagList (List[bool]): List of boolean flags to indicate if the derivation markup is for update

    Returns:
        List[str]: List of IRIs of the created asynchronous derivations
    """
    return self.derivation_client.bulkCreateAsyncDerivations(entitiesList, agentIRIList, inputsList, forUpdateFlagList)

createAsyncDerivationForNewInfo ¶

createAsyncDerivationForNewInfo(agentIRI: str, inputsAndDerivations: List[str]) -> str

Create an asynchronous derivation markup for new information. The derivation will be marked as "Requested" with a timestamp of 0. The outputs of the derivation will be computed by the agent in due course. Note that all IRIs in the inputsAndDerivations list will be directly connected to the derivation as its inputs. Therefore, existing IRIs of the derivation outputs should be provided if applicable, instead of the IRI of that derivation.

Parameters:

Name	Type	Description	Default
`agentIRI`	`str`	IRI of the agent that the derivation isDerivedUsing	required
`inputsAndDerivations`	`List[str]`	List of inputs and derivations that the derivation isDerivedFrom	required

Returns:

Name	Type	Description
`str`	`str`	IRI of the created asynchronous derivation for new information

Source code in JPS_BASE_LIB/python_wrapper/twa/kg_operations/derivation_client.py

def createAsyncDerivationForNewInfo(
    self,
    agentIRI: str,
    inputsAndDerivations: List[str]
) -> str:
    """
    Create an asynchronous derivation markup for new information. The derivation will be marked as "Requested" with
    a timestamp of 0. The outputs of the derivation will be computed by the agent in due course. Note that all IRIs in
    the `inputsAndDerivations` list will be directly connected to the derivation as its inputs. Therefore, existing IRIs
    of the derivation outputs should be provided if applicable, instead of the IRI of that derivation.

    Args:
        agentIRI (str): IRI of the agent that the derivation isDerivedUsing
        inputsAndDerivations (List[str]): List of inputs and derivations that the derivation isDerivedFrom

    Returns:
        str: IRI of the created asynchronous derivation for new information
    """
    return self.derivation_client.createAsyncDerivationForNewInfo(agentIRI, inputsAndDerivations)

bulkCreateAsyncDerivationsForNewInfo ¶

bulkCreateAsyncDerivationsForNewInfo(agentIRIList: List[str], inputsAndDerivationsList: List[List[str]]) -> List[str]

Create multiple asynchronous derivations for new information in one go. The derivations will be marked as "Requested" with a timestamp of 0. The outputs of the derivations will be computed by the agents in due course. Note that all IRIs in the inputsAndDerivationsList list will be directly connected to the derivations as their inputs. Therefore, existing IRIs of the derivation outputs should be provided if applicable, instead of the IRI of that derivation.

Parameters:

Name	Type	Description	Default
`agentIRIList`	`List[str]`	List of agents that the derivations isDerivedUsing	required
`inputsAndDerivationsList`	`List[List[str]]`	List of list of inputs and derivations that the derivations isDerivedFrom	required

Returns:

Type	Description
`List[str]`	List[str]: List of IRIs of the created asynchronous derivations for new information

Source code in JPS_BASE_LIB/python_wrapper/twa/kg_operations/derivation_client.py

def bulkCreateAsyncDerivationsForNewInfo(
    self,
    agentIRIList: List[str],
    inputsAndDerivationsList: List[List[str]]
) -> List[str]:
    """
    Create multiple asynchronous derivations for new information in one go. The derivations will be marked as
    "Requested" with a timestamp of 0. The outputs of the derivations will be computed by the agents in due course.
    Note that all IRIs in the `inputsAndDerivationsList` list will be directly connected to the derivations as their
    inputs. Therefore, existing IRIs of the derivation outputs should be provided if applicable, instead of the IRI
    of that derivation.

    Args:
        agentIRIList (List[str]): List of agents that the derivations isDerivedUsing
        inputsAndDerivationsList (List[List[str]]): List of list of inputs and derivations that the derivations isDerivedFrom

    Returns:
        List[str]: List of IRIs of the created asynchronous derivations for new information
    """
    return self.derivation_client.bulkCreateAsyncDerivationsForNewInfo(agentIRIList, inputsAndDerivationsList)

validateDerivations ¶

validateDerivations() -> bool

This checks for any circular dependency and ensures that all the linked inputs have a suitable timestamp attached. NOTE however, this method does not check for everything, e.g. instances having appropriate rdf:types, and the agent design.

Returns:

Name	Type	Description
`bool`	`bool`	Whether the created derivations are valid, i.e. no circular dependency, timestamps correctly attached

Source code in JPS_BASE_LIB/python_wrapper/twa/kg_operations/derivation_client.py

def validateDerivations(self) -> bool:
    """
    This checks for any circular dependency and ensures that all the linked inputs have a suitable timestamp attached.
    NOTE however, this method does not check for everything, e.g. instances having appropriate rdf:types, and the agent design.

    Returns:
        bool: Whether the created derivations are valid, i.e. no circular dependency, timestamps correctly attached
    """
    return self.derivation_client.validateDerivations()

createOntoAgentInstance ¶

createOntoAgentInstance(ontoAgentServiceIRI: str, ontoAgentOperationHttpUrl: str, inputTypes: List[str], outputTypes: List[str])

Register an OntoAgent instance in the triple store.

Parameters:

Name	Type	Description	Default
`ontoAgentServiceIRI`	`str`	IRI of the agent's OntoAgent:Service	required
`ontoAgentOperationHttpUrl`	`str`	IRI of the agent's operation HTTP URL	required
`inputTypes`	`List[str]`	List of IRI of the agent's input types	required
`outputTypes`	`List[str]`	List of IRI of the agent's output types	required

Source code in JPS_BASE_LIB/python_wrapper/twa/kg_operations/derivation_client.py

def createOntoAgentInstance(
    self,
    ontoAgentServiceIRI: str,
    ontoAgentOperationHttpUrl: str,
    inputTypes: List[str],
    outputTypes: List[str]
):
    """
    Register an OntoAgent instance in the triple store.

    Args:
        ontoAgentServiceIRI (str): IRI of the agent's OntoAgent:Service
        ontoAgentOperationHttpUrl (str): IRI of the agent's operation HTTP URL
        inputTypes (List[str]): List of IRI of the agent's input types
        outputTypes (List[str]): List of IRI of the agent's output types
    """
    self.derivation_client.createOntoAgentInstance(
        ontoAgentServiceIRI, ontoAgentOperationHttpUrl, inputTypes, outputTypes)

addTimeInstance ¶

addTimeInstance(entities: List[str])

Add a time instance to each entity in the list of entities.

Parameters:

Name	Type	Description	Default
`entities`	`List[str]`	List of IRIs of entities to add time instance to	required

Source code in JPS_BASE_LIB/python_wrapper/twa/kg_operations/derivation_client.py

def addTimeInstance(self, entities: List[str]):
    """
    Add a time instance to each entity in the list of entities.

    Args:
        entities (List[str]): List of IRIs of entities to add time instance to
    """
    self.derivation_client.addTimeInstance(entities)

addTimeInstanceCurrentTimestamp ¶

addTimeInstanceCurrentTimestamp(entities: List[str])

Add time instance with current timestamp to the given entities

Parameters:

Name	Type	Description	Default
`entities`	`List[str]`	List of IRIs of entities to add time instance to	required

Source code in JPS_BASE_LIB/python_wrapper/twa/kg_operations/derivation_client.py

def addTimeInstanceCurrentTimestamp(self, entities: List[str]):
    """
    Add time instance with current timestamp to the given entities

    Args:
        entities (List[str]): List of IRIs of entities to add time instance to
    """
    self.derivation_client.addTimeInstanceCurrentTimestamp(entities)

updateTimestamp ¶

updateTimestamp(entity: str)

Update the timestamp of the entity to the current time.

Parameters:

Name	Type	Description	Default
`entity`	`str`	IRI of the entity to update the timestamp of	required

Source code in JPS_BASE_LIB/python_wrapper/twa/kg_operations/derivation_client.py

def updateTimestamp(self, entity: str):
    """
    Update the timestamp of the entity to the current time.

    Args:
        entity (str): IRI of the entity to update the timestamp of
    """
    self.derivation_client.updateTimestamp(entity)

updateTimestamps ¶

updateTimestamps(entities: List[str])

Update the timestamp of all entities in the list.

Parameters:

Name	Type	Description	Default
`entities`	`List[str]`	List of IRIs of entities to update the timestamp for	required

Source code in JPS_BASE_LIB/python_wrapper/twa/kg_operations/derivation_client.py

def updateTimestamps(self, entities: List[str]):
    """
    Update the timestamp of all entities in the list.

    Args:
        entities (List[str]): List of IRIs of entities to update the timestamp for
    """
    self.derivation_client.updateTimestamps(entities)

dropTimestampsOf ¶

dropTimestampsOf(entities: List[str])

Drop the timestamp of the given entities.

Parameters:

Name	Type	Description	Default
`entities`	`List[str]`	List of IRIs of entities to drop the timestamp of	required

Source code in JPS_BASE_LIB/python_wrapper/twa/kg_operations/derivation_client.py

def dropTimestampsOf(self, entities: List[str]):
    """
    Drop the timestamp of the given entities.

    Args:
        entities (List[str]): List of IRIs of entities to drop the timestamp of
    """
    self.derivation_client.dropTimestampsOf(entities)

getDerivations ¶

getDerivations(agentIRI: str) -> List[str]

Get the derivations of the given agent.

Parameters:

Name	Type	Description	Default
`agentIRI`	`str`	IRI of the agent	required

Returns:

Name	Type	Description
`list`	`List[str]`	List of IRIs of the derivations that isDerivedUsing the given agent

Source code in JPS_BASE_LIB/python_wrapper/twa/kg_operations/derivation_client.py

def getDerivations(self, agentIRI: str) -> List[str]:
    """
    Get the derivations of the given agent.

    Args:
        agentIRI (str): IRI of the agent

    Returns:
        list: List of IRIs of the derivations that isDerivedUsing the given agent
    """
    return list(self.derivation_client.getDerivations(agentIRI))

getDerivationsOf ¶

getDerivationsOf(entities: List[str]) -> Dict[str, str]

Get the derivations of the given entities.

Parameters:

Name	Type	Description	Default
`entities`	`List[str]`	List of entities	required

Returns:

Type	Description
`Dict[str, str]`	Dict[str, str]: The dictionary with the entity IRIs as keys and the derivation IRIs as values

Source code in JPS_BASE_LIB/python_wrapper/twa/kg_operations/derivation_client.py

def getDerivationsOf(self, entities: List[str]) -> Dict[str, str]:
    """
    Get the derivations of the given entities.

    Args:
        entities (List[str]): List of entities

    Returns:
        Dict[str, str]: The dictionary with the entity IRIs as keys and the derivation IRIs as values
    """
    return self.derivation_client.getDerivationsOf(entities)

unifiedUpdateDerivation ¶

unifiedUpdateDerivation(derivationIRI: str)

Unified update derivation method. This methods updates the specified derivation and all its upstream derivations.

Parameters:

Name	Type	Description	Default
`derivationIRI`	`str`	IRI of the derivation to be updated	required

Source code in JPS_BASE_LIB/python_wrapper/twa/kg_operations/derivation_client.py

def unifiedUpdateDerivation(self, derivationIRI: str):
    """
    Unified update derivation method.
    This methods updates the specified derivation and all its upstream derivations.

    Args:
        derivationIRI (str): IRI of the derivation to be updated
    """
    self.derivation_client.unifiedUpdateDerivation(derivationIRI)

updatePureSyncDerivation ¶

updatePureSyncDerivation(derivationIRI: str)

Update the specified synchronous derivation and all its upstream derivations.

Parameters:

Name	Type	Description	Default
`derivationIRI`	`str`	IRI of the derivation to be updated	required

Source code in JPS_BASE_LIB/python_wrapper/twa/kg_operations/derivation_client.py

def updatePureSyncDerivation(self, derivationIRI: str):
    """
    Update the specified *synchronous* derivation and all its upstream derivations.

    Args:
        derivationIRI (str): IRI of the derivation to be updated
    """
    self.derivation_client.updatePureSyncDerivation(derivationIRI)

updatePureSyncDerivations ¶

updatePureSyncDerivations(derivationIRIs: List[str])

Update the specified list of synchronous derivations and all their upstream derivations.

Parameters:

Name	Type	Description	Default
`derivationIRIs`	`List[str]`	List of IRIs of the derivations to be updated	required

Source code in JPS_BASE_LIB/python_wrapper/twa/kg_operations/derivation_client.py

def updatePureSyncDerivations(self, derivationIRIs: List[str]):
    """
    Update the specified list of *synchronous* derivations and all their upstream derivations.

    Args:
        derivationIRIs (List[str]): List of IRIs of the derivations to be updated
    """
    self.derivation_client.updatePureSyncDerivations(derivationIRIs)

updatePureSyncDerivationsInParallel ¶

updatePureSyncDerivationsInParallel(derivationIRIs: List[str])

Update the specified list of synchronous derivations and all their upstream derivations in parallel.

Parameters:

Name	Type	Description	Default
`derivationIRIs`	`List[str]`	List of IRIs of the derivations to be updated	required

Source code in JPS_BASE_LIB/python_wrapper/twa/kg_operations/derivation_client.py

def updatePureSyncDerivationsInParallel(self, derivationIRIs: List[str]):
    """
    Update the specified list of *synchronous* derivations and all their upstream derivations in parallel.

    Args:
        derivationIRIs (List[str]): List of IRIs of the derivations to be updated
    """
    self.derivation_client.updatePureSyncDerivationsInParallel(derivationIRIs)

updateAllSyncDerivations ¶

updateAllSyncDerivations()

Update all synchronous derivations in the knowledge graph.

Source code in JPS_BASE_LIB/python_wrapper/twa/kg_operations/derivation_client.py

def updateAllSyncDerivations(self):
    """
    Update all *synchronous* derivations in the knowledge graph.
    """
    self.derivation_client.updateAllSyncDerivations()

updateMixedAsyncDerivation ¶

updateMixedAsyncDerivation(derivationIRI: str)

Update a directed acyclic graph (DAG) of pure asynchronous derivations or asynchronous derivations depending on synchronous derivations.

Parameters:

Name	Type	Description	Default
`derivationIRI`	`str`	IRI of the derivation to be updated	required

Source code in JPS_BASE_LIB/python_wrapper/twa/kg_operations/derivation_client.py

def updateMixedAsyncDerivation(self, derivationIRI: str):
    """
    Update a directed acyclic graph (DAG) of pure asynchronous derivations or asynchronous derivations depending on synchronous derivations.

    Args:
        derivationIRI (str): IRI of the derivation to be updated
    """
    self.derivation_client.updateMixedAsyncDerivation(derivationIRI)