Continuous Time Markov Chain (CTMC)

In a CTMC, each transition from \(s\) to \(s'\) is associated to a exponential probability distribution of parameter \(R(s,s')\). The probability of this transition to be triggered within \(\tau \in \mathbb{R}_{>0}\) time-units is \(1 - e^{- R(s,s') \, \tau}\). When, from a state \(s\), there are more than one outgoing transition, we are in presence of a race condition. In this case, the first transition to be triggered determines which observation is generated as well as the next state of the CTMC. According to these dynamics, the time spent in state \(s\) before any transition occurs, called waiting time (sometimes dwell time), is exponentially distributed with parameter \(E(s) = \sum_{s' \in S} R(s,s')\), called exit-rate of \(s\).

Example

In the following picture, the values on the transitions are the parameters \(R(s,s')\). The observations, which are red, yellow, blue are represented by the color of the states. The values on each state correspond to the expected waiting time before leaving this state: it is the inverse of the exit-rate.

Creation

>>> import jajapy as ja
>>> labelling = ['red','red','yellow','blue','blue']
>>> transitions = [(0,1,0.08),(0,2,0.12),(1,1,0.3),(1,2,0.7),
                   (2,0,0.2),(2,3,0.1),(2,4,0.2),(3,3,0.8),
                   (3,1,0.1),(3,4,0.1),(4,2,0.25)]

>>> model = ja.createCTMC(transitions,labelling,initial_state=0,name="My_CTMC")

We can also generate a random CTMC

>>> random_model = CTMC_random(number_states=5,
                               alphabet=['red','yellow','blue'],
                               random_initial_state=False,
                               min_exit_rate_time = 1.0,
                               max_exit_rate_time = 5.0)

Exploration

>>> model.e(0)
0.2
>>> model.expected_time(0)
5.0
>>> model.l(0,2)
0.12
>>> model.l(0,3)
0.0
>>> model.tau(0,2,'red')
0.6
>>> model.tau(0,2,'yellow')
0.0
>>> model.lkl(0,1.0)
0.1637461506155964
>>> # which is equal to 0.2*exp(-0.2*1.0), and 0.2 == model.e(0)
>>> model.getAlphabet()
['init','red','yellow','blue']
>>> model.getLabel(2)
'yellow'

Running

>>> model.run(5) # returns a list of 5 observations plus the initial one without the waiting times
['init','red', 'yellow', 'blue', 'blue', 'yellow', 'blue']
>>> model.run(5,timed=True) # returns a list of 5 observations plus the intial one with the waiting times
['init',0.5336343726140618,'red', 1.806338759176371, 'red', 0.05237298538963968,
'red', 0.5556197125154162, 'red', 0.35592063754974723, 'red', 0.6101068433945884, 'red']
>>> s = model.generateSet(10,5,timed=True) # returns a Set containing 10 traces of size 5
>>> s.sequences
[['init', 0.10788094099666436, 'red', 1.6789349613457798, 'red', 0.27255276944298595,
'red', 0.42350152604909247, 'yellow', 0.20220224981867935, 'blue', 9.704572104709351, 'yellow'],
['init', 1.6378986557150426, 'red', 4.1800880483401555, 'yellow', 1.8656306179724005,
'red', 9.427748853607259, 'yellow', 0.21286758511326231, 'red', 2.278263556837526, 'red'],
['init', 4.514275634368433, 'red', 1.2355021955427197, 'yellow', 0.31248093363627166,
'red', 8.824559530285878, 'red', 0.4529508092055165, 'red', 0.07274692217838562, 'red'],
['init', 0.04764578868145291, 'red', 0.4723999201503642, 'yellow', 0.6743349646745602,
'red', 0.7452790320928603, 'yellow', 0.7757445460815905, 'blue', 3.1265714080061575, 'yellow'],
['init', 0.2618927236991998, 'red', 8.945809803848748, 'yellow', 1.3509604669493112,
'blue', 2.1740236675910465, 'yellow', 7.560013081702193, 'blue', 0.6371556825230754, 'yellow'],
['init', 0.555526612944214, 'red', 2.1083387658424524, 'red', 1.5993222803300493,
'yellow', 1.1866495308662353, 'blue', 0.8608021252016191, 'yellow', 1.6199990800624533, 'blue'],
['init', 0.1643966933585648, 'red', 1.3025805686943215, 'yellow', 2.6383860396412477,
'red', 18.014055444939096, 'red', 1.2034307567991924, 'yellow', 0.06152914313554882, 'red'],
['init', 1.1424402840333279, 'red', 0.7250888175057543, 'red', 0.23039664620400085,
'red', 0.8155342281890401, 'yellow', 0.06167580348263705, 'red', 3.2313274235968423, 'red'],
['init', 1.349974406242494, 'red', 4.7893235714423605, 'yellow', 2.5583220293445823,
'red', 3.1703454235936244, 'red', 0.98555132768311, 'yellow', 0.6707954977567278, 'blue'],
['init', 0.6118189872487105, 'red', 7.057133997012888, 'yellow', 0.7762876536605751,
'blue', 0.6567357090100437, 'yellow', 3.3142910438173105, 'blue', 1.397231565238264, 'blue']]
>>> s.times # each of them appears only once
[ 1, 1, 1, 1, 1, 1, 1, 1, 1, 1]

Analysis

>>> model.logLikelihood(s) # loglikelihood of this set of traces under this model
-12.185488019773823
>>> # the result is quite low since it is a timed data set.

Saving/Loading

>>> model.save("my_ctmc.txt")
>>> the_same_model = ja.loadCTMC("my_ctmc.txt")

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 CTMCs can be composed to create one PCTMC. Jajapy synchronous composition follows the Prism formalism: it allows transitions to be labelled with actions. These actions can be used to force two or more modules to make transitions simultaneously (i.e. to synchronise). The rate of this transition is equal to the product of the two individual rates. To do so, Jajapy uses the synchronous_transitions attributes of the CTMC objects.

In the following example the synchronous transitions are the coloured one, the colour being the action.

>>> labelling = ['red','green','orange']
>>> transitions = [(2,0,2.0)]
>>> sync_trans = [(0,'ped_red',1,0.5), (1,'button',2,1.0)]
>>>     model1 = createCTMC(transitions,labelling,[0.6,0.4,0.0],"Car_tl",sync_trans)
>>>
>>>labelling = ['red','green']
>>> transitions = [(1,0,0.1)]
>>> sync_trans = [(0,'button',1,0.5), (0,'ped_red',0,10.0)]
>>>     model2 = createCTMC(transitions,labelling,0,"Ped_tl",sync_trans)
>>>
>>> composition = synchronousCompositionCTMCs(model1, model2)

Model

class jajapy.CTMC(matrix: ndarray, labelling: list, name: str = 'unknown_CTMC', synchronous_transitions: list = [])

Class representing a CTMC.

Creates an CTMC.

Parameters

matrixndarray: A (N x N) ndarray (with N the nb of states). Represents the transition matrix. matrix[s1][s2] is the rate associated to the transition from s1 to s2.
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.
namestr, optional: Name of the model. Default is “unknow_CTMC”
synchronous_transitions: list, optional.: This is useful only for synchronously composing this CTMC 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.

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'

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.

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.

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.

toMC(name: str = 'unknown_MC') → MC

Returns the equivalent untimed MC.

Parameters

namestr, optional: Name of the output model. By default “unknown_MC”

Returns

MC: An equivalent untimed model.

toStormpy()

Returns the equivalent stormpy sparse model. The output object will be a stormpy.SparseDtmc.

Returns

stormpy.SparseDtmc: The same model in stormpy format.

updateLabelling(new_labels: list) → None

Update the states labelling.

Parameters

new_labels: list: a list of new labels. The length of the list must be equal to the number of states.

Other Functions

jajapy.createCTMC(transitions: list, labelling: list, initial_state, name: str = 'unknown_CTMC', synchronous_transitions: list = []) → CTMC

An user-friendly way to create a CTMC.

Parameters

transitions[ list of tuples (int, int, float)]: Each tuple represents a transition as follow: (source state ID, destination state ID, rate).
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.
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).
namestr, optional: Name of the model. Default is “unknow_CTMC”
synchronous_transitions: list, optional.: This is useful only for synchronously composing this CTMC with another one. List of (source_state <int>, action <str>, dest_state <int>, rate <float>). Default is an empty list.

Returns

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

Examples

>>> model = createCTMC([(0,1,1.0),(1,0,0.3),(1,1,0.2)],['b','a'],0,"My_MC")
>>> print(model)
Name: My_MC
Initial state: s0
----STATE 0--b----
Exepected waiting time: 1.0
s0 -> s1 : lambda = 1.0
----STATE 1--a----
Exepected waiting time: 2.0
s1 -> s0 : lambda = 0.3
s1 -> s1 : lambda = 0.2

jajapy.loadCTMC(file_path: str) → CTMC

Load an CTMC saved into a text file.

Parameters

file_pathstr: Location of the text file.

Returns

outputCTMC: The CTMC saved in file_path.

jajapy.CTMC_random(nb_states: int, labelling: list, min_exit_rate_time: int, max_exit_rate_time: int, self_loop: bool = True, random_initial_state: bool = True, sseed: Optional[int] = None) → CTMC

Generates a random CTMC. All the rates will be between 0 and 1. All the exit rates will be integers.

Parameters

nb_statesint: Number of states.
labellinglist of str: List of observations.
min_exit_rate_time: int: Minimum exit rate for the states (included).
max_exit_rate_time: int: Maximum exit rate for the states (included).
self_loop: bool, optional: Wether or not there will be self loop in the output model. Default is True.
random_initial_state: bool, optional: If set to True we will start in each state with a random probability, otherwise we will always start in state 0. Default is True.
sseedint, optional: the seed value.

Returns

CTMC: A pseudo-randomly generated CTMC.

Examples

>>> model = CTMC_random(2,['a','b'],1,5)
>>> print(model)
Name: CTMC_random_2_states
Initial state: s2
----STATE 0--a----
Exepected waiting time: 2.0
s0 -> s0 : lambda = 0.38461538461538464
s0 -> s1 : lambda = 0.11538461538461539
----STATE 1--b----
Exepected waiting time: 4.0
s1 -> s0 : lambda = 0.13636363636363635
s1 -> s1 : lambda = 0.11363636363636363
----STATE 2--init----
Exepected waiting time: 1.0
s2 -> s0 : lambda = 0.2
s2 -> s1 : lambda = 0.8

jajapy.synchronousCompositionCTMCs(m1: CTMC, m2: CTMC, name: str = 'unknown_composition') → PCTMC

Returns the synchornous compotision of m1 and m2. The output model is a PCTMC since some parameters are shared by multiple transitions.

Parameters

m1CTMC: First CTMC to compose with.
m2CTMC: Second CTMC to compose with.

Returns

PCTMC: Synchronous composition of m1 and m2.