Parametric Continuous Time Markov Chain (PCTMC)

A PCTMC is a CTMC where the transition rates can be expressed as polynomial functions over a set of parameters.

Example

The model above represents the tandem queueing network where c=3. The yellow model corresponds to the serverC, and the green one the SeverM. For each yellow state, the first value corresponds to the value of sc and the second one the value of phi. The red transitions are the synchronous transitions (the route transitions in the Prism file), while the black transitions are the asynchronous one.

Creation

We can load the Prism file describing a PCTMC or a composition of PCMTCs as follow:

>>> import jajapy as ja
>>> model = ja.loadPrism("tandem_3.sm")

We can also create it manually using the createPCTMC function (the following will create a model for serverC only):

>>> labelling = ['sc_0_ph_1','sc_0_ph_2',
>>>             'sc_1_ph_1','sc_1_ph_2',
>>>             'sc_2_ph_1','sc_2_ph_2',
>>>             'sc_3_ph_1','sc_3_ph_2',]
>>> transitions = [(0,2,'lambda'),(2,4,'lambda'),(4,6,'lambda'),
>>>             (1,3,'lambda'),(3,5,'lambda'),(5,7,'lambda'),
>>>             (2,3,'mu1a'),(4,5,'mu1a'),(6,7,'mu1a')]
>>> sync_trans = [(2,'route',0,'mu1b'),(4,'route',2,'mu1b'),(6,'route',4,'mu1b'),
>>>             (3,'route',0,'mu2'),(5,'route',2,'mu2'),(7,'route',4,'mu2')]
>>> parameters = ['lambda','mu1a','mu1b','mu2']
>>> initial_state = 0
>>> parameter_instantiation = {'lambda':6.0}
>>> model = createPCTMC(transitions,labelling,parameters,initial_state,parameter_instantiation,sync_trans)

It is not possible to create a random PCTMC. However, we can ask Jajapy to randomly instantiate the parameters:

>>> model.isInstantiated()
False
>>> model.randomInstantiation(min_val=0.5, max_val=5.0)
>>> model.isInstantiated()
True

Converting from/to Stormpy

>>> stormpy_sparse_model = model.toStormpy() # the next line is equivalent
>>> stormpy_sparse_model = ja.jajapyModeltoStormpy(model)
>>> same_model == ja.stormpyModeltoJajapy(stormpy_sparse_model)

Converting from/to Prism

>>> model.savePrism("my_ctmc.sm")
>>> same_model = ja.loadPrism("my_ctmc.sm")

Synchronous composition

Two PCTMCs (or more) can be composed to create one PCTMC. PCTMCs composition works as CTMCs composition.

>>> composition = synchronousCompositionPCTMCs([pctmc1, pctmc2, pctmc3])

Model

class jajapy.PCTMC(matrix: ndarray, labelling: list, transition_expr: list, parameter_values: dict, parameter_indexes: list, parameter_str: list, name: str = 'unknow_PCTMC', synchronous_transitions: list = [])

Class representing a PCTMC.

Creates an PCTMC.

Parameters

matrixndarray: Represents the transition matrix. matrix[i,j] is the index, in transition_str, of the symbolic value of the transition from i to j.
labelling: list of str: A list of N observations (with N the nb of states). If labelling[s] == o then state of ID s is labelled by o. Each state has exactly one label.
transition_str: list of str: Contains the symbolic value for each transition.
parameter_values: dict: Contains the value for each instantiated parameter. parameter_values[i] is the instantiation for parameter i. If the ith parameter is not instantiated, parameter_values[i] == nan.
parameter_indexes: list of ndarray: Contains the indexes of each transition using each parameter. parameter_indexes[i] = array([[0,1],[2,1]]) means that parameter i is used by the transition from state 0 to state 1 and from state 2 to state 1.
parameter_str: list: Contains the name of each parameter. parameter_str[i] is the name of parameter i. Parameter i doesn’t have a name if parameter_str[i]==None.
namestr, optional: Name of the model. Default is “unknow_PCTMC”
synchronous_transitions: list, optional.: This is useful only for synchronously composing this PCTMC with another one. List of (source_state <int>, action <str>, dest_state <int>, rate <float>). Default is an empty list.

e(s: int) → float

Returns the exit rate of state s, i.e. the sum of all the rates in this state.

Returns

sint: A state ID.
float: An exit rate.

evaluateTransition(i: int, j: int, parameter_values: Optional[dict] = None)

Returns the value of the transition from state i to j. If parameter_values is given, the parameters are instantiated by parameter_values, otherwise by self.parameter_values. If at least one of the paramater in this transition is not instantiated, returns the symbolic representation of the transition.

Parameters

iint: source state ID.
jint: destination state ID.
parameter_values: dict, optional: dictionary with the instantiation of the parameters. The keys are the parameter names (str).

Returns

float or symbolic representation: The value of the transition from state i to j. If at least one of the paramater this transition is not instantiated, returns the symbolic representation of the transition.

expected_time(s: int) → float

Returns the expected waiting time in s, i.e. the inverse of the exit rate.

Parameters

sint: A state ID.

Returns

float: expected waiting time in this state.

generateSet(set_size: int, param, distribution=None, min_size=None, timed: bool = False) → Set

Generates a set (training set / test set) containing set_size traces.

Parameters

set_size: int: number of traces in the output set.
param: a list, an int or a float.: the parameter(s) for the distribution. See “distribution”.
distribution: str, optional: If distribution=='geo' then the sequence length will be distributed by a geometric law such that the expected length is min_size+(1/param). If distribution==None param can be an int, in this case all the seq will have the same length (param), or param can be a list of int. Default is None.
min_size: int, optional: see “distribution”. Default is None.
timed: bool, optional: Only for timed model. Generate timed or non-timed traces. Default is False.

Returns

output: Set: a set (training set / test set).

Examples

>>> set1 = model.generateSet(100,10)
>>> # set1 contains 100 traces of length 10
>>> set2 = model.generate(100, 1/4, "geo", min_size=6)
>>> # set2 contains 100 traces. The length of the traces is distributed following
>>> # a geometric distribution with parameter 1/4. All the traces contains at
>>> # least 6 observations, hence the average length of a trace is 6+(1/4)**(-1) = 10.

getAlphabet() → list

Returns the alphabet of this model.

Returns

list of str: The alphabet of this model

Example

>>> model.getAlphabet()
['a','b','c','d','done']

getLabel(state: int) → str

Returns the label of state.

Parameters

stateint: a state ID

Returns

str: a label

Example

>>> model.getLabel(2)
'Label-of-state-2'

instantiate(parameters: list, values: list) → bool

Set all the parameters in parameters to the values values.

Parameters

parameterslist of string: List of all parameters to set. This list must contain parameter names.
valueslist of float: List of values. parameters[i] will be set to values[i].

Returns

bool: True if the instantiation is valid (no transition has a negative evaluation) and then applied, False if the instantiation is not valid (and then ignored).

involvedParameters(i: int, j: int = -1) → list

Returns the parameters involved in the transition from i to j. If j is not set, it returns the parameters involved in all the transitions leaving i.

Parameters

iint: source state ID.
jint, optional: destination state ID.

Returns

list of str: list of parameter names.

isInstantiated(state1: Optional[int] = None, state2: Optional[int] = None, param: Optional[str] = None) → bool

If nothing is set, checks if all the parameters are instantiated. If state1 is set, checks if all the transitions leaving this state are instantiated. If state1 and state2 are set, checks if the transitions from state1 to state2 is instantiated. If param is set, checks if the parameter param is instantiated.

Parameters

state1int, optional.: state ID.
state2int, optional.: state ID.
paramstr, optional: Parameter name.

Returns

bool: True if all the requested entity(ies) is/are instantiated. See above.

l(s1: int, s2: int, obs: str) → float

Returns the rate associated to the transition from state s1, seeing obs, to state s2.

Parameters

s1int: A state ID.
s2int: A state ID.
obsstr: An observation.

Returns

float: A rate.

lkl(s: int, t: float) → float

Returns the likelihood of staying in s state for a duration t.

Parameters

sint: A state ID.
tfloat: A waiting time.

Returns

float: A Likelihood.

logLikelihood(traces: Set) → float

Computes the loglikelihood of traces under this model.

Parameters

tracesSet: a set of traces.

Returns

float: the loglikelihood of traces under this model.

next(state: int) → tuple

Return a state-observation pair according to the distributions described by matrix

Returns

output(int, str): A state-observation pair.

parameterIndexes(p: str) → list

Returns the list of all transitions involving p. If the model doesn’t have any parameter p, returns an empty list.

Parameters

pstr: parameter name.

Returns

list: The list of all transitions involving p. If the model doesn’t have any parameter p, returns an empty list.

Examples

>>> m.parameterIndexes('x')
[[4,0],[2,3]]
>>> # the transitions from state 4 to 0 and from 2 to 3 involve 'x'.
>>> m.transitionExpression(4,0)
3*x

parameterValue(p: str) → float

Returns the value of the parameter p. If p is not instantiated, returns numpy.nan. If the model doesn’t have any parameter p, returns numpy.nan.

Parameters

pstr: parameter name.

Returns

float: The value of p if p is instantiated, numpy.nan otherwise. If the model doesn’t have any parameter p, returns numpy.nan.

pi(s: int) → float

Return the probability of starting in state s.

Parameters

s: int: state ID.

Returns

outputfloat: the probability of starting in state s.

randomInstantiation(parameters: Optional[list] = None, min_val: Optional[float] = None, max_val: Optional[float] = None, sseed: Optional[int] = None) → None

Randomly instantiated the parameters given in parameters. If parameters is not set it instantiates all the non-instantiated parameters in the model.

Parameters

parameterslist of string, optional.: List of parameters names.
min_valfloat, optional: Minimal value for the randomly instantiated parameters. If not set and if the model has at least two instantiated parameters, this value is equal to the parameters with the smallest instantiation. If not set and if the model has less than two instantiated parameters, this value is equal to 0.1.
max_valfloat, optional: Maximal value for the randomly instantiated parameters. If not set and if the model has at least two instantiated parameters, this value is equal to the parameters with the highest instantiation. If not set and if the model has less than two instantiated parameters, this value is equal to 5.0.
sseedint, optional: the seed value.

rename(name: str) → None

Change the name of the model.

Parameters

namestr: new name.

run(number_steps: int, timed: bool = False) → list

Simulates a run of length number_steps of the model and return the sequence of observations generated. If timed it returns a list of pairs waiting time-observation.

Parameters

number_steps: int: length of the simulation.
timed: bool, optional: Wether or not it returns also the waiting times. Default is False.

Returns

output: list of str: trace generated by the run.

save(file_path: str) → None

Save the model into a text file.

Parameters

file_pathstr: path of the output file.

Examples

>>> model.save("my_model.txt")

savePrism(file_path: str) → None

Save this model into file_path in the Prism format.

Parameters

file_pathstr: Path of the output file.

tau(s1: int, s2: int, obs: str) → float

Returns the probability to move from s1 to s2 generating observation obs.

Parameters

s1int: A state ID.
s2int: A state ID.
obsstr: An observation.

Returns

float: A probability.

transitionExpression(i: int, j: int)

Returns the symbolic representation of the transition.

Parameters

iint: source state ID.
jint: destination state ID.

Returns

symbolic representation: The symbolic representation of the transition.

transitionValue(i: int, j: int)

Returns the value of the transition from state i to j. If at least one of the paramater in this transition is not instantiated, returns the symbolic representation of the transition.

Parameters

iint: source state ID.
jint: destination state ID.

Returns

float or symbolic representation: The value of the transition from state i to j. If at least one of the paramater in this transition is not instantiated, returns the symbolic representation of the transition.

Other Functions

jajapy.createPCTMC(transitions: list, labelling: list, parameters: list, initial_state, parameter_instantiation: dict = {}, synchronous_transitions=[], name: str = 'unknown_PCTMC') → PCTMC

An user-friendly way to create a PCTMC.

Parameters

transitions[ list of tuples (int, int, float or str)]: Each tuple represents a transition as follow: (source state ID, destination state ID, probability). The probability can be explicitly given (then it’s a float), or a parameter (then it’s the name of the parameter).
labelling: list of str: A list of N observations (with N the nb of states). If labelling[s] == o then state of ID s is labelled by o. Each state has exactly one label.
parameters: list of str.: A list containing all the parameters name.
initial_stateint or list of float: Determine which state is the initial one (then it’s the id of the state), or what are the probability to start in each state (then it’s a list of probabilities).
parameter_instantiation: dict: An instantion for some (or all) parameters. parameter_instantiation == {‘p’:0.5} means that parameter p should be instantiated to 0.5. The other parameters are not instantiated.
synchronous_transitions: list, optional.: This is useful only for synchronously composing this PCTMC with another one. List of (source_state <int>, action <str>, dest_state <int>, rate <float>). Default is an empty list.
namestr, optional: Name of the model. Default is “unknow_PCTMC”

Returns

PCTMC: the PCTMC describes by transitions, labelling, and initial_state.

jajapy.loadPCTMC(file_path: str) → PCTMC

Load an PCTMC saved into a text file.

Parameters

file_pathstr: Location of the text file.

Returns

outputPCTMC: The PCTMC saved in file_path.

Examples

>>> model = loadPCTMC("my_model.txt")

jajapy.synchronousCompositionPCTMCs(ms: list, name: str = 'unknown_composition') → PCTMC

Returns the synchornous compotision of the PCTMCs in ms.

Parameters

mslist of PCTMCs: List of PCTMCs to compose.
namestr, optional.: Name of the output model. Default is unknown_composition.

Returns

PCTMC: Synchronous composition of m1 and m2.