API Reference

Backend

ml4cps.automata

Automaton

Bases: CPSComponent

Automaton class is the main class for modeling various kinds of hybrid systems.

Source code in ml4cps/automata/base.py

class Automaton (CPSComponent):
    """
    Automaton class is the main class for modeling various kinds of hybrid systems.
    """

    def __init__(self, states: list = None, transitions: list = None,
                 unknown_state: str = 'raise', id="", initial_q=(), initial_r=None, final_q=(), super_states=(), decision_states=(),
                 **kwargs):
        """
        Class initialization from lists of elements.
        :param states: Discrete states / modes of continuous behavior.
        :param events: The events that trigger state transitions.
        :param transitions: The transition information. If a collection of dicts then dict should contain "source",
        "dest" and "event". The other attributes will be added as data of that transition. Alternatively, a collection
        of tuples can be used of the form (source, event, dest, *).
        :param unknown_state: The name of unknown states during "play in", if "raise", an exception will be raised.
        """
        self._G = nx.MultiDiGraph()
        if initial_q and isinstance(initial_q, str):
            initial_q = [initial_q]
        self.q0 = OrderedDict.fromkeys(initial_q)
        self.initial_r = initial_r
        if final_q and isinstance(final_q, str):
            final_q = [final_q]
        self.final_q = OrderedDict.fromkeys(final_q)
        self.previous_node_positions = None
        self.UNKNOWN_STATE = unknown_state
        if super_states is not None:
            if type(super_states) is str:
                self.__super_states = [super_states]
            else:
                self.__super_states = list(super_states)

            # self._G.add_nodes_from(self.__super_states)

        if decision_states is not None:
            if type(decision_states) is str:
                self.decision_states = [decision_states]
            else:
                self.decision_states = list(decision_states)

        if states is not None:
            self._G.add_nodes_from(states)

        if transitions is not None:
            for tr in transitions:
                if type(tr) is dict:
                    self._G.add_edge(tr.pop('source'), tr.pop('destination'), event=tr.pop('event'), **tr)
                else:
                    self._G.add_edge(tr[0], tr[2], event=tr[1])

        if 'discr_state_names' not in kwargs:
            kwargs['discr_state_names'] = ['Mode']
        elif type(kwargs['discr_state_names']) is str:
            kwargs['discr_state_names'] = [kwargs['discr_state_names']]
        CPSComponent.__init__(self, id, **kwargs)

    @property
    def Sigma(self):
        Sigma = []
        for x in self.transitions:
            if len(x) >= 3 and 'event' in x[2]:
                new_el = x[2]['event']
                if new_el not in Sigma:
                    Sigma.append(new_el)
        return Sigma

    @property
    def num_events(self):
        return len(self.Sigma)

    @property
    def discrete_states(self):
        return self._G.nodes

    @property
    def transitions(self):
        return self._G.edges(data=True)

    @property
    def discrete_state(self):
        if self._q is not None:
            if len(self._q) == 0:
                return ()
            return self._q[0]
        return None

    @discrete_state.setter
    def discrete_state(self, value):
        self._q = (value,)

    @property
    def state(self):
        """
        Automata discrete state is uni-variate.
        :return:
        """
        if len(self._q) < 1:
            raise Exception(f'State of "{self.id}" is empty.')
        return self._q[0], self._xt, self._xk

    @state.setter
    def state(self, state):
        if type(state) is not tuple:
            new_states = (state, (), ())
        elif len(state) < 3:
            new_states = [(), (), ()]
            for i, v in enumerate(state):
                new_states[i] = v
        else:
            new_states = state

        if new_states[0] not in self.discrete_states and new_states[0] not in self.__super_states and new_states[0] != self.UNKNOWN_STATE:
            raise ValueError(f'State "{new_states[0]}" is not a valid state. Valid states are: {self.discrete_states}')

        self._q = (new_states[0],)
        self._xt = new_states[1]
        self._xk = new_states[2]

    @property
    def num_modes(self):
        """
        Returns the number of modes in the automaton.
        :return: number of states.
        """
        return self._G.number_of_nodes()

    @property
    def num_transitions(self):
        """
        Returns the number of transitions in the automaton.
        :return: number of transitions.
        """
        return self._G.number_of_edges()

    def merge(self, q1, q2):
        """
        If two states are compatible, they are merged with the function merge. The transitions
        of the automaton, the in- and outdegree of the states and the number of transitions
        happening are adjusted.
        """
        intr = self.in_transitions(q2)
        # outtr_q1 = list(self.out_transitions(q1))
        outtr_q2 = list(self.out_transitions(q2))
        # nx.contracted_nodes(self._G, w, v, copy=False)
        # set event

        for tr in intr:
            if tr[0] != q2:
                self.add_single_transition(tr[0], q1, tr[2], timing=tr[-1]['timing'])
        # if q1=='q0' and q2 == 'q192':
        #     print('found')
        for tr in outtr_q2:
            dest = tr[1]
            if dest == q2:
                dest = q1
            self.add_single_transition(q1, dest, tr[2], timing=tr[-1]['timing'])
        self.remove_state(q2)

    def determinize(self, s, state_index, verbose=False):
        # out_tr = list(sorted(self.out_transitions(s), key=lambda x: state_index[x[1]]))
        if verbose:
            print('Determinize', s)
        # ind = sorted(, key=lambda x: state_index[x])
        # state_index_list = [None] * len(state_index)
        # for k, v in state_index.items():
        #     state_index_list[v] = k
        # if verbose:
        #     pprint.pprint(state_index_list[state_index[s]:], compact=True)

        to_analyze = [s]
        while to_analyze:  # state_ind < len(state_index_list):
            # print(state_ind)
            s = to_analyze.pop()
            # if not self.is_state(s):
            #     continue

            conflicting_transitions = {}
            to_merge = None
            for out_tr in self.out_transitions(s):
                if out_tr[2] in conflicting_transitions:
                    to_merge = (conflicting_transitions[out_tr[2]], out_tr[1])
                    break
                else:
                    conflicting_transitions[out_tr[2]] = out_tr[1]
            # out_tr = list(self.out_transitions(st))
            if to_merge is not None:
                if state_index[to_merge[0]] < state_index[to_merge[1]]:
                    merge_into = to_merge[0]
                    to_be_merged = to_merge[1]
                else:
                    merge_into = to_merge[1]
                    to_be_merged = to_merge[0]

                to_analyze.append(s)
                to_analyze.append(merge_into)
                if verbose:
                    print('Merge to determinize: ', to_be_merged, 'into', merge_into)
                self.merge(merge_into, to_be_merged)

        # to_determinize = []
        # for i, tr in enumerate(out_tr):
        #     if self.is_state(tr[1]):
        #         for tr2 in out_tr[:i:-1]:
        #             if self.is_state(tr2[1]) and tr[2] == tr2[2]:
        #                 # print(tr[1], ',', tr2[1])
        #                 if state_index[tr2[1]] > state_index[tr[1]]:
        #                     print('Merge to determinize: ', tr2[1], 'into', tr[1])
        #                     self.merge(tr[1], tr2[1])
        #                     to_determinize.append(tr[1])
        #                     # self.determinize(tr[1], state_index)
        #                 else:
        #                     print('Merge to determinize: ', tr[1], 'into', tr2[1])
        #                     self.merge(tr2[1], tr[1])
        #                     to_determinize.append(tr2[1])
        #                     # self.determinize(tr2[1], state_index)
        # for s in to_determinize:
        #     self.determinize(s, state_index)

    def accepts(self, string, return_states=False):
        state_path = []
        current_state = next(iter(self.q0))
        state_path.append(current_state)
        for symbol in string:
            trans = self.out_transitions(current_state, symbol)
            if trans is None or not trans:
                if return_states:
                    return False, state_path
                else:
                    return False
            elif len(trans) > 1:
                raise Exception(f'Too many transitions from {current_state} given symbol {symbol}.')
            else:
                current_state = trans[0][1]
                state_path.append(current_state)
        accepted = current_state in self.final_q or self.final_q is None or len(self.final_q) == 0
        if return_states:
            return accepted, state_path
        else:
            return accepted

    def generate(self, number_of_sequences=1, return_states=False, prob_to_accept=0.5):
        if number_of_sequences > 1:
            res = [self.generate(return_states=return_states, prob_to_accept=prob_to_accept)
                   for _ in range(number_of_sequences)]
            return res
        state_path = []
        current_state = next(iter(self.q0))
        event_path = []
        state_path.append(current_state)
        while True:
            if current_state in self.final_q:
                if random.random() < prob_to_accept:
                    break
            trans = self.out_transitions(current_state)
            if trans is None or not trans:
                if current_state not in self.final_q:
                    raise Exception(f'State "{current_state}" is not an accepting state, but no possible transitions.')
                break
            else:
                new_trans = random.choices(list(trans), weights=[x[3].get('prob', 1) for x in trans])
                current_state = new_trans[0][1]
                current_event = new_trans[0][3]['event']
                state_path.append(current_state)
                event_path.append(current_event)

        if return_states:
            return event_path, state_path
        else:
            return event_path


    def try_merge_states(self, state1, state2, try_fun=None):
        """Merge state2 into state1 and update transitions."""
        old_G = self._G
        make_final = state2 in self.final_q and state1 not in self.final_q
        make_initial = state2 in self.q0 and state1 not in self.q0
        self._G = nx.contracted_nodes(self._G, state1, state2)
        # make it deterministic again
        events = [x[3]['event'] for x in self.out_transitions(state1)]
        deterministic = len(set(events)) == len(events)

        if deterministic:
            if make_initial:
                self.add_initial_state(state1)
            if make_final:
                self.add_final_state(state1)

            if try_fun is not None and not try_fun(self):
                self._G = old_G
                if make_final:
                    self.final_q.pop(state1)
                if make_initial:
                    self.q0.pop(state1)
        else:
            self._G = old_G


    def remove_transition(self, source, dest):
        """
        Remove the transition(s) from source to dest.
        :param source:
        :param dest:
        :return:
        """
        self._G.remove_edge(source, dest)

    def is_decision(self, state, overall_state):
        return state in self.decision_states

    def get_alternatives(self, state, system_state):
        if self.is_decision(state, system_state):
            trans = [(x[3]['event'], dict()) for x in self.out_transitions(state)]
            return [(None, None)] + trans
        else:
            return None

    def remove_rare_transitions(self, min_p=0, min_num=0, keep_from_initial=False, keep_states=False, keep=None):
        self.learn_transition_probabilities()

        for source, dest, event, data in self.get_transitions():
            if keep_from_initial and source in self.q0:
                continue
            if (len(data['timing']) <= min_num or data['probability'] < min_p) and \
                    ((keep is None) or (keep is not None and source not in keep and dest not in keep)):
                self.remove_transition(source, dest)

        if not keep_states:
            for s in list(self.discrete_states):
                # if s in self.q0:
                #     continue
                if len(self.in_transitions(s)) == 0 and len(self.out_transitions(s)) == 0:
                    self.remove_state(s)

            # if self.DummyInitial:
            #     for s in list(self.InitialState):
            #         if len(self.out_transitions(s)) == 0:
            #             self.States.pop(s)
            #             self.InitialState.pop(s)

        # recalculate probabilities
        if min_p:
            self.learn_transition_probabilities()

    def learn_transition_probabilities(self):
        for s in self.discrete_states:
            total_num = sum([len(data['timing']) for s, d, e, data in self.out_transitions(s)])
            for s, d, e, data in self.out_transitions(s):
                data['probability'] = len(data['timing']) / total_num

    def state_is_deterministic(self, q):
        events = set()
        for tr in self.out_transitions(q):
            if tr[2] in events:
                return False
            else:
                events.add(tr[2])
        return True

    def update_timing_boundaries(self, source, destination, event, newTiming):
        edge_data = self._G.get_edge_data(source, destination, event)
        try:
            if newTiming < edge_data['minTiming']:
                edge_data['minTiming'] = newTiming
            elif newTiming > edge_data['maxTiming']:
                edge_data['maxTiming'] = newTiming
        except KeyError:
            edge_data['minTiming'] = newTiming
            edge_data['maxTiming'] = newTiming

    def is_deterministic(self):
        for q in self.discrete_states:
            if not self.state_is_deterministic(q):
                print('State', q, 'not deterministic:')
                for tr in sorted(self.out_transitions(q), key=lambda x: x[2]):
                    print(tr[0], '->', tr[2], '->', tr[1])
                return False
        return True

    def add_single_transition(self, s, d, e, timing=None):
        edge_data = self._G.get_edge_data(s, d, e)
        if edge_data is None:
            if timing is None:
                self._G.add_edge(s, d, key=e)
            else:
                try:
                    timing = list(timing)
                    self._G.add_edge(s, d, key=e, event=e, timing=timing)
                except:
                    self._G.add_edge(s, d, key=e, event=e, timing=[timing])
        elif timing is not None:
            try:
                timing = list(timing)
                edge_data['timing'] += timing
            except:
                edge_data['timing'].append(timing)

    def add_state_data(self, s: str, d: object):
        """
    Add state data to a state s the automaton.
        :param s: state
        :param d: data to be added to s
        :return:
        """
        self.Q[s] = d

    def add_state(self, new_state, **kwargs):
        """
    Add state to the automaton.
        :param new_state: State to be added.
        """
        self._G.add_node(new_state, **kwargs)

    def add_states_from(self, new_state, **kwargs):
        """
    Add multiple states to the automaton.
        :param new_state: States to be added.
        """
        self._G.add_nodes_from(new_state, **kwargs)

    def add_transitions_from(self, list_of_tuples, **other):
        """
    Add multiple transition.
        :param list_of_tuples: List of transitions in the form (source_state, destination_state, event, ...<unused>...).
        """
        self._G.add_edges_from(list_of_tuples, **other)

    def add_transition(self, s, d, e, **other):
        """
    Add multiple transition.
        :param list_of_tuples: List of transitions in the form (source_state, destination_state, event, ...<unused>...).
        """
        self._G.add_edge(s, d, e, **other, event=e)

    def add_initial_state(self, states):
        """
    Add initial state(s) of the automaton.
        :param states: States to add.
        """
        if type(states) is str:
            states = (states,)
        for s in states:
            if s is not None and s not in self.q0:
                self.q0[s] = None
            self._G.add_node(s)

    def add_final_state(self, states):
        """
    Add final state(s) of the automaton.
        :param states: States to add.
        """
        if type(states) is str:
            states = (states,)
        for s in states:
            if s is not None and s not in self.final_q:
                self.final_q[s] = None
            self._G.add_node(s)

    def is_transition(self, s, d, e):
        """
    Check if a transition (s,d,e) exists in the automaton.
        :param s: Source.
        :param d: Destination.
        :param e: Event.
        :return:
        """
        transitions = [trans for trans in self.T.values() if trans['source'] == s and
                       trans['destination'] == d and trans['event'] == e]

        is_transition = len(transitions) != 0
        return is_transition

    # def num_occur(self, q, e):
    #     tr = self.get_transition(q, e=e)
    #     if tr[-1]:
    #         return len(tr[-1]['timing'])
    #     else:
    #         return ""

    def num_occur(self, tr):
        return len(tr[-1]['timing'])

    def num_timings(self):
        return sum(len(tr[-1]['timing']) for tr in self.get_transitions())

    def get_num_in(self, q):
        """
        Returns the number of in transitions of state q in the automaton.
        :return: number of transitions.
        """
        if self._G.has_node(q):
            return self._G.in_degree(q)
        else:
            raise Exception(f'State {q} not in the automaton.')

    def get_num_out(self, q):
        """
        Returns the number of out transitions of state q in the automaton.
        :return: number of transitions.
        """
        if self._G.has_node(q):
            return self._G.out_degree(q)
        else:
            raise Exception(f'State {q} not in the automaton.')

    def is_state(self, q):
        return self._G.has_node(q)

    def remove_state(self, s):
        self._G.remove_node(s)
        if s in self.q0:
            self.q0.pop(s)

    def in_transitions(self, s, event=None):
        """
    Get all incoming transitions of state s.
        :param s:
        :return:
        """
        if event is None:
            return self._G.in_edges(s, data=True, keys=True)
        else:
            return [e for e in self._G.in_edges(s, data=True, keys=True) if e[3]['event'] == event]

    def out_transitions(self, s, event=None):
        """
    Get all outgoing transitions of state s.
        :param event:
        :param s:
        :return:
        """
        if event is None:
            return self._G.out_edges(s, data=True, keys=True)
        else:
            return [e for e in self._G.out_edges(s, data=True, keys=True) if e[3]['event'] == event]

    def discrete_event_dynamics(self, q, xt, xk, p) -> tuple:
        e = p["event"]
        new_q = self.UNKNOWN_STATE
        possible_destinations = set(d for s, d, ev in self._G.out_edges(q, data='event') if ev == e)
        if len(possible_destinations) == 1:
            new_q = possible_destinations.pop()
        else:
            stoch_dest = list((d, data.get('p', 0)) for s, d, data in self._G.out_edges(q, data=True) if data['event'] == e)
            if stoch_dest:
                new_q = random.choices([x[0] for x in stoch_dest], weights=[x[1] for x in stoch_dest])[0]
            else: # try to revocer
                dests = set(d for s, d, ev in self._G.edges(data='event') if ev == e)
                if len(dests) == 1:
                    new_q = dests.pop()
        return (new_q,), None, None

    def guards(self, q, x):
        pass

    def timed_event(self, q, xc, xd):
        possible_destinations = list(dict(ev, dest=d) for s, d, ev in self._G.out_edges(q, data=True) if s == q)
        if possible_destinations:
            if len(possible_destinations) == 1:
                dest = possible_destinations[0]
            else:
                dest = np.random.choice(possible_destinations, p=[p['prob']/sum(x['prob'] for x in possible_destinations) if 'prob' in p else 1/len(possible_destinations) for p in possible_destinations])
            if 'time' in dest:
                time = dest['time']
                if callable(time):
                    time = time()
                return time, dest.get('destination', dest['dest'])
        return None, None

    def get_transition(self, s, d=None, e=None, if_more_than_one='raise'):
        """
        Get all transitions with source state s, destination state __d. In case when e is provided, the returned list
        contains transitions where event is e.
        :param if_more_than_one:
        :param s: Source state.
        :param d: Destination state.
        :param e: Event.
        :return:
        """
        transitions = self._G.out_edges(s, keys=True, data=True)
        if e is None and d is not None:
            transitions = [trans for trans in transitions if trans[1] == d]
        elif d is None and e is not None:
            transitions = [trans for trans in transitions if trans[2] == e]
        else:
            transitions = [trans for trans in transitions if trans[1] == d and trans[2] == e]

        if len(transitions) > 1:
            if if_more_than_one == 'raise':
                raise Exception('There are multiple transitions which satisfy the condition.')
            else:
                return transitions
        elif len(transitions) == 0:
            return None
        else:
            return transitions[0]

    def rename_events(self, prefix="e_"):
        """
    Rename events to become e_0, e_1... The old id is stored in the field 'old_symbol' of the state data.
        """
        i = 0
        new_events_dict = OrderedDict()
        for k, v in self.Sigma.items():
            new_key = f'{prefix}{i}'
            new_value = v
            if new_value is None:
                new_value = {}
            new_value['old_symbol'] = k
            i += 1
            new_events_dict[new_key] = new_value
            for t in self.T.values():
                if t['event'] == k:
                    t['event'] = new_key
        # self.Sigma = new_events_dict

    def step(self, q, x0, t, u):
        """
    Simulates one time step of continuous behavior from t to t+dt. Underlying function is solve_ivp with method is 'RK23'.
        :param x0: Initial state vector.
        :param t: Time at start of the step simulation.
        :param u: Arguments passed.....
        :return: Time t+dt, value of state at t+dt
        """
        s = solve_ivp(self.flow, t_span=(t, t + self.dt), y0=x0, method='RK23', args=u)
        xc = s.y[:, -1]
        t = s.t[-1]

        xk = self.time_discrete_dynamics(t, q, x0)
        return t, np.concatenate(xc, xk)

    def __str__(self):
        """
    String representation of the automaton.
        :return:
        """
        return f"""Automaton:
    Number of states: {self.num_modes}
    Number of transitions: {self.num_transitions}
    Initial state(s): {list(self.q0.keys())}
    Final state(s): {list(self.final_q.keys())}"""

    def flow(self, q, p, x, u):
        """
    Flow equation gives derivative of the continuous variables.
        :param q: Current discrete state of the model.
        :param p: Stochastic parameters generated on entry to state current_q.
        :param x: Current continuous state of the model.
        :param u: Calculated internal i signals.
        :return: Derivative of the continuous state x.
        """
        pass

    def inv(self, t, q, x, y, z, p):
        """
    Invariants.
        :param t:
        :param q:
        :param x:
        :param y:
        :param z:
        :param p:
        """
        pass

    def get_transitions(self):
        return list(self._G.edges(data=True, keys=True))

    def print_state(self, v):
        """Prints outgoing transitions of a state v.

        Args:
            v (state): 

        Returns:
            String: Description of the outgoing transitions of the state.
        """
        s = f'<b>{str(v)}</b>'
        for tr in self.out_transitions(v):
            s += f"<br>{tr[2]} -> {tr[1]} [{self.num_occur(tr[0], tr[2])}]"
        return s

    def sample_initial(self):
        if len(self.q0) == 0:
            current_q = np.random.choice(list(self.discrete_states.keys()), 1)[-1]
            warnings.warn(
                'Initial state not defined, sampling initial state uniformly from the set of all states.')
        else:
            current_q = np.random.choice(list(self.q0.keys()), 1)[-1]
        return current_q

    # def simulate(self, finish_time=100, current_q=None):
    #     """
    #     Simulates behaviour of the system.
    #     :param finish_time: Time when simulation finishes.
    #     :return: generated data.
    #     """
    #
    #     if current_q is None:
    #         if len(self.q0) == 0:
    #             current_q = np.random.choice(list(self.discrete_states.keys()), 1)[-1]
    #             warnings.warn(
    #                 'Initial state not defined, sampling initial state uniformly from the set of all states.')
    #         else:
    #             current_q = np.random.choice(list(self.q0.keys()), 1)[-1]
    #
    #     state_sequence = []
    #     data_state = []
    #     t = 0
    #     last_x = None
    #     last_output = None
    #     states = []
    #     data = []
    #     current_e = None
    #
    #     while True:
    #         cont_state, cont_time, cont_output = self.__step_continuous()
    #         last_x = dict(cont_state.iloc[-1])
    #         last_output = dict(cont_output.iloc[-1])
    #         cont_time = cont_time + t - cont_time.iloc[0, 0]
    #         clock = cont_time.iloc[-1, 0] - cont_time.iloc[0, 0]
    #
    #         current_state = current_q
    #
    #         observed_current_state = current_state
    #         state_sequence.append(
    #             pd.DataFrame(np.full((cont_time.size - 1, 3), (current_state, observed_current_state, current_e)),
    #                          index=cont_time.iloc[0:-1, 0],
    #                          columns=['State', 'Observed State', 'Event']))
    #
    #         data_state.append(cont_output.iloc[:-1].set_index(cont_time.iloc[0:-1, 0]))
    #
    #         tr = self.out_transitions(current_state)
    #
    #         if len(tr) == 0:
    #             break
    #         elif len(tr) != 1:
    #             tr = np.random.choice(tr, 1)[-1]
    #             warnings.warn('Multiple transitions can occur.')
    #         else:
    #             tr = tr[0]
    #
    #         last_q = current_q
    #         current_e = tr.event
    #         self.apply_sim_event(current_e)
    #         if cont_time.size == 0:
    #             break
    #         t += clock
    #
    #         if t >= finish_time:
    #             break
    #
    #         states.append(pd.concat(state_sequence, axis=0))
    #         data.append(pd.concat(data_state, axis=0))
    #     return states, data

    def predict_state(self, data_collection, time_col_name=None, discr_col_names=None):
        for data in data_collection:
            data["StateEstimate"] = None
            data["Event"] = None

            prev_discr_state = None
            prev_time = None

            if discr_col_names is None:
                discr_col_names = data.columns
            if time_col_name is not None:
                discr_col_names -= time_col_name

            for row in data[discr_col_names].itertuples(index=True):
                if time_col_name is not None:
                    time = data[time_col_name].iloc[row[0]]
                else:
                    time = row[0]
                discr_state = row[1:]
                if prev_discr_state is not None and prev_discr_state != discr_state:
                    event = np.asarray(discr_state) - np.asarray(prev_discr_state)
                    event = ' '.join(str(x) for x in event)
                    data.loc[row[0], "Event"] = event

                data.loc[row[0], "StateEstimate"] = signal_vector_to_state(discr_state)
                prev_discr_state = discr_state
                prev_time = time
        return data_collection


    def read_event(self, t, e, clear_p=False, keep_p=None, **kwargs):
        if keep_p:
            for k in keep_p:
                kwargs[k] = self._p[k]
        if clear_p or keep_p:
            self._p = kwargs
        else:
            self._p.update(kwargs)
        self._p.pop('Error Message', None)
        self._t = t

        self._p['event'] = e

        new_q, self._xt, self._xk = self.discrete_event_dynamics(self._q, self._xt, self._xk, self._p)
        new_q = new_q[0]

        if new_q == self.UNKNOWN_STATE:
            text2 = '{}->{}->{}'.format(self.discrete_state, e, new_q)
            self._p['Error Message'] = text2
        else:
            dl = self.decision_logic
            # d = self.check_decisions(self.x, self.overall_system.state)
            if dl and not self.overall_system.is_unknown(self):
                try:
                    state_key = json.dumps(self.overall_system.get_choice_state(self), sort_keys=True)
                except:
                    print('Problem getting choice state')
                old_e = self.choices_set.get(state_key, None)
                if old_e is not None:
                    if self._e not in old_e:
                        # warnings.warn(
                        #     '{}: Different decision for same state {}: {} vs {}'.format(t, state_key, old_e, self._e))
                        self.choices_set[state_key][self._e] = [str(t)]
                    else:
                        self.choices_set[state_key][self._e].append(str(t))
                else:
                    self.choices_set[state_key] = {self._e: [str(t)]}
                # if self._e not in (x[0] for x in d):
                #     warnings.warn('{}: Not allowed decision in state {}: {}'.format(t, state_key, self._e))

        self.discrete_state = new_q
        self._past_t.append(t)
        self._past_p.append(self._p)
        if len(self._discrete_state_data):
            self._discrete_state_data[-1]["Finish"] = t
        self._discrete_state_data.append(dict(Time=t, Event=e, Mode=self.discrete_state, **self._p))
        self._discrete_output_data.append([t, *self._d, self._p.get('event', None)])

num_modes property

Returns the number of modes in the automaton. :return: number of states.

num_transitions property

Returns the number of transitions in the automaton. :return: number of transitions.

state property writable

Automata discrete state is uni-variate. :return:

__init__(states=None, transitions=None, unknown_state='raise', id='', initial_q=(), initial_r=None, final_q=(), super_states=(), decision_states=(), **kwargs)

Class initialization from lists of elements. :param states: Discrete states / modes of continuous behavior. :param events: The events that trigger state transitions. :param transitions: The transition information. If a collection of dicts then dict should contain "source", "dest" and "event". The other attributes will be added as data of that transition. Alternatively, a collection of tuples can be used of the form (source, event, dest, *). :param unknown_state: The name of unknown states during "play in", if "raise", an exception will be raised.

Source code in ml4cps/automata/base.py

def __init__(self, states: list = None, transitions: list = None,
             unknown_state: str = 'raise', id="", initial_q=(), initial_r=None, final_q=(), super_states=(), decision_states=(),
             **kwargs):
    """
    Class initialization from lists of elements.
    :param states: Discrete states / modes of continuous behavior.
    :param events: The events that trigger state transitions.
    :param transitions: The transition information. If a collection of dicts then dict should contain "source",
    "dest" and "event". The other attributes will be added as data of that transition. Alternatively, a collection
    of tuples can be used of the form (source, event, dest, *).
    :param unknown_state: The name of unknown states during "play in", if "raise", an exception will be raised.
    """
    self._G = nx.MultiDiGraph()
    if initial_q and isinstance(initial_q, str):
        initial_q = [initial_q]
    self.q0 = OrderedDict.fromkeys(initial_q)
    self.initial_r = initial_r
    if final_q and isinstance(final_q, str):
        final_q = [final_q]
    self.final_q = OrderedDict.fromkeys(final_q)
    self.previous_node_positions = None
    self.UNKNOWN_STATE = unknown_state
    if super_states is not None:
        if type(super_states) is str:
            self.__super_states = [super_states]
        else:
            self.__super_states = list(super_states)

        # self._G.add_nodes_from(self.__super_states)

    if decision_states is not None:
        if type(decision_states) is str:
            self.decision_states = [decision_states]
        else:
            self.decision_states = list(decision_states)

    if states is not None:
        self._G.add_nodes_from(states)

    if transitions is not None:
        for tr in transitions:
            if type(tr) is dict:
                self._G.add_edge(tr.pop('source'), tr.pop('destination'), event=tr.pop('event'), **tr)
            else:
                self._G.add_edge(tr[0], tr[2], event=tr[1])

    if 'discr_state_names' not in kwargs:
        kwargs['discr_state_names'] = ['Mode']
    elif type(kwargs['discr_state_names']) is str:
        kwargs['discr_state_names'] = [kwargs['discr_state_names']]
    CPSComponent.__init__(self, id, **kwargs)

__str__()

String representation of the automaton. :return:

Source code in ml4cps/automata/base.py

def __str__(self):
    """
String representation of the automaton.
    :return:
    """
    return f"""Automaton:
Number of states: {self.num_modes}
Number of transitions: {self.num_transitions}
Initial state(s): {list(self.q0.keys())}
Final state(s): {list(self.final_q.keys())}"""

add_final_state(states)

Add final state(s) of the automaton. :param states: States to add.

Source code in ml4cps/automata/base.py

def add_final_state(self, states):
    """
Add final state(s) of the automaton.
    :param states: States to add.
    """
    if type(states) is str:
        states = (states,)
    for s in states:
        if s is not None and s not in self.final_q:
            self.final_q[s] = None
        self._G.add_node(s)

add_initial_state(states)

Add initial state(s) of the automaton. :param states: States to add.

Source code in ml4cps/automata/base.py

def add_initial_state(self, states):
    """
Add initial state(s) of the automaton.
    :param states: States to add.
    """
    if type(states) is str:
        states = (states,)
    for s in states:
        if s is not None and s not in self.q0:
            self.q0[s] = None
        self._G.add_node(s)

add_state(new_state, **kwargs)

Add state to the automaton. :param new_state: State to be added.

Source code in ml4cps/automata/base.py

def add_state(self, new_state, **kwargs):
    """
Add state to the automaton.
    :param new_state: State to be added.
    """
    self._G.add_node(new_state, **kwargs)

add_state_data(s, d)

Add state data to a state s the automaton. :param s: state :param d: data to be added to s :return:

Source code in ml4cps/automata/base.py

def add_state_data(self, s: str, d: object):
    """
Add state data to a state s the automaton.
    :param s: state
    :param d: data to be added to s
    :return:
    """
    self.Q[s] = d

add_states_from(new_state, **kwargs)

Add multiple states to the automaton. :param new_state: States to be added.

Source code in ml4cps/automata/base.py

def add_states_from(self, new_state, **kwargs):
    """
Add multiple states to the automaton.
    :param new_state: States to be added.
    """
    self._G.add_nodes_from(new_state, **kwargs)

add_transition(s, d, e, **other)

Add multiple transition. :param list_of_tuples: List of transitions in the form (source_state, destination_state, event, ......).

Source code in ml4cps/automata/base.py

def add_transition(self, s, d, e, **other):
    """
Add multiple transition.
    :param list_of_tuples: List of transitions in the form (source_state, destination_state, event, ...<unused>...).
    """
    self._G.add_edge(s, d, e, **other, event=e)

add_transitions_from(list_of_tuples, **other)

Add multiple transition. :param list_of_tuples: List of transitions in the form (source_state, destination_state, event, ......).

Source code in ml4cps/automata/base.py

def add_transitions_from(self, list_of_tuples, **other):
    """
Add multiple transition.
    :param list_of_tuples: List of transitions in the form (source_state, destination_state, event, ...<unused>...).
    """
    self._G.add_edges_from(list_of_tuples, **other)

flow(q, p, x, u)

Flow equation gives derivative of the continuous variables. :param q: Current discrete state of the model. :param p: Stochastic parameters generated on entry to state current_q. :param x: Current continuous state of the model. :param u: Calculated internal i signals. :return: Derivative of the continuous state x.

Source code in ml4cps/automata/base.py

def flow(self, q, p, x, u):
    """
Flow equation gives derivative of the continuous variables.
    :param q: Current discrete state of the model.
    :param p: Stochastic parameters generated on entry to state current_q.
    :param x: Current continuous state of the model.
    :param u: Calculated internal i signals.
    :return: Derivative of the continuous state x.
    """
    pass

get_num_in(q)

Returns the number of in transitions of state q in the automaton. :return: number of transitions.

Source code in ml4cps/automata/base.py

def get_num_in(self, q):
    """
    Returns the number of in transitions of state q in the automaton.
    :return: number of transitions.
    """
    if self._G.has_node(q):
        return self._G.in_degree(q)
    else:
        raise Exception(f'State {q} not in the automaton.')

get_num_out(q)

Returns the number of out transitions of state q in the automaton. :return: number of transitions.

Source code in ml4cps/automata/base.py

def get_num_out(self, q):
    """
    Returns the number of out transitions of state q in the automaton.
    :return: number of transitions.
    """
    if self._G.has_node(q):
        return self._G.out_degree(q)
    else:
        raise Exception(f'State {q} not in the automaton.')

get_transition(s, d=None, e=None, if_more_than_one='raise')

Get all transitions with source state s, destination state __d. In case when e is provided, the returned list contains transitions where event is e. :param if_more_than_one: :param s: Source state. :param d: Destination state. :param e: Event. :return:

Source code in ml4cps/automata/base.py

def get_transition(self, s, d=None, e=None, if_more_than_one='raise'):
    """
    Get all transitions with source state s, destination state __d. In case when e is provided, the returned list
    contains transitions where event is e.
    :param if_more_than_one:
    :param s: Source state.
    :param d: Destination state.
    :param e: Event.
    :return:
    """
    transitions = self._G.out_edges(s, keys=True, data=True)
    if e is None and d is not None:
        transitions = [trans for trans in transitions if trans[1] == d]
    elif d is None and e is not None:
        transitions = [trans for trans in transitions if trans[2] == e]
    else:
        transitions = [trans for trans in transitions if trans[1] == d and trans[2] == e]

    if len(transitions) > 1:
        if if_more_than_one == 'raise':
            raise Exception('There are multiple transitions which satisfy the condition.')
        else:
            return transitions
    elif len(transitions) == 0:
        return None
    else:
        return transitions[0]

in_transitions(s, event=None)

Get all incoming transitions of state s. :param s: :return:

Source code in ml4cps/automata/base.py

def in_transitions(self, s, event=None):
    """
Get all incoming transitions of state s.
    :param s:
    :return:
    """
    if event is None:
        return self._G.in_edges(s, data=True, keys=True)
    else:
        return [e for e in self._G.in_edges(s, data=True, keys=True) if e[3]['event'] == event]

inv(t, q, x, y, z, p)

Invariants. :param t: :param q: :param x: :param y: :param z: :param p:

Source code in ml4cps/automata/base.py

def inv(self, t, q, x, y, z, p):
    """
Invariants.
    :param t:
    :param q:
    :param x:
    :param y:
    :param z:
    :param p:
    """
    pass

is_transition(s, d, e)

Check if a transition (s,d,e) exists in the automaton. :param s: Source. :param d: Destination. :param e: Event. :return:

Source code in ml4cps/automata/base.py

def is_transition(self, s, d, e):
    """
Check if a transition (s,d,e) exists in the automaton.
    :param s: Source.
    :param d: Destination.
    :param e: Event.
    :return:
    """
    transitions = [trans for trans in self.T.values() if trans['source'] == s and
                   trans['destination'] == d and trans['event'] == e]

    is_transition = len(transitions) != 0
    return is_transition

merge(q1, q2)

If two states are compatible, they are merged with the function merge. The transitions of the automaton, the in- and outdegree of the states and the number of transitions happening are adjusted.

Source code in ml4cps/automata/base.py

def merge(self, q1, q2):
    """
    If two states are compatible, they are merged with the function merge. The transitions
    of the automaton, the in- and outdegree of the states and the number of transitions
    happening are adjusted.
    """
    intr = self.in_transitions(q2)
    # outtr_q1 = list(self.out_transitions(q1))
    outtr_q2 = list(self.out_transitions(q2))
    # nx.contracted_nodes(self._G, w, v, copy=False)
    # set event

    for tr in intr:
        if tr[0] != q2:
            self.add_single_transition(tr[0], q1, tr[2], timing=tr[-1]['timing'])
    # if q1=='q0' and q2 == 'q192':
    #     print('found')
    for tr in outtr_q2:
        dest = tr[1]
        if dest == q2:
            dest = q1
        self.add_single_transition(q1, dest, tr[2], timing=tr[-1]['timing'])
    self.remove_state(q2)

out_transitions(s, event=None)

Get all outgoing transitions of state s. :param event: :param s: :return:

Source code in ml4cps/automata/base.py

def out_transitions(self, s, event=None):
    """
Get all outgoing transitions of state s.
    :param event:
    :param s:
    :return:
    """
    if event is None:
        return self._G.out_edges(s, data=True, keys=True)
    else:
        return [e for e in self._G.out_edges(s, data=True, keys=True) if e[3]['event'] == event]

print_state(v)

Prints outgoing transitions of a state v.

Parameters:

Name	Type	Description	Default
v	state		required

Returns:

Name	Type	Description
String		Description of the outgoing transitions of the state.

Source code in ml4cps/automata/base.py

def print_state(self, v):
    """Prints outgoing transitions of a state v.

    Args:
        v (state): 

    Returns:
        String: Description of the outgoing transitions of the state.
    """
    s = f'<b>{str(v)}</b>'
    for tr in self.out_transitions(v):
        s += f"<br>{tr[2]} -> {tr[1]} [{self.num_occur(tr[0], tr[2])}]"
    return s

remove_transition(source, dest)

Remove the transition(s) from source to dest. :param source: :param dest: :return:

Source code in ml4cps/automata/base.py

def remove_transition(self, source, dest):
    """
    Remove the transition(s) from source to dest.
    :param source:
    :param dest:
    :return:
    """
    self._G.remove_edge(source, dest)

rename_events(prefix='e_')

Rename events to become e_0, e_1... The old id is stored in the field 'old_symbol' of the state data.

Source code in ml4cps/automata/base.py

def rename_events(self, prefix="e_"):
    """
Rename events to become e_0, e_1... The old id is stored in the field 'old_symbol' of the state data.
    """
    i = 0
    new_events_dict = OrderedDict()
    for k, v in self.Sigma.items():
        new_key = f'{prefix}{i}'
        new_value = v
        if new_value is None:
            new_value = {}
        new_value['old_symbol'] = k
        i += 1
        new_events_dict[new_key] = new_value
        for t in self.T.values():
            if t['event'] == k:
                t['event'] = new_key

step(q, x0, t, u)

Simulates one time step of continuous behavior from t to t+dt. Underlying function is solve_ivp with method is 'RK23'. :param x0: Initial state vector. :param t: Time at start of the step simulation. :param u: Arguments passed..... :return: Time t+dt, value of state at t+dt

Source code in ml4cps/automata/base.py

def step(self, q, x0, t, u):
    """
Simulates one time step of continuous behavior from t to t+dt. Underlying function is solve_ivp with method is 'RK23'.
    :param x0: Initial state vector.
    :param t: Time at start of the step simulation.
    :param u: Arguments passed.....
    :return: Time t+dt, value of state at t+dt
    """
    s = solve_ivp(self.flow, t_span=(t, t + self.dt), y0=x0, method='RK23', args=u)
    xc = s.y[:, -1]
    t = s.t[-1]

    xk = self.time_discrete_dynamics(t, q, x0)
    return t, np.concatenate(xc, xk)

try_merge_states(state1, state2, try_fun=None)

Merge state2 into state1 and update transitions.

Source code in ml4cps/automata/base.py

def try_merge_states(self, state1, state2, try_fun=None):
    """Merge state2 into state1 and update transitions."""
    old_G = self._G
    make_final = state2 in self.final_q and state1 not in self.final_q
    make_initial = state2 in self.q0 and state1 not in self.q0
    self._G = nx.contracted_nodes(self._G, state1, state2)
    # make it deterministic again
    events = [x[3]['event'] for x in self.out_transitions(state1)]
    deterministic = len(set(events)) == len(events)

    if deterministic:
        if make_initial:
            self.add_initial_state(state1)
        if make_final:
            self.add_final_state(state1)

        if try_fun is not None and not try_fun(self):
            self._G = old_G
            if make_final:
                self.final_q.pop(state1)
            if make_initial:
                self.q0.pop(state1)
    else:
        self._G = old_G

CPSComponent

Bases: PythonModel, Simulator

General hybrid system class based on scipy and simpy.

Source code in ml4cps/cps/base.py

class CPSComponent(PythonModel, sim.Simulator):
    """
    General hybrid system class based on scipy and simpy.
    """

    def __init__(self, id, t=0, initial_q=(), initial_xt: list = (), initial_xk: list = (), dt=-1., p=None,
                 cont_state_names=None, discr_state_names=None, discr_output_names=None,
                 cont_output_names=None, unknown_state="Unknown"):
        sim.Simulator.__init__(self)
        PythonModel.__init__(self)

        self._parent_system = None
        self.decision_logic = None
        self.id = id
        self.dt = dt

        # State variables
        self._t = t
        self._q = initial_q
        self._xt = initial_xt
        self._xk = initial_xk

        # Other variables
        self._y = None
        self._u = None
        self._d = ()
        self._p = {} if p is None else p

        # simpy events
        self._pending_events = []
        self._block_event = None

        # List to append the past values
        self._past_p = []
        self._past_t = []
        self._discrete_state_data = []
        self._discrete_output_data = []
        self._continuous_state_data = []
        self._continuous_output_data = []

        # Variable names
        self.cont_state_names = cont_state_names
        self.cont_output_names = cont_output_names
        self.discrete_state_names = discr_state_names
        self.discrete_output_names = discr_output_names

        self.UNKNOWN_STATE = unknown_state

        self.output_y = None
        self.output_d = None



    # def copy(self):
    #     new_copy = CPSComponent(self.id, self._t, initial_q=self._q, initial_xt=self._xt,
    #                             initial_xk = self._xk, dt = self.dt, p=self._p,
    #     cont_state_names = copy.copy(self.cont_state_names), discr_state_names = copy.copy(self.discrete_state_names),
    #                             discr_output_names=copy.copy(self.discrete_output_names),
    #     cont_output_names=copy.copy(self.cont_output_names), unknown_state = copy.copy(self.UNKNOWN_STATE))
    #     return new_copy

    def copy(self):
        new_copy = copy.deepcopy(self)
        return new_copy

    @property
    def parent_system(self):
        return self._parent_system

    @property
    def full_id(self):
        full_id = self.id
        s = self.parent_system
        while s.parent_system:
            full_id = s.id + "." + full_id
            s = s.parent_system
        return full_id

    @property
    def overall_system(self):
        s = self
        while s.parent_system:
            s = s.parent_system
        return s

    @property
    def state(self):
        """
        System state is a vector of all state variables.
        :return:
        """
        return self._q, self._xt, self._xk

    @state.setter
    def state(self, value):
        self._q = value[0]
        self._xt = value[1]
        self._xk = value[2]

    @property
    def discrete_state(self):
        return self._q

    @discrete_state.setter
    def discrete_state(self, value):
        self._q = value

    def is_decision(self, state, overall_state):
        return False

    def __str__(self):
        return '{}: {}'.format(self.id, self._q)

    def _step_continuous(self, clock_start, clock_finish):
        return self.__step_continuous(clock_start, clock_finish)

    def simulation_process_simpy(self, env, max_time, verbose=False):
        """
            The function simulates single concurrent thread of one component (CPSComponent).
            Parameters
            ----------
            env : simpy.Environment
                It is simpy environment used to synchronize multiple parallel threads during simulation.
            max_time : float
                It is the maximum time that the simulation will perform.
            verbose : bool
                Should the function print execution logs (default is False).
        """
        env.process(self.discrete_dynamics_process(env, max_time, verbose=verbose))
        env.process(self.continuous_dynamics_process(env, max_time, verbose=verbose))
        env.process(self.discrete_time_dynamics_process(env, max_time, verbose=verbose))

    def discrete_event_dynamics(self, q, xt, xk, p) -> tuple:
        pass

    def discrete_dynamics_process(self, env, max_time, verbose=False):
        # self._env = env
        self._u = self.input(self._q, 0)
        if self._p is None:
            self._p = self.context(self._q, None, None, 0)
        self._past_t = [0]
        self._past_p = [self._p]


        self._discrete_state_data = []
        self._discrete_output_data = []

        while True:
            discr_state = dict(zip(self.discrete_state_names, self._q))
            if self.output_d is None:
                self._d = self.discrete_state
            else:
                self._d = self.output_d(self.discrete_state, self._xt, self._xk)
            self._discrete_state_data.append(dict(Time=env.now, **discr_state, **self._p))
            self._discrete_output_data.append([env.now, *self._d])

            # Stop the simulation if the max time is reached
            if env.now > max_time:
                if verbose:
                    print(f'Stop because of maximum time {max_time} is reached.')
                break
            # Stop if the stopping condition of the overall system is met
            elif self.overall_system.finish_condition():
                if verbose:
                    print('Stop because stop condition of the overall system is met.')
                break

            # DETERMINE THE NEXT EVENT
            # 1. CHECK IF TIMED EVENT
            try:
                event_delay, new_state_value = self.timed_event(*self.state)
            except Exception as ex:
                print(f"{self.id}: Exception during self.timed_transition in state {self._q}")
                raise ex

            # 2. IF NOT TIMED THEN WAIT
            if event_delay is None:
                if verbose:
                    print('{}: Waiting in: {}'.format(self.id, self._q))
                if self._block_event is None:
                    self._block_event = env.event()
                    yield self._block_event
                else:
                    raise Exception("{}: State: {}. Exception during block event.".format(self.id, self._q))
                event = self._block_event.value
                self._block_event = None

                self._p['event'] = event
                new_q, xt, xk = self.discrete_event_dynamics(self._q, self._xt, self._xk, self._p)
                if verbose:
                    print(f'{self._q}->{event}->{new_q}')
                if xt is not None:
                    self._xt = xt
                if xk is not None:
                    self._xk = xk

                self._q = new_q
            else:
                if verbose:
                    print('Waiting in: {} for {}. New state: {}'.format(self._q, event_delay, new_state_value))
                yield env.timeout(event_delay)
                self._q = (new_state_value,)

            try:
                self.on_entry(self._q, self._p)
            except Exception as ex:
                print(f"{self.id}: Exception during discrete event dynamics '{self._q}'->")
                raise ex

    def discrete_time_dynamics_process(self, env, max_time, verbose=False):
        while (self._xk is not None and len(self._xk) > 0) or self.output_y is not None:
            self._continuous_state_data.append((self._t, *self._xk))
            self._continuous_output_data.append((self._t, *self.output_y(*self.state)))

            # Stop the simulation if the max time is reached
            if env.now > max_time:
                if verbose:
                    print(f'Stop because of maximum time {max_time} is reached.')
                break
            # Stop if the stopping condition of the overall system is met
            elif self.overall_system.finish_condition():
                if verbose:
                    print('Stop because stop condition of the overall system is met.')
                break

            yield env.timeout(self.dt)
            xk = self.time_discrete_dynamics(self._q[0], self._p, self._xk, self._u)
            self._xk = xk
            self._t = env.now
            grds = self.guards(self._q[0], self._xk)
            if grds is not None:
                self.apply_sim_event(grds, env)


    def guards(self, q, x):
        pass

    def continuous_dynamics_process(self, env, max_time, verbose=False):
        # try:
        #     if len(self._xt) > 0 or len(self._xk) > 0:  # there is time-continuous state variable
        #         self.__step_continuous(*self.state)
        # except Exception as ex:
        #     print_exc()
        #     raise Exception('Exception during self.__step_continuous in state {}'.format(self._q))

        # EXECUTE THE NEXT EVENT
        if False:
            yield
        # if verbose:
        #     print('--------------------')
        #
        # self._past_t.append(env.now)
        # self._past_p.append(self._p)
        #
        # self._discrete_state_data[-1]['Finish'] = env.now
        # discr_state = dict(zip(self.discrete_state_names, self._q))
        # self._discrete_state_data.append(
        #     dict(Time=env.now, Finish=None, **discr_state, **self._p))
        # self._discrete_output_data.append([env.now, *self._d])
        #
        # if verbose:
        #     print('{}: Time: {} State: {}'.format(self.id, env.now, self._q))
        #
        # if self._xt is not None:
        #     self._continuous_state_data.append([env.now, *self._xt, *self._xk])
        # if self._y is not None:
        #     self._continuous_output_data.append([env.now, *self._y])
        # if self._xt is not None:
        #     self._xt = self.__step_continuous(env.now - clock_start)
        # self._d, self._y = self.output(self._q, 0, self._xt, self._xk, self._u, self._p)
        #
        # print('--------------------')
        # print('{}: Simulation finished'.format(self.id))

    def simulate(self, finish_time, verbose=False, save_choices=False):
        """Simulates behaviour of the system until the finish_time is reached.

        Parameters
        ----------
        finish_time : Time when simulation finishes.
        verbose : Should the execution log be printed in detail (default is False).
        save_choices : Should the choices for each component be saved to json files after
        the simulation is finished (default is False).
        """

        env = simpy.Environment()
        finish_time = float(finish_time)

        if verbose:
            print('Simulation started: ')

        self.simulation_process_simpy(env, finish_time, verbose)
        env.run(until=finish_time)

        discr_state_data = {}
        discr_output_data = {}
        cont_state_data = {}
        cont_output_data = {}

        try:
            if len(self._xt) > 0 and len(self._xt) > 0 and finish_time > self._t:  # there is time-continuous state variable
                self._step_continuous(self._t, finish_time)
        except Exception as ex:
            print_exc()
            raise Exception('Exception during self._step_continuous in state {}'.format(self._q))

        try:
            self.finish(env.now)

            discr_state_data = pd.DataFrame(self._discrete_state_data).set_index("Time")
            discr_output_data = tools.data_list_to_dataframe(None, self._discrete_output_data,
                                                                   self.discrete_output_names, 'd')
            try:
                cont_state_data = tools.data_list_to_dataframe(self.id, self._continuous_state_data,
                                                               self.cont_state_names, 'x')
            except Exception as ex:
                print_exc()
            cont_output_data = tools.data_list_to_dataframe(self.id, self._continuous_output_data,
                                                                     self.cont_output_names, 'y')
        except Exception as ex:
            print_exc()
            warnings.warn('Simulation failed.')

        if verbose:
            print('Simulation finished.')
        return discr_state_data, discr_output_data, cont_state_data, cont_output_data, env.now

    def get_sim_state(self):
        return self._q, self._p, self._y, self._block_event

    def set_sim_state(self, q, e, p, y, block_event):
        self._q = q
        self._p = p
        self._y = y
        self._block_event = block_event

    def finish_condition(self):
        pass

    def reinitialize(self, t, state=None):
        if state is not None:
            self.state = state

        self._p = self.context(self._q, None, None, 0)
        self._u = self.input(self._q, 0)
        self._past_t = [t]
        self._past_p = [self._p]
        # self.output(self._q, 0, self._xt, self._xk, self._u, self._p)
        self._continuous_state_data = []
        self._continuous_output_data = []
        try:
            discr_state = dict(zip(self.discrete_state_names, self._q))
        except:
            raise Exception('Exception during discrete state initialization')

        self._discrete_state_data = [dict(Time=t, **discr_state, **self._p)]
        self._discrete_output_data = []
        self._block_event = None
        self._pending_events = []
        self._t = t

    def get_execution_data(self):
        try:
            data = pd.DataFrame(self._discrete_state_data) #, columns=['Time', 'Finish', 'State', 'Event'] + list(self._p.keys()))
        except ValueError as ve:
            warnings.warn(str(self._discrete_state_data))
            warnings.warn('Discrete state data not available due to value error.')
            for i, row in enumerate(self._discrete_state_data):
                for k, v in row.items():
                    if not is_scalar(v):
                        print(f"Non-scalar at row {i}, key '{k}'")
                        print("Type:", type(v))
                        print("Length:", len(v) if hasattr(v, "__len__") else "N/A")
                        break
            raise ve

        data['Finish'] = data['Time'].shift(-1)
        data['Duration'] = pd.to_timedelta(data['Finish'] - data['Time']).dt.total_seconds()
        return data

    # def clear_execution_data(self):
    #     self.discrete_state_data = None
    #     data = pd.DataFrame(self._discrete_state_data) #, columns=['Time', 'Finish', 'State', 'Event'] + list(self._p.keys()))
    #     data['Finish'] = data['Time'].shift(-1)
    #     data['Duration'] = pd.to_timedelta(data['Finish'] - data['Time']).dt.total_seconds()
    #     return data

    def update_input(self, u):  # Set i if you want from outside
        self._u = u

    def output_d(self, q, xt, xk):
        return ()

    def output_y(self, q, xt, xk):
        return ()

    def input(self, q, clock) -> tuple:
        pass

    def context(self, q, past_t, past_p, t) -> tuple:
        return dict()

    def time_continuous_dynamics(self, t, xt, xk, q, u, p, y):
        pass

    def time_discrete_dynamics(self, q, p, x, u):
        pass

    def on_entry(self, q, context):
        pass

    def __step_continuous(self, clock_start, clock_finish):
        """
    Simulates one time step of continuous behavior from t to t+dt. Underlying function is solve_ivp with method is 'RK23'.
        :param x0: Initial state vector.
        :param t: Time at start of the step simulation.
        :param u: Arguments passed.....
        :return: Time t+dt, value of state at t+dt
        """

        s = solve_ivp(self.time_continuous_dynamics, t_eval=np.arange(clock_start + self.dt, clock_finish + self.dt, self.dt),
                      t_span=(clock_start, clock_finish + self.dt), y0=self._xt, method='RK23',
                      args=(self._xk, self._q, self._u, self._p, self._y))

        self._xk = self.time_discrete_dynamics(self._q, self._p, self._xt, self._u)
        self._xt = s.y[-1]

        # self._continuous_state_data += np.concatenate([s.t.T[None], s.y]).T.tolist()

    def invariants(self, q, clock, xc, xd, y):
        pass

    def timed_event(self, q, xc, xd):
        """
        Calculates if and when the next time event will happen and the new state values.
        :param t: Current time.
        :param q: Discrete-event part of the state.
        :param xc: Time-continuous part of the state.
        :param xd: Time-discrete part of the state.
        :return: Time delay of the timed event, new state value.
        """
        return None, None

    def wait(self, q):
        pass

    def apply_sim_event(self, e, env=None):
        """
        The event e is applied in this component's simpy execution, this means that the process must wait for an event.
        :param e: Event to apply.
        :return:
        """
        if not self._block_event:
            # self._block_event = self._env.now
            self._block_event = env.now
        # if not self._block_event:
        #     raise Exception("Block event is not set, but event is applied.")
        elif self._block_event.triggered:
            # pass
            # if self._e != e:
            raise Exception('Should not happen')
        else:
            self._block_event.succeed(value=e)

    def finish(self, t, **kwargs):
        if self._p is None:
            self._p = kwargs
        else:
            self._p.update(kwargs)
        self._t = t
        # self._e = None
        # self._q = None
        self._past_p.append(self._p)
        # self._discrete_state_data[-1]['Finish'] = t

    def get_alternatives(self, state, system_state):
        return None

    def get_decision_state(self, state, overall_state):
        return state

state property writable

System state is a vector of all state variables. :return:

__step_continuous(clock_start, clock_finish)

Simulates one time step of continuous behavior from t to t+dt. Underlying function is solve_ivp with method is 'RK23'. :param x0: Initial state vector. :param t: Time at start of the step simulation. :param u: Arguments passed..... :return: Time t+dt, value of state at t+dt

Source code in ml4cps/cps/base.py

def __step_continuous(self, clock_start, clock_finish):
    """
Simulates one time step of continuous behavior from t to t+dt. Underlying function is solve_ivp with method is 'RK23'.
    :param x0: Initial state vector.
    :param t: Time at start of the step simulation.
    :param u: Arguments passed.....
    :return: Time t+dt, value of state at t+dt
    """

    s = solve_ivp(self.time_continuous_dynamics, t_eval=np.arange(clock_start + self.dt, clock_finish + self.dt, self.dt),
                  t_span=(clock_start, clock_finish + self.dt), y0=self._xt, method='RK23',
                  args=(self._xk, self._q, self._u, self._p, self._y))

    self._xk = self.time_discrete_dynamics(self._q, self._p, self._xt, self._u)
    self._xt = s.y[-1]

apply_sim_event(e, env=None)

The event e is applied in this component's simpy execution, this means that the process must wait for an event. :param e: Event to apply. :return:

Source code in ml4cps/cps/base.py

def apply_sim_event(self, e, env=None):
    """
    The event e is applied in this component's simpy execution, this means that the process must wait for an event.
    :param e: Event to apply.
    :return:
    """
    if not self._block_event:
        # self._block_event = self._env.now
        self._block_event = env.now
    # if not self._block_event:
    #     raise Exception("Block event is not set, but event is applied.")
    elif self._block_event.triggered:
        # pass
        # if self._e != e:
        raise Exception('Should not happen')
    else:
        self._block_event.succeed(value=e)

simulate(finish_time, verbose=False, save_choices=False)

Simulates behaviour of the system until the finish_time is reached.

Parameters

finish_time : Time when simulation finishes. verbose : Should the execution log be printed in detail (default is False). save_choices : Should the choices for each component be saved to json files after the simulation is finished (default is False).

Source code in ml4cps/cps/base.py

def simulate(self, finish_time, verbose=False, save_choices=False):
    """Simulates behaviour of the system until the finish_time is reached.

    Parameters
    ----------
    finish_time : Time when simulation finishes.
    verbose : Should the execution log be printed in detail (default is False).
    save_choices : Should the choices for each component be saved to json files after
    the simulation is finished (default is False).
    """

    env = simpy.Environment()
    finish_time = float(finish_time)

    if verbose:
        print('Simulation started: ')

    self.simulation_process_simpy(env, finish_time, verbose)
    env.run(until=finish_time)

    discr_state_data = {}
    discr_output_data = {}
    cont_state_data = {}
    cont_output_data = {}

    try:
        if len(self._xt) > 0 and len(self._xt) > 0 and finish_time > self._t:  # there is time-continuous state variable
            self._step_continuous(self._t, finish_time)
    except Exception as ex:
        print_exc()
        raise Exception('Exception during self._step_continuous in state {}'.format(self._q))

    try:
        self.finish(env.now)

        discr_state_data = pd.DataFrame(self._discrete_state_data).set_index("Time")
        discr_output_data = tools.data_list_to_dataframe(None, self._discrete_output_data,
                                                               self.discrete_output_names, 'd')
        try:
            cont_state_data = tools.data_list_to_dataframe(self.id, self._continuous_state_data,
                                                           self.cont_state_names, 'x')
        except Exception as ex:
            print_exc()
        cont_output_data = tools.data_list_to_dataframe(self.id, self._continuous_output_data,
                                                                 self.cont_output_names, 'y')
    except Exception as ex:
        print_exc()
        warnings.warn('Simulation failed.')

    if verbose:
        print('Simulation finished.')
    return discr_state_data, discr_output_data, cont_state_data, cont_output_data, env.now

simulation_process_simpy(env, max_time, verbose=False)

The function simulates single concurrent thread of one component (CPSComponent). Parameters

---

env : simpy.Environment It is simpy environment used to synchronize multiple parallel threads during simulation. max_time : float It is the maximum time that the simulation will perform. verbose : bool Should the function print execution logs (default is False).

Source code in ml4cps/cps/base.py

def simulation_process_simpy(self, env, max_time, verbose=False):
    """
        The function simulates single concurrent thread of one component (CPSComponent).
        Parameters
        ----------
        env : simpy.Environment
            It is simpy environment used to synchronize multiple parallel threads during simulation.
        max_time : float
            It is the maximum time that the simulation will perform.
        verbose : bool
            Should the function print execution logs (default is False).
    """
    env.process(self.discrete_dynamics_process(env, max_time, verbose=verbose))
    env.process(self.continuous_dynamics_process(env, max_time, verbose=verbose))
    env.process(self.discrete_time_dynamics_process(env, max_time, verbose=verbose))

timed_event(q, xc, xd)

Calculates if and when the next time event will happen and the new state values. :param t: Current time. :param q: Discrete-event part of the state. :param xc: Time-continuous part of the state. :param xd: Time-discrete part of the state. :return: Time delay of the timed event, new state value.

Source code in ml4cps/cps/base.py

def timed_event(self, q, xc, xd):
    """
    Calculates if and when the next time event will happen and the new state values.
    :param t: Current time.
    :param q: Discrete-event part of the state.
    :param xc: Time-continuous part of the state.
    :param xd: Time-discrete part of the state.
    :return: Time delay of the timed event, new state value.
    """
    return None, None

automata

The module provides learning algorithms for creation of different kinds of automata.

Authors: - Nemanja Hranisavljevic, hranisan@hsu-hh.de, nemanja@ai4cps.com - Tom Westermann, tom.westermann@hsu-hh.de, tom@ai4cps.com

FnDetectChangePoints(xout, udout, xout_shifts)

Detects change points in output and input variables, filters them, and constructs a trace structure. This function analyzes the provided output (xout) and input (udout) data to detect change points using the findChangePoints function. It processes both output and input variables, aggregates and filters the detected change points, and returns a structured trace dictionary containing the processed data and change point information. Args: xout (np.ndarray): Output data array of shape (n_samples, n_outputs). udout (np.ndarray): Input data array of shape (n_samples, n_inputs). xout_shifts (np.ndarray): Shifted output data array, used for further processing. Returns: dict: A dictionary with the following keys: - 'x': Filtered output data array. - 'xs': Filtered shifted output data array. - 'chpoints': Array of global change points detected across all variables. - 'chpoints_per_var': List of arrays, each containing change points for a specific variable. - 'ud': Filtered input data array. - 'labels_num': Empty list (reserved for numeric labels). - 'labels_trace': Empty list (reserved for trace labels). Notes: - Relies on global variables: num_var, num_ud, max_deriv, and chp_depths. - Uses external functions: findChangePoints and filterChangePoints. - The function is intended for use in time-series or sequential data analysis where detecting significant changes in variable values is required.

Source code in ml4cps/automata/learn.py

def FnDetectChangePoints(xout, udout, xout_shifts):
    """
    Detects change points in output and input variables, filters them, and constructs a trace structure.
    This function analyzes the provided output (`xout`) and input (`udout`) data to detect change points
    using the `findChangePoints` function. It processes both output and input variables, aggregates and filters
    the detected change points, and returns a structured trace dictionary containing the processed data and
    change point information.
    Args:
        xout (np.ndarray): Output data array of shape (n_samples, n_outputs).
        udout (np.ndarray): Input data array of shape (n_samples, n_inputs).
        xout_shifts (np.ndarray): Shifted output data array, used for further processing.
    Returns:
        dict: A dictionary with the following keys:
            - 'x': Filtered output data array.
            - 'xs': Filtered shifted output data array.
            - 'chpoints': Array of global change points detected across all variables.
            - 'chpoints_per_var': List of arrays, each containing change points for a specific variable.
            - 'ud': Filtered input data array.
            - 'labels_num': Empty list (reserved for numeric labels).
            - 'labels_trace': Empty list (reserved for trace labels).
    Notes:
        - Relies on global variables: `num_var`, `num_ud`, `max_deriv`, and `chp_depths`.
        - Uses external functions: `findChangePoints` and `filterChangePoints`.
        - The function is intended for use in time-series or sequential data analysis where detecting
          significant changes in variable values is required.
    """

    global num_var, num_ud, max_deriv, chp_depths

    # Initialize global variables
    chpoints = []  # Global changepoints
    chp_depths = np.zeros(max_deriv + 1)  # Purely for debugging
    chp_var = [None] * (num_var + num_ud)  # Local changepoints (per variable)

    # Detect change points for output variables
    for i in range(num_var):
        new_chp = findChangePoints(xout[:, i::num_var], 0, 1, xout.shape[0], max_deriv)
        chpoints = np.union1d(chpoints, new_chp)
        chp_var[i] = np.sort(new_chp)

    # Detect change points for input variables
    for i in range(num_ud):
        new_chp = findChangePoints(udout[:, i], 0, 1, udout.shape[0], 0)
        chpoints = np.union1d(chpoints, new_chp)
        chp_var[num_var + i] = new_chp

    # Filter changepoints
    xout, udout, xout_shifts, chpoints, chp_var = filterChangePoints(xout, udout, xout_shifts, chpoints, chp_var)

    # Create the trace structure
    trace = {
        'x': xout,
        'xs': xout_shifts,
        'chpoints': chpoints,
        'chpoints_per_var': chp_var,
        'ud': udout,
        'labels_num': [],
        'labels_trace': []
    }
    return trace

FnShiftAndDiff(xout, udout, norm_coeff, num_var, num_ud, max_deriv, Ts)

Normalizes and processes state and input data by generating shifted versions and computing derivatives. This function performs the following operations: 1. Normalizes the state output xout using the provided normalization coefficients. 2. Generates shifted duplicates of xout up to the specified maximum derivative order. 3. Computes numerical derivatives of xout up to max_deriv order and appends them to the state data. 4. Strips the initial entries from xout and the shifted data to account for the derivative computation. 5. Normalizes the input data udout (if present) and strips initial entries to match the processed state data. Args: xout (np.ndarray): State output data of shape (n_samples, num_var). udout (np.ndarray): Input data of shape (n_samples, num_ud). norm_coeff (np.ndarray): Normalization coefficients of shape (num_var + num_ud, 1). num_var (int): Number of state variables. num_ud (int): Number of input variables. max_deriv (int): Maximum order of derivatives to compute. Ts (float): Sampling time interval. Returns: Tuple[np.ndarray, np.ndarray, pd.DataFrame]: - xout (np.ndarray): Processed and augmented state data with derivatives, shape (n_samples - max_deriv, ...). - udout (np.ndarray): Normalized input data, shape (n_samples - max_deriv, num_ud). - xout_shifts (pd.DataFrame): Shifted duplicates of the normalized state data, shape (n_samples - max_deriv, ...). Notes: - The function assumes that xout and udout are NumPy arrays and that xout has at least max_deriv rows. - The normalization coefficients should be provided for both state and input variables. - The function uses zero-padding for shifted and derivative computations.

Source code in ml4cps/automata/learn.py

def FnShiftAndDiff(xout, udout, norm_coeff, num_var, num_ud, max_deriv, Ts):
    """
    Normalizes and processes state and input data by generating shifted versions and computing derivatives.
    This function performs the following operations:
    1. Normalizes the state output `xout` using the provided normalization coefficients.
    2. Generates shifted duplicates of `xout` up to the specified maximum derivative order.
    3. Computes numerical derivatives of `xout` up to `max_deriv` order and appends them to the state data.
    4. Strips the initial entries from `xout` and the shifted data to account for the derivative computation.
    5. Normalizes the input data `udout` (if present) and strips initial entries to match the processed state data.
    Args:
        xout (np.ndarray): State output data of shape (n_samples, num_var).
        udout (np.ndarray): Input data of shape (n_samples, num_ud).
        norm_coeff (np.ndarray): Normalization coefficients of shape (num_var + num_ud, 1).
        num_var (int): Number of state variables.
        num_ud (int): Number of input variables.
        max_deriv (int): Maximum order of derivatives to compute.
        Ts (float): Sampling time interval.
    Returns:
        Tuple[np.ndarray, np.ndarray, pd.DataFrame]:
            - xout (np.ndarray): Processed and augmented state data with derivatives, shape (n_samples - max_deriv, ...).
            - udout (np.ndarray): Normalized input data, shape (n_samples - max_deriv, num_ud).
            - xout_shifts (pd.DataFrame): Shifted duplicates of the normalized state data, shape (n_samples - max_deriv, ...).
    Notes:
        - The function assumes that `xout` and `udout` are NumPy arrays and that `xout` has at least `max_deriv` rows.
        - The normalization coefficients should be provided for both state and input variables.
        - The function uses zero-padding for shifted and derivative computations.
    """

    # Normalize xout
    xout = xout / norm_coeff

    xout_shifts = pd.DataFrame(np.tile(xout, (1, max_deriv + 1)))

    # Calculate shifted duplicates
    for shift in range(max_deriv + 1):
        # Fill with zeros for the first 'shift' rows, then shift the xout
        xout_shifts[:, shift * num_var:(shift + 1) * num_var] = np.vstack(
            (np.zeros((shift, num_var)), xout[:-shift if shift != 0 else None, :num_var])
        )

    # Calculate derivatives up to the max_deriv order
    for deriv in range(1, max_deriv + 1):
        for curr_var in range(num_var):
            pos_last_deriv = (deriv - 1) * num_var + curr_var
            # Compute the derivative and append it to xout
            derivative = np.vstack((np.zeros((deriv, 1)), np.diff(xout[deriv - 1:, pos_last_deriv], axis=0) / Ts))
            xout = np.hstack((xout, derivative))

    # Strip entries from the front of xout and xout_shifts to match the size after derivation
    xout = xout[max_deriv:]
    xout_shifts = xout_shifts[max_deriv:]

    # Normalize udout using normalization factors (derivatives not needed)
    if num_ud != 0:
        for j in range(num_ud):
            udout[:, j] = udout[:, j] / norm_coeff[num_var + j, 0]
        # Strip entries from the front of udout to match xout
        udout = udout[max_deriv:]

    return xout, udout, xout_shifts

FnTraceToTrainingData(trace, num_var, num_ud, useTime)

Converts a trace dictionary into training data suitable for machine learning models. Parameters: trace (dict): A dictionary containing the following keys: - 'x' (np.ndarray): Array of system variables over time (shape: [timesteps, num_var]). - 'ud' (np.ndarray): Array of user-defined variables over time (shape: [timesteps, num_ud]). - 'labels_trace' (np.ndarray): Array of state labels for each segment. - 'chpoints' (list or np.ndarray): Indices indicating change points (state switches) in the trace. num_var (int): Number of system variables to include in the feature vector. num_ud (int): Number of user-defined variables to include in the feature vector. useTime (bool): If True, includes the time since the last state switch as a feature. Returns: X (np.ndarray): Feature matrix where each row corresponds to a time step and contains: - Current state label - System variables - User-defined variables (if num_ud > 0) - Time since last state switch (if useTime is True) Y (np.ndarray): Array of next state labels (class labels) for each feature vector in X. states (np.ndarray): Array of state labels for each time step in the trace. Notes: - The function skips the last time step in the trace for feature construction, as it cannot form a (X, Y) pair. - The function assumes that the trace data is properly aligned and that 'chpoints' and 'labels_trace' are consistent.

Source code in ml4cps/automata/learn.py

def FnTraceToTrainingData(trace, num_var, num_ud, useTime):
    """
    Converts a trace dictionary into training data suitable for machine learning models.
    Parameters:
        trace (dict): A dictionary containing the following keys:
            - 'x' (np.ndarray): Array of system variables over time (shape: [timesteps, num_var]).
            - 'ud' (np.ndarray): Array of user-defined variables over time (shape: [timesteps, num_ud]).
            - 'labels_trace' (np.ndarray): Array of state labels for each segment.
            - 'chpoints' (list or np.ndarray): Indices indicating change points (state switches) in the trace.
        num_var (int): Number of system variables to include in the feature vector.
        num_ud (int): Number of user-defined variables to include in the feature vector.
        useTime (bool): If True, includes the time since the last state switch as a feature.
    Returns:
        X (np.ndarray): Feature matrix where each row corresponds to a time step and contains:
            - Current state label
            - System variables
            - User-defined variables (if num_ud > 0)
            - Time since last state switch (if useTime is True)
        Y (np.ndarray): Array of next state labels (class labels) for each feature vector in X.
        states (np.ndarray): Array of state labels for each time step in the trace.
    Notes:
        - The function skips the last time step in the trace for feature construction, as it cannot form a (X, Y) pair.
        - The function assumes that the trace data is properly aligned and that 'chpoints' and 'labels_trace' are consistent.
    """


    # Preallocate arrays for samples
    states = np.zeros((trace['x'].shape[0] - 1, 1))
    values = np.zeros((trace['x'].shape[0] - 1, num_var + num_ud))
    timeSwitch = np.zeros((trace['x'].shape[0] - 1, 1))

    # Variables to keep track of states
    lastswitch = 1
    indxStates = 0

    # Loop through trace data to generate feature vectors and class labels
    for indx in range(trace['x'].shape[0] - 1):
        # Update index associated with system mode switch
        if indxStates + 1 < len(trace['chpoints']) and indx >= trace['chpoints'][indxStates + 1]:
            indxStates += 1
            lastswitch = indx

        # Save current states and values for the feature vector
        states[indx] = trace['labels_trace'][indxStates]
        values[indx, :num_var] = trace['x'][indx, :num_var]
        if num_ud != 0:
            values[indx, num_var:num_var + num_ud] = trace['ud'][indx, :num_ud]

        timeSwitch[indx] = indx - lastswitch

    # Create matrices containing feature vectors and corresponding class labels
    points = np.arange(states.shape[0] - 1)
    if useTime:
        X = np.hstack([states[points], values[points, :], timeSwitch[points]])
    else:
        X = np.hstack([states[points], values[points, :]])

    Y = states[points + 1]

    return X, Y, states

build_pta(data, event_col='event', boundaries=1)

Builds a Prefix Tree Acceptor (PTA) from a collection of event sequences. This function constructs a PTA by iterating through each sequence of events in the provided data. It adds states and transitions to the automaton based on the observed event sequences, and sets the depth, in-degree, and out-degree of the states. The PTA is useful for learning automata from positive examples. Args: data (iterable): An iterable of event sequences. Each sequence can be a pandas DataFrame, pandas Series, or string. If a DataFrame, it should contain at least a time column and an event column. event_col (str, optional): The name of the column containing event labels in the input DataFrame or Series. Defaults to 'event'. boundaries (int or dict, optional): Not currently used in the function, but intended for handling event boundaries or timing constraints. Defaults to 1. Returns: Automaton: The constructed Prefix Tree Acceptor representing the input event sequences. Notes: - The function expects the presence of an Automaton class with methods for adding states, transitions, and final states. - If a sequence is a string, it is converted to a pandas Series of characters. - Timing information (dt) is calculated as the difference between consecutive time steps. - The function skips empty sequences.

Source code in ml4cps/automata/learn.py

def build_pta(data, event_col='event', boundaries=1):
    """
    Builds a Prefix Tree Acceptor (PTA) from a collection of event sequences.
    This function constructs a PTA by iterating through each sequence of events in the provided data. 
    It adds states and transitions to the automaton based on the observed event sequences, 
    and sets the depth, in-degree, and out-degree of the states. The PTA is useful for learning 
    automata from positive examples.
    Args:
        data (iterable): An iterable of event sequences. Each sequence can be a pandas DataFrame, 
            pandas Series, or string. If a DataFrame, it should contain at least a time column and 
            an event column.
        event_col (str, optional): The name of the column containing event labels in the input 
            DataFrame or Series. Defaults to 'event'.
        boundaries (int or dict, optional): Not currently used in the function, but intended for 
            handling event boundaries or timing constraints. Defaults to 1.
    Returns:
        Automaton: The constructed Prefix Tree Acceptor representing the input event sequences.
    Notes:
        - The function expects the presence of an `Automaton` class with methods for adding states, 
          transitions, and final states.
        - If a sequence is a string, it is converted to a pandas Series of characters.
        - Timing information (dt) is calculated as the difference between consecutive time steps.
        - The function skips empty sequences.
    """

    pta = Automaton()
    pta.add_initial_state('q0')
    for seq in data:
        if len(seq) == 0:
            continue
        if not isinstance(seq, pd.DataFrame):
            if isinstance(seq, str):
                seq = pd.Series(list(seq))
            if isinstance(seq, pd.Series):
                seq.name = event_col
            seq = pd.DataFrame(seq).reset_index(drop=False)
        old_t = seq[seq.columns[0]].iloc[0]
        curr_stat = "q0"
        time_col = seq.columns[0]
        seq = seq[[time_col, event_col]] #.iloc[1:]
        for t, event in seq.itertuples(index=False, name=None):
            dt = t - old_t
            # if event in boundaries and curr_stat != "q0":
            #     sub_event = 1 + next(ii for ii, tt in enumerate(boundaries[event]) if dt >= tt)
            #     event = event + "'" * sub_event
            dest = pta.get_transition(curr_stat, e=event)
            if dest is None:
                dest = f"q{pta.num_modes}"
            else:
                dest = dest[1]
            pta.add_single_transition(curr_stat, dest, event, timing=dt)
            curr_stat = dest
            old_t = t
        pta.add_final_state(curr_stat)
    return pta

computeDistance(der)

Computes a distance metric over a sliding window for a given derivative array. For each position in the input array der, the function calculates the sum of absolute differences between two windows of size windowSize before and after the current position, after normalizing each window by subtracting its first element. The result is an array of distances. Parameters: der (np.ndarray): The input array (e.g., derivative values) over which to compute the distance. Returns: np.ndarray: An array containing the computed distances for each valid position.

Source code in ml4cps/automata/learn.py

def computeDistance(der):
    """
    Computes a distance metric over a sliding window for a given derivative array.
    For each position in the input array `der`, the function calculates the sum of absolute differences
    between two windows of size `windowSize` before and after the current position, after normalizing
    each window by subtracting its first element. The result is an array of distances.
    Parameters:
        der (np.ndarray): The input array (e.g., derivative values) over which to compute the distance.
    Returns:
        np.ndarray: An array containing the computed distances for each valid position.
    """

    global windowSize
    dist = np.zeros(windowSize)
    for i in range(windowSize, len(der) - windowSize):
        before = der[(i - windowSize):i]
        after = der[(i + 1):(i + windowSize + 1)]
        dist_new = np.sum(np.abs((before - before[0]) - (after - after[0])))
        dist = np.append(dist, dist_new)
    return dist

filterChangePoints(xout, udout, xout_shifts, chpoints, chp_var)

Filters and synchronizes detected changepoints in time series data across multiple variables. This function processes global and local changepoint indices to ensure consistency and remove redundant or closely spaced changepoints. It updates the provided data arrays and changepoint lists accordingly. Args: xout (np.ndarray): Output variable time series data (samples x variables). udout (np.ndarray): Input variable time series data (samples x variables), or an empty array if not used. xout_shifts (np.ndarray): Shifted output variable data (samples x variables). chpoints (list or np.ndarray): List of global changepoint indices. chp_var (list of np.ndarray): List containing arrays of changepoint indices for each variable. Returns: tuple: - xout (np.ndarray): Filtered output variable data. - udout (np.ndarray): Filtered input variable data. - xout_shifts (np.ndarray): Filtered shifted output variable data. - chpoints (np.ndarray): Filtered and synchronized global changepoint indices. - chp_var (list of np.ndarray): Updated list of changepoint indices for each variable. Notes: - Uses global variables: windowSize, num_var, num_ud. - Assumes that changepoints are sorted and that there are at least two changepoints. - The function modifies chp_var in place.

Source code in ml4cps/automata/learn.py

def filterChangePoints(xout, udout, xout_shifts, chpoints, chp_var):
    """
    Filters and synchronizes detected changepoints in time series data across multiple variables.
    This function processes global and local changepoint indices to ensure consistency and remove redundant or closely spaced changepoints. It updates the provided data arrays and changepoint lists accordingly.
    Args:
        xout (np.ndarray): Output variable time series data (samples x variables).
        udout (np.ndarray): Input variable time series data (samples x variables), or an empty array if not used.
        xout_shifts (np.ndarray): Shifted output variable data (samples x variables).
        chpoints (list or np.ndarray): List of global changepoint indices.
        chp_var (list of np.ndarray): List containing arrays of changepoint indices for each variable.
    Returns:
        tuple:
            - xout (np.ndarray): Filtered output variable data.
            - udout (np.ndarray): Filtered input variable data.
            - xout_shifts (np.ndarray): Filtered shifted output variable data.
            - chpoints (np.ndarray): Filtered and synchronized global changepoint indices.
            - chp_var (list of np.ndarray): Updated list of changepoint indices for each variable.
    Notes:
        - Uses global variables: windowSize, num_var, num_ud.
        - Assumes that changepoints are sorted and that there are at least two changepoints.
        - The function modifies chp_var in place.
    """

    global windowSize, num_var, num_ud

    # Filter global changepoints detected on multiple output variables
    chpoints = filterindx(chpoints, windowSize)

    if chpoints[-1] - chpoints[-2] < 2 * windowSize:
        xout = xout[:chpoints[-2], :]
        xout_shifts = xout_shifts[:chpoints[-2], :]
        if num_ud != 0:
            udout = udout[:chpoints[-2], :]
        chpoints = chpoints[:-1]
        for i in range(num_var + num_ud):
            current_chps = np.array(chp_var[i])
            current_chps = current_chps[:-1]
            current_chps = np.append(current_chps, chpoints[-1])
            chp_var[i] = current_chps

    # Ensure consistency between global and local changepoint sets
    for i in range(num_var + num_ud):
        current_chps = np.array(chp_var[i])
        for j in range(len(current_chps)):
            idx = np.argmin(np.abs(chpoints - current_chps[j]))
            current_chps[j] = chpoints[idx]
        if current_chps[-2] == current_chps[-1]:
            current_chps = current_chps[:-1]
        chp_var[i] = current_chps

    return xout, udout, xout_shifts, chpoints, chp_var

filterindx(indx, windw)

Filters out indices from the input array that are within a specified window of each other. Given a sorted array of indices, this function removes any index that is within windw distance from its predecessor, keeping only the first occurrence in each window. Parameters: indx (array-like): A sorted array or list of integer indices. windw (int): The minimum allowed distance between consecutive indices. Returns: numpy.ndarray: The filtered array of indices, where no two indices are within windw of each other. Example: >>> filterindx(np.array([1, 2, 3, 10, 12]), 2) array([ 1, 10, 12])

Source code in ml4cps/automata/learn.py

def filterindx(indx, windw):
    """
    Filters out indices from the input array that are within a specified window of each other.
    Given a sorted array of indices, this function removes any index that is within `windw` distance
    from its predecessor, keeping only the first occurrence in each window.
    Parameters:
        indx (array-like): A sorted array or list of integer indices.
        windw (int): The minimum allowed distance between consecutive indices.
    Returns:
        numpy.ndarray: The filtered array of indices, where no two indices are within `windw` of each other.
    Example:
        >>> filterindx(np.array([1, 2, 3, 10, 12]), 2)
        array([ 1, 10, 12])
    """
    n = 0
    while n < len(indx) - 1:
        id1 = indx[n]
        while n + 1 < len(indx) and indx[n + 1] - id1 <= windw:
            indx = np.delete(indx, n + 1)
        n += 1
    return indx

findChangePoints(xout, depth, starting, ending, max_depth)

Recursively detects change points in a multi-dimensional signal using a hierarchical approach. This function analyzes a segment of the input array xout at a given depth (dimension), computes a distance metric to identify potential change points (peaks), and then recursively searches for further change points in subsegments at deeper levels. The recursion stops when the maximum depth is reached or the segment is too small. Parameters

---

xout : np.ndarray The input array containing the signal or features to analyze. Expected shape is (n_samples, n_features). depth : int The current depth (dimension) being analyzed. starting : int The starting index of the segment to analyze. ending : int The ending index (exclusive) of the segment to analyze. max_depth : int The maximum depth (dimension) to analyze. Returns

---

np.ndarray An array of indices representing detected change points within the specified segment. Notes

---

• Uses global variables windowSize and chp_depths for windowing and tracking change points per depth.
• Utilizes computeDistance, find_peaks, and filterindx helper functions.
• At the top level (depth == 0), prepends and appends boundary indices to the result.

Source code in ml4cps/automata/learn.py

def findChangePoints(xout, depth, starting, ending, max_depth):
    """
    Recursively detects change points in a multi-dimensional signal using a hierarchical approach.
    This function analyzes a segment of the input array `xout` at a given `depth` (dimension),
    computes a distance metric to identify potential change points (peaks), and then recursively
    searches for further change points in subsegments at deeper levels. The recursion stops when
    the maximum depth is reached or the segment is too small.
    Parameters
    ----------
    xout : np.ndarray
        The input array containing the signal or features to analyze. Expected shape is (n_samples, n_features).
    depth : int
        The current depth (dimension) being analyzed.
    starting : int
        The starting index of the segment to analyze.
    ending : int
        The ending index (exclusive) of the segment to analyze.
    max_depth : int
        The maximum depth (dimension) to analyze.
    Returns
    -------
    np.ndarray
        An array of indices representing detected change points within the specified segment.
    Notes
    -----
    - Uses global variables `windowSize` and `chp_depths` for windowing and tracking change points per depth.
    - Utilizes `computeDistance`, `find_peaks`, and `filterindx` helper functions.
    - At the top level (depth == 0), prepends and appends boundary indices to the result.
    """

    global windowSize, chp_depths

    locs = []
    if depth > max_depth or ending - starting - 1 < 2 * windowSize:
        return locs

    der = xout[starting:ending, depth]
    dist = computeDistance(der)

    # Find peaks in distance to detect change points
    _, locsDist = find_peaks(dist, height=5)
    locsHere = np.sort(locsDist + starting - 1)
    locsHere = filterindx(locsHere, 1.5 * windowSize)
    chp_depths[depth] += len(locsHere)
    locs.extend(locsHere)

    locsHere = np.concatenate([[starting - windowSize // 2], locsHere, [ending + windowSize // 2]])
    for i in range(len(locsHere) - 1):
        newStart = int(locsHere[i] + windowSize / 2)
        newEnd = int(locsHere[i + 1] - windowSize / 2)
        locsNew = findChangePoints(xout, depth + 1, newStart, newEnd, max_depth)
        locs.extend(locsNew)

    if depth == 0:
        locs = np.concatenate([[1], locs, [len(der)]])

    return np.array(locs)

rpni(positive_samples, negative_samples)

Implements the RPNI (Regular Positive and Negative Inference) algorithm for learning a DFA from positive and negative samples. Args: positive_samples (Iterable[str]): A collection of strings that should be accepted by the learned DFA. negative_samples (Iterable[str]): A collection of strings that should be rejected by the learned DFA. Returns: DFA: A deterministic finite automaton (DFA) that accepts all positive samples and rejects all negative samples. Notes: - The function first constructs a Prefix Tree Acceptor (PTA) from the positive samples. - It then attempts to merge states in the DFA, ensuring that no negative sample is accepted after each merge. - The merging process is guided by the constraint that all negative samples must be rejected.

Source code in ml4cps/automata/learn.py

def rpni(positive_samples, negative_samples):
    """
    Implements the RPNI (Regular Positive and Negative Inference) algorithm for learning a DFA from positive and negative samples.
    Args:
        positive_samples (Iterable[str]): A collection of strings that should be accepted by the learned DFA.
        negative_samples (Iterable[str]): A collection of strings that should be rejected by the learned DFA.
    Returns:
        DFA: A deterministic finite automaton (DFA) that accepts all positive samples and rejects all negative samples.
    Notes:
        - The function first constructs a Prefix Tree Acceptor (PTA) from the positive samples.
        - It then attempts to merge states in the DFA, ensuring that no negative sample is accepted after each merge.
        - The merging process is guided by the constraint that all negative samples must be rejected.
    """


    # Process positive samples
    dfa = build_pta(positive_samples)

    # Process negative samples
    # Function to check if the DFA rejects all negative samples
    def rejects_negative_samples(mdl):
        for sample in negative_samples:
            if mdl.accepts(sample):
                return False
        return True

    # Attempt to merge states while preserving rejection of negative samples
    states = list(dfa.discrete_states)
    for i in range(len(states)):
        for j in range(i + 1, len(states)):
            state1 = states[i]
            state2 = states[j]
            if dfa.is_state(state2):
                dfa.try_merge_states(state1, state2, rejects_negative_samples)
                print('Compare')

    return dfa

simple_learn_from_event_logs(data, initial=True, count_repetition=True, verbose=False)

Simple algorithm to learn a timed automaton from event log data. This function constructs a timed automaton by iterating over sequences of timestamped events. Each event sequence is treated as a trace, and transitions are created between states based on event occurrences and their timing. States are determined by the emitted events, optionally including repetition counts. The automaton can be initialized with an explicit initial state, and transitions can account for repeated events. Args: data (list or pandas.Series): A list of event sequences, where each sequence is a pandas Series with timestamps as indices and event labels as values. initial (bool, optional): If True, adds an explicit 'initial' state to the automaton. Defaults to True. count_repetition (bool, optional): If True, distinguishes states and transitions by counting consecutive repetitions of the same event. Defaults to True. verbose (bool, optional): If True, prints detailed information about the learning process. Defaults to False. Returns: Automaton: The learned timed automaton object. Notes: - Each sequence in data should be a pandas Series indexed by timestamps. - If a sequence contains fewer than two events, it is skipped. - The function assumes the existence of an Automaton class with add_initial_state and add_single_transition methods.

Source code in ml4cps/automata/learn.py

def simple_learn_from_event_logs(data, initial=True, count_repetition=True, verbose=False):
    """
    Simple algorithm to learn a timed automaton from event log data.
    This function constructs a timed automaton by iterating over sequences of timestamped events.
    Each event sequence is treated as a trace, and transitions are created between states based on event occurrences and their timing.
    States are determined by the emitted events, optionally including repetition counts.
    The automaton can be initialized with an explicit initial state, and transitions can account for repeated events.
    Args:
        data (list or pandas.Series): A list of event sequences, where each sequence is a pandas Series with timestamps as indices and event labels as values.
        initial (bool, optional): If True, adds an explicit 'initial' state to the automaton. Defaults to True.
        count_repetition (bool, optional): If True, distinguishes states and transitions by counting consecutive repetitions of the same event. Defaults to True.
        verbose (bool, optional): If True, prints detailed information about the learning process. Defaults to False.
    Returns:
        Automaton: The learned timed automaton object.
    Notes:
        - Each sequence in `data` should be a pandas Series indexed by timestamps.
        - If a sequence contains fewer than two events, it is skipped.
        - The function assumes the existence of an `Automaton` class with `add_initial_state` and `add_single_transition` methods.

    """

    # Here the state is determined by the events it emits, but only the first event is taken as transition
    if type(data) is not list:
        data = [data]

    a = Automaton(id='Simple')
    sequence = 0
    if verbose:
        print('***Timed automaton learning from event logs***')

    for d in data:
        sequence += 1
        print('Sequence #{}'.format(sequence))
        if len(d) < 2:
            print('Skipping because num events: 0')
            continue
        print('Duration: {}'.format(d.index[-1] - d.index[0]))

        event_rpt = 0
        state_event = ''

        old_event_rpt = 0
        old_state_event = ''

        t_old = d.index[0]
        if initial:
            a.add_initial_state('initial')

        if type(d) is pd.DataFrame:
            d = tools.create_events_from_concurent_logs(d)

        for t, event in d:
            if state_event == event:
                event_rpt += 1
            else:
                state_event = event
                event_rpt = 0

            delta_t = t - t_old
            if old_state_event == '':
                source = 'initial'
            else:
                if count_repetition and old_event_rpt:
                    source = f'{old_state_event}#{old_event_rpt}'
                else:
                    source = old_state_event

            if count_repetition and event_rpt:
                dest = f'{state_event}#{event_rpt}'
            else:
                dest = state_event

            if source != 'initial' or initial:
                a.add_single_transition(source, dest, event, delta_t)
            t_old = t
            old_state_event = state_event
            old_event_rpt = event_rpt
            if verbose:
                print(source, dest, event, delta_t)
    return a

simple_learn_from_signal_updates(data, sig_names, initial=True, verbose=False)

Learns a timed automaton from sequences of signal updates. This function processes a list of dataframes, each representing a sequence of signal updates over time. For each sequence, it constructs states based on the values of the specified signals and adds transitions to an Automaton object whenever a signal value changes. Args: data (list of pandas.DataFrame): List of dataframes, each containing time-stamped signal updates. The first column is assumed to be the time column, and subsequent columns correspond to signal names. sig_names (list of str): List of signal names to track and use for state construction. initial (bool, optional): If True, adds an initial state to the automaton for each sequence. Defaults to True. verbose (bool, optional): If True, prints detailed information about the learning process. Defaults to False. Returns: Automaton: The learned automaton with states and transitions based on the observed signal updates. Notes: - Each state is represented as a dictionary mapping signal names to their current values. - Transitions are added only when all signal values are set (i.e., not None). - The event label for each transition is formatted as '<-'. - The transition is annotated with the time difference (delta_t) between consecutive events.

Source code in ml4cps/automata/learn.py

def simple_learn_from_signal_updates(data, sig_names, initial=True, verbose=False):
    """
    Learns a timed automaton from sequences of signal updates.
    This function processes a list of dataframes, each representing a sequence of signal updates over time.
    For each sequence, it constructs states based on the values of the specified signals and adds transitions
    to an Automaton object whenever a signal value changes.
    Args:
        data (list of pandas.DataFrame): List of dataframes, each containing time-stamped signal updates.
            The first column is assumed to be the time column, and subsequent columns correspond to signal names.
        sig_names (list of str): List of signal names to track and use for state construction.
        initial (bool, optional): If True, adds an initial state to the automaton for each sequence. Defaults to True.
        verbose (bool, optional): If True, prints detailed information about the learning process. Defaults to False.
    Returns:
        Automaton: The learned automaton with states and transitions based on the observed signal updates.
    Notes:
        - Each state is represented as a dictionary mapping signal names to their current values.
        - Transitions are added only when all signal values are set (i.e., not None).
        - The event label for each transition is formatted as '<signal_name><-<value>'.
        - The transition is annotated with the time difference (delta_t) between consecutive events.
    """

    a = Automaton()
    sequence = 0
    if verbose:
        print('***Timed automaton learning from variable changes***')

    for d in data:
        time_col = d.columns[0]
        sequence += 1
        print('Sequence #{}'.format(sequence))
        if len(d) < 2:
            print('Skipping because num events: 0')
            continue
        print('Duration: {}'.format(d[time_col].iloc[-1] - d[time_col].iloc[0]))

        t_old = d[time_col].iloc[0]
        if initial:
            a.add_initial_state('initial')

        state = dict.fromkeys(sig_names)
        for t, signal, value in d.itertuples(index=False, name=None):
            event = f'{signal}<-{value}'
            all_values_are_set = all(value is not None for value in state.values())

            delta_t = t - t_old
            t_old = t
            source = pprint.pformat(state)
            state[signal] = value
            dest = pprint.pformat(state)

            if all_values_are_set:
                a.add_single_transition(source, dest, event, delta_t)
    return a

simple_learn_from_signal_vectors(data, drop_no_changes=False, verbose=False)

Learns a timed automaton from a list of signal vector dataframes. This function processes sequences of signal vectors (as pandas DataFrames), detects changes in the specified signal columns, and constructs a timed automaton by adding transitions for each detected event. Args: data (list of pandas.DataFrame): List of DataFrames, each representing a sequence of signal vectors. The first column is assumed to be the time column, and the remaining columns are signal values. sig_names (list of str): List of column names in the DataFrame that correspond to the signals to be considered for state transitions. drop_no_changes (bool, optional): If True, rows where no signal changes occur are dropped before processing. Default is False. verbose (bool, optional): If True, prints detailed information about the learning process. Default is False. Returns: Automaton: An Automaton object constructed from the observed transitions in the input data. Notes: - Each transition in the automaton corresponds to a change in the signal vector, with the event label representing the difference between consecutive signal vectors and the transition time as the time delta. - The function assumes that the Automaton class and its add_single_transition method are defined elsewhere.

Source code in ml4cps/automata/learn.py

def simple_learn_from_signal_vectors(data, drop_no_changes=False, verbose=False):
    """
    Learns a timed automaton from a list of signal vector dataframes.
    This function processes sequences of signal vectors (as pandas DataFrames), detects changes in the specified
    signal columns, and constructs a timed automaton by adding transitions for each detected event.
    Args:
        data (list of pandas.DataFrame): 
            List of DataFrames, each representing a sequence of signal vectors. 
            The first column is assumed to be the time column, and the remaining columns are signal values.
        sig_names (list of str): 
            List of column names in the DataFrame that correspond to the signals to be considered for state transitions.
        drop_no_changes (bool, optional): 
            If True, rows where no signal changes occur are dropped before processing. Default is False.
        verbose (bool, optional): 
            If True, prints detailed information about the learning process. Default is False.
    Returns:
        Automaton: 
            An Automaton object constructed from the observed transitions in the input data.
    Notes:
        - Each transition in the automaton corresponds to a change in the signal vector, with the event label
          representing the difference between consecutive signal vectors and the transition time as the time delta.
        - The function assumes that the Automaton class and its add_single_transition method are defined elsewhere.
    """

    a = Automaton()
    sequence = 0
    if verbose:
        print('***Timed automaton learning from variable changes***')

    dummy_initial = 'initial'
    a.add_initial_state(dummy_initial)

    for d in data:
        sig_names = d.columns
        if drop_no_changes:
            d = tools.remove_timestamps_without_change(d)
        sequence += 1
        if verbose:
            print('Sequence #{}'.format(sequence))
            if len(d) < 2:
                print('Skipping because num events: 0')
                continue
            print('Duration: {}'.format(d.index[-1] - d.index[0]))

        previous_state = d[sig_names].iloc[:-1]
        dest_state = d[sig_names].iloc[1:]
        mask = ~d[sig_names].isin([0, 1])
        has_invalid = mask.any().any()
        if has_invalid:
            event = d[sig_names].apply(lambda x: ' '.join(x.astype(str)).replace(".0", ""), 1).iloc[1:]
        else:
            event = d[sig_names].diff().apply(lambda x: ' '.join(x.astype(str)).replace(".0", ""), 1).iloc[1:]
        deltat = d.index.diff()[1:]

        obs_ind = 0
        for source, dest, ev, dt in zip(previous_state.itertuples(index=False, name=None),
                                        dest_state.itertuples(index=False, name=None), event, deltat):
            obs_ind += 1
            source = pprint.pformat(source, compact=True, width=10000).replace(".0", "")
            dest = pprint.pformat(dest, compact=True, width=10000).replace(".0", "")

            if obs_ind == 1:
                a.add_single_transition(dummy_initial, source, "Start", 0)

            a.add_single_transition(source, dest, ev, dt)

            if obs_ind == len(previous_state):
                a.add_final_state(dest)
    return a

The module provides a class Automaton which inherits CPSComponent and implements the dynamics of different kinds of automata.

Authors: - Nemanja Hranisavljevic, hranisan@hsu-hh.de, nemanja@ai4cps.com - Tom Westermann, tom.westermann@hsu-hh.de, tom@ai4cps.com

Automaton

Bases: CPSComponent

Automaton class is the main class for modeling various kinds of hybrid systems.

Source code in ml4cps/automata/base.py

class Automaton (CPSComponent):
    """
    Automaton class is the main class for modeling various kinds of hybrid systems.
    """

    def __init__(self, states: list = None, transitions: list = None,
                 unknown_state: str = 'raise', id="", initial_q=(), initial_r=None, final_q=(), super_states=(), decision_states=(),
                 **kwargs):
        """
        Class initialization from lists of elements.
        :param states: Discrete states / modes of continuous behavior.
        :param events: The events that trigger state transitions.
        :param transitions: The transition information. If a collection of dicts then dict should contain "source",
        "dest" and "event". The other attributes will be added as data of that transition. Alternatively, a collection
        of tuples can be used of the form (source, event, dest, *).
        :param unknown_state: The name of unknown states during "play in", if "raise", an exception will be raised.
        """
        self._G = nx.MultiDiGraph()
        if initial_q and isinstance(initial_q, str):
            initial_q = [initial_q]
        self.q0 = OrderedDict.fromkeys(initial_q)
        self.initial_r = initial_r
        if final_q and isinstance(final_q, str):
            final_q = [final_q]
        self.final_q = OrderedDict.fromkeys(final_q)
        self.previous_node_positions = None
        self.UNKNOWN_STATE = unknown_state
        if super_states is not None:
            if type(super_states) is str:
                self.__super_states = [super_states]
            else:
                self.__super_states = list(super_states)

            # self._G.add_nodes_from(self.__super_states)

        if decision_states is not None:
            if type(decision_states) is str:
                self.decision_states = [decision_states]
            else:
                self.decision_states = list(decision_states)

        if states is not None:
            self._G.add_nodes_from(states)

        if transitions is not None:
            for tr in transitions:
                if type(tr) is dict:
                    self._G.add_edge(tr.pop('source'), tr.pop('destination'), event=tr.pop('event'), **tr)
                else:
                    self._G.add_edge(tr[0], tr[2], event=tr[1])

        if 'discr_state_names' not in kwargs:
            kwargs['discr_state_names'] = ['Mode']
        elif type(kwargs['discr_state_names']) is str:
            kwargs['discr_state_names'] = [kwargs['discr_state_names']]
        CPSComponent.__init__(self, id, **kwargs)

    @property
    def Sigma(self):
        Sigma = []
        for x in self.transitions:
            if len(x) >= 3 and 'event' in x[2]:
                new_el = x[2]['event']
                if new_el not in Sigma:
                    Sigma.append(new_el)
        return Sigma

    @property
    def num_events(self):
        return len(self.Sigma)

    @property
    def discrete_states(self):
        return self._G.nodes

    @property
    def transitions(self):
        return self._G.edges(data=True)

    @property
    def discrete_state(self):
        if self._q is not None:
            if len(self._q) == 0:
                return ()
            return self._q[0]
        return None

    @discrete_state.setter
    def discrete_state(self, value):
        self._q = (value,)

    @property
    def state(self):
        """
        Automata discrete state is uni-variate.
        :return:
        """
        if len(self._q) < 1:
            raise Exception(f'State of "{self.id}" is empty.')
        return self._q[0], self._xt, self._xk

    @state.setter
    def state(self, state):
        if type(state) is not tuple:
            new_states = (state, (), ())
        elif len(state) < 3:
            new_states = [(), (), ()]
            for i, v in enumerate(state):
                new_states[i] = v
        else:
            new_states = state

        if new_states[0] not in self.discrete_states and new_states[0] not in self.__super_states and new_states[0] != self.UNKNOWN_STATE:
            raise ValueError(f'State "{new_states[0]}" is not a valid state. Valid states are: {self.discrete_states}')

        self._q = (new_states[0],)
        self._xt = new_states[1]
        self._xk = new_states[2]

    @property
    def num_modes(self):
        """
        Returns the number of modes in the automaton.
        :return: number of states.
        """
        return self._G.number_of_nodes()

    @property
    def num_transitions(self):
        """
        Returns the number of transitions in the automaton.
        :return: number of transitions.
        """
        return self._G.number_of_edges()

    def merge(self, q1, q2):
        """
        If two states are compatible, they are merged with the function merge. The transitions
        of the automaton, the in- and outdegree of the states and the number of transitions
        happening are adjusted.
        """
        intr = self.in_transitions(q2)
        # outtr_q1 = list(self.out_transitions(q1))
        outtr_q2 = list(self.out_transitions(q2))
        # nx.contracted_nodes(self._G, w, v, copy=False)
        # set event

        for tr in intr:
            if tr[0] != q2:
                self.add_single_transition(tr[0], q1, tr[2], timing=tr[-1]['timing'])
        # if q1=='q0' and q2 == 'q192':
        #     print('found')
        for tr in outtr_q2:
            dest = tr[1]
            if dest == q2:
                dest = q1
            self.add_single_transition(q1, dest, tr[2], timing=tr[-1]['timing'])
        self.remove_state(q2)

    def determinize(self, s, state_index, verbose=False):
        # out_tr = list(sorted(self.out_transitions(s), key=lambda x: state_index[x[1]]))
        if verbose:
            print('Determinize', s)
        # ind = sorted(, key=lambda x: state_index[x])
        # state_index_list = [None] * len(state_index)
        # for k, v in state_index.items():
        #     state_index_list[v] = k
        # if verbose:
        #     pprint.pprint(state_index_list[state_index[s]:], compact=True)

        to_analyze = [s]
        while to_analyze:  # state_ind < len(state_index_list):
            # print(state_ind)
            s = to_analyze.pop()
            # if not self.is_state(s):
            #     continue

            conflicting_transitions = {}
            to_merge = None
            for out_tr in self.out_transitions(s):
                if out_tr[2] in conflicting_transitions:
                    to_merge = (conflicting_transitions[out_tr[2]], out_tr[1])
                    break
                else:
                    conflicting_transitions[out_tr[2]] = out_tr[1]
            # out_tr = list(self.out_transitions(st))
            if to_merge is not None:
                if state_index[to_merge[0]] < state_index[to_merge[1]]:
                    merge_into = to_merge[0]
                    to_be_merged = to_merge[1]
                else:
                    merge_into = to_merge[1]
                    to_be_merged = to_merge[0]

                to_analyze.append(s)
                to_analyze.append(merge_into)
                if verbose:
                    print('Merge to determinize: ', to_be_merged, 'into', merge_into)
                self.merge(merge_into, to_be_merged)

        # to_determinize = []
        # for i, tr in enumerate(out_tr):
        #     if self.is_state(tr[1]):
        #         for tr2 in out_tr[:i:-1]:
        #             if self.is_state(tr2[1]) and tr[2] == tr2[2]:
        #                 # print(tr[1], ',', tr2[1])
        #                 if state_index[tr2[1]] > state_index[tr[1]]:
        #                     print('Merge to determinize: ', tr2[1], 'into', tr[1])
        #                     self.merge(tr[1], tr2[1])
        #                     to_determinize.append(tr[1])
        #                     # self.determinize(tr[1], state_index)
        #                 else:
        #                     print('Merge to determinize: ', tr[1], 'into', tr2[1])
        #                     self.merge(tr2[1], tr[1])
        #                     to_determinize.append(tr2[1])
        #                     # self.determinize(tr2[1], state_index)
        # for s in to_determinize:
        #     self.determinize(s, state_index)

    def accepts(self, string, return_states=False):
        state_path = []
        current_state = next(iter(self.q0))
        state_path.append(current_state)
        for symbol in string:
            trans = self.out_transitions(current_state, symbol)
            if trans is None or not trans:
                if return_states:
                    return False, state_path
                else:
                    return False
            elif len(trans) > 1:
                raise Exception(f'Too many transitions from {current_state} given symbol {symbol}.')
            else:
                current_state = trans[0][1]
                state_path.append(current_state)
        accepted = current_state in self.final_q or self.final_q is None or len(self.final_q) == 0
        if return_states:
            return accepted, state_path
        else:
            return accepted

    def generate(self, number_of_sequences=1, return_states=False, prob_to_accept=0.5):
        if number_of_sequences > 1:
            res = [self.generate(return_states=return_states, prob_to_accept=prob_to_accept)
                   for _ in range(number_of_sequences)]
            return res
        state_path = []
        current_state = next(iter(self.q0))
        event_path = []
        state_path.append(current_state)
        while True:
            if current_state in self.final_q:
                if random.random() < prob_to_accept:
                    break
            trans = self.out_transitions(current_state)
            if trans is None or not trans:
                if current_state not in self.final_q:
                    raise Exception(f'State "{current_state}" is not an accepting state, but no possible transitions.')
                break
            else:
                new_trans = random.choices(list(trans), weights=[x[3].get('prob', 1) for x in trans])
                current_state = new_trans[0][1]
                current_event = new_trans[0][3]['event']
                state_path.append(current_state)
                event_path.append(current_event)

        if return_states:
            return event_path, state_path
        else:
            return event_path


    def try_merge_states(self, state1, state2, try_fun=None):
        """Merge state2 into state1 and update transitions."""
        old_G = self._G
        make_final = state2 in self.final_q and state1 not in self.final_q
        make_initial = state2 in self.q0 and state1 not in self.q0
        self._G = nx.contracted_nodes(self._G, state1, state2)
        # make it deterministic again
        events = [x[3]['event'] for x in self.out_transitions(state1)]
        deterministic = len(set(events)) == len(events)

        if deterministic:
            if make_initial:
                self.add_initial_state(state1)
            if make_final:
                self.add_final_state(state1)

            if try_fun is not None and not try_fun(self):
                self._G = old_G
                if make_final:
                    self.final_q.pop(state1)
                if make_initial:
                    self.q0.pop(state1)
        else:
            self._G = old_G


    def remove_transition(self, source, dest):
        """
        Remove the transition(s) from source to dest.
        :param source:
        :param dest:
        :return:
        """
        self._G.remove_edge(source, dest)

    def is_decision(self, state, overall_state):
        return state in self.decision_states

    def get_alternatives(self, state, system_state):
        if self.is_decision(state, system_state):
            trans = [(x[3]['event'], dict()) for x in self.out_transitions(state)]
            return [(None, None)] + trans
        else:
            return None

    def remove_rare_transitions(self, min_p=0, min_num=0, keep_from_initial=False, keep_states=False, keep=None):
        self.learn_transition_probabilities()

        for source, dest, event, data in self.get_transitions():
            if keep_from_initial and source in self.q0:
                continue
            if (len(data['timing']) <= min_num or data['probability'] < min_p) and \
                    ((keep is None) or (keep is not None and source not in keep and dest not in keep)):
                self.remove_transition(source, dest)

        if not keep_states:
            for s in list(self.discrete_states):
                # if s in self.q0:
                #     continue
                if len(self.in_transitions(s)) == 0 and len(self.out_transitions(s)) == 0:
                    self.remove_state(s)

            # if self.DummyInitial:
            #     for s in list(self.InitialState):
            #         if len(self.out_transitions(s)) == 0:
            #             self.States.pop(s)
            #             self.InitialState.pop(s)

        # recalculate probabilities
        if min_p:
            self.learn_transition_probabilities()

    def learn_transition_probabilities(self):
        for s in self.discrete_states:
            total_num = sum([len(data['timing']) for s, d, e, data in self.out_transitions(s)])
            for s, d, e, data in self.out_transitions(s):
                data['probability'] = len(data['timing']) / total_num

    def state_is_deterministic(self, q):
        events = set()
        for tr in self.out_transitions(q):
            if tr[2] in events:
                return False
            else:
                events.add(tr[2])
        return True

    def update_timing_boundaries(self, source, destination, event, newTiming):
        edge_data = self._G.get_edge_data(source, destination, event)
        try:
            if newTiming < edge_data['minTiming']:
                edge_data['minTiming'] = newTiming
            elif newTiming > edge_data['maxTiming']:
                edge_data['maxTiming'] = newTiming
        except KeyError:
            edge_data['minTiming'] = newTiming
            edge_data['maxTiming'] = newTiming

    def is_deterministic(self):
        for q in self.discrete_states:
            if not self.state_is_deterministic(q):
                print('State', q, 'not deterministic:')
                for tr in sorted(self.out_transitions(q), key=lambda x: x[2]):
                    print(tr[0], '->', tr[2], '->', tr[1])
                return False
        return True

    def add_single_transition(self, s, d, e, timing=None):
        edge_data = self._G.get_edge_data(s, d, e)
        if edge_data is None:
            if timing is None:
                self._G.add_edge(s, d, key=e)
            else:
                try:
                    timing = list(timing)
                    self._G.add_edge(s, d, key=e, event=e, timing=timing)
                except:
                    self._G.add_edge(s, d, key=e, event=e, timing=[timing])
        elif timing is not None:
            try:
                timing = list(timing)
                edge_data['timing'] += timing
            except:
                edge_data['timing'].append(timing)

    def add_state_data(self, s: str, d: object):
        """
    Add state data to a state s the automaton.
        :param s: state
        :param d: data to be added to s
        :return:
        """
        self.Q[s] = d

    def add_state(self, new_state, **kwargs):
        """
    Add state to the automaton.
        :param new_state: State to be added.
        """
        self._G.add_node(new_state, **kwargs)

    def add_states_from(self, new_state, **kwargs):
        """
    Add multiple states to the automaton.
        :param new_state: States to be added.
        """
        self._G.add_nodes_from(new_state, **kwargs)

    def add_transitions_from(self, list_of_tuples, **other):
        """
    Add multiple transition.
        :param list_of_tuples: List of transitions in the form (source_state, destination_state, event, ...<unused>...).
        """
        self._G.add_edges_from(list_of_tuples, **other)

    def add_transition(self, s, d, e, **other):
        """
    Add multiple transition.
        :param list_of_tuples: List of transitions in the form (source_state, destination_state, event, ...<unused>...).
        """
        self._G.add_edge(s, d, e, **other, event=e)

    def add_initial_state(self, states):
        """
    Add initial state(s) of the automaton.
        :param states: States to add.
        """
        if type(states) is str:
            states = (states,)
        for s in states:
            if s is not None and s not in self.q0:
                self.q0[s] = None
            self._G.add_node(s)

    def add_final_state(self, states):
        """
    Add final state(s) of the automaton.
        :param states: States to add.
        """
        if type(states) is str:
            states = (states,)
        for s in states:
            if s is not None and s not in self.final_q:
                self.final_q[s] = None
            self._G.add_node(s)

    def is_transition(self, s, d, e):
        """
    Check if a transition (s,d,e) exists in the automaton.
        :param s: Source.
        :param d: Destination.
        :param e: Event.
        :return:
        """
        transitions = [trans for trans in self.T.values() if trans['source'] == s and
                       trans['destination'] == d and trans['event'] == e]

        is_transition = len(transitions) != 0
        return is_transition

    # def num_occur(self, q, e):
    #     tr = self.get_transition(q, e=e)
    #     if tr[-1]:
    #         return len(tr[-1]['timing'])
    #     else:
    #         return ""

    def num_occur(self, tr):
        return len(tr[-1]['timing'])

    def num_timings(self):
        return sum(len(tr[-1]['timing']) for tr in self.get_transitions())

    def get_num_in(self, q):
        """
        Returns the number of in transitions of state q in the automaton.
        :return: number of transitions.
        """
        if self._G.has_node(q):
            return self._G.in_degree(q)
        else:
            raise Exception(f'State {q} not in the automaton.')

    def get_num_out(self, q):
        """
        Returns the number of out transitions of state q in the automaton.
        :return: number of transitions.
        """
        if self._G.has_node(q):
            return self._G.out_degree(q)
        else:
            raise Exception(f'State {q} not in the automaton.')

    def is_state(self, q):
        return self._G.has_node(q)

    def remove_state(self, s):
        self._G.remove_node(s)
        if s in self.q0:
            self.q0.pop(s)

    def in_transitions(self, s, event=None):
        """
    Get all incoming transitions of state s.
        :param s:
        :return:
        """
        if event is None:
            return self._G.in_edges(s, data=True, keys=True)
        else:
            return [e for e in self._G.in_edges(s, data=True, keys=True) if e[3]['event'] == event]

    def out_transitions(self, s, event=None):
        """
    Get all outgoing transitions of state s.
        :param event:
        :param s:
        :return:
        """
        if event is None:
            return self._G.out_edges(s, data=True, keys=True)
        else:
            return [e for e in self._G.out_edges(s, data=True, keys=True) if e[3]['event'] == event]

    def discrete_event_dynamics(self, q, xt, xk, p) -> tuple:
        e = p["event"]
        new_q = self.UNKNOWN_STATE
        possible_destinations = set(d for s, d, ev in self._G.out_edges(q, data='event') if ev == e)
        if len(possible_destinations) == 1:
            new_q = possible_destinations.pop()
        else:
            stoch_dest = list((d, data.get('p', 0)) for s, d, data in self._G.out_edges(q, data=True) if data['event'] == e)
            if stoch_dest:
                new_q = random.choices([x[0] for x in stoch_dest], weights=[x[1] for x in stoch_dest])[0]
            else: # try to revocer
                dests = set(d for s, d, ev in self._G.edges(data='event') if ev == e)
                if len(dests) == 1:
                    new_q = dests.pop()
        return (new_q,), None, None

    def guards(self, q, x):
        pass

    def timed_event(self, q, xc, xd):
        possible_destinations = list(dict(ev, dest=d) for s, d, ev in self._G.out_edges(q, data=True) if s == q)
        if possible_destinations:
            if len(possible_destinations) == 1:
                dest = possible_destinations[0]
            else:
                dest = np.random.choice(possible_destinations, p=[p['prob']/sum(x['prob'] for x in possible_destinations) if 'prob' in p else 1/len(possible_destinations) for p in possible_destinations])
            if 'time' in dest:
                time = dest['time']
                if callable(time):
                    time = time()
                return time, dest.get('destination', dest['dest'])
        return None, None

    def get_transition(self, s, d=None, e=None, if_more_than_one='raise'):
        """
        Get all transitions with source state s, destination state __d. In case when e is provided, the returned list
        contains transitions where event is e.
        :param if_more_than_one:
        :param s: Source state.
        :param d: Destination state.
        :param e: Event.
        :return:
        """
        transitions = self._G.out_edges(s, keys=True, data=True)
        if e is None and d is not None:
            transitions = [trans for trans in transitions if trans[1] == d]
        elif d is None and e is not None:
            transitions = [trans for trans in transitions if trans[2] == e]
        else:
            transitions = [trans for trans in transitions if trans[1] == d and trans[2] == e]

        if len(transitions) > 1:
            if if_more_than_one == 'raise':
                raise Exception('There are multiple transitions which satisfy the condition.')
            else:
                return transitions
        elif len(transitions) == 0:
            return None
        else:
            return transitions[0]

    def rename_events(self, prefix="e_"):
        """
    Rename events to become e_0, e_1... The old id is stored in the field 'old_symbol' of the state data.
        """
        i = 0
        new_events_dict = OrderedDict()
        for k, v in self.Sigma.items():
            new_key = f'{prefix}{i}'
            new_value = v
            if new_value is None:
                new_value = {}
            new_value['old_symbol'] = k
            i += 1
            new_events_dict[new_key] = new_value
            for t in self.T.values():
                if t['event'] == k:
                    t['event'] = new_key
        # self.Sigma = new_events_dict

    def step(self, q, x0, t, u):
        """
    Simulates one time step of continuous behavior from t to t+dt. Underlying function is solve_ivp with method is 'RK23'.
        :param x0: Initial state vector.
        :param t: Time at start of the step simulation.
        :param u: Arguments passed.....
        :return: Time t+dt, value of state at t+dt
        """
        s = solve_ivp(self.flow, t_span=(t, t + self.dt), y0=x0, method='RK23', args=u)
        xc = s.y[:, -1]
        t = s.t[-1]

        xk = self.time_discrete_dynamics(t, q, x0)
        return t, np.concatenate(xc, xk)

    def __str__(self):
        """
    String representation of the automaton.
        :return:
        """
        return f"""Automaton:
    Number of states: {self.num_modes}
    Number of transitions: {self.num_transitions}
    Initial state(s): {list(self.q0.keys())}
    Final state(s): {list(self.final_q.keys())}"""

    def flow(self, q, p, x, u):
        """
    Flow equation gives derivative of the continuous variables.
        :param q: Current discrete state of the model.
        :param p: Stochastic parameters generated on entry to state current_q.
        :param x: Current continuous state of the model.
        :param u: Calculated internal i signals.
        :return: Derivative of the continuous state x.
        """
        pass

    def inv(self, t, q, x, y, z, p):
        """
    Invariants.
        :param t:
        :param q:
        :param x:
        :param y:
        :param z:
        :param p:
        """
        pass

    def get_transitions(self):
        return list(self._G.edges(data=True, keys=True))

    def print_state(self, v):
        """Prints outgoing transitions of a state v.

        Args:
            v (state): 

        Returns:
            String: Description of the outgoing transitions of the state.
        """
        s = f'<b>{str(v)}</b>'
        for tr in self.out_transitions(v):
            s += f"<br>{tr[2]} -> {tr[1]} [{self.num_occur(tr[0], tr[2])}]"
        return s

    def sample_initial(self):
        if len(self.q0) == 0:
            current_q = np.random.choice(list(self.discrete_states.keys()), 1)[-1]
            warnings.warn(
                'Initial state not defined, sampling initial state uniformly from the set of all states.')
        else:
            current_q = np.random.choice(list(self.q0.keys()), 1)[-1]
        return current_q

    # def simulate(self, finish_time=100, current_q=None):
    #     """
    #     Simulates behaviour of the system.
    #     :param finish_time: Time when simulation finishes.
    #     :return: generated data.
    #     """
    #
    #     if current_q is None:
    #         if len(self.q0) == 0:
    #             current_q = np.random.choice(list(self.discrete_states.keys()), 1)[-1]
    #             warnings.warn(
    #                 'Initial state not defined, sampling initial state uniformly from the set of all states.')
    #         else:
    #             current_q = np.random.choice(list(self.q0.keys()), 1)[-1]
    #
    #     state_sequence = []
    #     data_state = []
    #     t = 0
    #     last_x = None
    #     last_output = None
    #     states = []
    #     data = []
    #     current_e = None
    #
    #     while True:
    #         cont_state, cont_time, cont_output = self.__step_continuous()
    #         last_x = dict(cont_state.iloc[-1])
    #         last_output = dict(cont_output.iloc[-1])
    #         cont_time = cont_time + t - cont_time.iloc[0, 0]
    #         clock = cont_time.iloc[-1, 0] - cont_time.iloc[0, 0]
    #
    #         current_state = current_q
    #
    #         observed_current_state = current_state
    #         state_sequence.append(
    #             pd.DataFrame(np.full((cont_time.size - 1, 3), (current_state, observed_current_state, current_e)),
    #                          index=cont_time.iloc[0:-1, 0],
    #                          columns=['State', 'Observed State', 'Event']))
    #
    #         data_state.append(cont_output.iloc[:-1].set_index(cont_time.iloc[0:-1, 0]))
    #
    #         tr = self.out_transitions(current_state)
    #
    #         if len(tr) == 0:
    #             break
    #         elif len(tr) != 1:
    #             tr = np.random.choice(tr, 1)[-1]
    #             warnings.warn('Multiple transitions can occur.')
    #         else:
    #             tr = tr[0]
    #
    #         last_q = current_q
    #         current_e = tr.event
    #         self.apply_sim_event(current_e)
    #         if cont_time.size == 0:
    #             break
    #         t += clock
    #
    #         if t >= finish_time:
    #             break
    #
    #         states.append(pd.concat(state_sequence, axis=0))
    #         data.append(pd.concat(data_state, axis=0))
    #     return states, data

    def predict_state(self, data_collection, time_col_name=None, discr_col_names=None):
        for data in data_collection:
            data["StateEstimate"] = None
            data["Event"] = None

            prev_discr_state = None
            prev_time = None

            if discr_col_names is None:
                discr_col_names = data.columns
            if time_col_name is not None:
                discr_col_names -= time_col_name

            for row in data[discr_col_names].itertuples(index=True):
                if time_col_name is not None:
                    time = data[time_col_name].iloc[row[0]]
                else:
                    time = row[0]
                discr_state = row[1:]
                if prev_discr_state is not None and prev_discr_state != discr_state:
                    event = np.asarray(discr_state) - np.asarray(prev_discr_state)
                    event = ' '.join(str(x) for x in event)
                    data.loc[row[0], "Event"] = event

                data.loc[row[0], "StateEstimate"] = signal_vector_to_state(discr_state)
                prev_discr_state = discr_state
                prev_time = time
        return data_collection


    def read_event(self, t, e, clear_p=False, keep_p=None, **kwargs):
        if keep_p:
            for k in keep_p:
                kwargs[k] = self._p[k]
        if clear_p or keep_p:
            self._p = kwargs
        else:
            self._p.update(kwargs)
        self._p.pop('Error Message', None)
        self._t = t

        self._p['event'] = e

        new_q, self._xt, self._xk = self.discrete_event_dynamics(self._q, self._xt, self._xk, self._p)
        new_q = new_q[0]

        if new_q == self.UNKNOWN_STATE:
            text2 = '{}->{}->{}'.format(self.discrete_state, e, new_q)
            self._p['Error Message'] = text2
        else:
            dl = self.decision_logic
            # d = self.check_decisions(self.x, self.overall_system.state)
            if dl and not self.overall_system.is_unknown(self):
                try:
                    state_key = json.dumps(self.overall_system.get_choice_state(self), sort_keys=True)
                except:
                    print('Problem getting choice state')
                old_e = self.choices_set.get(state_key, None)
                if old_e is not None:
                    if self._e not in old_e:
                        # warnings.warn(
                        #     '{}: Different decision for same state {}: {} vs {}'.format(t, state_key, old_e, self._e))
                        self.choices_set[state_key][self._e] = [str(t)]
                    else:
                        self.choices_set[state_key][self._e].append(str(t))
                else:
                    self.choices_set[state_key] = {self._e: [str(t)]}
                # if self._e not in (x[0] for x in d):
                #     warnings.warn('{}: Not allowed decision in state {}: {}'.format(t, state_key, self._e))

        self.discrete_state = new_q
        self._past_t.append(t)
        self._past_p.append(self._p)
        if len(self._discrete_state_data):
            self._discrete_state_data[-1]["Finish"] = t
        self._discrete_state_data.append(dict(Time=t, Event=e, Mode=self.discrete_state, **self._p))
        self._discrete_output_data.append([t, *self._d, self._p.get('event', None)])

num_modes property

Returns the number of modes in the automaton. :return: number of states.

num_transitions property

Returns the number of transitions in the automaton. :return: number of transitions.

state property writable

Automata discrete state is uni-variate. :return:

__init__(states=None, transitions=None, unknown_state='raise', id='', initial_q=(), initial_r=None, final_q=(), super_states=(), decision_states=(), **kwargs)

Class initialization from lists of elements. :param states: Discrete states / modes of continuous behavior. :param events: The events that trigger state transitions. :param transitions: The transition information. If a collection of dicts then dict should contain "source", "dest" and "event". The other attributes will be added as data of that transition. Alternatively, a collection of tuples can be used of the form (source, event, dest, *). :param unknown_state: The name of unknown states during "play in", if "raise", an exception will be raised.

Source code in ml4cps/automata/base.py

def __init__(self, states: list = None, transitions: list = None,
             unknown_state: str = 'raise', id="", initial_q=(), initial_r=None, final_q=(), super_states=(), decision_states=(),
             **kwargs):
    """
    Class initialization from lists of elements.
    :param states: Discrete states / modes of continuous behavior.
    :param events: The events that trigger state transitions.
    :param transitions: The transition information. If a collection of dicts then dict should contain "source",
    "dest" and "event". The other attributes will be added as data of that transition. Alternatively, a collection
    of tuples can be used of the form (source, event, dest, *).
    :param unknown_state: The name of unknown states during "play in", if "raise", an exception will be raised.
    """
    self._G = nx.MultiDiGraph()
    if initial_q and isinstance(initial_q, str):
        initial_q = [initial_q]
    self.q0 = OrderedDict.fromkeys(initial_q)
    self.initial_r = initial_r
    if final_q and isinstance(final_q, str):
        final_q = [final_q]
    self.final_q = OrderedDict.fromkeys(final_q)
    self.previous_node_positions = None
    self.UNKNOWN_STATE = unknown_state
    if super_states is not None:
        if type(super_states) is str:
            self.__super_states = [super_states]
        else:
            self.__super_states = list(super_states)

        # self._G.add_nodes_from(self.__super_states)

    if decision_states is not None:
        if type(decision_states) is str:
            self.decision_states = [decision_states]
        else:
            self.decision_states = list(decision_states)

    if states is not None:
        self._G.add_nodes_from(states)

    if transitions is not None:
        for tr in transitions:
            if type(tr) is dict:
                self._G.add_edge(tr.pop('source'), tr.pop('destination'), event=tr.pop('event'), **tr)
            else:
                self._G.add_edge(tr[0], tr[2], event=tr[1])

    if 'discr_state_names' not in kwargs:
        kwargs['discr_state_names'] = ['Mode']
    elif type(kwargs['discr_state_names']) is str:
        kwargs['discr_state_names'] = [kwargs['discr_state_names']]
    CPSComponent.__init__(self, id, **kwargs)

__str__()

String representation of the automaton. :return:

Source code in ml4cps/automata/base.py

def __str__(self):
    """
String representation of the automaton.
    :return:
    """
    return f"""Automaton:
Number of states: {self.num_modes}
Number of transitions: {self.num_transitions}
Initial state(s): {list(self.q0.keys())}
Final state(s): {list(self.final_q.keys())}"""

add_final_state(states)

Add final state(s) of the automaton. :param states: States to add.

Source code in ml4cps/automata/base.py

def add_final_state(self, states):
    """
Add final state(s) of the automaton.
    :param states: States to add.
    """
    if type(states) is str:
        states = (states,)
    for s in states:
        if s is not None and s not in self.final_q:
            self.final_q[s] = None
        self._G.add_node(s)

add_initial_state(states)

Add initial state(s) of the automaton. :param states: States to add.

Source code in ml4cps/automata/base.py

def add_initial_state(self, states):
    """
Add initial state(s) of the automaton.
    :param states: States to add.
    """
    if type(states) is str:
        states = (states,)
    for s in states:
        if s is not None and s not in self.q0:
            self.q0[s] = None
        self._G.add_node(s)

add_state(new_state, **kwargs)

Add state to the automaton. :param new_state: State to be added.

Source code in ml4cps/automata/base.py

def add_state(self, new_state, **kwargs):
    """
Add state to the automaton.
    :param new_state: State to be added.
    """
    self._G.add_node(new_state, **kwargs)

add_state_data(s, d)

Add state data to a state s the automaton. :param s: state :param d: data to be added to s :return:

Source code in ml4cps/automata/base.py

def add_state_data(self, s: str, d: object):
    """
Add state data to a state s the automaton.
    :param s: state
    :param d: data to be added to s
    :return:
    """
    self.Q[s] = d

add_states_from(new_state, **kwargs)

Add multiple states to the automaton. :param new_state: States to be added.

Source code in ml4cps/automata/base.py

def add_states_from(self, new_state, **kwargs):
    """
Add multiple states to the automaton.
    :param new_state: States to be added.
    """
    self._G.add_nodes_from(new_state, **kwargs)

add_transition(s, d, e, **other)

Add multiple transition. :param list_of_tuples: List of transitions in the form (source_state, destination_state, event, ......).

Source code in ml4cps/automata/base.py

def add_transition(self, s, d, e, **other):
    """
Add multiple transition.
    :param list_of_tuples: List of transitions in the form (source_state, destination_state, event, ...<unused>...).
    """
    self._G.add_edge(s, d, e, **other, event=e)

add_transitions_from(list_of_tuples, **other)

Add multiple transition. :param list_of_tuples: List of transitions in the form (source_state, destination_state, event, ......).

Source code in ml4cps/automata/base.py

def add_transitions_from(self, list_of_tuples, **other):
    """
Add multiple transition.
    :param list_of_tuples: List of transitions in the form (source_state, destination_state, event, ...<unused>...).
    """
    self._G.add_edges_from(list_of_tuples, **other)

flow(q, p, x, u)

Flow equation gives derivative of the continuous variables. :param q: Current discrete state of the model. :param p: Stochastic parameters generated on entry to state current_q. :param x: Current continuous state of the model. :param u: Calculated internal i signals. :return: Derivative of the continuous state x.

Source code in ml4cps/automata/base.py

def flow(self, q, p, x, u):
    """
Flow equation gives derivative of the continuous variables.
    :param q: Current discrete state of the model.
    :param p: Stochastic parameters generated on entry to state current_q.
    :param x: Current continuous state of the model.
    :param u: Calculated internal i signals.
    :return: Derivative of the continuous state x.
    """
    pass

get_num_in(q)

Returns the number of in transitions of state q in the automaton. :return: number of transitions.

Source code in ml4cps/automata/base.py

def get_num_in(self, q):
    """
    Returns the number of in transitions of state q in the automaton.
    :return: number of transitions.
    """
    if self._G.has_node(q):
        return self._G.in_degree(q)
    else:
        raise Exception(f'State {q} not in the automaton.')

get_num_out(q)

Returns the number of out transitions of state q in the automaton. :return: number of transitions.

Source code in ml4cps/automata/base.py

def get_num_out(self, q):
    """
    Returns the number of out transitions of state q in the automaton.
    :return: number of transitions.
    """
    if self._G.has_node(q):
        return self._G.out_degree(q)
    else:
        raise Exception(f'State {q} not in the automaton.')

get_transition(s, d=None, e=None, if_more_than_one='raise')

Get all transitions with source state s, destination state __d. In case when e is provided, the returned list contains transitions where event is e. :param if_more_than_one: :param s: Source state. :param d: Destination state. :param e: Event. :return:

Source code in ml4cps/automata/base.py

def get_transition(self, s, d=None, e=None, if_more_than_one='raise'):
    """
    Get all transitions with source state s, destination state __d. In case when e is provided, the returned list
    contains transitions where event is e.
    :param if_more_than_one:
    :param s: Source state.
    :param d: Destination state.
    :param e: Event.
    :return:
    """
    transitions = self._G.out_edges(s, keys=True, data=True)
    if e is None and d is not None:
        transitions = [trans for trans in transitions if trans[1] == d]
    elif d is None and e is not None:
        transitions = [trans for trans in transitions if trans[2] == e]
    else:
        transitions = [trans for trans in transitions if trans[1] == d and trans[2] == e]

    if len(transitions) > 1:
        if if_more_than_one == 'raise':
            raise Exception('There are multiple transitions which satisfy the condition.')
        else:
            return transitions
    elif len(transitions) == 0:
        return None
    else:
        return transitions[0]

in_transitions(s, event=None)

Get all incoming transitions of state s. :param s: :return:

Source code in ml4cps/automata/base.py

def in_transitions(self, s, event=None):
    """
Get all incoming transitions of state s.
    :param s:
    :return:
    """
    if event is None:
        return self._G.in_edges(s, data=True, keys=True)
    else:
        return [e for e in self._G.in_edges(s, data=True, keys=True) if e[3]['event'] == event]

inv(t, q, x, y, z, p)

Invariants. :param t: :param q: :param x: :param y: :param z: :param p:

Source code in ml4cps/automata/base.py

def inv(self, t, q, x, y, z, p):
    """
Invariants.
    :param t:
    :param q:
    :param x:
    :param y:
    :param z:
    :param p:
    """
    pass

is_transition(s, d, e)

Check if a transition (s,d,e) exists in the automaton. :param s: Source. :param d: Destination. :param e: Event. :return:

Source code in ml4cps/automata/base.py

def is_transition(self, s, d, e):
    """
Check if a transition (s,d,e) exists in the automaton.
    :param s: Source.
    :param d: Destination.
    :param e: Event.
    :return:
    """
    transitions = [trans for trans in self.T.values() if trans['source'] == s and
                   trans['destination'] == d and trans['event'] == e]

    is_transition = len(transitions) != 0
    return is_transition

merge(q1, q2)

If two states are compatible, they are merged with the function merge. The transitions of the automaton, the in- and outdegree of the states and the number of transitions happening are adjusted.

Source code in ml4cps/automata/base.py

def merge(self, q1, q2):
    """
    If two states are compatible, they are merged with the function merge. The transitions
    of the automaton, the in- and outdegree of the states and the number of transitions
    happening are adjusted.
    """
    intr = self.in_transitions(q2)
    # outtr_q1 = list(self.out_transitions(q1))
    outtr_q2 = list(self.out_transitions(q2))
    # nx.contracted_nodes(self._G, w, v, copy=False)
    # set event

    for tr in intr:
        if tr[0] != q2:
            self.add_single_transition(tr[0], q1, tr[2], timing=tr[-1]['timing'])
    # if q1=='q0' and q2 == 'q192':
    #     print('found')
    for tr in outtr_q2:
        dest = tr[1]
        if dest == q2:
            dest = q1
        self.add_single_transition(q1, dest, tr[2], timing=tr[-1]['timing'])
    self.remove_state(q2)

out_transitions(s, event=None)

Get all outgoing transitions of state s. :param event: :param s: :return:

Source code in ml4cps/automata/base.py

def out_transitions(self, s, event=None):
    """
Get all outgoing transitions of state s.
    :param event:
    :param s:
    :return:
    """
    if event is None:
        return self._G.out_edges(s, data=True, keys=True)
    else:
        return [e for e in self._G.out_edges(s, data=True, keys=True) if e[3]['event'] == event]

print_state(v)

Prints outgoing transitions of a state v.

Parameters:

Name	Type	Description	Default
v	state		required

Returns:

Name	Type	Description
String		Description of the outgoing transitions of the state.

Source code in ml4cps/automata/base.py

def print_state(self, v):
    """Prints outgoing transitions of a state v.

    Args:
        v (state): 

    Returns:
        String: Description of the outgoing transitions of the state.
    """
    s = f'<b>{str(v)}</b>'
    for tr in self.out_transitions(v):
        s += f"<br>{tr[2]} -> {tr[1]} [{self.num_occur(tr[0], tr[2])}]"
    return s

remove_transition(source, dest)

Remove the transition(s) from source to dest. :param source: :param dest: :return:

Source code in ml4cps/automata/base.py

def remove_transition(self, source, dest):
    """
    Remove the transition(s) from source to dest.
    :param source:
    :param dest:
    :return:
    """
    self._G.remove_edge(source, dest)

rename_events(prefix='e_')

Rename events to become e_0, e_1... The old id is stored in the field 'old_symbol' of the state data.

Source code in ml4cps/automata/base.py

def rename_events(self, prefix="e_"):
    """
Rename events to become e_0, e_1... The old id is stored in the field 'old_symbol' of the state data.
    """
    i = 0
    new_events_dict = OrderedDict()
    for k, v in self.Sigma.items():
        new_key = f'{prefix}{i}'
        new_value = v
        if new_value is None:
            new_value = {}
        new_value['old_symbol'] = k
        i += 1
        new_events_dict[new_key] = new_value
        for t in self.T.values():
            if t['event'] == k:
                t['event'] = new_key

step(q, x0, t, u)

Simulates one time step of continuous behavior from t to t+dt. Underlying function is solve_ivp with method is 'RK23'. :param x0: Initial state vector. :param t: Time at start of the step simulation. :param u: Arguments passed..... :return: Time t+dt, value of state at t+dt

Source code in ml4cps/automata/base.py

def step(self, q, x0, t, u):
    """
Simulates one time step of continuous behavior from t to t+dt. Underlying function is solve_ivp with method is 'RK23'.
    :param x0: Initial state vector.
    :param t: Time at start of the step simulation.
    :param u: Arguments passed.....
    :return: Time t+dt, value of state at t+dt
    """
    s = solve_ivp(self.flow, t_span=(t, t + self.dt), y0=x0, method='RK23', args=u)
    xc = s.y[:, -1]
    t = s.t[-1]

    xk = self.time_discrete_dynamics(t, q, x0)
    return t, np.concatenate(xc, xk)

try_merge_states(state1, state2, try_fun=None)

Merge state2 into state1 and update transitions.

Source code in ml4cps/automata/base.py

def try_merge_states(self, state1, state2, try_fun=None):
    """Merge state2 into state1 and update transitions."""
    old_G = self._G
    make_final = state2 in self.final_q and state1 not in self.final_q
    make_initial = state2 in self.q0 and state1 not in self.q0
    self._G = nx.contracted_nodes(self._G, state1, state2)
    # make it deterministic again
    events = [x[3]['event'] for x in self.out_transitions(state1)]
    deterministic = len(set(events)) == len(events)

    if deterministic:
        if make_initial:
            self.add_initial_state(state1)
        if make_final:
            self.add_final_state(state1)

        if try_fun is not None and not try_fun(self):
            self._G = old_G
            if make_final:
                self.final_q.pop(state1)
            if make_initial:
                self.q0.pop(state1)
    else:
        self._G = old_G

cps

The module provides base classes to represent the dynamics of cyber-physical systems (CPS): - CPS, and, - CPSComponent.

Author: Nemanja Hranisavljevic, hranisan@hsu-hh.de

CPS

CPS class represents a cyber-physical system. It is used to model the hierarchy in the system as each CPS can contain a mixture of other CPS and CPS components. The leaves of the hierarchy are only CPS component objects which define their dynamics and communication with other components.

Attributes:

Name	Type	Description
id	str	Identifier of the object.
com	OrderedDict	Property. A collection of components.

Source code in ml4cps/cps/base.py

class CPS:
    """
    CPS class represents a cyber-physical system. It is used to model the hierarchy in the system as each CPS can
    contain a mixture of other CPS and CPS components. The leaves of the hierarchy are only CPS component objects
    which define their dynamics and communication with other components.

    Attributes:
        id (str): Identifier of the object.
        com (OrderedDict): Property. A collection of components.
    """

    def __init__(self, sys_id, components):
        """
        Initializes CPS with the given attributes.

        Args:
            sys_id (str): ID of the system.
            components (iterable): Child components of the system.
        """
        super().__init__()
        # self._env = simpy.Environment()  # It is a simpy environment used internally for simulation
        self.id = sys_id
        self._parent_system = None  # It is always referenced two-way from children to parent and vice versa
        self._com = OrderedDict()  # "Private" in order to avoid setting value without reference in other way
        for s in components:
            s._parent_system = self
            self._com[s.id] = s

    def copy(self):
        new_copy = copy.deepcopy(self)
        return new_copy


    @property
    def parent_system(self):
        """
        Gets the parent system by accessing _parent_system private attribute.

        Returns:
            (CPS): The parent system.
        """
        return self._parent_system

    @parent_system.setter
    def parent_system(self, parent_system):
        """
        Args:
            parent_system (CPS): The new parent system.

        Raises:
            AttributeError: If we cannot set child of the parent system or if the component with the same id is already
            a child of the parent system.
        """
        if id in parent_system._com:
            raise AttributeError("Component with that id is already a child of the parent system.")
        if self._parent_system is not None and self.id in self._parent_system:
            self._parent_system.pop(self.id)
        try:
            self._parent_system = parent_system
            parent_system._com[id] = self
        except AttributeError as ex:
            raise ex

    def __getitem__(self, key):
        """
        Gets the component with the given key in the collection.

        Args:
            key (string): The key where the value will be stored. Must be a hashable type (e.g., int, str).
        Return:
            (CPS, CPSComponent): Returned component or subsystem.

        """
        return self._com[key]

    def __setitem__(self, key, value):
        self._com[key] = value
        self._parent_system[value] = self

    # Method to get components as in dict.
    def items(self):
        return self._com.items()

    @property
    def state(self):
        return {k: x.state for k, x in self._com.items()}

    @state.setter
    def state(self, value): # If some component is not specified in value dict it will stay the same
        for k, v in value.items():
            self[k].state = v

    @property
    def overall_system(self):
        s = self
        while s.parent_system:
            s = s.parent_system
        return s

    def get_components(self, exclude=None):
        """Returns a list of all direct child components except those with ids possibly provided in exclude.

        Parameters
        ----------
        exclude : iterable
            A list of component ids to exclude in the returned list (default is None).
        """
        if exclude is not None and type(exclude) is str:
            exclude = [exclude]
        return list(c for c in self._com.values() if exclude is None or c.id not in exclude)

    def set_child_component(self, id, com):
        """Set component with the id.

        Parameters
        ----------
        id : str
            ID of the component to add.
        com : (CPS, CPSComponent)
            Component of subsystem to add.
        """
        self._com[id] = com
        com.parent_system = self

    def get_all_components(self, exclude=None):
        """Returns a list of all child components (not only direct) except those with ids possibly provided in exclude.

        Parameters
        ----------
        exclude : iterable
            A list of component ids to exclude in the returned list.
        """
        comp = []
        for c in self._com.values():
            if issubclass(type(c), CPS):
                comp += c.get_all_components(exclude=exclude)
            elif exclude is None or c.id not in exclude:
                comp.append(c)
        return comp

    def get_component(self, name):
        for k, c in self._com.items():
            if type(c) is CPS:
                cc = c.get_component(name)
                if cc is not None:
                    return cc
            else:
                if k == name:
                    return c
        return None

    def get_component_by_full_id(self, full_id):
        ids = full_id.split('.')
        c = self
        for ii in ids:
            c = c[ii]
        return c

    def get_execution_data(self, flat=False):
        exe_data = OrderedDict()
        if flat:
            for c in self.get_all_components():
                exe_data[c.full_id] = c.get_execution_data()
        else:
            for k, c in self.com.items():
                exe_data[c.id] = c.get_execution_data()
        return exe_data

    def reinitialize(self, t=0, state=None):
        """
        The function re-initializes the CPS components of this CPS with the given state values.
        :param t: Current time to set to the components.
        :param state: State of the CPS (it's components) given as a dictionary of dictionaries ... of tuples (according
        to the CPS hierarchy). The tuplus are created as values concatenated values of discrete-event state variables,
        time-continuous state variables and time-discrete state variables.

        :return:
        """
        if state is None:
            state = {}
        for k, v in self._com.items():
            v.reinitialize(t, state.get(k, None))

    def simulate(self, finish_time, verbose=False, save_choices=False):
        """Simulates behaviour of the system until the finish_time is reached.

        Parameters
        ----------
        finish_time : Time when simulation finishes.
        verbose : Should the execution log be printed in detail (default is False).
        save_choices : Should the choices for each component be saved to json files after
        the simulation is finished (default is False).
        """

        env = simpy.Environment()
        finish_time = float(finish_time)
        print('Simulation started: ')

        for s in self.get_all_components():
            s.simulation_process_simpy(env, finish_time, verbose)

        env.run(until=finish_time)

        stateflow_data = {}
        discr_output_data = {}
        cont_state_data = {}
        cont_output_data = {}

        decisions = {}
        for s in self.get_all_components():
            try:
                if len(s._xt) > 0 and finish_time > s._t:  # there is time-continuous state variable
                    s._step_continuous(s._t, finish_time)
            except Exception as ex:
                print_exc()
                raise Exception('Exception during self._step_continuous in state {}'.format(s._q))

            if save_choices and len(s.choices_set) > 0:
                out_file = open(s.id + str(env.now) + ".json", "w")
                json.dump(s.choices_set, out_file, indent=6)
            try:
                s.finish(env.now)
                # if s.parent_system.shared_alternatives_set:
                #     if s.parent_system.id not in decisions:
                #         decisions[s.parent_system.id] = s.choices_set
                # else:
                #     decisions[s.id] = s.choices_set
                stateflow_data[s.id] = pd.DataFrame(s._discrete_state_data)
                discr_output_data[s.id] = tools.data_list_to_dataframe(None, s._discrete_output_data,
                                                                       s.discrete_output_names,
                                                                       'd', 'e')
                try:
                    cont_state_data[s.id] = tools.data_list_to_dataframe(s.id, s._continuous_state_data, s.cont_state_names,
                                                                     'x')
                except Exception as ex:
                    print_exc()
                cont_output_data[s.id] = tools.data_list_to_dataframe(s.id, s._continuous_output_data,
                                                                      s.cont_output_names, 'y')
            except Exception as ex:
                print_exc()
                warnings.warn('Simulation failed.')
        print('Simulation finished.')
        return stateflow_data, discr_output_data, cont_state_data, cont_output_data, env.now

    def finish_condition(self):
        pass

parent_system property writable

Gets the parent system by accessing _parent_system private attribute.

Returns:

Type	Description
CPS	The parent system.

__getitem__(key)

Gets the component with the given key in the collection.

Parameters:

Name	Type	Description	Default
key	string	The key where the value will be stored. Must be a hashable type (e.g., int, str).	required

Return: (CPS, CPSComponent): Returned component or subsystem.

Source code in ml4cps/cps/base.py

def __getitem__(self, key):
    """
    Gets the component with the given key in the collection.

    Args:
        key (string): The key where the value will be stored. Must be a hashable type (e.g., int, str).
    Return:
        (CPS, CPSComponent): Returned component or subsystem.

    """
    return self._com[key]

__init__(sys_id, components)

Initializes CPS with the given attributes.

Parameters:

Name	Type	Description	Default
sys_id	str	ID of the system.	required
components	iterable	Child components of the system.	required

Source code in ml4cps/cps/base.py

def __init__(self, sys_id, components):
    """
    Initializes CPS with the given attributes.

    Args:
        sys_id (str): ID of the system.
        components (iterable): Child components of the system.
    """
    super().__init__()
    # self._env = simpy.Environment()  # It is a simpy environment used internally for simulation
    self.id = sys_id
    self._parent_system = None  # It is always referenced two-way from children to parent and vice versa
    self._com = OrderedDict()  # "Private" in order to avoid setting value without reference in other way
    for s in components:
        s._parent_system = self
        self._com[s.id] = s

get_all_components(exclude=None)

Returns a list of all child components (not only direct) except those with ids possibly provided in exclude.

Parameters

exclude : iterable A list of component ids to exclude in the returned list.

Source code in ml4cps/cps/base.py

def get_all_components(self, exclude=None):
    """Returns a list of all child components (not only direct) except those with ids possibly provided in exclude.

    Parameters
    ----------
    exclude : iterable
        A list of component ids to exclude in the returned list.
    """
    comp = []
    for c in self._com.values():
        if issubclass(type(c), CPS):
            comp += c.get_all_components(exclude=exclude)
        elif exclude is None or c.id not in exclude:
            comp.append(c)
    return comp

get_components(exclude=None)

Returns a list of all direct child components except those with ids possibly provided in exclude.

Parameters

exclude : iterable A list of component ids to exclude in the returned list (default is None).

Source code in ml4cps/cps/base.py

def get_components(self, exclude=None):
    """Returns a list of all direct child components except those with ids possibly provided in exclude.

    Parameters
    ----------
    exclude : iterable
        A list of component ids to exclude in the returned list (default is None).
    """
    if exclude is not None and type(exclude) is str:
        exclude = [exclude]
    return list(c for c in self._com.values() if exclude is None or c.id not in exclude)

reinitialize(t=0, state=None)

The function re-initializes the CPS components of this CPS with the given state values. :param t: Current time to set to the components. :param state: State of the CPS (it's components) given as a dictionary of dictionaries ... of tuples (according to the CPS hierarchy). The tuplus are created as values concatenated values of discrete-event state variables, time-continuous state variables and time-discrete state variables.

:return:

Source code in ml4cps/cps/base.py

def reinitialize(self, t=0, state=None):
    """
    The function re-initializes the CPS components of this CPS with the given state values.
    :param t: Current time to set to the components.
    :param state: State of the CPS (it's components) given as a dictionary of dictionaries ... of tuples (according
    to the CPS hierarchy). The tuplus are created as values concatenated values of discrete-event state variables,
    time-continuous state variables and time-discrete state variables.

    :return:
    """
    if state is None:
        state = {}
    for k, v in self._com.items():
        v.reinitialize(t, state.get(k, None))

set_child_component(id, com)

Set component with the id.

Parameters

id : str ID of the component to add. com : (CPS, CPSComponent) Component of subsystem to add.

Source code in ml4cps/cps/base.py

def set_child_component(self, id, com):
    """Set component with the id.

    Parameters
    ----------
    id : str
        ID of the component to add.
    com : (CPS, CPSComponent)
        Component of subsystem to add.
    """
    self._com[id] = com
    com.parent_system = self

simulate(finish_time, verbose=False, save_choices=False)

Simulates behaviour of the system until the finish_time is reached.

Parameters

finish_time : Time when simulation finishes. verbose : Should the execution log be printed in detail (default is False). save_choices : Should the choices for each component be saved to json files after the simulation is finished (default is False).

Source code in ml4cps/cps/base.py

def simulate(self, finish_time, verbose=False, save_choices=False):
    """Simulates behaviour of the system until the finish_time is reached.

    Parameters
    ----------
    finish_time : Time when simulation finishes.
    verbose : Should the execution log be printed in detail (default is False).
    save_choices : Should the choices for each component be saved to json files after
    the simulation is finished (default is False).
    """

    env = simpy.Environment()
    finish_time = float(finish_time)
    print('Simulation started: ')

    for s in self.get_all_components():
        s.simulation_process_simpy(env, finish_time, verbose)

    env.run(until=finish_time)

    stateflow_data = {}
    discr_output_data = {}
    cont_state_data = {}
    cont_output_data = {}

    decisions = {}
    for s in self.get_all_components():
        try:
            if len(s._xt) > 0 and finish_time > s._t:  # there is time-continuous state variable
                s._step_continuous(s._t, finish_time)
        except Exception as ex:
            print_exc()
            raise Exception('Exception during self._step_continuous in state {}'.format(s._q))

        if save_choices and len(s.choices_set) > 0:
            out_file = open(s.id + str(env.now) + ".json", "w")
            json.dump(s.choices_set, out_file, indent=6)
        try:
            s.finish(env.now)
            # if s.parent_system.shared_alternatives_set:
            #     if s.parent_system.id not in decisions:
            #         decisions[s.parent_system.id] = s.choices_set
            # else:
            #     decisions[s.id] = s.choices_set
            stateflow_data[s.id] = pd.DataFrame(s._discrete_state_data)
            discr_output_data[s.id] = tools.data_list_to_dataframe(None, s._discrete_output_data,
                                                                   s.discrete_output_names,
                                                                   'd', 'e')
            try:
                cont_state_data[s.id] = tools.data_list_to_dataframe(s.id, s._continuous_state_data, s.cont_state_names,
                                                                 'x')
            except Exception as ex:
                print_exc()
            cont_output_data[s.id] = tools.data_list_to_dataframe(s.id, s._continuous_output_data,
                                                                  s.cont_output_names, 'y')
        except Exception as ex:
            print_exc()
            warnings.warn('Simulation failed.')
    print('Simulation finished.')
    return stateflow_data, discr_output_data, cont_state_data, cont_output_data, env.now

CPSComponent

Bases: PythonModel, Simulator

General hybrid system class based on scipy and simpy.

Source code in ml4cps/cps/base.py

class CPSComponent(PythonModel, sim.Simulator):
    """
    General hybrid system class based on scipy and simpy.
    """

    def __init__(self, id, t=0, initial_q=(), initial_xt: list = (), initial_xk: list = (), dt=-1., p=None,
                 cont_state_names=None, discr_state_names=None, discr_output_names=None,
                 cont_output_names=None, unknown_state="Unknown"):
        sim.Simulator.__init__(self)
        PythonModel.__init__(self)

        self._parent_system = None
        self.decision_logic = None
        self.id = id
        self.dt = dt

        # State variables
        self._t = t
        self._q = initial_q
        self._xt = initial_xt
        self._xk = initial_xk

        # Other variables
        self._y = None
        self._u = None
        self._d = ()
        self._p = {} if p is None else p

        # simpy events
        self._pending_events = []
        self._block_event = None

        # List to append the past values
        self._past_p = []
        self._past_t = []
        self._discrete_state_data = []
        self._discrete_output_data = []
        self._continuous_state_data = []
        self._continuous_output_data = []

        # Variable names
        self.cont_state_names = cont_state_names
        self.cont_output_names = cont_output_names
        self.discrete_state_names = discr_state_names
        self.discrete_output_names = discr_output_names

        self.UNKNOWN_STATE = unknown_state

        self.output_y = None
        self.output_d = None



    # def copy(self):
    #     new_copy = CPSComponent(self.id, self._t, initial_q=self._q, initial_xt=self._xt,
    #                             initial_xk = self._xk, dt = self.dt, p=self._p,
    #     cont_state_names = copy.copy(self.cont_state_names), discr_state_names = copy.copy(self.discrete_state_names),
    #                             discr_output_names=copy.copy(self.discrete_output_names),
    #     cont_output_names=copy.copy(self.cont_output_names), unknown_state = copy.copy(self.UNKNOWN_STATE))
    #     return new_copy

    def copy(self):
        new_copy = copy.deepcopy(self)
        return new_copy

    @property
    def parent_system(self):
        return self._parent_system

    @property
    def full_id(self):
        full_id = self.id
        s = self.parent_system
        while s.parent_system:
            full_id = s.id + "." + full_id
            s = s.parent_system
        return full_id

    @property
    def overall_system(self):
        s = self
        while s.parent_system:
            s = s.parent_system
        return s

    @property
    def state(self):
        """
        System state is a vector of all state variables.
        :return:
        """
        return self._q, self._xt, self._xk

    @state.setter
    def state(self, value):
        self._q = value[0]
        self._xt = value[1]
        self._xk = value[2]

    @property
    def discrete_state(self):
        return self._q

    @discrete_state.setter
    def discrete_state(self, value):
        self._q = value

    def is_decision(self, state, overall_state):
        return False

    def __str__(self):
        return '{}: {}'.format(self.id, self._q)

    def _step_continuous(self, clock_start, clock_finish):
        return self.__step_continuous(clock_start, clock_finish)

    def simulation_process_simpy(self, env, max_time, verbose=False):
        """
            The function simulates single concurrent thread of one component (CPSComponent).
            Parameters
            ----------
            env : simpy.Environment
                It is simpy environment used to synchronize multiple parallel threads during simulation.
            max_time : float
                It is the maximum time that the simulation will perform.
            verbose : bool
                Should the function print execution logs (default is False).
        """
        env.process(self.discrete_dynamics_process(env, max_time, verbose=verbose))
        env.process(self.continuous_dynamics_process(env, max_time, verbose=verbose))
        env.process(self.discrete_time_dynamics_process(env, max_time, verbose=verbose))

    def discrete_event_dynamics(self, q, xt, xk, p) -> tuple:
        pass

    def discrete_dynamics_process(self, env, max_time, verbose=False):
        # self._env = env
        self._u = self.input(self._q, 0)
        if self._p is None:
            self._p = self.context(self._q, None, None, 0)
        self._past_t = [0]
        self._past_p = [self._p]


        self._discrete_state_data = []
        self._discrete_output_data = []

        while True:
            discr_state = dict(zip(self.discrete_state_names, self._q))
            if self.output_d is None:
                self._d = self.discrete_state
            else:
                self._d = self.output_d(self.discrete_state, self._xt, self._xk)
            self._discrete_state_data.append(dict(Time=env.now, **discr_state, **self._p))
            self._discrete_output_data.append([env.now, *self._d])

            # Stop the simulation if the max time is reached
            if env.now > max_time:
                if verbose:
                    print(f'Stop because of maximum time {max_time} is reached.')
                break
            # Stop if the stopping condition of the overall system is met
            elif self.overall_system.finish_condition():
                if verbose:
                    print('Stop because stop condition of the overall system is met.')
                break

            # DETERMINE THE NEXT EVENT
            # 1. CHECK IF TIMED EVENT
            try:
                event_delay, new_state_value = self.timed_event(*self.state)
            except Exception as ex:
                print(f"{self.id}: Exception during self.timed_transition in state {self._q}")
                raise ex

            # 2. IF NOT TIMED THEN WAIT
            if event_delay is None:
                if verbose:
                    print('{}: Waiting in: {}'.format(self.id, self._q))
                if self._block_event is None:
                    self._block_event = env.event()
                    yield self._block_event
                else:
                    raise Exception("{}: State: {}. Exception during block event.".format(self.id, self._q))
                event = self._block_event.value
                self._block_event = None

                self._p['event'] = event
                new_q, xt, xk = self.discrete_event_dynamics(self._q, self._xt, self._xk, self._p)
                if verbose:
                    print(f'{self._q}->{event}->{new_q}')
                if xt is not None:
                    self._xt = xt
                if xk is not None:
                    self._xk = xk

                self._q = new_q
            else:
                if verbose:
                    print('Waiting in: {} for {}. New state: {}'.format(self._q, event_delay, new_state_value))
                yield env.timeout(event_delay)
                self._q = (new_state_value,)

            try:
                self.on_entry(self._q, self._p)
            except Exception as ex:
                print(f"{self.id}: Exception during discrete event dynamics '{self._q}'->")
                raise ex

    def discrete_time_dynamics_process(self, env, max_time, verbose=False):
        while (self._xk is not None and len(self._xk) > 0) or self.output_y is not None:
            self._continuous_state_data.append((self._t, *self._xk))
            self._continuous_output_data.append((self._t, *self.output_y(*self.state)))

            # Stop the simulation if the max time is reached
            if env.now > max_time:
                if verbose:
                    print(f'Stop because of maximum time {max_time} is reached.')
                break
            # Stop if the stopping condition of the overall system is met
            elif self.overall_system.finish_condition():
                if verbose:
                    print('Stop because stop condition of the overall system is met.')
                break

            yield env.timeout(self.dt)
            xk = self.time_discrete_dynamics(self._q[0], self._p, self._xk, self._u)
            self._xk = xk
            self._t = env.now
            grds = self.guards(self._q[0], self._xk)
            if grds is not None:
                self.apply_sim_event(grds, env)


    def guards(self, q, x):
        pass

    def continuous_dynamics_process(self, env, max_time, verbose=False):
        # try:
        #     if len(self._xt) > 0 or len(self._xk) > 0:  # there is time-continuous state variable
        #         self.__step_continuous(*self.state)
        # except Exception as ex:
        #     print_exc()
        #     raise Exception('Exception during self.__step_continuous in state {}'.format(self._q))

        # EXECUTE THE NEXT EVENT
        if False:
            yield
        # if verbose:
        #     print('--------------------')
        #
        # self._past_t.append(env.now)
        # self._past_p.append(self._p)
        #
        # self._discrete_state_data[-1]['Finish'] = env.now
        # discr_state = dict(zip(self.discrete_state_names, self._q))
        # self._discrete_state_data.append(
        #     dict(Time=env.now, Finish=None, **discr_state, **self._p))
        # self._discrete_output_data.append([env.now, *self._d])
        #
        # if verbose:
        #     print('{}: Time: {} State: {}'.format(self.id, env.now, self._q))
        #
        # if self._xt is not None:
        #     self._continuous_state_data.append([env.now, *self._xt, *self._xk])
        # if self._y is not None:
        #     self._continuous_output_data.append([env.now, *self._y])
        # if self._xt is not None:
        #     self._xt = self.__step_continuous(env.now - clock_start)
        # self._d, self._y = self.output(self._q, 0, self._xt, self._xk, self._u, self._p)
        #
        # print('--------------------')
        # print('{}: Simulation finished'.format(self.id))

    def simulate(self, finish_time, verbose=False, save_choices=False):
        """Simulates behaviour of the system until the finish_time is reached.

        Parameters
        ----------
        finish_time : Time when simulation finishes.
        verbose : Should the execution log be printed in detail (default is False).
        save_choices : Should the choices for each component be saved to json files after
        the simulation is finished (default is False).
        """

        env = simpy.Environment()
        finish_time = float(finish_time)

        if verbose:
            print('Simulation started: ')

        self.simulation_process_simpy(env, finish_time, verbose)
        env.run(until=finish_time)

        discr_state_data = {}
        discr_output_data = {}
        cont_state_data = {}
        cont_output_data = {}

        try:
            if len(self._xt) > 0 and len(self._xt) > 0 and finish_time > self._t:  # there is time-continuous state variable
                self._step_continuous(self._t, finish_time)
        except Exception as ex:
            print_exc()
            raise Exception('Exception during self._step_continuous in state {}'.format(self._q))

        try:
            self.finish(env.now)

            discr_state_data = pd.DataFrame(self._discrete_state_data).set_index("Time")
            discr_output_data = tools.data_list_to_dataframe(None, self._discrete_output_data,
                                                                   self.discrete_output_names, 'd')
            try:
                cont_state_data = tools.data_list_to_dataframe(self.id, self._continuous_state_data,
                                                               self.cont_state_names, 'x')
            except Exception as ex:
                print_exc()
            cont_output_data = tools.data_list_to_dataframe(self.id, self._continuous_output_data,
                                                                     self.cont_output_names, 'y')
        except Exception as ex:
            print_exc()
            warnings.warn('Simulation failed.')

        if verbose:
            print('Simulation finished.')
        return discr_state_data, discr_output_data, cont_state_data, cont_output_data, env.now

    def get_sim_state(self):
        return self._q, self._p, self._y, self._block_event

    def set_sim_state(self, q, e, p, y, block_event):
        self._q = q
        self._p = p
        self._y = y
        self._block_event = block_event

    def finish_condition(self):
        pass

    def reinitialize(self, t, state=None):
        if state is not None:
            self.state = state

        self._p = self.context(self._q, None, None, 0)
        self._u = self.input(self._q, 0)
        self._past_t = [t]
        self._past_p = [self._p]
        # self.output(self._q, 0, self._xt, self._xk, self._u, self._p)
        self._continuous_state_data = []
        self._continuous_output_data = []
        try:
            discr_state = dict(zip(self.discrete_state_names, self._q))
        except:
            raise Exception('Exception during discrete state initialization')

        self._discrete_state_data = [dict(Time=t, **discr_state, **self._p)]
        self._discrete_output_data = []
        self._block_event = None
        self._pending_events = []
        self._t = t

    def get_execution_data(self):
        try:
            data = pd.DataFrame(self._discrete_state_data) #, columns=['Time', 'Finish', 'State', 'Event'] + list(self._p.keys()))
        except ValueError as ve:
            warnings.warn(str(self._discrete_state_data))
            warnings.warn('Discrete state data not available due to value error.')
            for i, row in enumerate(self._discrete_state_data):
                for k, v in row.items():
                    if not is_scalar(v):
                        print(f"Non-scalar at row {i}, key '{k}'")
                        print("Type:", type(v))
                        print("Length:", len(v) if hasattr(v, "__len__") else "N/A")
                        break
            raise ve

        data['Finish'] = data['Time'].shift(-1)
        data['Duration'] = pd.to_timedelta(data['Finish'] - data['Time']).dt.total_seconds()
        return data

    # def clear_execution_data(self):
    #     self.discrete_state_data = None
    #     data = pd.DataFrame(self._discrete_state_data) #, columns=['Time', 'Finish', 'State', 'Event'] + list(self._p.keys()))
    #     data['Finish'] = data['Time'].shift(-1)
    #     data['Duration'] = pd.to_timedelta(data['Finish'] - data['Time']).dt.total_seconds()
    #     return data

    def update_input(self, u):  # Set i if you want from outside
        self._u = u

    def output_d(self, q, xt, xk):
        return ()

    def output_y(self, q, xt, xk):
        return ()

    def input(self, q, clock) -> tuple:
        pass

    def context(self, q, past_t, past_p, t) -> tuple:
        return dict()

    def time_continuous_dynamics(self, t, xt, xk, q, u, p, y):
        pass

    def time_discrete_dynamics(self, q, p, x, u):
        pass

    def on_entry(self, q, context):
        pass

    def __step_continuous(self, clock_start, clock_finish):
        """
    Simulates one time step of continuous behavior from t to t+dt. Underlying function is solve_ivp with method is 'RK23'.
        :param x0: Initial state vector.
        :param t: Time at start of the step simulation.
        :param u: Arguments passed.....
        :return: Time t+dt, value of state at t+dt
        """

        s = solve_ivp(self.time_continuous_dynamics, t_eval=np.arange(clock_start + self.dt, clock_finish + self.dt, self.dt),
                      t_span=(clock_start, clock_finish + self.dt), y0=self._xt, method='RK23',
                      args=(self._xk, self._q, self._u, self._p, self._y))

        self._xk = self.time_discrete_dynamics(self._q, self._p, self._xt, self._u)
        self._xt = s.y[-1]

        # self._continuous_state_data += np.concatenate([s.t.T[None], s.y]).T.tolist()

    def invariants(self, q, clock, xc, xd, y):
        pass

    def timed_event(self, q, xc, xd):
        """
        Calculates if and when the next time event will happen and the new state values.
        :param t: Current time.
        :param q: Discrete-event part of the state.
        :param xc: Time-continuous part of the state.
        :param xd: Time-discrete part of the state.
        :return: Time delay of the timed event, new state value.
        """
        return None, None

    def wait(self, q):
        pass

    def apply_sim_event(self, e, env=None):
        """
        The event e is applied in this component's simpy execution, this means that the process must wait for an event.
        :param e: Event to apply.
        :return:
        """
        if not self._block_event:
            # self._block_event = self._env.now
            self._block_event = env.now
        # if not self._block_event:
        #     raise Exception("Block event is not set, but event is applied.")
        elif self._block_event.triggered:
            # pass
            # if self._e != e:
            raise Exception('Should not happen')
        else:
            self._block_event.succeed(value=e)

    def finish(self, t, **kwargs):
        if self._p is None:
            self._p = kwargs
        else:
            self._p.update(kwargs)
        self._t = t
        # self._e = None
        # self._q = None
        self._past_p.append(self._p)
        # self._discrete_state_data[-1]['Finish'] = t

    def get_alternatives(self, state, system_state):
        return None

    def get_decision_state(self, state, overall_state):
        return state

state property writable

System state is a vector of all state variables. :return:

__step_continuous(clock_start, clock_finish)

Simulates one time step of continuous behavior from t to t+dt. Underlying function is solve_ivp with method is 'RK23'. :param x0: Initial state vector. :param t: Time at start of the step simulation. :param u: Arguments passed..... :return: Time t+dt, value of state at t+dt

Source code in ml4cps/cps/base.py

def __step_continuous(self, clock_start, clock_finish):
    """
Simulates one time step of continuous behavior from t to t+dt. Underlying function is solve_ivp with method is 'RK23'.
    :param x0: Initial state vector.
    :param t: Time at start of the step simulation.
    :param u: Arguments passed.....
    :return: Time t+dt, value of state at t+dt
    """

    s = solve_ivp(self.time_continuous_dynamics, t_eval=np.arange(clock_start + self.dt, clock_finish + self.dt, self.dt),
                  t_span=(clock_start, clock_finish + self.dt), y0=self._xt, method='RK23',
                  args=(self._xk, self._q, self._u, self._p, self._y))

    self._xk = self.time_discrete_dynamics(self._q, self._p, self._xt, self._u)
    self._xt = s.y[-1]

apply_sim_event(e, env=None)

The event e is applied in this component's simpy execution, this means that the process must wait for an event. :param e: Event to apply. :return:

Source code in ml4cps/cps/base.py

def apply_sim_event(self, e, env=None):
    """
    The event e is applied in this component's simpy execution, this means that the process must wait for an event.
    :param e: Event to apply.
    :return:
    """
    if not self._block_event:
        # self._block_event = self._env.now
        self._block_event = env.now
    # if not self._block_event:
    #     raise Exception("Block event is not set, but event is applied.")
    elif self._block_event.triggered:
        # pass
        # if self._e != e:
        raise Exception('Should not happen')
    else:
        self._block_event.succeed(value=e)

simulate(finish_time, verbose=False, save_choices=False)

Simulates behaviour of the system until the finish_time is reached.

Parameters

finish_time : Time when simulation finishes. verbose : Should the execution log be printed in detail (default is False). save_choices : Should the choices for each component be saved to json files after the simulation is finished (default is False).

Source code in ml4cps/cps/base.py

def simulate(self, finish_time, verbose=False, save_choices=False):
    """Simulates behaviour of the system until the finish_time is reached.

    Parameters
    ----------
    finish_time : Time when simulation finishes.
    verbose : Should the execution log be printed in detail (default is False).
    save_choices : Should the choices for each component be saved to json files after
    the simulation is finished (default is False).
    """

    env = simpy.Environment()
    finish_time = float(finish_time)

    if verbose:
        print('Simulation started: ')

    self.simulation_process_simpy(env, finish_time, verbose)
    env.run(until=finish_time)

    discr_state_data = {}
    discr_output_data = {}
    cont_state_data = {}
    cont_output_data = {}

    try:
        if len(self._xt) > 0 and len(self._xt) > 0 and finish_time > self._t:  # there is time-continuous state variable
            self._step_continuous(self._t, finish_time)
    except Exception as ex:
        print_exc()
        raise Exception('Exception during self._step_continuous in state {}'.format(self._q))

    try:
        self.finish(env.now)

        discr_state_data = pd.DataFrame(self._discrete_state_data).set_index("Time")
        discr_output_data = tools.data_list_to_dataframe(None, self._discrete_output_data,
                                                               self.discrete_output_names, 'd')
        try:
            cont_state_data = tools.data_list_to_dataframe(self.id, self._continuous_state_data,
                                                           self.cont_state_names, 'x')
        except Exception as ex:
            print_exc()
        cont_output_data = tools.data_list_to_dataframe(self.id, self._continuous_output_data,
                                                                 self.cont_output_names, 'y')
    except Exception as ex:
        print_exc()
        warnings.warn('Simulation failed.')

    if verbose:
        print('Simulation finished.')
    return discr_state_data, discr_output_data, cont_state_data, cont_output_data, env.now

simulation_process_simpy(env, max_time, verbose=False)

The function simulates single concurrent thread of one component (CPSComponent). Parameters

---

env : simpy.Environment It is simpy environment used to synchronize multiple parallel threads during simulation. max_time : float It is the maximum time that the simulation will perform. verbose : bool Should the function print execution logs (default is False).

Source code in ml4cps/cps/base.py

def simulation_process_simpy(self, env, max_time, verbose=False):
    """
        The function simulates single concurrent thread of one component (CPSComponent).
        Parameters
        ----------
        env : simpy.Environment
            It is simpy environment used to synchronize multiple parallel threads during simulation.
        max_time : float
            It is the maximum time that the simulation will perform.
        verbose : bool
            Should the function print execution logs (default is False).
    """
    env.process(self.discrete_dynamics_process(env, max_time, verbose=verbose))
    env.process(self.continuous_dynamics_process(env, max_time, verbose=verbose))
    env.process(self.discrete_time_dynamics_process(env, max_time, verbose=verbose))

timed_event(q, xc, xd)

Calculates if and when the next time event will happen and the new state values. :param t: Current time. :param q: Discrete-event part of the state. :param xc: Time-continuous part of the state. :param xd: Time-discrete part of the state. :return: Time delay of the timed event, new state value.

Source code in ml4cps/cps/base.py

def timed_event(self, q, xc, xd):
    """
    Calculates if and when the next time event will happen and the new state values.
    :param t: Current time.
    :param q: Discrete-event part of the state.
    :param xc: Time-continuous part of the state.
    :param xd: Time-discrete part of the state.
    :return: Time delay of the timed event, new state value.
    """
    return None, None

Simulation.

Author: - Nemanja Hranisavljevic, hranisan@hsu-hh.de, nemanja@ai4cps.com

Simulator

Source code in ml4cps/cps/sim.py

class Simulator():
    def simulate(self, finish_time):
        pass

    def reinitialize(self, time, state):
        pass

    def stop_condition(self, t):
        """
        Simulation stop condition which can be overridden by the subclasses.
        """
        return False

stop_condition(t)

Simulation stop condition which can be overridden by the subclasses.

Source code in ml4cps/cps/sim.py

def stop_condition(self, t):
    """
    Simulation stop condition which can be overridden by the subclasses.
    """
    return False

discretization

The module provides base classes to represent the dynamics of cyber-physical systems (CPS): CPS and CPSComponent.

Author: Nemanja Hranisavljevic, hranisan@hsu-hh.de

EqualFrequencyDiscretizer

Bases: TimeSeriesDiscretizer

A class that implements the equal-frequency interval (EFI) discretization method. It divides each variable (column) into intervals such that each interval contains approximately the same number of data points.

Source code in ml4cps/discretization/discretization.py

class EqualFrequencyDiscretizer(TimeSeriesDiscretizer):
    """
    A class that implements the equal-frequency interval (EFI) discretization method. It divides each variable
    (column) into intervals such that each interval contains approximately the same number of data points.
    """
    def __init__(self):
        self.intervals = None

    def train(self, data, number_of_intervals=10):
        """
        Estimate model parameters, thresholds that divide each variable into equal-frequency intervals.
        :param data: Data to calculate model parameters from.
        :param number_of_intervals: Number of equal-frequency intervals per variable.
        :return:
        """
        data = [pd.DataFrame(d) for d in data]
        combined_data = pd.concat(data, axis=0)  # Combine all datasets for quantile calculation
        self.intervals = {}
        for col in combined_data.columns:
            intvl = pd.qcut(combined_data[col], q=number_of_intervals, duplicates='drop', retbins=True)[1]
            if len(np.unique(intvl)) != len(intvl):
                print(1)
            if len(intvl) < number_of_intervals + 1:
                expanded_arr = np.empty(number_of_intervals+1, dtype=combined_data[col].dtype)
                expanded_arr[:] = np.nan
                expanded_arr[:len(intvl)] = intvl
                intvl = expanded_arr
            self.intervals[col] = intvl

        self.intervals = pd.DataFrame(self.intervals)

    def discretize(self, df, return_str=False, append_discr=None):
        """
        Discretize data into equal-frequency intervals.
        :param df: Data to discretize.
        :param return_str: Whether to return discretized data as a concatenated string per row.
        :return: Discretized data.
        """
        df = pd.DataFrame(df)
        binned_df = df.copy()

        for col in self.intervals.columns:
            bins = self.intervals[col]
            if bins is None:
                continue

            col_min, col_max = bins.iloc[0], bins.iloc[-1]

            # Check for out-of-range values and adjust
            if (df[col] < col_min).any() or (df[col] > col_max).any():
                print(f"Warning: Column '{col}' contains values outside the range [{col_min}, {col_max}].")
                binned_df[col] = np.clip(df[col], col_min, col_max)
            else:
                binned_df[col] = df[col]

            # Apply binning with predefined bin edges
            binned_df[col] = pd.cut(
                binned_df[col],
                bins=bins.dropna(),
                labels=False,
                include_lowest=True
            )

        if append_discr is not None:
            binned_df = np.hstack((binned_df, append_discr))

        binned_df = pd.DataFrame(binned_df)
        if return_str:
            binned_df = binned_df.round(0).astype("Int64").astype(str)
            binned_df = binned_df.agg(','.join, axis=1).to_numpy()

        return binned_df

discretize(df, return_str=False, append_discr=None)

Discretize data into equal-frequency intervals. :param df: Data to discretize. :param return_str: Whether to return discretized data as a concatenated string per row. :return: Discretized data.

Source code in ml4cps/discretization/discretization.py

def discretize(self, df, return_str=False, append_discr=None):
    """
    Discretize data into equal-frequency intervals.
    :param df: Data to discretize.
    :param return_str: Whether to return discretized data as a concatenated string per row.
    :return: Discretized data.
    """
    df = pd.DataFrame(df)
    binned_df = df.copy()

    for col in self.intervals.columns:
        bins = self.intervals[col]
        if bins is None:
            continue

        col_min, col_max = bins.iloc[0], bins.iloc[-1]

        # Check for out-of-range values and adjust
        if (df[col] < col_min).any() or (df[col] > col_max).any():
            print(f"Warning: Column '{col}' contains values outside the range [{col_min}, {col_max}].")
            binned_df[col] = np.clip(df[col], col_min, col_max)
        else:
            binned_df[col] = df[col]

        # Apply binning with predefined bin edges
        binned_df[col] = pd.cut(
            binned_df[col],
            bins=bins.dropna(),
            labels=False,
            include_lowest=True
        )

    if append_discr is not None:
        binned_df = np.hstack((binned_df, append_discr))

    binned_df = pd.DataFrame(binned_df)
    if return_str:
        binned_df = binned_df.round(0).astype("Int64").astype(str)
        binned_df = binned_df.agg(','.join, axis=1).to_numpy()

    return binned_df

train(data, number_of_intervals=10)

Estimate model parameters, thresholds that divide each variable into equal-frequency intervals. :param data: Data to calculate model parameters from. :param number_of_intervals: Number of equal-frequency intervals per variable. :return:

Source code in ml4cps/discretization/discretization.py

def train(self, data, number_of_intervals=10):
    """
    Estimate model parameters, thresholds that divide each variable into equal-frequency intervals.
    :param data: Data to calculate model parameters from.
    :param number_of_intervals: Number of equal-frequency intervals per variable.
    :return:
    """
    data = [pd.DataFrame(d) for d in data]
    combined_data = pd.concat(data, axis=0)  # Combine all datasets for quantile calculation
    self.intervals = {}
    for col in combined_data.columns:
        intvl = pd.qcut(combined_data[col], q=number_of_intervals, duplicates='drop', retbins=True)[1]
        if len(np.unique(intvl)) != len(intvl):
            print(1)
        if len(intvl) < number_of_intervals + 1:
            expanded_arr = np.empty(number_of_intervals+1, dtype=combined_data[col].dtype)
            expanded_arr[:] = np.nan
            expanded_arr[:len(intvl)] = intvl
            intvl = expanded_arr
        self.intervals[col] = intvl

    self.intervals = pd.DataFrame(self.intervals)

EqualWidthDiscretizer

Bases: TimeSeriesDiscretizer

A class that implements equal-width interval (EWI) discretization method. It calculates range of every variable (column) of the input data into a predefined number of equal-width intervals.

Source code in ml4cps/discretization/discretization.py

class EqualWidthDiscretizer (TimeSeriesDiscretizer):
    """
    A class that implements equal-width interval (EWI) discretization method. It calculates range of every variable
    (column) of the input data into a predefined number of equal-width intervals.
    """
    def __init__(self):
        self.intervals = None
        self.min_d = None
        self.max_d = None

    def train(self, data, number_of_intervals=10):
        """
        Estimate model parameters, thresholds that divide each variable into equal-width intervals.
        :param data: Data to calculate model parameters from.
        :param number_of_intervals: Number of equal-width intervals per variable.
        :return:
        """
        data = [pd.DataFrame(d) for d in data]
        min_d = min(d.min(axis=0) for d in data)
        max_d = max(d.max(axis=0) for d in data)
        self.min_d = min_d
        self.max_d = max_d
        self.intervals = pd.DataFrame(np.linspace(self.min_d, self.max_d, number_of_intervals), columns=data[0].columns)

    def discretize(self, df, return_str=False, append_discr=None):
        """
        Discretize data into equal width intervals.
        :param data: Data to discretize.
        :return: Discretized data.
        """

        df = pd.DataFrame(df)
        binned_df = df.copy()
        for col in self.intervals.columns:
            col_min=self.min_d[col]
            col_max=self.max_d[col]
            bins=self.intervals[col]

            # Check for out-of-range values and adjust
            if (df[col] < col_min).any() or (df[col] > col_max).any():
                print(f"Warning: Column '{col}' contains values outside the range [{col_min}, {col_max}].")
                binned_df[col] = np.clip(df[col], col_min, col_max)
            else:
                binned_df[col] = df[col]

            # Apply binning with predefined bin edges
            try:
                binned_df[col] = pd.cut(
                    binned_df[col],
                    bins=bins,
                    labels=False,
                    include_lowest=True)
            except:
                print(1)

        if append_discr is not None:
            binned_df = np.hstack((binned_df, append_discr))

        binned_df = pd.DataFrame(binned_df)
        if return_str:
            binned_df = binned_df.round(0).astype("Int64").astype(str)
            binned_df = binned_df.agg(','.join, axis=1).to_numpy()
        return binned_df

discretize(df, return_str=False, append_discr=None)

Discretize data into equal width intervals. :param data: Data to discretize. :return: Discretized data.

Source code in ml4cps/discretization/discretization.py

def discretize(self, df, return_str=False, append_discr=None):
    """
    Discretize data into equal width intervals.
    :param data: Data to discretize.
    :return: Discretized data.
    """

    df = pd.DataFrame(df)
    binned_df = df.copy()
    for col in self.intervals.columns:
        col_min=self.min_d[col]
        col_max=self.max_d[col]
        bins=self.intervals[col]

        # Check for out-of-range values and adjust
        if (df[col] < col_min).any() or (df[col] > col_max).any():
            print(f"Warning: Column '{col}' contains values outside the range [{col_min}, {col_max}].")
            binned_df[col] = np.clip(df[col], col_min, col_max)
        else:
            binned_df[col] = df[col]

        # Apply binning with predefined bin edges
        try:
            binned_df[col] = pd.cut(
                binned_df[col],
                bins=bins,
                labels=False,
                include_lowest=True)
        except:
            print(1)

    if append_discr is not None:
        binned_df = np.hstack((binned_df, append_discr))

    binned_df = pd.DataFrame(binned_df)
    if return_str:
        binned_df = binned_df.round(0).astype("Int64").astype(str)
        binned_df = binned_df.agg(','.join, axis=1).to_numpy()
    return binned_df

train(data, number_of_intervals=10)

Estimate model parameters, thresholds that divide each variable into equal-width intervals. :param data: Data to calculate model parameters from. :param number_of_intervals: Number of equal-width intervals per variable. :return:

Source code in ml4cps/discretization/discretization.py

def train(self, data, number_of_intervals=10):
    """
    Estimate model parameters, thresholds that divide each variable into equal-width intervals.
    :param data: Data to calculate model parameters from.
    :param number_of_intervals: Number of equal-width intervals per variable.
    :return:
    """
    data = [pd.DataFrame(d) for d in data]
    min_d = min(d.min(axis=0) for d in data)
    max_d = max(d.max(axis=0) for d in data)
    self.min_d = min_d
    self.max_d = max_d
    self.intervals = pd.DataFrame(np.linspace(self.min_d, self.max_d, number_of_intervals), columns=data[0].columns)

KMeansDiscretizer

Bases: TimeSeriesDiscretizer

A class that implements K-means discretization. It clusters the data values of each variable into a predefined number of clusters using the K-means algorithm.

Source code in ml4cps/discretization/discretization.py

class KMeansDiscretizer(TimeSeriesDiscretizer):
    """
    A class that implements K-means discretization. It clusters the data values of each variable
    into a predefined number of clusters using the K-means algorithm.
    """
    def __init__(self):
        self.cluster_centers = None
        self.kmeans_models = None

    def train(self, data, number_of_clusters_per_var=10):
        """
        Train the K-means discretizer by fitting K-means models for each variable (column) in the data.
        :param data: List of DataFrames to calculate model parameters from.
        :param number_of_clusters_per_var: Number of clusters (intervals) per variable.
        """
        data = [pd.DataFrame(d) for d in data]
        combined_data = pd.concat(data, ignore_index=True)  # Combine all datasets for training
        self.kmeans_models = {}
        self.cluster_centers = {}

        # Train a K-means model for each column
        for col in combined_data.columns:
            kmeans = KMeans(n_clusters=number_of_clusters_per_var, random_state=0)
            kmeans.fit(combined_data[[col]])
            self.kmeans_models[col] = kmeans
            self.cluster_centers[col] = np.sort(kmeans.cluster_centers_.flatten())  # Sort centers for logical ordering

    def discretize(self, df, return_str=False, append_discr=None):
        """
        Discretize data into clusters determined by K-means.
        :param df: DataFrame to discretize.
        :param return_str: Whether to return discretized data as concatenated strings.
        :return: Discretized data.
        """
        df = pd.DataFrame(df)
        discretized_df = df.copy()

        for col in df.columns:
            kmeans = self.kmeans_models.get(col)
            if not kmeans:
                raise ValueError(f"No K-means model trained for column '{col}'.")
            # Assign each value to the closest cluster center
            discretized_df[col] = kmeans.predict(df[[col]])

        if append_discr is not None:
            discretized_df = np.hstack((discretized_df, append_discr))

        discretized_df = pd.DataFrame(discretized_df)
        if return_str:
            discretized_df = discretized_df.round(0).astype("Int64").astype(str)
            discretized_df = discretized_df.agg(','.join, axis=1).to_numpy()
        return discretized_df

discretize(df, return_str=False, append_discr=None)

Discretize data into clusters determined by K-means. :param df: DataFrame to discretize. :param return_str: Whether to return discretized data as concatenated strings. :return: Discretized data.

Source code in ml4cps/discretization/discretization.py

def discretize(self, df, return_str=False, append_discr=None):
    """
    Discretize data into clusters determined by K-means.
    :param df: DataFrame to discretize.
    :param return_str: Whether to return discretized data as concatenated strings.
    :return: Discretized data.
    """
    df = pd.DataFrame(df)
    discretized_df = df.copy()

    for col in df.columns:
        kmeans = self.kmeans_models.get(col)
        if not kmeans:
            raise ValueError(f"No K-means model trained for column '{col}'.")
        # Assign each value to the closest cluster center
        discretized_df[col] = kmeans.predict(df[[col]])

    if append_discr is not None:
        discretized_df = np.hstack((discretized_df, append_discr))

    discretized_df = pd.DataFrame(discretized_df)
    if return_str:
        discretized_df = discretized_df.round(0).astype("Int64").astype(str)
        discretized_df = discretized_df.agg(','.join, axis=1).to_numpy()
    return discretized_df

train(data, number_of_clusters_per_var=10)

Train the K-means discretizer by fitting K-means models for each variable (column) in the data. :param data: List of DataFrames to calculate model parameters from. :param number_of_clusters_per_var: Number of clusters (intervals) per variable.

Source code in ml4cps/discretization/discretization.py

def train(self, data, number_of_clusters_per_var=10):
    """
    Train the K-means discretizer by fitting K-means models for each variable (column) in the data.
    :param data: List of DataFrames to calculate model parameters from.
    :param number_of_clusters_per_var: Number of clusters (intervals) per variable.
    """
    data = [pd.DataFrame(d) for d in data]
    combined_data = pd.concat(data, ignore_index=True)  # Combine all datasets for training
    self.kmeans_models = {}
    self.cluster_centers = {}

    # Train a K-means model for each column
    for col in combined_data.columns:
        kmeans = KMeans(n_clusters=number_of_clusters_per_var, random_state=0)
        kmeans.fit(combined_data[[col]])
        self.kmeans_models[col] = kmeans
        self.cluster_centers[col] = np.sort(kmeans.cluster_centers_.flatten())  # Sort centers for logical ordering

MultivariateKMeansDiscretizer

Bases: TimeSeriesDiscretizer

A class that implements multivariate K-means discretization. It clusters the data based on all variables (columns) together into a predefined number of clusters.

Source code in ml4cps/discretization/discretization.py

class MultivariateKMeansDiscretizer(TimeSeriesDiscretizer):
    """
    A class that implements multivariate K-means discretization. It clusters the data
    based on all variables (columns) together into a predefined number of clusters.
    """

    def __init__(self):
        self.kmeans_model = None
        self.cluster_centers = None

    def train(self, data, number_of_clusters=10):
        """
        Train the K-means discretizer by fitting a single K-means model for all variables.
        :param data: List of DataFrames to calculate model parameters from.
        :param number_of_clusters: Number of clusters.
        """
        data = [pd.DataFrame(d) for d in data]
        combined_data = pd.concat(data, ignore_index=True)  # Combine all datasets for training
        self.kmeans_model = KMeans(n_clusters=number_of_clusters, random_state=0)
        self.kmeans_model.fit(combined_data)
        self.cluster_centers = self.kmeans_model.cluster_centers_

    def discretize(self, df, return_str=False, append_discr=None):
        """
        Discretize data into clusters determined by K-means.
        :param df: DataFrame to discretize.
        :param return_str: Whether to return discretized data as concatenated strings.
        :return: Discretized data (cluster labels).
        """
        df = pd.DataFrame(df)
        if self.kmeans_model is None:
            raise ValueError("The K-means model has not been trained yet.")

        # Predict cluster labels for the entire dataset
        cluster_labels = self.kmeans_model.predict(df)

        if append_discr is not None:
            if cluster_labels.ndim == 1:
                cluster_labels = cluster_labels.reshape(-1, 1)
            cluster_labels = np.hstack((cluster_labels, append_discr))

        cluster_labels = pd.DataFrame(cluster_labels)
        if return_str:
            cluster_labels = cluster_labels.round(0).astype("Int64").astype(str)
            cluster_labels = cluster_labels.agg(','.join, axis=1).to_numpy()
        return cluster_labels

discretize(df, return_str=False, append_discr=None)

Discretize data into clusters determined by K-means. :param df: DataFrame to discretize. :param return_str: Whether to return discretized data as concatenated strings. :return: Discretized data (cluster labels).

Source code in ml4cps/discretization/discretization.py

def discretize(self, df, return_str=False, append_discr=None):
    """
    Discretize data into clusters determined by K-means.
    :param df: DataFrame to discretize.
    :param return_str: Whether to return discretized data as concatenated strings.
    :return: Discretized data (cluster labels).
    """
    df = pd.DataFrame(df)
    if self.kmeans_model is None:
        raise ValueError("The K-means model has not been trained yet.")

    # Predict cluster labels for the entire dataset
    cluster_labels = self.kmeans_model.predict(df)

    if append_discr is not None:
        if cluster_labels.ndim == 1:
            cluster_labels = cluster_labels.reshape(-1, 1)
        cluster_labels = np.hstack((cluster_labels, append_discr))

    cluster_labels = pd.DataFrame(cluster_labels)
    if return_str:
        cluster_labels = cluster_labels.round(0).astype("Int64").astype(str)
        cluster_labels = cluster_labels.agg(','.join, axis=1).to_numpy()
    return cluster_labels

train(data, number_of_clusters=10)

Train the K-means discretizer by fitting a single K-means model for all variables. :param data: List of DataFrames to calculate model parameters from. :param number_of_clusters: Number of clusters.

Source code in ml4cps/discretization/discretization.py

def train(self, data, number_of_clusters=10):
    """
    Train the K-means discretizer by fitting a single K-means model for all variables.
    :param data: List of DataFrames to calculate model parameters from.
    :param number_of_clusters: Number of clusters.
    """
    data = [pd.DataFrame(d) for d in data]
    combined_data = pd.concat(data, ignore_index=True)  # Combine all datasets for training
    self.kmeans_model = KMeans(n_clusters=number_of_clusters, random_state=0)
    self.kmeans_model.fit(combined_data)
    self.cluster_centers = self.kmeans_model.cluster_centers_

TimeSeriesDiscretizer

Abstract class that encapsulates methods used for the discretization of time series.

Source code in ml4cps/discretization/discretization.py

class TimeSeriesDiscretizer:
    """
        Abstract class that encapsulates methods used for the discretization of time series.
    """
    def train(self, data, *args):
        pass

    def discretize(self, data):
        pass

vis

The module provides methods to visualize various kinds of data, such as time series or automata graphs.

Authors: - Nemanja Hranisavljevic, hranisan@hsu-hh.de, nemanja@ai4cps.com - Tom Westermann, tom.westermann@hsu-hh.de, tom@ai4cps.com

plot2d(df, x=None, y=None, mode='markers', hovercolumns=None, figure=False, **args)

Creates a 2D scatter or line plot using Plotly based on the provided DataFrame columns. Parameters: df (pd.DataFrame): The input DataFrame containing the data to plot. x (str, optional): The column name to use for the x-axis. y (str, optional): The column name to use for the y-axis. mode (str, optional): The Plotly scatter mode (e.g., 'markers', 'lines'). Defaults to 'markers'. hovercolumns (list of str, optional): List of column names to include in the hover tooltip. figure (bool, optional): If True, returns a Plotly Figure object; otherwise, returns a Scatter trace. Defaults to False. **args: Additional keyword arguments passed to the Plotly Scatter constructor. Returns: plotly.graph_objs._scatter.Scatter or plotly.graph_objs._figure.Figure: The generated Plotly Scatter trace or Figure, depending on the 'figure' parameter. Example: plot2d(df, x='feature1', y='feature2', hovercolumns=['label'], mode='markers', figure=True)

Source code in ml4cps/vis.py

def plot2d(df, x=None, y=None, mode='markers', hovercolumns=None, figure=False, **args):
    """
    Creates a 2D scatter or line plot using Plotly based on the provided DataFrame columns.
    Parameters:
        df (pd.DataFrame): The input DataFrame containing the data to plot.
        x (str, optional): The column name to use for the x-axis.
        y (str, optional): The column name to use for the y-axis.
        mode (str, optional): The Plotly scatter mode (e.g., 'markers', 'lines'). Defaults to 'markers'.
        hovercolumns (list of str, optional): List of column names to include in the hover tooltip.
        figure (bool, optional): If True, returns a Plotly Figure object; otherwise, returns a Scatter trace. Defaults to False.
        **args: Additional keyword arguments passed to the Plotly Scatter constructor.
    Returns:
        plotly.graph_objs._scatter.Scatter or plotly.graph_objs._figure.Figure:
            The generated Plotly Scatter trace or Figure, depending on the 'figure' parameter.
    Example:
        plot2d(df, x='feature1', y='feature2', hovercolumns=['label'], mode='markers', figure=True)
    """

    hovertemplate = f"{x}: %{{x}}<br>{y}: %{{y}}"
    customdata = None
    if hovercolumns:
        customdata = df[hovercolumns]
        for ind, c in enumerate(hovercolumns):
            hovertemplate += f"<br>{c}: %{{customdata[{ind}]}}"

    trace = go.Scatter(x=df[x], y=df[y], mode=mode, customdata=customdata, hovertemplate=hovertemplate, **args)
    if figure:
        return go.Figure(data=trace)
    return trace

plot3d(df, x=None, y=None, z=None, mode='markers', hovercolumns=None, **args)

Creates a 3D scatter plot using Plotly's Scatter3d, with customizable axes, hover information, and additional plot arguments. Parameters: df (pandas.DataFrame): The data source containing columns for x, y, z, and optional hover data. x (str, optional): The column name in df to use for the x-axis. y (str, optional): The column name in df to use for the y-axis. z (str, optional): The column name in df to use for the z-axis. mode (str, optional): Plotly scatter mode (e.g., 'markers', 'lines'). Defaults to 'markers'. hovercolumns (list of str, optional): List of column names in df to include in the hover tooltip. **args: Additional keyword arguments passed to go.Scatter3d. Returns: plotly.graph_objs._scatter3d.Scatter3d: A Plotly 3D scatter plot object configured with the specified data and options.

Source code in ml4cps/vis.py

def plot3d(df, x=None, y=None, z=None, mode='markers', hovercolumns=None, **args):
    """
    Creates a 3D scatter plot using Plotly's Scatter3d, with customizable axes, hover information, and additional plot arguments.
    Parameters:
        df (pandas.DataFrame): The data source containing columns for x, y, z, and optional hover data.
        x (str, optional): The column name in `df` to use for the x-axis.
        y (str, optional): The column name in `df` to use for the y-axis.
        z (str, optional): The column name in `df` to use for the z-axis.
        mode (str, optional): Plotly scatter mode (e.g., 'markers', 'lines'). Defaults to 'markers'.
        hovercolumns (list of str, optional): List of column names in `df` to include in the hover tooltip.
        **args: Additional keyword arguments passed to `go.Scatter3d`.
    Returns:
        plotly.graph_objs._scatter3d.Scatter3d: A Plotly 3D scatter plot object configured with the specified data and options.
    """

    hovertemplate = f"{x}: %{{x}}<br>{y}: %{{y}}<br>{z}: %{{z}}"
    customdata = None
    if hovercolumns:
        customdata = df[hovercolumns]
        for ind, c in enumerate(hovercolumns):
            hovertemplate += f"<br>{c}: %{{customdata[{ind}]}}"
    return go.Scatter3d(x=df[x], y=df[y], z=df[z], mode=mode, customdata=customdata, hovertemplate=hovertemplate, **args)

plot_2d_contour_from_fun(fun, rangex=None, rangey=None, th=50, **kwargs)

Plots a 2D contour of a function over a specified range. Parameters: fun (callable): A function that takes a 2D array of shape (n_points, 2) and returns a 1D array of function values. rangex (tuple, optional): The range for the x-axis as (min, max). Defaults to (-5, 5) if not provided. rangey (tuple, optional): The range for the y-axis as (min, max). Defaults to (-5, 5) if not provided. th (int, optional): Unused parameter, kept for compatibility. Defaults to 50. **kwargs: Additional keyword arguments passed to the plotly.graph_objs.Contour constructor. Returns: plotly.graph_objs.Contour: A Plotly contour plot object representing the function values over the specified range.

Source code in ml4cps/vis.py

def plot_2d_contour_from_fun(fun, rangex=None, rangey=None, th=50, **kwargs):
    """
    Plots a 2D contour of a function over a specified range.
    Parameters:
        fun (callable): A function that takes a 2D array of shape (n_points, 2) and returns a 1D array of function values.
        rangex (tuple, optional): The range for the x-axis as (min, max). Defaults to (-5, 5) if not provided.
        rangey (tuple, optional): The range for the y-axis as (min, max). Defaults to (-5, 5) if not provided.
        th (int, optional): Unused parameter, kept for compatibility. Defaults to 50.
        **kwargs: Additional keyword arguments passed to the plotly.graph_objs.Contour constructor.
    Returns:
        plotly.graph_objs.Contour: A Plotly contour plot object representing the function values over the specified range.
    """

    if rangex is None:
        rangex = (-5, 5)

    if rangey is None:
        rangey = (-5, 5)

    x = np.linspace(rangex[0], rangex[-1], 100)
    y = np.linspace(rangey[0], rangey[-1], 100)
    [dx, dy] = np.meshgrid(x, y)
    d = np.column_stack([dx.flatten(), dy.flatten()])
    f = fun(d)

    contours = list(f)
    contours.sort()
    contours = contours[0:1000:]
    return go.Contour(x=x, y=y, z=np.reshape(f, dx.shape), contours=dict(coloring='lines'), **kwargs)

plot_cps(cps, dash_id=None, node_labels=False, edge_labels=True, node_size=40, node_font_size=20, edge_font_size=16, edge_text_max_width=None, output='cyto', dash_port=8050, height='100%', minZoom=0.5, maxZoom=2, **kwargs)

Plots all the components of a CPS in the same figure. :param cps: CPS to plot. :param node_labels: Should node labels be plotted. :param edge_labels: Should edge labels be plotted. :param node_size: What is the size of the nodes in the figure. :param node_font_size: The font size of the node labels. :param edge_font_size: The font size of the edge labels. :param edge_text_max_width: Max width of the edge labels. :param output: Should output be plotted as a dash.Cytoscape component ("cyto"), or should dash server be run ("dash"). :param dash_port: If temporary dash server is run, what port to use. :param kwargs: Other paramters are forwarded to the Cytoscape component. :return:

Source code in ml4cps/vis.py

def plot_cps(cps: CPS, dash_id=None, node_labels=False, edge_labels=True, node_size=40, node_font_size=20,
             edge_font_size=16, edge_text_max_width=None, output="cyto", dash_port=8050, height='100%',
             minZoom=0.5, maxZoom=2, **kwargs):
    """
    Plots all the components of a CPS in the same figure.
    :param cps: CPS to plot.
    :param node_labels: Should node labels be plotted.
    :param edge_labels: Should edge labels be plotted.
    :param node_size: What is the size of the nodes in the figure.
    :param node_font_size: The font size of the node labels.
    :param edge_font_size: The font size of the edge labels.
    :param edge_text_max_width: Max width of the edge labels.
    :param output: Should output be plotted as a dash.Cytoscape component ("cyto"), or should dash server be run
    ("dash").
    :param dash_port: If temporary dash server is run, what port to use.
    :param kwargs: Other paramters are forwarded to the Cytoscape component.
    :return:
    """
    elements = dict(nodes=[], edges=[])

    for comid, com in cps.items():
        els = dict(edges=[], nodes=com)
        elements['nodes'].append({'data': {'id': comid, 'label': comid}, 'classes': 'parent'})
        for x in els['nodes']:
            if type(x) is str or type(x) is list:
                x = {'data': {'id': x}}
            x['data']['group'] = comid
            x['data']['parent'] = comid
            x['data']['label'] = f"{x['data']['id']}"
            x['data']['id'] = f"{comid}-{x['data']['id']}"
            elements['nodes'].append(x)
        for x in els['edges']:
            x['data']['source'] = f"{comid}-{x['data']['source']}"
            x['data']['target'] = f"{comid}-{x['data']['target']}"
            elements['edges'].append(x)

    node_style = {'width': node_size,
                  'height': node_size}
    if node_labels:
        node_style['label'] = 'data(label)'
        node_style['font-size'] = node_font_size
        node_style['text-wrap'] = 'wrap'
        node_style['text-max-width'] = 50

    edge_style = {
        # The default curve style does not work with certain arrows
        'curve-style': 'bezier',
        'target-arrow-shape': 'triangle',
        'target-arrow-size': 3,
        'width': 1,
        'font-color': 'black',
        'text-wrap': 'wrap',
        'font-size': edge_font_size,
        'text-max-width': edge_text_max_width
    }
    if edge_labels:
        edge_style['label'] = 'data(label)'

    stylesheet = [
        {
            'selector': 'node',
            'style': node_style
        },
        {
            'selector': 'edge',
            'style': edge_style
        }]

    network = cyto.Cytoscape(
        id=dash_id if dash_id is not None else cps.id,
        layout={
            'name': 'grid',
            'padding': 10,  # Padding around the graph layout
            'nodeOverlap': 20,  # Adjust to reduce overlap
            'nodeRepulsion': 50,  # Increase repulsion for better separation
            'idealEdgeLength': 10,  # Increase edge length to spread nodes
            'componentSpacing': 50,  # Spacing between disconnected components
            'nodeDimensionsIncludeLabels': True,  # Include label sizes in layout
            'nestingFactor': 0.7,  # Factor to apply to compounds when calculating layout
            "spacingFactor": 1.5
        },
        # layout={
        #     "name": "breadthfirst",
        #     "directed": True,
        #     "spacingFactor": 1.5
        # },
        maxZoom=maxZoom,
        minZoom=minZoom,
        stylesheet=stylesheet,
        elements=elements, style={'width': '100%', 'height': height},
        **kwargs)

    modal_state_data = dbc.Modal(children=[dbc.ModalHeader("Timings"),
                                           dbc.ModalBody(html.Div(children=[]))],
                                 id=f"{id}-modal-state-data")
    modal_transition_data = dbc.Modal(children=[dbc.ModalHeader("Timings"),
                                                dbc.ModalBody(html.Div(children=[]))],
                                      id=f"{id}-modal-transition-data")
    # network = html.Div([network, modal_state_data, modal_transition_data])
    if output == "notebook":
        app = Dash(__name__)
        app.layout = html.Div(children=[network])
        app.run(mode='inline')
        return
    if output == "dash":
        app = Dash(__name__)
        app.layout = html.Div(children=[network], style={'width': '100%',
                                                         'height': '100%',
                                                         'margin': '0',
                                                         'padding': '0'})

        # Function to start the Dash server
        def run_dash():
            app.run(port=dash_port, debug=False, use_reloader=False)  # Start the Dash server

        # Function to open the browser
        def open_browser():
            time.sleep(1)  # Give the server a second to start
            webbrowser.open(f"http://127.0.0.1:{dash_port}/")  # Open the Dash app in the browser

        # Start the Dash server in a separate thread
        server_thread = threading.Thread(target=run_dash)
        server_thread.daemon = True  # Allows the program to exit even if this thread is running
        server_thread.start()

        # Open the Dash app in the default browser
        open_browser()
        server_thread.join(timeout=1)

    return network

plot_cps_component(cps, id=None, node_labels=False, center_node_labels=False, event_label=True, show_transition_freq=False, show_transition_timing=False, font_size=6, edge_font_size=6, edge_text_max_width=None, init_label=False, limit_interval_precision=None, show_transition_data=False, transition_data_keys=True, node_size=20, output='cyto', dash_port=8050, min_zoom=0.5, split_edges_diff_event=False, max_zoom=1, min_edge_thickness=0.1, max_edge_thickness=4, freq_as_edge_thickness=False, color='black', title_text=None, layout_name='breadthfirst', layout_spacingFactor=1, hide_nodes=None)

Visualizes a component of a Cyber-Physical System (CPS) as a graph using Dash Cytoscape.
This function generates a graphical representation of the discrete states and transitions of a CPS,
with various customization options for node and edge appearance, labels, and output format.
The visualization can be rendered as Dash Cytoscape elements, in a Dash app, or as a notebook widget.
Parameters:
    cps: object
        The CPS object containing discrete states, transitions, and related data.
    id: str, optional
        The unique identifier for the Cytoscape component (default: "graph").
    node_labels: bool, optional
        Whether to display labels on nodes (default: False).
    center_node_labels: bool, optional
        Whether to center node labels (default: False).
    edge_labels: bool, optional
        Whether to display labels on edges (default: True).
    show_transition_freq: bool, optional
        Whether to show transition frequency on edge labels (default: False).
    edge_font_size: int, optional
        Font size for edge labels (default: 6).
    edge_text_max_width: int or None, optional
        Maximum width for edge label text wrapping (default: None).
    init_label: bool, optional
        Whether to label initial state transitions as 'init' (default: False).
    show_transition_data: bool or list, optional
        Whether to display additional transition data on edge labels. If a list, only specified keys are shown (default: False).
    node_size: int, optional
        Size of the nodes (default: 20).
    output: str, optional
        Output format: "cyto" (Dash Cytoscape Div), "elements" (raw elements), "notebook" (inline Dash app), or "dash" (Dash app in browser) (default: "cyto").
    dash_port: int, optional
        Port for running the Dash app (default: 8050).
    min_zoom: float, optional
        Minimum zoom level for the Cytoscape component (default: 0.5).
    max_zoom: float, optional
        Maximum zoom level for the Cytoscape component (default: 1).
    min_edge_thickness: float, optional
        Minimum edge thickness for frequency-based scaling (default: 0.1).
    max_edge_thickness: float, optional
        Maximum edge thickness for frequency-based scaling (default: 4).
    freq_as_edge_thickness: bool, optional
        Whether to scale edge thickness based on transition frequency (default: False).
    color: str, optional
        Color for nodes and edges (default: "black"). If "hsu", uses a preset color.
    title_text: str or Dash component, optional
        Title text or component to display above the graph (default: None).
Returns:
    Dash component, dict, or Dash app:
        - If output == "cyto": returns a Dash html.Div containing the Cytoscape graph.
        - If output == "elements": returns a dict with 'nodes' and 'edges'.
        - If output == "notebook": runs and displays a Dash app inline (for Jupyter).
        - If output == "dash": runs a Dash app in the browser and returns the app instance.
Notes:
    - Requires Dash, dash_cytoscape, dash_bootstrap_components, and pandas.
    - The function supports interactive modals for displaying timing data on states and transitions.
    - Threading is used to launch the Dash app in browser mode without blocking the main program.
# 1. 'grid'            → Places nodes in a simple rectangular grid.

2. 'random' → Randomly positions nodes; useful for testing.

3. 'circle' → Arranges nodes evenly around a circle.

4. 'concentric' → Places nodes in concentric circles, often by degree or weight.

5. 'breadthfirst' → Hierarchical layout (tree-like), good for state machines or DAGs.

Optional params: directed=True, padding=

6. 'cose' → Force-directed layout (spring simulation). Great for organic graphs.

Optional params: idealEdgeLength, nodeRepulsion, gravity, numIter

7. 'cose-bilkent' → Improved force-directed layout with better stability and aesthetics.

(Requires: cyto.load_extra_layouts())

8. 'cola' → Constraint-based force-directed layout; handles larger graphs well.

(Requires: cyto.load_extra_layouts())

9. 'euler' → Physically simulated layout; looks natural and dynamic.

(Requires: cyto.load_extra_layouts())

10. 'avsdf' → Circular layout optimized to reduce edge crossings.

(Requires: cyto.load_extra_layouts())

11. 'spread' → Distributes disconnected components evenly across space.

(Requires: cyto.load_extra_layouts())

12. 'klay' → Layered (hierarchical) layout, excellent for flowcharts or process models.

(Requires: cyto.load_extra_layouts())

13. 'dagre' → Directed acyclic graph layout, ideal for workflows and automata.

(Requires: cyto.load_extra_layouts())

Optional params: rankDir='TB' (top-bottom), 'LR' (left-right), etc.

Source code in ml4cps/vis.py

def plot_cps_component(cps, id=None, node_labels=False, center_node_labels=False, event_label=True,
                       show_transition_freq=False, show_transition_timing=False, font_size=6, edge_font_size=6,
                       edge_text_max_width=None, init_label=False, limit_interval_precision=None,
                       show_transition_data=False, transition_data_keys=True, node_size=20, output="cyto",
                       dash_port=8050, min_zoom=0.5, split_edges_diff_event=False,
                       max_zoom=1, min_edge_thickness=0.1, max_edge_thickness=4, freq_as_edge_thickness=False,
                       color="black", title_text=None, layout_name='breadthfirst', layout_spacingFactor=1,
                       hide_nodes=None):
    """    
    Visualizes a component of a Cyber-Physical System (CPS) as a graph using Dash Cytoscape.
    This function generates a graphical representation of the discrete states and transitions of a CPS,
    with various customization options for node and edge appearance, labels, and output format.
    The visualization can be rendered as Dash Cytoscape elements, in a Dash app, or as a notebook widget.
    Parameters:
        cps: object
            The CPS object containing discrete states, transitions, and related data.
        id: str, optional
            The unique identifier for the Cytoscape component (default: "graph").
        node_labels: bool, optional
            Whether to display labels on nodes (default: False).
        center_node_labels: bool, optional
            Whether to center node labels (default: False).
        edge_labels: bool, optional
            Whether to display labels on edges (default: True).
        show_transition_freq: bool, optional
            Whether to show transition frequency on edge labels (default: False).
        edge_font_size: int, optional
            Font size for edge labels (default: 6).
        edge_text_max_width: int or None, optional
            Maximum width for edge label text wrapping (default: None).
        init_label: bool, optional
            Whether to label initial state transitions as 'init' (default: False).
        show_transition_data: bool or list, optional
            Whether to display additional transition data on edge labels. If a list, only specified keys are shown (default: False).
        node_size: int, optional
            Size of the nodes (default: 20).
        output: str, optional
            Output format: "cyto" (Dash Cytoscape Div), "elements" (raw elements), "notebook" (inline Dash app), or "dash" (Dash app in browser) (default: "cyto").
        dash_port: int, optional
            Port for running the Dash app (default: 8050).
        min_zoom: float, optional
            Minimum zoom level for the Cytoscape component (default: 0.5).
        max_zoom: float, optional
            Maximum zoom level for the Cytoscape component (default: 1).
        min_edge_thickness: float, optional
            Minimum edge thickness for frequency-based scaling (default: 0.1).
        max_edge_thickness: float, optional
            Maximum edge thickness for frequency-based scaling (default: 4).
        freq_as_edge_thickness: bool, optional
            Whether to scale edge thickness based on transition frequency (default: False).
        color: str, optional
            Color for nodes and edges (default: "black"). If "hsu", uses a preset color.
        title_text: str or Dash component, optional
            Title text or component to display above the graph (default: None).
    Returns:
        Dash component, dict, or Dash app:
            - If output == "cyto": returns a Dash html.Div containing the Cytoscape graph.
            - If output == "elements": returns a dict with 'nodes' and 'edges'.
            - If output == "notebook": runs and displays a Dash app inline (for Jupyter).
            - If output == "dash": runs a Dash app in the browser and returns the app instance.
    Notes:
        - Requires Dash, dash_cytoscape, dash_bootstrap_components, and pandas.
        - The function supports interactive modals for displaying timing data on states and transitions.
        - Threading is used to launch the Dash app in browser mode without blocking the main program.
    # 1. 'grid'            → Places nodes in a simple rectangular grid.
# 2. 'random'          → Randomly positions nodes; useful for testing.
# 3. 'circle'          → Arranges nodes evenly around a circle.
# 4. 'concentric'      → Places nodes in concentric circles, often by degree or weight.
# 5. 'breadthfirst'    → Hierarchical layout (tree-like), good for state machines or DAGs.
#                        Optional params: directed=True, padding=<int>
# 6. 'cose'            → Force-directed layout (spring simulation). Great for organic graphs.
#                        Optional params: idealEdgeLength, nodeRepulsion, gravity, numIter
# 7. 'cose-bilkent'    → Improved force-directed layout with better stability and aesthetics.
#                        (Requires: cyto.load_extra_layouts())
# 8. 'cola'            → Constraint-based force-directed layout; handles larger graphs well.
#                        (Requires: cyto.load_extra_layouts())
# 9. 'euler'           → Physically simulated layout; looks natural and dynamic.
#                        (Requires: cyto.load_extra_layouts())
# 10. 'avsdf'          → Circular layout optimized to reduce edge crossings.
#                        (Requires: cyto.load_extra_layouts())
# 11. 'spread'         → Distributes disconnected components evenly across space.
#                        (Requires: cyto.load_extra_layouts())
# 12. 'klay'           → Layered (hierarchical) layout, excellent for flowcharts or process models.
#                        (Requires: cyto.load_extra_layouts())
# 13. 'dagre'          → Directed acyclic graph layout, ideal for workflows and automata.
#                        (Requires: cyto.load_extra_layouts())
#                        Optional params: rankDir='TB' (top-bottom), 'LR' (left-right), etc.
    """


    if id is None:
        id = "graph"

    if color == "hsu":
       color = "#B8234F"
    nodes = []
    edges = []
    for n in cps.discrete_states:
        if n in cps.final_q:
            classes = ['final']
            if hide_nodes and n in hide_nodes:
                classes.append('hidden')
            nodes.append(dict(data={'id': n, 'label': n.replace(' ','')}, classes=classes))
        else:
            classes = []
            if hide_nodes and n in hide_nodes:
                classes.append('hidden')
            nodes.append(dict(data={'id': n, 'label': n.replace(' ','')}, classes=classes))

        if n in cps.q0:
            nodes.append(dict(data={'id': f"q0{n}", 'label': f"q0{n}"}, classes='q0'))
            if init_label:
                edges.append(dict(data=dict(label='init', source=f"q0{n}", target=n)))
            elif event_label:
                edges.append(dict(data=dict(label=cps.initial_r, source=f"q0{n}", target=n)))
            else:
                edges.append(dict(data=dict(source=f"q0{n}", target=n)))

    for e in cps.get_transitions():
        if hide_nodes and (e[0] in hide_nodes or e[1] in hide_nodes):
            continue

        if 'timing' in e[3]:
            freq = len(e[3]['timing'])
            import numbers


            timings = [float(x) if isinstance(x, numbers.Number) else pd.Timedelta(x).total_seconds() for x in e[3]['timing']]
        else:
            freq = 0
            timings = []

        edge = dict(data={'source': e[0],
                          'target': e[1],
                          'label': f'{e[3]["event"]}' if event_label else '',
                          'timing': timings,
                          'freq': freq})

        existing_edge = next((x for x in edges if x['data']['source'] == edge['data']['source'] and
                             x['data']['target'] == edge['data']['target']), None)
        if existing_edge is None or split_edges_diff_event:
            if show_transition_freq:
                edge['data']['label'] += f' #{freq}'
            if show_transition_timing:
                if limit_interval_precision is None:
                    edge['data']['label'] += f' [{min(timings)},{max(timings)}]'
                else:
                    edge['data']['label'] += f' [{min(timings):.{limit_interval_precision}f},{max(timings):.{limit_interval_precision}f}]'

            if show_transition_data:
                edge_data = e[3]
                ev = edge_data.pop('event', None)
                if isinstance(show_transition_data, list):
                    edge_data = {k: v for k, v in edge_data.items() if k in show_transition_data}
                # edge['data']['label'] += "\n"
                if transition_data_keys:
                    edge['data']['label'] += " ".join(f"{key}: {value}" for key, value in edge_data.items())
                else:
                    edge['data']['label'] += " ".join(f"{value}" for key, value in edge_data.items())
            edges.append(edge)
        else: # existing_edge
            if show_transition_freq:
                existing_edge['data']['label'] += f' #{freq}]'
            if show_transition_timing:
                existing_edge['data']['label'] += f' [{min(timings):.{limit_interval_precision}f},{max(timings):.{limit_interval_precision}f}]'

            if show_transition_data or event_label:
                edge_data = e[3]
                ev = edge_data.pop('event', None)
                if isinstance(show_transition_data, list):
                    edge_data = {k: v for k, v in edge_data.items() if k in show_transition_data}
                existing_edge['data']['label'] += f" ,{ev} " + "; ".join(f"{key} = {value}" for key, value in edge_data.items())

    # Normalize thickness to the range [1, 10]
    thickness_values = [edge["data"].get("freq", 1) for edge in edges]
    min_thickness = min(thickness_values) if thickness_values else 0
    max_thickness = max(thickness_values) if thickness_values else 0

    if max_thickness == min_thickness:
        max_thickness += 1

    for edge in edges:
        raw_thickness = edge["data"].get("freq", 1)
        edge["data"]["thickness"] = ((raw_thickness - min_thickness) / (max_thickness - min_thickness) *
                                     (max_edge_thickness - min_edge_thickness) + min_edge_thickness)

    elements = dict(nodes=nodes, edges=edges)

    if output == "elements":
        return elements

    node_style = {'width': node_size,
                  'height': node_size,
                  'border-width': 1,
                  'border-color': color,
                  'background-color': 'transparent',
                  "font-family": "serif",
                  'background-opacity': 0}
    if node_labels:
        node_style['label'] = 'data(label)'
        node_style['font-size'] = font_size
        node_style['font-style'] = "italic"
        node_style['text-wrap'] = 'wrap'
        node_style['text-max-width'] = 50
    if center_node_labels:
        node_style['text-halign'] = 'center'
        node_style['text-valign'] = 'center'


    edge_style = {
                'curve-style': 'bezier',
                'background-color': 'white',  # Inner fill
                'target-arrow-shape': 'triangle',
                'target-arrow-color': color,
                'target-arrow-size': 3,
                'text-background-color': '#ffffff',
                'text-background-opacity': 1,
                'text-background-shape': 'roundrectangle',
                'color': "#B8234F",
                'width': 1,
                'font-style': 'italic',
                'font-family': "serif",
                'text-wrap': 'wrap',
                'font-size': edge_font_size,
                'text-max-width': edge_text_max_width,
                'line-color': color
    }

    if freq_as_edge_thickness:
        edge_style['width'] = 'data(thickness)'

    edge_style['label'] = 'data(label)'

    stylesheet = [
        {
            'selector': 'node',
            'style': node_style
        },
        {
            'selector': '.q0',
            'style': {
                    'width': 1,  # Small width to make it look like a point
                    'height': 1,  # Small height to make it look like a point
                    'label': '',  # No label to keep it minimal
                    'border-width': 0  # No border
                }
        },
        {
            'selector': '.final',
            'style': {
                'border-width': 3  # No border
            }
        },
        {
            'selector': 'edge',
            'style': edge_style
        },
        {
            "selector": ".hidden",
            "style": {"visibility": "hidden"}
        }
    ]

    network = cyto.Cytoscape(
        id=id,
        layout={'name': layout_name, "fit": True, "spacingFactor": layout_spacingFactor},
        maxZoom=max_zoom,
        minZoom=min_zoom,
        style={'width': '100%', 'height': '100%'}, stylesheet=stylesheet,
        elements=elements)

    modal_state_data = dbc.Modal(children=[dbc.ModalHeader("Timings"),
                                           dbc.ModalBody(html.Div(children=[]))],
                                 id=f"{id}-modal-state-data")
    modal_transition_data = dbc.Modal(children=[dbc.ModalHeader("Timings"),
                                                dbc.ModalBody(html.Div(children=[]))],
                                 id=f"{id}-modal-transition-data")
    network = html.Div([title_text, network, modal_state_data, modal_transition_data], style={'width': '100%', 'height': '100%'})

    if output == "notebook":
        app = Dash(__name__)
        app.layout = html.Div(children=[network], style={'width': '100%',
                                                         'height': '100vh',
                                                         'margin': '0',
                                                         'padding': '0'})
        app.run(mode='inline', port=dash_port)
        return None
    elif output == "dash":
        app = Dash(__name__)
        app.layout = html.Div(children=[network], style={'width': '100%',
                                                         'height': '100vh',
                                                         'margin': '0',
                                                         'padding': '0'})

        # Function to start the Dash server
        def run_dash():
            app.run(port=dash_port, debug=False, use_reloader=False)  # Start the Dash server

        # Function to open the browser
        def open_browser():
            time.sleep(1)  # Give the server a second to start
            webbrowser.open(f"http://127.0.0.1:{dash_port}/")  # Open the Dash app in the browser

        # Start the Dash server in a separate thread
        server_thread = threading.Thread(target=run_dash)
        server_thread.daemon = True  # Allows the program to exit even if this thread is running
        server_thread.start()

        # Open the Dash app in the default browser
        open_browser()
        server_thread.join(timeout=10)
        return app
    else:
        return network

plot_cps_plotly(cps, layout='dot', marker_size=20, node_positions=None, show_events=True, show_num_occur=False, show_state_label=True, font_size=10, plot_self_transitions=True, use_previos_node_positions=False, **kwargs)

Visualizes a Cyber-Physical System (CPS) state-transition graph using Plotly. This function generates an interactive Plotly figure representing the states and transitions of a CPS. Nodes represent system states, and edges represent transitions. Various layout algorithms and display options are supported. Args: cps: The CPS object containing the state-transition graph. Must have attributes _G (networkx graph), get_transitions(), print_state(), num_occur(), and previous_node_positions. layout (str, optional): Layout algorithm for node positioning. Options are "dot" (default), "spectral", "kamada_kawai", or "fruchterman_reingold". marker_size (int, optional): Size of the node markers. Default is 20. node_positions (dict, optional): Precomputed node positions as a dictionary {node: (x, y)}. If None, positions are computed using the selected layout. show_events (bool, optional): Whether to display event labels on transitions. Default is True. show_num_occur (bool, optional): Whether to display the number of occurrences for each transition. Default is False. show_state_label (bool, optional): Whether to display state labels on nodes. Default is True. font_size (int, optional): Font size for transition/event labels. Default is 10. plot_self_transitions (bool, optional): Whether to plot self-loop transitions. Default is True. use_previos_node_positions (bool, optional): If True and node_positions is None, reuse positions from cps.previous_node_positions. Default is False. **kwargs: Additional keyword arguments passed to the layout function (e.g., for networkx layouts). Returns: plotly.graph_objs.Figure: A Plotly figure object representing the CPS state-transition graph. Notes: - Requires Plotly, NetworkX, and pydotplus (for "dot" layout). - The CPS object must provide the required methods and attributes as described above. - Edge and node styling can be further customized by modifying the function.

Source code in ml4cps/vis.py

def plot_cps_plotly(cps, layout="dot", marker_size=20, node_positions=None, show_events=True, show_num_occur=False,
                show_state_label=True, font_size=10, plot_self_transitions=True, use_previos_node_positions=False,
                **kwargs):
    """
    Visualizes a Cyber-Physical System (CPS) state-transition graph using Plotly.
    This function generates an interactive Plotly figure representing the states and transitions of a CPS.
    Nodes represent system states, and edges represent transitions. Various layout algorithms and display options
    are supported.
    Args:
        cps: The CPS object containing the state-transition graph. Must have attributes `_G` (networkx graph),
            `get_transitions()`, `print_state()`, `num_occur()`, and `previous_node_positions`.
        layout (str, optional): Layout algorithm for node positioning. Options are "dot" (default), "spectral",
            "kamada_kawai", or "fruchterman_reingold".
        marker_size (int, optional): Size of the node markers. Default is 20.
        node_positions (dict, optional): Precomputed node positions as a dictionary {node: (x, y)}. If None,
            positions are computed using the selected layout.
        show_events (bool, optional): Whether to display event labels on transitions. Default is True.
        show_num_occur (bool, optional): Whether to display the number of occurrences for each transition. Default is False.
        show_state_label (bool, optional): Whether to display state labels on nodes. Default is True.
        font_size (int, optional): Font size for transition/event labels. Default is 10.
        plot_self_transitions (bool, optional): Whether to plot self-loop transitions. Default is True.
        use_previos_node_positions (bool, optional): If True and node_positions is None, reuse positions from
            `cps.previous_node_positions`. Default is False.
        **kwargs: Additional keyword arguments passed to the layout function (e.g., for networkx layouts).
    Returns:
        plotly.graph_objs.Figure: A Plotly figure object representing the CPS state-transition graph.
    Notes:
        - Requires Plotly, NetworkX, and pydotplus (for "dot" layout).
        - The CPS object must provide the required methods and attributes as described above.
        - Edge and node styling can be further customized by modifying the function.
    """
    # layout = 'kamada_kawai'  # TODO
    edge_scatter_lines = None
    annotations = []
    if node_positions is None:
        if use_previos_node_positions:
            node_positions = cps.previous_node_positions
        else:
            g = cps._G
            if layout == "dot":
                graph = pdp.graph_from_edges([('"' + tr[0] + '"', '"' + tr[1] + '"')
                                              for tr in g.edges], directed=True)
                # graph.set_node_defaults(shape='point')
                for nnn in g.nodes:
                    graph.add_node(pdp.Node(nnn, shape='point'))
                graph.set_prog('dot')
                graph = graph.create(format="dot")
                # graph.
                # graph.write_dot('temp.dot')
                # graph.write_svg('temp.svg')
                # graph = pdp.graph_from_dot_file('temp.dot')
                graph = pdp.graph_from_dot_data(graph)
                node_positions = {n.get_name().strip('"'): tuple(float(x) for x in n.get_pos()[1:-1].split(','))
                                  for n in graph.get_nodes() if
                                  n.get_name().strip('"') not in ['\\r\\n', 'node', 'graph']}
                edges = {e.obj_dict['points']: e.get_pos()[3:-1].split(' ')
                         for e in graph.get_edges()}  # [3:].split(",")

                # edge_shapes = []
                # edge_scatter_lines = []
                # for points, edg in edges.items():
                #     edg = [tuple(float(eee.replace('\r', '').replace('\n', '').replace('\\', '').strip())
                #                  for eee in e.split(",")) for e in edg]
                #     node_pos_start = node_positions[points[0].replace('"', '')]
                # edg.insert(0, node_pos_finish)ääääääääääääääääääääääääääääääääääääääääääääääääääääääää
                # node_pos_finish = node_positions[points[1].replace('"', '')]
                # control_points = ' '.join(','.join(map(str, e)) for e in edg[1:])
                # {node_pos_start[0]}, {node_pos_start[1]}
                # Cubic Bezier Curves
                # edge_shapes.append(dict(
                #     type="path",
                #     path=f"M {node_pos_start[0]},{node_pos_start[1]} C {control_points}", #{node_pos_finish[0]}, {node_pos_finish[1]}",
                #     line_color="MediumPurple",
                # ))

                # edg.append(node_pos_start)

                # edg.append((None, None))
                # annotations.append(dict(ax=node_pos_finish[0], ay=node_pos_finish[1], axref='x', ayref='y',
                #     x=edg[-2][0], y=edg[-2][1], xref='x', yref='y',
                #     showarrow=True, arrowhead=1, arrowsize=2, startarrowhead=0))
                # edge_scatter_lines.append(edg)
                # parse_path(edges)
                # points_from_path(edges)
            elif layout == 'spectral':
                node_positions = nx.spectral_layout(g, **kwargs)
            elif layout == 'kamada_kawai':
                node_positions = nx.kamada_kawai_layout(g, **kwargs)
            elif layout == 'fruchterman_reingold':
                node_positions = nx.fruchterman_reingold_layout(g, **kwargs)
        cps.previous_node_positions = node_positions
    node_x = []
    node_y = []
    for node in cps._G.nodes:
        x, y = node_positions[node]
        node_x.append(x)
        node_y.append(y)
    texts = []
    for v in cps._G.nodes:
        try:
            texts.append(cps.print_state(v))
        except:
            texts.append('Error printing state: ')
    if show_state_label:
        mode = 'markers+text'
    else:
        mode = 'markers'
    node_trace = go.Scatter(x=node_x, y=node_y, text=list(cps._G.nodes), mode=mode, textposition="top center",
                            hovertext=texts, hovertemplate='%{hovertext}<extra></extra>',
                            marker=dict(size=marker_size, line_width=1), showlegend=False)

    annotations = [dict(ax=node_positions[tr[0]][0], ay=node_positions[tr[0]][1], axref='x', ayref='y',
                        x=node_positions[tr[1]][0], y=node_positions[tr[1]][1], xref='x', yref='y',
                        showarrow=True, arrowhead=1, arrowsize=2) for tr in cps._G.edges]

    # annotations = []

    def fun(tr):
        if show_events and show_num_occur:
            return '<i>{} ({})</i>'.format(tr[2], cps.num_occur(tr))
        elif show_events:
            return '<i>{}</i>'.format(tr[2])
        elif show_num_occur:
            return '<i>{}</i>'.format(cps.num_occur(tr))

    if show_num_occur or show_events:
        annotations_text = [dict(x=(0.4 * node_positions[tr[0]][0] + 0.6 * node_positions[tr[1]][0]),
                                 y=(0.4 * node_positions[tr[0]][1] + 0.6 * node_positions[tr[1]][1]),
                                 xref='x', yref='y', text=fun(tr), font=dict(size=font_size, color='darkblue'),
                                 yshift=0, showarrow=False)  # , bgcolor='white')
                            for tr in cps.get_transitions() if plot_self_transitions or tr[0] != tr[1]]

        annotations += annotations_text

    traces = [node_trace]
    if edge_scatter_lines:
        edge_scatter_lines = list(chain(*edge_scatter_lines))
        edge_trace = go.Scatter(x=[xx[0] for xx in edge_scatter_lines], y=[xx[1] for xx in edge_scatter_lines],
                                mode='lines', showlegend=False, line=dict(color='black', width=1), hoverinfo=None,
                                hovertext=None, name='Transitions')
        traces.insert(0, edge_trace)

    fig = go.Figure(data=traces, layout=go.Layout(annotations=annotations,
                                                  paper_bgcolor='rgba(0,0,0,0)',
                                                  plot_bgcolor='rgba(0,0,0,0)'))

    fig.update_xaxes({'showgrid': False,  # thin lines in the background
                      'zeroline': False,  # thick line learn x=0
                      'visible': False})
    # 'fixedrange': True})  # numbers below)
    fig.update_yaxes({'showgrid': False,  # thin lines in the background
                      'zeroline': False,  # thick line learn x=0
                      'visible': False})
    # 'fixedrange': True})  # numbers below)
    fig.update_annotations(standoff=marker_size / 2, startstandoff=marker_size / 2)
    fig.update_layout(clickmode='event')
    return fig

plot_dash_frames(graph_frames, dash_port=8050)

Launches an interactive Dash web application to visualize a sequence of graph frames with a slider for manual frame selection. Args: graph_frames (list): A list of Dash components (e.g., Cytoscape graphs) representing different frames to display. dash_port (int, optional): The port number on which to run the Dash server. Defaults to 8050. Returns: dash.Dash: The Dash application instance. Side Effects: - Starts a Dash server in a separate thread. - Opens the default web browser to display the Dash app. - Waits for user input before returning. Notes: - The app displays the first frame by default and allows users to select other frames using a slider. - The function blocks until the user presses Enter in the console.

Source code in ml4cps/vis.py

def plot_dash_frames(graph_frames, dash_port=8050):
    """
    Launches an interactive Dash web application to visualize a sequence of graph frames with a slider for manual frame selection.
    Args:
        graph_frames (list): A list of Dash components (e.g., Cytoscape graphs) representing different frames to display.
        dash_port (int, optional): The port number on which to run the Dash server. Defaults to 8050.
    Returns:
        dash.Dash: The Dash application instance.
    Side Effects:
        - Starts a Dash server in a separate thread.
        - Opens the default web browser to display the Dash app.
        - Waits for user input before returning.
    Notes:
        - The app displays the first frame by default and allows users to select other frames using a slider.
        - The function blocks until the user presses Enter in the console.
    """

    app = Dash(__name__)
    app.layout = html.Div(children=graph_frames[0], style={'width': '100%',
                                                           'height': '100vh',
                                                           'margin': '0',
                                                           'padding': '0'})

    app.layout = html.Div([
        html.Div(graph_frames[0], id='cytoscape-graph'),

        # Slider for manual frame selection
        dcc.Slider(
            id='graph-slider',
            min=0,
            max=len(graph_frames) - 1,
            step=1,
            marks={i: str(i) for i in range(len(graph_frames))},  # Label frames
            value=0,  # Start at first frame
        ),
    ])


    # Callback to update Cytoscape graph when slider changes
    @app.callback(
        Output('cytoscape-graph', 'children'),
        Input('graph-slider', 'value')
    )
    def update_graph(frame_idx):
        print(frame_idx)
        return graph_frames[frame_idx]  # Update Cytoscape graph

    # Function to start the Dash server
    def run_dash():
        app.run_server(port=dash_port, debug=False, use_reloader=False)  # Start the Dash server

    # Function to open the browser
    def open_browser():
        time.sleep(1)  # Give the server a second to start
        webbrowser.open(f"http://127.0.0.1:{dash_port}/")  # Open the Dash app in the browser

    # Start the Dash server in a separate thread
    server_thread = threading.Thread(target=run_dash)
    server_thread.daemon = True  # Allows the program to exit even if this thread is running
    server_thread.start()

    # Open the Dash app in the default browser
    open_browser()
    # server_thread.join(timeout=1)

    input("Press Enter to continue...")
    return app

plot_execution_tree(graph, nodes_to_color, color, font_size=30)

Plots a system execution tree as a graph, where the horizontal position of nodes corresponds to their timestamps and the tree branches vertically. Args: graph (networkx.DiGraph): A directed graph where each node represents a system state, and edges represent transitions. Each node should have a 'label' (str) and 'weight' (int) attribute. Node names must be timestamp strings in the format "%d/%m/%Y, %H:%M:%S". nodes_to_color (list): List of node identifiers (timestamp strings) to be highlighted with a specific color. color (str): The color to use for highlighting nodes in nodes_to_color. font_size (int, optional): Font size for node labels in the visualization. Defaults to 30. Returns: cyto.Cytoscape: A Dash Cytoscape object representing the execution tree visualization, with nodes positioned by timestamp and colored as specified. Notes: - The function assumes the first node in graph.nodes is the starting node. - Node positions are determined by the time difference from the start node (x-axis) and their 'weight' attribute (y-axis). - Nodes in nodes_to_color are colored with the specified color; all others are gray. - Requires the cyto (Dash Cytoscape) library and datetime module.

Source code in ml4cps/vis.py

def plot_execution_tree(graph, nodes_to_color, color, font_size=30):
    """
    Plots a system execution tree as a graph, where the horizontal position of nodes corresponds to their timestamps and the tree branches vertically.
    Args:
        graph (networkx.DiGraph): A directed graph where each node represents a system state, and edges represent transitions. 
            Each node should have a 'label' (str) and 'weight' (int) attribute. Node names must be timestamp strings in the format "%d/%m/%Y, %H:%M:%S".
        nodes_to_color (list): List of node identifiers (timestamp strings) to be highlighted with a specific color.
        color (str): The color to use for highlighting nodes in `nodes_to_color`.
        font_size (int, optional): Font size for node labels in the visualization. Defaults to 30.
    Returns:
        cyto.Cytoscape: A Dash Cytoscape object representing the execution tree visualization, with nodes positioned by timestamp and colored as specified.
    Notes:
        - The function assumes the first node in `graph.nodes` is the starting node.
        - Node positions are determined by the time difference from the start node (x-axis) and their 'weight' attribute (y-axis).
        - Nodes in `nodes_to_color` are colored with the specified `color`; all others are gray.
        - Requires the `cyto` (Dash Cytoscape) library and `datetime` module.
    """

    # for ntd in nodes_to_delete:
    #     if ntd in graph:
    #         prenodes = list(graph.predecessors(ntd))
    #         sucnodes = list(graph.successors(ntd))
    #         preedges = list(graph.in_edges(ntd))
    #         sucedges = list(graph.out_edges(ntd))
    #         edgestodelete = preedges + sucedges
    #         if ((len(preedges) > 0) and (len(sucedges) > 0)):
    #             for prenode in prenodes:
    #                 for sucnode in sucnodes:
    #                     graph.add_edge(prenode, sucnode)
    #         if (len(edgestodelete) > 0):
    #             graph.remove_edges_from(edgestodelete)

    startstring = list(graph.nodes)[0]
    arr_elements = []
    num_of_nodes = graph.number_of_nodes()
    # vertical_height = num_of_states
    visited = set()
    stack = [startstring]
    while stack:
        node = stack.pop()
        if node not in visited:
            elemid = str(node)
            elemlabel = graph.nodes[node].get('label')
            datepos1 = datetime.strptime(startstring, "%d/%m/%Y, %H:%M:%S")
            datepos2 = datetime.strptime(node, "%d/%m/%Y, %H:%M:%S")
            nodeweight = graph.nodes[node].get('weight')
            ypos = 0
            if nodeweight == 0:
                ypos = num_of_states * 100
            else:
                ypos = (nodeweight - 1) * 200
            element = {
                'data': {
                    'id': elemid,
                    'label': elemlabel
                },
                'position': {
                    'x': (datepos2 - datepos1).total_seconds() / 7200,
                    'y': ypos
                },
                # 'locked': True
            }
            arr_elements.append(element)
            visited.add(node)
            stack.extend(neighbor for neighbor in graph.successors(node) if neighbor not in visited)
    for u, v in list(graph.edges):
        edge_element = {
            'data': {
                'source': u,
                'target': v
            }
        }
        arr_elements.append(edge_element)


    colorcode = ['gray'] * num_of_nodes
    for n in nodes_to_color:
        if n in graph:
            n_ind = list(graph.nodes).index(n)
            if (n_ind < num_of_nodes):
                colorcode[n_ind] = color
    new_stylesheet = []
    for i in range(0, num_of_nodes):
        new_stylesheet.append({
            'selector': f'node[id = "{list(graph.nodes)[i]}"]',
            'style': {
                'font-size': f'{font_size}px',
                'content': 'data(label)',
                'background-color': colorcode[i],
                'text-valign': 'top',
                'text-halign': 'center',
                # 'animate': True
            }
        })

    cytoscapeobj = cyto.Cytoscape(
        id='org-chart',
        layout={'name': 'preset'},
        style={'width': '2400px', 'height': '1200px'},
        elements=arr_elements,
        stylesheet=new_stylesheet
    )
    return cytoscapeobj

plot_state_transitions(ta, state, obs=None)

Visualizes the outgoing state transitions from a given state in a timed automaton, along with associated observation data. Parameters: ta: An object representing the timed automaton, expected to have an out_transitions(state) method that returns transitions from the given state. state: The current state for which outgoing transitions and associated observations are to be visualized. obs (optional): A pandas DataFrame containing observation data. Must include at least the columns 'Mode', 'q_next', 'Duration', 'Time', and optionally 'Vergussgruppe', 'HID', 'ChipID', 'Order', and 'ArtNr'. If None, the function raises NotImplemented. Returns: fig: A Plotly figure object containing subplots for each outgoing transition. For each transition, the function displays: - A scatter plot of observation durations over time, grouped by 'Vergussgruppe'. - A histogram of durations for each 'Vergussgruppe'. The subplots are arranged with shared axes and appropriate titles for each transition. Raises: NotImplemented: If obs is None. Notes: - The function expects certain columns to exist in the obs DataFrame. If missing, default values are assigned. - Colors for different 'Vergussgruppe' groups are assigned from DEFAULT_PLOTLY_COLORS. - The function uses Plotly's make_subplots, go.Scatter, and go.Histogram for visualization.

Source code in ml4cps/vis.py

def plot_state_transitions(ta, state, obs=None):
    """
    Visualizes the outgoing state transitions from a given state in a timed automaton, along with associated observation data.
    Parameters:
        ta: An object representing the timed automaton, expected to have an `out_transitions(state)` method that returns transitions from the given state.
        state: The current state for which outgoing transitions and associated observations are to be visualized.
        obs (optional): A pandas DataFrame containing observation data. Must include at least the columns 'Mode', 'q_next', 'Duration', 'Time', and optionally 'Vergussgruppe', 'HID', 'ChipID', 'Order', and 'ArtNr'. If None, the function raises NotImplemented.
    Returns:
        fig: A Plotly figure object containing subplots for each outgoing transition. For each transition, the function displays:
            - A scatter plot of observation durations over time, grouped by 'Vergussgruppe'.
            - A histogram of durations for each 'Vergussgruppe'.
        The subplots are arranged with shared axes and appropriate titles for each transition.
    Raises:
        NotImplemented: If `obs` is None.
    Notes:
        - The function expects certain columns to exist in the `obs` DataFrame. If missing, default values are assigned.
        - Colors for different 'Vergussgruppe' groups are assigned from `DEFAULT_PLOTLY_COLORS`.
        - The function uses Plotly's `make_subplots`, `go.Scatter`, and `go.Histogram` for visualization.
    """

    trans = ta.out_transitions(state)
    titles = []
    for k in trans:
        titles.append('State: {0} -> {1} -> {2}'.format(k[0], k[3]['event'], k[1]))
        titles.append('')

    fig = subplots.make_subplots(len(trans), 2, shared_xaxes=True, shared_yaxes=True,
                                 subplot_titles=titles, column_widths=[0.8, 0.2],
                                 horizontal_spacing=0.02, vertical_spacing=0.2)
    if obs is None:
        raise NotImplemented()
        # observations = self.get_transition_observations(state)

    obs = obs[obs['Mode'] == state]
    ind = 0
    for k in trans:
        v = obs[obs.q_next == k[1]]
        ind += 1
        ind_color = 0
        if len(v) == 0:
            continue
        # v['VG'] = 'Unknown'
        if 'Vergussgruppe' in v:
            v['Vergussgruppe'] = v['Vergussgruppe'].fillna('Unknown')
        else:
            v['Vergussgruppe'] = 'Unknown'

        v['Order'] = 'Unknown'
        v['ChipID'] = 'Unknown'
        v['Item'] = 'Unknown'
        v['ArtNr'] = 'Unknown'
        for vg, vv in v.groupby('Vergussgruppe'):
            vv = vv.to_dict('records')
            fig.add_trace(go.Histogram(y=[o['Duration'] for o in vv],
                                       name=vg,
                                       marker_color=DEFAULT_PLOTLY_COLORS[ind_color]), row=ind, col=2)
            ind_color += 1

        # Overlay both histograms
        fig.update_layout(barmode='overlay')
        # Reduce opacity to see both histograms
        fig.update_traces(opacity=0.5, row=ind, col=2)

        ind_color = 0
        # v = pd.DataFrame(v)

        v['Item'] = v['HID']
        for vg, vv in v.groupby('Vergussgruppe'):
            vv = vv.to_dict('records')
            hovertext = [
                'Timing: {}s<br>Zähler: {}<br>ChipID: {}<br>Order: {}<br>VG: {}<br>ArtNr: {}'.format(o['Duration'],
                                                                                                     o['Item'],
                                                                                                     o['ChipID'],
                                                                                                     o['Order'],
                                                                                                     o['Vergussgruppe'],
                                                                                                     o['ArtNr'])
                for o in vv]
            fig.add_trace(go.Scatter(x=[o['Time'] for o in vv], y=[o['Duration'] for o in vv],
                                     marker=dict(size=6, symbol="circle", color=DEFAULT_PLOTLY_COLORS[ind_color]),
                                     name=vg,
                                     mode="markers",
                                     hovertext=hovertext), row=ind, col=1)
            ind_color += 1
        fig.update_xaxes(showticklabels=True, row=ind, col=1)
    fig.update_layout(showlegend=False, margin=dict(b=0, t=30), width=1200)
    return fig

plot_stateflow(stateflow, color_mapping=None, state_col='State', task_col='Task', bar_height=12, start_column='Start', finish_column='Finish', return_figure=False, description_col='Description', idle_states=None)

Visualizes state transitions over time for one or more tasks/stations as a Gantt-like interactive timeline.

Parameters: - stateflow (DataFrame or dict): DataFrame with state transitions, or a dictionary of DataFrames per station. - color_mapping (dict, optional): Mapping of state names to colors. If None, default colors are used. - state_col (str): Column name indicating the state (default: 'State'). - bar_height (int): Height of the timeline bars (default: 12). - start_column (str): Column name with start timestamps (default: 'Start'). - finish_column (str): Column name with end timestamps (default: 'Finish'). - return_figure (bool): If True, returns a Plotly Figure. Otherwise, returns a list of Plotly traces. - description_col (str or list): Column(s) to include in the hover tooltip (default: 'Description'). - idle_states (str or list): State(s) to exclude from the plot (e.g., 'IDLE').

Returns: - Plotly Figure or list of traces, depending on return_figure.

Example

fig = plot_stateflow(df, state_col='Mode', start_column='StartTime', finish_column='EndTime', return_figure=True) fig.show()

This function is ideal for visualizing process flows, machine states, or event-based logs with time intervals.

Source code in ml4cps/vis.py

def plot_stateflow(stateflow, color_mapping=None, state_col='State', task_col='Task', bar_height=12,
                   start_column='Start', finish_column='Finish', return_figure=False, description_col='Description',
                   idle_states=None):
    """
    Visualizes state transitions over time for one or more tasks/stations as a Gantt-like interactive timeline.

    Parameters:
    - stateflow (DataFrame or dict): DataFrame with state transitions, or a dictionary of DataFrames per station.
    - color_mapping (dict, optional): Mapping of state names to colors. If None, default colors are used.
    - state_col (str): Column name indicating the state (default: 'State').
    - bar_height (int): Height of the timeline bars (default: 12).
    - start_column (str): Column name with start timestamps (default: 'Start').
    - finish_column (str): Column name with end timestamps (default: 'Finish').
    - return_figure (bool): If True, returns a Plotly Figure. Otherwise, returns a list of Plotly traces.
    - description_col (str or list): Column(s) to include in the hover tooltip (default: 'Description').
    - idle_states (str or list): State(s) to exclude from the plot (e.g., 'IDLE').

    Returns:
    - Plotly Figure or list of traces, depending on `return_figure`.

    Example:
        fig = plot_stateflow(df, state_col='Mode', start_column='StartTime', finish_column='EndTime', return_figure=True)
        fig.show()

    This function is ideal for visualizing process flows, machine states, or event-based logs with time intervals.
    """

    if idle_states is None:
        idle_states = []
    if type(idle_states) is str:
        idle_states = [idle_states]

    if isinstance(stateflow, dict):
        stateflow_df_list = []
        for station, s in stateflow.items():
            if s.size:
                sf = s[(~s[state_col].isin(idle_states))]
                        # ((start_plot <= s.Time) &
                        #  (s.Time <= finish_plot)) |
                        # ((start_plot <= s.Finish) & (s.Finish <= finish_plot)))
                sf[task_col] = station
                # if sf.size > 0 and pd.isnull(sf['Finish'].iloc[-1]):
                #     sf['Finish'].iloc[-1] = pd.to_datetime(finish_plot)
                # s['Finish'] = pd.to_datetime(s['Finish'])
                stateflow_df_list.append(sf)
            else:
                stateflow_df_list.append(pd.DataFrame([]))
        stateflow_df = pd.concat(stateflow_df_list)
    else:
        stateflow_df = stateflow

    if stateflow_df.shape[0] == 0:
        if return_figure:
            return go.Figure()
        else:
            return []

    if description_col is not None and type(description_col) is str:
        description_col = [description_col]
    if color_mapping is None:
        color_mapping = {}
        items = list(stateflow_df[state_col].unique())
        for k, i in enumerate(items):
            color_mapping[i] = colors.qualitative.Dark24[k % 24]

    stateflow_df['Duration'] = stateflow_df[finish_column] - stateflow_df[start_column]
    if state_col not in stateflow_df:
        stateflow_df[state_col] = None
    stateflow_df[state_col] = stateflow_df[state_col].replace([None], [''])

    traces = []
    for name, g in stateflow_df.groupby(state_col):
        if name is None or name == '':
            continue
        x = []
        y = []
        hovertext = []
        custom_data = []
        text = []
        for k, row in g.iterrows():  # , g[item_col], g.Source, g.Destination):
            x1, x2, tsk = row[start_column], row[finish_column], row[task_col]
            x.append(x1)
            x.append(x2)
            x.append(None)
            y.append(tsk)
            y.append(tsk)
            y.append(None)
            dauer = x2 - x1
            if type(x1) in [datetime.datetime, pd.Timestamp]:
                x1_str = x1.strftime("%d.%m %H:%M:%S")
            else:
                x1_str = x1
            if type(x2) in [datetime.datetime, pd.Timestamp]:
                x2_str = x2.strftime("%d.%m %H:%M:%S")
            else:
                x2_str = x2

            ht = 'Start: {}<br>Finish: {}<br>Duration: {}'.format(x1_str, x2_str, dauer)
            if description_col is not None:
                for dc in description_col:
                    if dc in row:
                        ht += '<br>{}: {}'.format(dc, row[dc])
            for k, val in row.items():
                if not pd.isnull(val) and k not in [state_col, finish_column, start_column, task_col, 'Duration']:
                    ht += f'<br>{k}: {val}'
            hovertext.append(ht)

            custom_data.append(dict(Start=x1, Finish=x2, State=name, Task=tsk, Source=row.get('Quelle', None)))

        color = color_mapping.get(name, "black")
        traces.append(go.Scatter(x=x, y=y, line=dict(width=bar_height), name=name, line_color=color,
                                   hoverinfo='skip', mode='lines', legendgroup=name, showlegend=True, opacity=0.8))
        traces.append(go.Scatter(x=np.asarray(g[start_column] + g.Duration / 2), y=g.Task, mode='text+markers',
                                   marker=dict(size=5, color=color), name=name,
                                   showlegend=False, opacity=0.8, customdata=custom_data,
                                   hovertext=hovertext, text=text, textfont=dict(size=10, color='olive'),
                                   hovertemplate=f'<extra></extra><b>{name}</b><br>%{{hovertext}}'))

    if return_figure:
        fig = go.Figure(data=traces)
        return fig
    else:
        return traces

plot_timeseries(data, timestamp=None, mode_data=None, discrete=False, title=None, use_columns=None, height=None, limit_num_points=None, names=None, xaxis_title=None, customdata=None, iterate_colors=True, y_title_font_size=14, opacity=1, vertical_spacing=0.005, sharey=False, bounds=None, plot_only_changes=False, yAxisLabelOffset=False, marker_size=4, showlegend=False, mode='lines+markers', mode_height=0.2, x_title=None, **kwargs)

Using plotly library, plots each variable (column) in a collection of dataframe as subplots, one after another.

Arguments: yAxisLabelOffset (bool): if True, adds an offset to the plots y-axis labels. Improves readability on long subplot names.

Returns:

Name	Type	Description
fig	Figure

Source code in ml4cps/vis.py

def plot_timeseries(data, timestamp=None, mode_data=None, discrete=False, title=None, use_columns=None, height=None,
                    limit_num_points=None, names=None, xaxis_title=None, customdata=None, iterate_colors=True,
                    y_title_font_size=14, opacity=1, vertical_spacing=0.005, sharey=False, bounds=None,
                    plot_only_changes=False, yAxisLabelOffset=False, marker_size=4, showlegend=False,
                    mode='lines+markers', mode_height=0.2, x_title=None, **kwargs):
    """
    Using plotly library, plots each variable (column) in a collection of dataframe as subplots, one after another.

    Arguments:
    yAxisLabelOffset (bool): if True, adds an offset to the plots y-axis labels. Improves readability on long subplot names.

    Returns:
        fig (plotly.Figure):
    """

    if limit_num_points is None or limit_num_points < 0:
        limit_num_points = np.inf
    if customdata is not None:
        customdata = customdata.fillna('')
    if type(data) is not list:
        data = [data]

    if len(data) == 0:
        return go.Figure()

    # if not panda data frame
    for i in range(len(data)):
        if not isinstance(data[i], pd.DataFrame):
            data[i] = pd.DataFrame(data[i])

    # if no timestamp is in the data
    if timestamp is not None:
        if type(timestamp) is str or type(timestamp) is int:
            for i in range(len(data)):
                data[i] = data[i].set_index(timestamp)

    if height is None:
        height = max(800, len(data[0].columns) * 60)

    if use_columns is None:
        columns = data[0].columns
    else:
        columns = use_columns

    num_rows = len(columns)
    categories = []
    if mode_data is not None:
        num_rows += 1
        fig = make_subplots(rows=num_rows, cols=1,
                            row_heights=[mode_height] + [(1 - mode_height)/(num_rows - 1)]*(num_rows - 1),
                            shared_xaxes=True, vertical_spacing=vertical_spacing,
                        shared_yaxes=sharey)
    else:
        fig = make_subplots(rows=num_rows, cols=1, shared_xaxes=True, vertical_spacing=vertical_spacing,
                            shared_yaxes=sharey)



    if mode_data is not None:
        if not isinstance(mode_data, list):
            mode_data = [mode_data]
        k = -1
        for md in mode_data:
            k += 1
            if iterate_colors:
                color = DEFAULT_PLOTLY_COLORS[k % len(DEFAULT_PLOTLY_COLORS)]
            else:
                color = DEFAULT_PLOTLY_COLORS[0]

            if names:
                trace_name = names[k]
            else:
                trace_name = str(k)

            if isinstance(md, np.ndarray):
                md = pd.DataFrame({timestamp:  np.arange(md.shape[0]), 'Mode': md})
            elif isinstance(md, pd.Series):
                md = pd.DataFrame(md)
            categories.append(md['Mode'].drop_duplicates())
            time_temp = md['Time'] if 'Time' in md else md.index
            fig.add_trace(go.Scatter(x=time_temp, y=md['Mode'], mode='markers+lines',
                                       name=trace_name, legendgroup=trace_name, line_shape='hv',
                                       marker=dict(line_color=color, color=color, line_width=2, size=marker_size),
                                       customdata=customdata, showlegend=showlegend, **kwargs), row=1, col=1)
        i = 1
    else:
        i = 0
    for col_ind in range(len(columns)):
        i += 1
        k = -1
        for trace_ind, d in enumerate(data):
            col_name = columns[col_ind]

            if names:
                trace_name = names[trace_ind]
            else:
                trace_name = str(trace_ind)

            hovertemplate = f"<b>Time:</b> %{{x}}<br><b>Event:</b> %{{y}}"
            if customdata is not None:
                hovertemplate += "<br><b>Context:</b>"
                for ind, c in enumerate(customdata.columns):
                    hovertemplate += f"<br>&nbsp;&nbsp;&nbsp;&nbsp;<em>{c}:</em> %{{customdata[{ind}]}}"

            k += 1
            if iterate_colors:
                color = DEFAULT_PLOTLY_COLORS[k % len(DEFAULT_PLOTLY_COLORS)]
            else:
                color = DEFAULT_PLOTLY_COLORS[0]

            color = f'rgba{color[3:-1]}, {str(opacity)})'
            if len(d.index.names) > 1:
                t = d.index.get_level_values(d.index.names[-1]).to_numpy()
            else:
                t = d.index.values
            if d[col_name].dtype == tuple:
                sig = d[col_name].astype(str).to_numpy()
            else:
                sig = d[col_name].to_numpy()
            if discrete:
                ind = min(limit_num_points, d.shape[0])
                if plot_only_changes:
                    ind = np.nonzero(np.not_equal(sig[0:ind - 1], sig[1:ind]))[0] + 1
                    # sig = __d[col][0:min(limit_num_points, __d.shape[0])]
                    ind = np.insert(ind, 0, 0)
                    t = t[ind]
                    sig = sig[ind]
                    if customdata is not None:
                        customdata = customdata[ind]
                else:
                    t = t[0:ind]
                    sig = sig[0:ind]
                    if customdata is not None:
                        customdata = customdata[0:ind]

                fig.add_trace(go.Scatter(x=t, y=sig, mode='markers', name=trace_name, legendgroup=trace_name,
                                           marker=dict(line_color=color, color=color, line_width=2, size=marker_size),
                                           customdata=customdata, hovertemplate=hovertemplate,
                                           showlegend=(showlegend and col_ind == 0 and mode_data is None), **kwargs),
                              row=i, col=1)
            else:
                ind = min(limit_num_points, d.shape[0])
                fig.add_trace(go.Scatter(x=t[0:ind], y=sig[0:ind], mode=mode, name=trace_name, legendgroup=trace_name,
                                           customdata=customdata,
                                           line=dict(color=color, shape='linear'),
                                           showlegend=(showlegend and col_ind == 0 and mode_data is None), **kwargs), row=i, col=1)
            fig.update_yaxes(title_text=str(col_name), row=i, col=1, title_font=dict(size=y_title_font_size),
                             categoryorder='category ascending')
        if i % 2 == 0:
            fig.update_yaxes(side="right", row=i, col=1)
        if yAxisLabelOffset == True:
            fig.update_yaxes(title_standoff=10 * i, row=i, col=1)
        if xaxis_title is not None:
            fig.update_xaxes(title=xaxis_title)
        if bounds is not None:
            upper_col = bounds[0].iloc[:, col_ind]
            lower_vol = bounds[1].iloc[:, col_ind]
            upper_bound = go.Scatter(
                name='Upper Bound',
                x=bounds[0].index.get_level_values(-1),
                y=upper_col,
                mode='lines',
                marker=dict(color="#444"),
                line=dict(width=0),
                showlegend=False)
            lower_bound = go.Scatter(
                name='Lower Bound',
                x=bounds[1].index.get_level_values(-1),
                y=lower_vol,
                marker=dict(color="#444"),
                line=dict(width=0),
                mode='lines',
                fillcolor='rgba(68, 68, 68, 0.3)',
                fill='tonexty',
                showlegend=False)
            fig.add_trace(upper_bound, row=i, col=1)
            fig.add_trace(lower_bound, row=i, col=1)

    if title is not None:
        fig.update_layout(title={'text': title, 'x': 0.5}, autosize=True, height=height + 180, showlegend=showlegend)

    if mode_data is not None:
        categories = pd.concat(categories).drop_duplicates().to_list()
        fig.update_yaxes(
            categoryorder='array',
            categoryarray=categories,
            row=1, col=1
        )

    if x_title:
        for i in reversed(range(1, 100)):
            key = f"xaxis{i if i > 1 else ''}"
            if key in fig.layout:
                fig.layout[key].title = "Time [s]"
                break
    return fig

plot_transition(self, s, d)

Plots the transition histogram between two states. Retrieves the transition data between the source state s and destination state d, and generates a Plotly figure visualizing the timing distribution of the transition. The plot includes a title, an annotation indicating the transition, and a histogram of the transition timings. Args: s: The source state identifier. d: The destination state identifier. Returns: plotly.graph_objs._figure.Figure: A Plotly Figure object containing the histogram of transition timings.

Source code in ml4cps/vis.py

def plot_transition(self, s, d):
    """
    Plots the transition histogram between two states.
    Retrieves the transition data between the source state `s` and destination state `d`,
    and generates a Plotly figure visualizing the timing distribution of the transition.
    The plot includes a title, an annotation indicating the transition, and a histogram
    of the transition timings.
    Args:
        s: The source state identifier.
        d: The destination state identifier.
    Returns:
        plotly.graph_objs._figure.Figure: A Plotly Figure object containing the histogram
        of transition timings.
    """

    trans = self.get_transition(s, d)
    titles = '{0} -> {1} -> {2}'.format(trans[0], trans[2], trans[1])
    fig = go.Figure()
    fig.update_layout(title=trans[2], font=dict(size=6))
    fig.add_annotation(
        xref="x domain",
        yref="y domain",
        x=0.5,
        y=0.9,
        text= '{} -> {}'.format(trans[0], trans[1]))
    v = trans[3]['timing']
    fig.add_trace(go.Histogram(x=[o.total_seconds() for o in v],
                               name='Timings'))
    return fig

view_graphviz(self, layout='dot', marker_size=20, node_positions=None, show_events=True, show_num_occur=False, show_state_label=True, font_size=10, plot_self_transitions=True, use_previos_node_positions=False, **kwargs)

Visualizes the internal graph structure using Graphviz and returns a pydot graph object. Parameters: layout (str): The layout algorithm to use for node positioning (default: "dot"). marker_size (int): Size of the node markers in the visualization (default: 20). node_positions (dict or None): Optional dictionary mapping node names to (x, y) positions. If None, positions are computed. show_events (bool): Whether to display event labels on transitions (default: True). show_num_occur (bool): Whether to display the number of occurrences for each transition (default: False). show_state_label (bool): Whether to display state labels on nodes (default: True). font_size (int): Font size for labels and annotations (default: 10). plot_self_transitions (bool): Whether to plot self-loop transitions (default: True). use_previos_node_positions (bool): Whether to reuse previously computed node positions (default: False). **kwargs: Additional keyword arguments for customization. Returns: pdp.Dot: A pydot graph object representing the visualized graph. Notes: - Node positions are either computed using Graphviz or taken from the provided/previous positions. - Annotations for transitions can include event names and/or occurrence counts. - The function prepares the graph for further rendering or export, but does not display it directly.

Source code in ml4cps/vis.py

def view_graphviz(self, layout="dot", marker_size=20, node_positions=None, show_events=True, show_num_occur=False,
                show_state_label=True, font_size=10, plot_self_transitions=True, use_previos_node_positions=False,
                **kwargs):
    """
    Visualizes the internal graph structure using Graphviz and returns a pydot graph object.
    Parameters:
        layout (str): The layout algorithm to use for node positioning (default: "dot").
        marker_size (int): Size of the node markers in the visualization (default: 20).
        node_positions (dict or None): Optional dictionary mapping node names to (x, y) positions. If None, positions are computed.
        show_events (bool): Whether to display event labels on transitions (default: True).
        show_num_occur (bool): Whether to display the number of occurrences for each transition (default: False).
        show_state_label (bool): Whether to display state labels on nodes (default: True).
        font_size (int): Font size for labels and annotations (default: 10).
        plot_self_transitions (bool): Whether to plot self-loop transitions (default: True).
        use_previos_node_positions (bool): Whether to reuse previously computed node positions (default: False).
        **kwargs: Additional keyword arguments for customization.
    Returns:
        pdp.Dot: A pydot graph object representing the visualized graph.
    Notes:
        - Node positions are either computed using Graphviz or taken from the provided/previous positions.
        - Annotations for transitions can include event names and/or occurrence counts.
        - The function prepares the graph for further rendering or export, but does not display it directly.
    """

    graph = None
    if node_positions is None:
        if use_previos_node_positions:
            node_positions = self.previous_node_positions
        else:
            g = self._G
            graph = pdp.graph_from_edges([('"' + tr[0] + '"', '"' + tr[1] + '"') for tr in g.edges], directed=True)
            for nnn in g.nodes:
                graph.add_node(pdp.Node(nnn, shape='point'))
            graph.set_prog('dot')
            graph = graph.create(format="dot")
            graph = pdp.graph_from_dot_data(graph)
            node_positions = {n.get_name().strip('"'): tuple(float(x) for x in n.get_pos()[1:-1].split(','))
                              for n in graph.get_nodes() if
                              n.get_name().strip('"') not in ['\\r\\n', 'node', 'graph']}
        self.previous_node_positions = node_positions
    node_x = []
    node_y = []
    for node in self._G.nodes:
        x, y = node_positions[node]
        node_x.append(x)
        node_y.append(y)
    texts = []
    for v in self._G.nodes:
        try:
            texts.append(self.print_state(v))
        except:
            texts.append('Error printing state: ')

    annotations = [dict(ax=node_positions[tr[0]][0], ay=node_positions[tr[0]][1], axref='x', ayref='y',
                        x=node_positions[tr[1]][0], y=node_positions[tr[1]][1], xref='x', yref='y',
                        showarrow=True, arrowhead=1, arrowsize=2) for tr in self._G.edges]
    def fun(tr):
        if show_events and show_num_occur:
            return '<i>{} ({})</i>'.format(tr[2], self.num_occur(tr[0], tr[2]))
        elif show_events:
            return '<i>{}</i>'.format(tr[2])
        elif show_num_occur:
            return '<i>{}</i>'.format(self.num_occur(tr[0], tr[2]))

    if show_num_occur or show_events:
        annotations_text = [dict(x=(0.4 * node_positions[tr[0]][0] + 0.6 * node_positions[tr[1]][0]),
                                 y=(0.4 * node_positions[tr[0]][1] + 0.6 * node_positions[tr[1]][1]),
                                 xref='x', yref='y', text=fun(tr), font=dict(size=font_size, color='darkblue'),
                                 yshift=0, showarrow=False)
                            for tr in self.get_transitions() if plot_self_transitions or tr[0] != tr[1]]

        annotations += annotations_text

    graph = pdp.Dot(graph_type='digraph')
    for tr in self._G.edges:
        graph.add_edge(pdp.Edge('"' + tr[0] + '"', '"' + tr[1] + '"', label=tr[2]))
    for nnn in self._G.nodes:
        graph.add_node(pdp.Node(nnn, shape='box'))
    return graph

tools

Various methods to transform data.

binary_ordinal_encode(column, order)

Encodes a pandas Series with binary ordinal encoding based on the specified order.

Parameters:

Name	Type	Description	Default
column	Series	The column to encode.	required
order	list	The ordered list of unique values in the column.	required

Returns:

Type	Description
	pd.DataFrame: The binary ordinal encoded DataFrame for the given column.

Source code in ml4cps/tools.py

def binary_ordinal_encode(column, order):
    """
    Encodes a pandas Series with binary ordinal encoding based on the specified order.

    Args:
        column (pd.Series): The column to encode.
        order (list): The ordered list of unique values in the column.

    Returns:
        pd.DataFrame: The binary ordinal encoded DataFrame for the given column.
    """
    num_levels = len(order)
    num_bits = num_levels.bit_length()

    # Create a dictionary mapping each level to its binary representation
    encoding_map = {value: list(map(int, format(i, f'0{num_bits}b'))) for i, value in enumerate(order)}

    # Apply the encoding to the column
    encoded_df = pd.DataFrame(column.map(encoding_map).tolist(),
                              columns=[f"{column.name}_bit_{i}" for i in range(num_bits)])
    return encoded_df

generate_random_walk(start_values, steps=100)

Generates a random walk process for multiple variables.

Parameters: - start_values (list): A list of starting values for each variable. - steps (int): Number of steps in the random walk.

Returns: - pd.DataFrame: DataFrame containing the random walk process for each variable.

Source code in ml4cps/tools.py

def generate_random_walk(start_values, steps=100):
    """
    Generates a random walk process for multiple variables.

    Parameters:
    - start_values (list): A list of starting values for each variable.
    - steps (int): Number of steps in the random walk.

    Returns:
    - pd.DataFrame: DataFrame containing the random walk process for each variable.
    """
    num_variables = len(start_values)

    # Generate random steps (Normal distribution with mean=0, std=1)
    random_steps = np.random.normal(loc=0, scale=1, size=(steps, num_variables))

    # Initialize the DataFrame with the start values
    random_walk = pd.DataFrame([start_values], columns=[f'Var_{i + 1}' for i in range(num_variables)])

    # Generate the random walk by cumulative sum of the random steps
    for i in range(steps):
        new_row = random_walk.iloc[-1].values + random_steps[i]
        new_row_df = pd.DataFrame([new_row], columns=random_walk.columns)
        random_walk = pd.concat([random_walk, new_row_df], ignore_index=True)

    return random_walk

remove_timestamps_without_change(data, sig_names=None)

Removes timestamps where no values changed in comparison to the previous timestamp.

Source code in ml4cps/tools.py

def remove_timestamps_without_change(data, sig_names=None):
    """Removes timestamps where no values changed in comparison to the previous timestamp."""
    if type(data) is list:
        return [remove_timestamps_without_change(d, sig_names=sig_names) for d in data]
    else:
        if type(data) is not pd.DataFrame:
            data = pd.DataFrame(data)
        if sig_names is None:
            sig = data.columns
        else:
            sig = sig_names
        ind = (data[sig] != data[sig].shift(1)).any(axis=1)
        dd = data.loc[ind]
        return dd.copy(deep=True)