`h1flow.h1flow`

Module Contents

Classes

Graph

A Graph is itself a NodeContainable, meaning it can be enclosed within a Node,

class h1flow.h1flow.Graph(node_validation_schema_name='NODE_VALIDATION_SCHEMA')

Bases: h1flow.h1step_containable.NodeContainable, h1st.trust.trustable.Trustable

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

property nodes(self) → types.SimpleNamespace: Gets all nodes of the graph in unspecified orders Gets a specific node by ID: nodes.<ID>

start(self) → 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(self, node: Union[h1flow.h1step.Node, h1flow.h1step_containable.NodeContainable, None] = None, yes: Union[h1flow.h1step.Node, h1flow.h1step_containable.NodeContainable, None] = None, no: Union[h1flow.h1step.Node, h1flow.h1step_containable.NodeContainable, None] = None, id: str = None) → Union[h1flow.h1step.Node, List[h1flow.h1step.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.core as h1
from h1st.core 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(self) → 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(self, 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.core as h1
from h1st.core 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(self, data) → Any: A shortcut function for the “execute” function with command=”predict”

visualize(self): Visualizes the flowchart for this graph

describe(self)

Returns a description of the model’s behavior and properties based on Who’s asking for what.

Parameters:
constituent : Constituency: The Constituency asking for the explanation Who aspect : The Aspect of the question. What

Returns:
out : Description of Model’s behavior and properties

explain(self)

Returns an explanation for a decision made by the Model based on Who’s asking and why.

Parameters:
constituent : Constituency: The Constituency asking for the explanation aspect : Aspect: The Aspect of the question (e.g., Accountable, Functional, Operational) decision : array-like: the input data of the decision to be explained

Returns:
out : Specific decision explanation (e.g., SHAP or LIME)

_wrap_and_add(self, node: Union[h1flow.h1step.Node, h1flow.h1step_containable.NodeContainable], id: str = None)

Wraps NodeContainable to a Node if needed. Override node’s id if an id provided. Then adds the node to the graph.

Parameters

node – a Node or ContainableNode
id – id for the node

Returns

the newly added node

_generate_id(self, node: h1flow.h1step.Node): Automatically generates an id for the node. If multiple instances of the same NodeContainable type are added, the ids of the node will be ClassName, ClassName2, ClassName3, etc

_connect_nodes(self, from_: h1flow.h1step.Node, to: h1flow.h1step.Node, edge_label=None) → NoReturn

Connects “from_” node to the “to” node

Parameters

from – the source node
to – the destination node

Edge_label

the label for the edge between from_ and to

_execute_one(self, command: str, data: Dict) → Dict

Executes the graph exactly 1 time

Parameters

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

Returns

result as a dictionary

_add_and_connect(self, node: Union[h1flow.h1step.Node, h1flow.h1step_containable.NodeContainable, None] = None, yes: Union[h1flow.h1step.Node, h1flow.h1step_containable.NodeContainable, None] = None, no: Union[h1flow.h1step.Node, h1flow.h1step_containable.NodeContainable, None] = None, id: str = None, from_: Union[h1flow.h1step.Node, None] = None) → Union[h1flow.h1step.Node, List[h1flow.h1step.Node]]: Adds node/yes/no nodes to self.nodes and connect from_node to newly added nodes

h1flow.h1flow

Module Contents

Classes

`h1flow.h1flow`