TS

Edge

class eBCSgen.TS.Edge.Edge(source, target, probability, label=None, encoded=False)

Bases: object

add_rate(rate)

Adds given rate to self.probability.

Used when joining two Edges between the same nodes.

Parameters:

rate – given rate expression

encode(encoding)

normalise(factor)

Normalises rate to probability and converts it to float or string (depending on parameters presence).

Parameters:

factor – given sum of all rates

recode(encoding_old: dict, encoding_new: dict) → Edge

Recodes the edge to new encoding

Parameters:

• encoding_old – the old encoding

• encoding_new – the new encoding

Returns:

new Edge in new encoding

to_PRISM_string(decoding) → str

Creates string representation for PRISM file.

Returns:

PRISM string representation

to_dict()

Exports the edge as a dict.

Returns:

dict representing the edge

to_vector(ordering)

eBCSgen.TS.Edge.edge_from_dict(d: dict) → Edge

Creates edge from given dict.

Parameters:

d – dict representing the edge

Returns:

Edge

eBCSgen.TS.Edge.truncate(f, n)

State

class eBCSgen.TS.State.Memory(level: int)

Bases: object

update_memory(label)

class eBCSgen.TS.State.Multiset(value: Counter)

Bases: object

check_intersection(other: Multiset)

reorder(indices: array) → Multiset

set_hell()

to_vector(ordering, is_hell)

validate_bound(bound)

class eBCSgen.TS.State.State(content, memory: Memory, is_hell=False)

Bases: object

check_AP(ap, ordering: tuple) → bool

Checks whether the State satisfies given AtomicProposition. Works only for vector variant.

Parameters:

• ap – given AtomicProposition

• ordering – position of corresponding Complex

Returns:

True if satisfied

check_intersection(other: State)

Check if the states have common intersection.

Parameters:

other – given other State

Returns:

True if intersection is nonempty

reorder(indices: array)

Changes order of individual values according to given indices. Works only for vector variant.

Parameters:

indices – array of indices

to_PRISM_string(apostrophe=False) → str

Creates string representation for PRISM file. Works only for vector variant.

Parameters:

apostrophe – indicates whether variables should be with the apostrophe

Returns:

PRISM string representation

to_vector(ordering)

Convert content from Multiset to Vector based on given ordering

Parameters:

ordering – given ordering of agents

update_state(consumed, produced, label, bound) → State

Creates a new state by subtracting consumed object and adding produced objects. If given bound is exceeded, special state labeled as hell is returned.

Parameters:

• consumed – consumed objects

• produced – produced objects

• label – label of rule/reaction used

• bound – maximal allowed bound on individual values

Returns:

resulting State

class eBCSgen.TS.State.Vector(value: array)

Bases: object

check_intersection(other: Vector)

filter_values(state: Vector) → int

Computed sum of individual values after intersection with given Vector. (used to indicate abstract agents in reaction-based setup).

Parameters:

state – given Vector

Returns:

resulting summation

reorder(indices: array) → Vector

set_hell()

to_ODE_string() → str

Each non-zero value transforms to form y[i] where i is its particular position. Finally, groups these strings to a sum of them (in string manner). Works only for vector variant.

Returns:

string symbolic representation of state

to_multiset(ordering)

validate_bound(bound)

TSworker

class eBCSgen.TS.TSworker.TSworker(ts, reactions, definitions, regulation)

Bases: Thread

join(timeout=None)

Wait until the thread terminates.

This blocks the calling thread until the thread whose join() method is called terminates – either normally or through an unhandled exception or until the optional timeout occurs.

When the timeout argument is present and not None, it should be a floating point number specifying a timeout for the operation in seconds (or fractions thereof). As join() always returns None, you must call is_alive() after join() to decide whether a timeout happened – if the thread is still alive, the join() call timed out.

When the timeout argument is not present or None, the operation will block until the thread terminates.

A thread can be join()ed many times.

join() raises a RuntimeError if an attempt is made to join the current thread as that would cause a deadlock. It is also an error to join() a thread before it has been started and attempts to do so raises the same exception.

run()

Method takes a state from pool of states to be states and:

• iteratively applies all rules on it

• checks whether newly created states (if any) were already states (present in self.ts.states_encoding); if not, they are added to self.states_to_process

• creates Edge from the source state to created ones (since ts.edges is a set, we don’t care about its presence)

• all outgoing Edges from the state are normalised to probability

TransitionSystem

class eBCSgen.TS.TransitionSystem.TransitionSystem(ordering: SortedList | None = None, bound=None)

Bases: object

change_hell()

Changes hell from inf to bound + 1.

change_to_vector_backend()

Changes backend from Multisets to Vectors by encoding them.

create_AP_labels(APs: list, include_init=True)

Creates label for each AtomicProposition. Moreover, goes through all states in ts.states_encoding and validates whether they satisfy given APs - if so, the particular label is assigned to the state.

Parameters:

APs – give AtomicProposition extracted from Formula

Returns:

dictionary of State_code -> set of labels and AP -> label

decode()

Flips encoding to continue in generating.

edges_to_PRISM(decoding)

Takes ordered edges grouped by source and for each group creates PRISM file representation

Returns:

list of strings for each group (source)

encode()

Assigns a unique code to each State for storing purposes

encode_edges()

Encodes every Vector in Edge according to the unique encoding.

filter_unused_agents()

There are cases when agents which are always 0 are used. This methods removed such agents and fixes encoding.

Returns:

minimalised TS

recode(new_encoding: dict)

Recodes the transition system according to the new encoding.

Parameters:

new_encoding – given new encoding

Returns:

new TransitionSystem

revert_encoding() → dict

Swaps encoding dictionary for decoding purposes.

Returns:

swapped dictionary

save_to_STORM_explicit(transitions_file: str, labels_file: str, state_labels: dict, AP_labels)

Save the TransitionSystem as explicit Storm file (no parameters).

Parameters:

• transitions_file – file for transitions

• labels_file – file for labels

• labels – labels representing atomic propositions assigned to states

save_to_json(output_file: str, params=None)

Save current TS as a JSON file.

Parameters:

• params – given set of unknown parameters

• output_file – given file to write to

save_to_prism(output_file: str, params: set, prism_formulas: list)

Save the TransitionSystem as a PRISM file (parameters present).

Parameters:

• output_file – output file name

• params – set of present parameters

• prism_formulas – definition of abstract Complexes

to_kripke(state_labels)

Create Kripke structure in format of pyModelChecking module.

Returns:

Kripke structure representation of the transition system

eBCSgen.TS.TransitionSystem.create_indices(ordering_1: SortedList, ordering_2: SortedList)

Creates indices np.array which represents how agents from ordering_1 have to be rearranged in order to fit agents from ordering_2. If such relation is not possible, return False.

Parameters:

• ordering_1 – first agents ordering

• ordering_2 – second agents ordering

Returns:

np.array of indices

VectorModel

class eBCSgen.TS.VectorModel.VectorModel(vector_reactions: set, init: State, ordering: SortedList, bound: int, regulation=None)

Bases: object

compute_bound()

Computes maximal bound from all reactions and initial state.

Returns:

maximal bound

deterministic_simulation(max_time: float, volume: float, step: float = 0.01) → DataFrame

Translates model to ODE and runs odeint solver for given max_time.

Parameters:

• max_time – end time of simulation

• volume – volume of the system

• step – distance between time points

Returns:

simulated data

generate_transition_system(ts: TransitionSystem | None = None, max_time: float = inf, max_size: float = inf) → TransitionSystem

Parallel implementation of Transition system generating.

The workload is distributed to Workers which take unprocessed States from the pool and process them.

If the given bound should be exceeded, a special infinite state is introduced.

The algorithm dynamically changes number of active workers using thread events. This is done according to the current volume of unprocessed states.

Returns:

generated Transition system

stochastic_simulation(max_time: float, runs: int, testing: bool = False) → DataFrame

Gillespie algorithm implementation.

Each step a random reaction is chosen by exponential distribution with density given as a sum of all possible rates in particular VectorState. Then such reaction is applied and next time is computed using Poisson distribution (random.expovariate).

Parameters:

• max_time – time when simulation ends

• runs – how many time the process should be repeated (then average behaviour is taken)

Returns:

simulated data

eBCSgen.TS.VectorModel.fake_expovariate(rate)

eBCSgen.TS.VectorModel.handle_number_of_threads(number, workers)

Estimated number of required workers for current volume of unprocessed states.

Parameters:

• number – volume of unprocessed states

• workers – available workers

VectorReaction

class eBCSgen.TS.VectorReaction.VectorReaction(source: State, target: State, rate: Rate, label=None)

Bases: object

evaluate_rate(state, definitions)

match(state, all=False)

reconstruct_complexes_from_match(match)

replace(aligned_match)

to_symbolic()

Transforms rate of the reaction to symbolic representation (used in ODEs).