h1st.core.graph module

class h1st.core.graph.Graph(node_validation_schema_name='NODE_VALIDATION_SCHEMA')

Bases: h1st.core.node_containable.NodeContainable

A Graph is itself a NodeContainable, meaning it can be enclosed within a Node, forming a hierarchy of Graphs

__init__(node_validation_schema_name='NODE_VALIDATION_SCHEMA')

Parameters

node_validation_schema_name – name of the variable in config.py to load schema definition fornode_validation_schema_name

property nodes

Gets all nodes of the graph in unspecified orders Gets a specific node by ID: nodes.<ID>

start() → h1st.core.graph.Graph

Initial action to begin adding nodes to a (fresh) Graph. A node with the id=’start’ will be automatically added to the Graph.

add(node: Optional[Union[h1st.core.node.Node, h1st.core.node_containable.NodeContainable]] = None, yes: Optional[Union[h1st.core.node.Node, h1st.core.node_containable.NodeContainable]] = None, no: Optional[Union[h1st.core.node.Node, h1st.core.node_containable.NodeContainable]] = None, id: str = None) → Union[h1st.core.node.Node, List[h1st.core.node.Node]]

Adds a new Node or NodeContainable to this graph. Period keeps a running preference to the current possition in the graph to be added If the object to be added is a NodeContainable then a new node will be automatically instanciated to contain that object and the node is added to this graph. The new node’s id can be specified or automatically inferred from the NodeContainable’s type.

Parameters

• node – Node or NodeContainable object to be added to the graph

• yes/no – Node or NodeContable object to be added to the graph following a conditional (Decision) node

• from – the node to which the new node will be connected

• id – the id of the new node

Returns

new added node if adding a single node Or new added node for yes branch if adding yes node only (without no node) following a condition node Or new added node for no branch if adding no node only (without yes node) following a condition node Or [new added node for yes branch, new added node for no branch] if adding both yes & no nodes following a condition node

Cyber Security example to handle injection and replacement attacks

import h1st as h1
from h1st import NodeContainable, Decision

class MyGraph(h1.Graph):
    def __init__(self):
        super().__init__()

        imc, rec = self
            .start()
            .add(GenerateWindowEvents())
            .add(Decision(InjectionEventClassifier()))
            .add(
                yes=InjectionMessageClassifier(),
                no=Decision(ReplacementEventClassifier())
            )

        rec.add(
            yes=ReplacementMessageClassifier(),
            no=BuildNormalResult()
        )

        self.end()

end() → h1st.core.graph.Graph

This method is required after adding all nodes to the graph. The end node with id=’end’ will be automatically added to the graph. All leaf nodes (without outgoing edges) will be automatically connected to the end node. Consolidate ids for nodes using the same NodeContainable type without provided id. Ids will be Xyz1, Xyz2, … with class Xyz inherits from NodeContainable

execute(command: str, data: Union[Dict, List[Dict]]) → Union[Dict, List[Dict]]

The graph will scan through nodes to invoke appropriate node’s function with name = value of command parameter. Everytime the graph invokes the appropreate function of the node, it will passing an accumulated dictionary as the input and merge result of the function into the accumulated dictionary.

Parameters

• command – for Node or NodeContainable object to decide which function will be invoked during executing the graph

• data – input data to execute. if data is a dictionary, the graph will execute one. if data is a list of dictionary, the graph will execute multiple time

Returns

single dictionary if the input is a single dictionary Or list of dictionary if the input is a list of dictionary

Example graph for Cyber Security and how to execute the graph

import h1st as h1
from h1st import NodeContainable, Decision

class MyGraph(h1.Graph):
    def __init__(self):
        imc, rec = self
            .start()
            .add(GenerateWindowEvents())
            .add(Decision(InjectionEventClassifier()))
            .add(
                yes=InjectionMessageClassifier(),
                no=Decision(ReplacementEventClassifier())
            )

        rec.add(
            yes=ReplacementMessageClassifier(),
            no=BuildNormalResult()
        )

        self.end()

g = MyGraph()
result = g.execute(command='predict', data={'df': my_dataframe})

predict(data) → Any

A shortcut function for the “execute” function with command=”predict”

visualize()

Visualizes the flowchart for this graph

describe()

explain()