Initial app commit

.gitignore

@@ -0,0 +1,30 @@
+# Compiled class file
+*.class
+
+# Log file
+*.log
+
+# BlueJ files
+*.ctxt
+
+# Mobile Tools for Java (J2ME)
+.mtj.tmp/
+
+# Package Files #
+*.jar
+*.war
+*.nar
+*.ear
+*.zip
+*.tar.gz
+*.rar
+
+# virtual machine crash logs, see http://www.java.com/en/download/help/error_hotspot.xml
+hs_err_pid*
+
+# VSCode files
+*.code-workspace
+.editorconfig
+
+# Macintosh filesystem
+.DS_Store

Inspect_Logic.py

@@ -0,0 +1,101 @@
+from io import StringIO
+import streamlit as st
+
+from smem_token_parser import SMEM_Parser, read_tokens_from_lines
+from smem_obj import SMEM_Obj, ObjType
+
+MIN_COLS = 4
+
+# Set up page config before any other streamlit commands
+st.set_page_config(page_title="Soar Agent Memory Inspector", layout="wide")
+st.title("Logic Inspector")
+
+
+def get_smem_root_from_file(smem_file):
+    # tokens = get_smem_tokens_from_local_file(smem_filename)
+    tokens = read_tokens_from_lines(smem_file)
+    if tokens == None:
+        st.error("Error reading file: '"+str(smem_file)+"'")
+        return None
+    parser = SMEM_Parser()
+    parser.parse_file(tokens)
+    return parser.get_context_root()
+
+if "col_obj_list" not in st.session_state:
+    st.session_state["col_obj_list"] = None
+
+file_upload_expander = st.expander(label="Select a file to inspect", expanded=(st.session_state.col_obj_list == None))
+file = file_upload_expander.file_uploader(" ")
+
+if file is not None:
+    if st.session_state.col_obj_list is None:
+        root = get_smem_root_from_file(StringIO(file.getvalue().decode("utf-8")))
+        if root:
+            st.session_state["col_obj_list"] = [None]*MIN_COLS
+            st.session_state.col_obj_list[0] = root
+            st.experimental_rerun()
+else:
+    st.session_state["col_obj_list"] = None
+
+# inpath = "/Users/bstearn1/Documents/Projects/OptumGitHub/Intelligent-Solutions/AICarePlan/agent/D-RULES_082222_smem.soar"
+# inpath = "SMEM Generator/test_pim_smem.soar"
+# st.session_state.col_obj_list[0] = get_smem_root_from_file(inpath)
+
+if st.session_state.col_obj_list is None:
+    st.stop()
+
+def add_col(index, obj):
+    st.session_state.col_obj_list = st.session_state.col_obj_list[:index+1]
+    st.session_state.col_obj_list.append(obj)
+    # st.session_state.current_tab_index = index+1
+    st.experimental_rerun()
+
+def get_header_str(obj_type):
+    if obj_type == ObjType.CONTEXT:
+        return "START"
+    elif obj_type == ObjType.COND_CONJ:
+        return "CONDITION"
+    elif obj_type == ObjType.OP:
+        return "CHOICE"
+    elif obj_type == ObjType.ACT_GROUP:
+        return "RESULT"
+    else:
+        return " "
+
+st.subheader("Click the buttons below to select conditions and to inspect resulting actions.")
+
+cols = st.columns(max(MIN_COLS,len(st.session_state.col_obj_list)))
+
+for i,col in enumerate(cols):
+    try:
+        if st.session_state.col_obj_list[i] == None:
+            break
+    except Exception as e:
+        # print("ERROR checking column "+str(i+1)+" of "+str(len(st.session_state.col_obj_list))+": "+str(e))
+        break
+    # Build the available objects to navigate under this col's object
+    obj = st.session_state.col_obj_list[i]
+    obj_str,obj_label,obj_desc = obj.to_string()
+    sub_objs = list(obj.get_child_objects(compact=True))
+    # print(obj.obj_type)
+    if len(sub_objs) > 0:
+        col.markdown("**"+get_header_str(sub_objs[0].obj_type)+"**",)
+    col.markdown("---")
+    # Show the object's main title and description
+    col.text(obj_str)
+    if obj_desc != None:
+        col.markdown("*"+obj_desc+"*")
+    for j,sub in enumerate(sub_objs):
+        sub_str,sub_label,sub_desc = sub.to_string()
+        if sub.obj_type == ObjType.OP:
+            subsub = list(sub.get_child_objects(compact=True))[0]
+            _,group_string,_ = subsub.to_string()
+            group_string = "\n * "+str(group_string).replace("AND", "\n * ")
+            col.markdown(group_string)
+            if col.button(sub_label, key="button"+str(i)+"-"+str(j)):
+                add_col(i,sub)
+        elif sub.obj_type == ObjType.ACT_GROUP:
+            col.markdown("*Output Details:* ")
+            col.markdown("\n * "+str(sub_str).replace(") AND", ")\n * "))
+        elif col.button(sub_label, key="button"+str(i)+"-"+str(j)):
+            add_col(i,sub)

README.md

@@ -5,7 +5,7 @@ colorFrom: indigo
 colorTo: yellow
 sdk: streamlit
 sdk_version: 1.15.2
-app_file: app.py
+app_file: Inspect_Logic.py
 pinned: false
 ---
 

smem_obj.py

@@ -0,0 +1,276 @@
+from enum import Enum
+
+class SMEM_Obj():
+    def __init__(self):
+        self.processed = False
+        self.obj_type = None
+        self.id_var = None
+        self.class_str = None
+        self.wme_list = []
+        self.description = None
+
+    def set_id_var(self, var):
+        self.id_var = var
+    
+    def add_wme(self, attr, val):
+        self.wme_list.append((attr,val))
+
+    def get_class(self):
+        for (attr,val) in self.wme_list:
+            if attr == "^class":
+                self.class_str = val
+                break
+        return self.class_str
+
+    def get_description(self):
+        if self.description != None:
+            return self.description
+        # Search for the description
+        for (attr,val) in self.wme_list:
+            if attr == "^description":
+                self.description = val
+                break
+        return self.description
+
+    def get_compact_printable_object(self):
+        """ Return this object if printable in compact mode, else return the set of next printable descendants.
+        """
+        if self.obj_type == None:
+            print("Warning: No obj_type for obj '"+self.id_var+"' in get_compact_printable_object().")
+            return None
+        retval = set()
+        returnSelf = False
+
+        if self.obj_type == ObjType.COND_PRIM:
+            for (attr,val) in self.wme_list:
+                if attr == "^supplies-condition":
+                    val.obj_type = ObjType.COND_CONJ
+                    retval.update(val.get_compact_printable_object())
+        elif self.obj_type == ObjType.COND_CONJ:
+            for (attr,val) in self.wme_list:
+                if attr == "^satisfies-op":
+                    val.obj_type = ObjType.OP
+                    returnSelf = True
+                    break
+                elif attr == "^activates-context":
+                    val.obj_type = ObjType.CONTEXT
+                    returnSelf = True
+                    break
+                elif attr == "^supplies-condition":
+                    val.obj_type = ObjType.COND_CONJ
+                    retval.update(val.get_compact_printable_object())
+        elif self.obj_type == ObjType.CONTEXT:
+            for (attr,val) in self.wme_list:
+                if attr == "^primes-condition":
+                    val.obj_type = ObjType.COND_PRIM
+                    retval.update(val.get_compact_printable_object())
+        else:
+            returnSelf = True
+        
+        if returnSelf:
+            retval = set()
+            retval.add(self)
+        
+        return retval
+
+    def get_child_objects(self, compact=True):
+        """ If compact=True, skip condition prims and contexts, only get conjunctions and action groups
+        """
+        if self.obj_type == None:
+            print("Warning: No obj_type for obj '"+self.id_var+"'")
+            return None
+        retval = set()
+        for (attr,val) in self.wme_list:
+            if self.obj_type == ObjType.CONTEXT:
+                if attr == "^primes-condition":
+                    val.obj_type = ObjType.COND_PRIM
+                    if compact:
+                        retval.update(val.get_compact_printable_object())
+                    else:
+                        retval.add(val)
+            elif self.obj_type == ObjType.COND_PRIM:
+                if attr == "^supplies-condition":
+                    val.obj_type = ObjType.COND_CONJ
+                    if compact:
+                        retval.update(val.get_compact_printable_object())
+                    else:
+                        retval.add(val)
+            elif self.obj_type == ObjType.COND_CONJ:
+                if attr == "^satisfies-op":
+                    val.obj_type = ObjType.OP
+                    retval.add(val)
+                elif attr == "^activates-context":
+                    val.obj_type = ObjType.CONTEXT
+                    if compact:
+                        retval.update(val.get_compact_printable_object())
+                    else:
+                        retval.add(val)
+                elif attr == "^supplies-condition":
+                    val.obj_type = ObjType.COND_CONJ
+                    if compact:
+                        retval.update(val.get_compact_printable_object())
+                    else:
+                        retval.add(val)
+            elif self.obj_type == ObjType.OP:
+                if attr == "^activates-action":
+                    val.obj_type = ObjType.ACT_GROUP
+                    retval.add(val)
+            elif self.obj_type == ObjType.ACT_GROUP:
+                if attr == "^action":
+                    val.obj_type = ObjType.ACT_PRIM
+                    retval.add(val)
+        return retval
+
+    def get_prim_cond_string(self, negate=False):
+        if len(self.wme_list) == 0:
+            return "", ""
+        if negate:
+            op_join_str = " is not "
+        else:
+            op_join_str = " is "
+        if self.class_str == None:
+            self.get_class()
+        
+        retval_long, retval_short, retval_desc = ["(", "(", None]
+        if self.class_str == "prim":
+            group_str, attr_str, val_str = ["","",""]
+            for (attr,val) in self.wme_list:
+                if attr == "^group":
+                    group_str = val
+                elif attr == "^attribute":
+                    attr_str = val
+                elif attr == "^value":
+                    val_str = val
+                elif attr == "^description":
+                    self.description = val
+                    retval_desc = val
+            retval_long += group_str+"."+attr_str+op_join_str+val_str
+            retval_short += attr_str+op_join_str+val_str
+        return retval_long+")", retval_short+")", retval_desc
+
+    def get_cond_string(self):
+        retval_long, retval_short, retval_desc = ["","",None]
+        # Compile the parent prim descriptions
+        parent_conds = []
+        # child_conjs = []
+        is_negation = (self.get_class() == "negation")
+        for (attr,val) in self.wme_list:
+            if attr == "^receives-condition":
+                parent_conds.append(val)
+            if attr == "^description":
+                self.description = val
+                retval_desc = val
+            # if attr == "^supplies-condition":
+            #     child_conjs.append(val)
+        
+        for i,cond in enumerate(parent_conds):
+            if i > 0:
+                retval_long += " \nAND "
+                retval_short += " AND "
+            if cond.obj_type == ObjType.COND_CONJ:
+                long, short, _ = cond.get_cond_string()
+            else:
+                long, short, _ = cond.get_prim_cond_string(negate=is_negation)
+            retval_long += long
+            retval_short += short
+        return retval_long, retval_short, retval_desc
+    
+    def get_op_string(self):
+        retval_long, retval_short, retval_desc = ["", "", None]
+        numeric_pref = None
+        for (attr,val) in self.wme_list:
+            if attr == "^name":
+                retval_long = val
+                retval_short = val
+            elif attr == "^numeric-preference":
+                try:
+                    numeric_pref = str(float(val)*100)+"%"
+                except Exception as e:
+                    print("ERROR in get_op_string() converting numeric_pref to string: "+str(e))
+                    numeric_pref = None
+            elif attr == "^description":
+                self.description = val
+                retval_desc = val
+        if numeric_pref:
+            retval_long += "\nPercentage: "+numeric_pref
+        # print(self.id_var+": "+retval)
+        return retval_long, retval_short, retval_desc
+
+    def get_prim_act_string(self):
+        if len(self.wme_list) == 0:
+            return ""
+        if self.class_str == None:
+            self.get_class()
+        retval_long = "("
+        retval_short = "("
+        if self.class_str == "prim":
+            type_str, attr_str, val_str = ["","",""]
+            for (attr,val) in self.wme_list:
+                if attr == "^type":
+                    type_str = val
+                elif attr == "^attribute":
+                    attr_str = val
+                elif attr == "^value":
+                    val_str = val
+            retval_long += type_str+"."+attr_str+" = "+val_str
+            retval_short += attr_str+" = "+val_str
+        elif self.class_str == "prim-message":
+            type_str, flag_str, msg_str = ["","",""]
+            for (attr,val) in self.wme_list:
+                if attr == "^type":
+                    type_str = val
+                elif attr == "^flag":
+                    flag_str = val
+                elif attr == "^message":
+                    msg_str = val
+            retval_long += type_str+".flag = "+flag_str+" AND "+type_str+'.message = "'+msg_str+'"'
+            retval_short += 'message: "'+msg_str+'"'
+        return retval_long+")", retval_short+")"
+    
+    def get_act_group_string(self):
+        retval_long, retval_short, retval_desc = ["", "", None]
+        # Compile the child prim action descriptions
+        child_prims = []
+        for (attr,val) in self.wme_list:
+            if attr == "^action":
+                child_prims.append(val)
+            elif attr == "^description":
+                self.description = val
+                retval_desc = val
+        for i,prim in enumerate(child_prims):
+            if i > 0:
+                retval_long += " AND "
+                retval_short += " AND "
+            long, short = prim.get_prim_act_string()
+            retval_long += long
+            retval_short += short
+        return retval_long, retval_short, retval_desc
+
+    def to_string(self):
+        long = self.id_var
+        short = self.id_var
+        description = None
+
+        if self.obj_type == ObjType.CONTEXT:
+            return "", "", self.get_description()
+        if self.obj_type == ObjType.COND_PRIM:
+            long, short, description = self.get_prim_cond_string()
+            # long = "IF "+long
+            # short = "IF "+short
+        if self.obj_type == ObjType.COND_CONJ:
+            long, short, description = self.get_cond_string()
+        if self.obj_type == ObjType.OP:
+            long, short, description = self.get_op_string()
+        if self.obj_type == ObjType.ACT_GROUP:
+            long, short, description = self.get_act_group_string()
+        if self.obj_type == ObjType.ACT_PRIM:
+            long, short = self.get_prim_act_string()
+        return long, short, description
+
+class ObjType(Enum):
+    CONTEXT = 0
+    COND_PRIM = 1
+    COND_CONJ = 2
+    OP = 3
+    ACT_GROUP = 4
+    ACT_PRIM = 5

smem_token_parser.py

@@ -0,0 +1,140 @@
+import sys
+from os import access, R_OK
+from os.path import isfile
+from collections import defaultdict
+import re
+
+from smem_obj import SMEM_Obj, ObjType
+
+
+def clean_smem_string_token(s):
+    return str(s).replace("|","")
+
+def removeComments(s):
+    """ Trim any '#' content from the given string
+    """
+    ind = str(s).find('#')
+    if (ind == -1):
+        return s
+    return s[:ind]
+
+def read_tokens_from_lines(file):
+    retval = []             # A list of all tokens from the "smem --add {}" contents.
+    inSmemAdd = False       # Whether the read is inside an "smem --add {}" command
+
+    try:
+        for line in file:
+            sline = line.lstrip()
+            # Don't use comments
+            if sline.startswith('#'):
+                continue
+            # Check if we're in an smem --add command
+            if not inSmemAdd:
+                if sline.startswith("smem --add {"):
+                    inSmemAdd = True
+                    # Only add from this line any content after the opening '{'
+                    sline = sline[12:]
+                else:
+                    continue
+            
+            # Get the tokens: split on whitespace, pipe quotes, parentheses, and brackets, but only exclude whitespace delimiters and pipe quotes
+            regex_pattern = '[\s+]|\\|(.+?)\\||(?<=\\))|(?<=\\()|(?<=\\{)|(?<=\\})'
+            tokens = re.split(regex_pattern, removeComments(sline).rstrip())
+
+            while None in tokens:
+                tokens.remove(None)
+            while '' in tokens:
+                tokens.remove('')
+
+            # Check for the closing character
+            try:
+                ind = tokens.index('}')
+                # If no exception, it was found
+                tokens = tokens[:ind]
+                inSmemAdd = False
+            except:
+                pass
+
+            # Add this line to the return list
+            if len(tokens) > 0:
+                retval.extend(tokens)
+    except Exception as e:
+        print("ERROR extracting tokens from the given file: "+str(e), file=sys.stderr)
+        return None
+    
+    return retval
+
+""" 
+This method scans a file that holds an 'smem --add{}' command and returns the relevant tokens from that file
+"""
+def get_smem_tokens_from_local_file(filename):
+    # Error check for reading the file
+    if not isfile(filename):
+        print("ERROR in get_smem_tokens_from_local_file(): File does not exist: '"+str(filename)+"'.", file=sys.stderr)
+        return None
+    if not access(filename, R_OK):
+        print("ERROR in get_smem_tokens_from_local_file(): File is not readable: '"+str(filename)+"'.", file=sys.stderr)
+        return None
+
+    # Get the file content
+    with open(filename) as file:
+        retval = read_tokens_from_lines(file)
+
+    # All done
+    return retval
+
+
+class SMEM_Parser():
+    
+    def __init__(self):
+        # Init the data structured needed to parse the smem file
+        self.smem_var_obj_map = defaultdict(SMEM_Obj)
+
+    def parse_file(self, smem_tokens):
+        # Iterate through the tokens
+        current_obj = None
+        current_attr = None
+        isNextWMEId = False
+        for token in smem_tokens:
+            # Skip empty tokens
+            if len(token) == 0:
+                continue
+            # Get WME ID
+            if isNextWMEId:
+                current_obj = self.smem_var_obj_map[token]
+                current_obj.set_id_var(token)
+                isNextWMEId = False
+                continue
+            # Get start of obj
+            if token == '(':
+                isNextWMEId = True
+                continue
+            # Get end of obj
+            if token == ')':
+                current_obj = None
+                current_attr = None
+                continue
+            # Get attributes
+            if token.startswith('^'):
+                current_attr = token
+                continue
+            # Get values
+            if current_obj != None and current_attr != None:
+                # Add the WME to the current object
+                if token.startswith('<'):
+                    token_val = self.smem_var_obj_map[token]
+                    token_val.set_id_var(token)
+                else:
+                    token_val = clean_smem_string_token(token)
+                current_obj.add_wme(current_attr, token_val)
+            else:
+                print("ERROR: Unexpected token '"+token+"'.")
+                return
+
+    def get_context_root(self):
+        for var,obj in self.smem_var_obj_map.items():
+            for (attr,val) in obj.wme_list:
+                if attr == "^context-root":
+                    obj.obj_type = ObjType.CONTEXT
+                    return obj
+        return None