expose enums in client.py

absbox/client.py

@@ -49,6 +49,20 @@ class RunResp(int, enum.Enum):
     LogResp = 2
 
 
+class MsgColor(str, enum.Enum):
+    Warning = "[bold yellow]"
+    Error = "[bold red]"
+    Success = "[bold green]"
+    Info = "[bold magenta]"
+
+
+class LibraryEndpoints(str, enum.Enum):
+    Token = "token"
+    Query = "query"
+    List = "list"
+    Run = "run"
+
+
 @dataclass
 class API:
     url: str
@@ -63,20 +77,20 @@ class API:
 
     def __post_init__(self):
         self.url = self.url.rstrip("/")
-        with console.status(f"[magenta]Connecting engine server -> {self.url}") as status:
+        with console.status(f"{MsgColor.Info.value}Connecting engine server -> {self.url}") as status:
             try:
                 _r = requests.get(f"{self.url}/{Endpoints.Version.value}", verify=False, timeout=5).text
             except (ConnectionRefusedError, ConnectionError):
-                console.print(f"❌[bold red]Error: Can't not connect to API server {self.url}")
+                console.print(f"❌{MsgColor.Error.value}Error: Can't not connect to API server {self.url}")
                 self.url = None
                 return
 
             self.server_info = self.server_info | json.loads(_r)
             engine_version = self.server_info['_version'].split(".")
             if self.check and (self.version[1] != engine_version[1]):
-                console.print(f"❌[bold red]Failed to init the api instance, lib support={self.version} but server version={self.server_info['_version']} , pls upgrade your api package by: pip -U absbox")
+                console.print(f"❌{MsgColor.Error.value}Failed to init the api instance, lib support={self.version} but server version={self.server_info['_version']} , pls upgrade your api package by: pip -U absbox")
                 return
-        console.print(f"✅[bold green]Connected, local lib:{'.'.join(self.version)}, server:{'.'.join(engine_version)}")
+        console.print(f"✅{MsgColor.Success.value}Connected, local lib:{'.'.join(self.version)}, server:{'.'.join(engine_version)}")
         self.session = requests.Session() 
 
     def build_run_deal_req(self, run_type, deal, perfAssump=None, nonPerfAssump=[]) -> str:
@@ -147,10 +161,10 @@ def run(self, deal,
 
         rawWarnMsg = []
         if multi_run_flag:
-            rawWarnMsgByScen = {k: [f"⚠[bold yellow]{_['contents']}" for _ in filter_by_tags(v[RunResp.LogResp.value],[ValidationMsg.Warning.value])] for k, v in result.items()}
+            rawWarnMsgByScen = {k: [f"⚠{MsgColor.Warning.value}{_['contents']}" for _ in filter_by_tags(v[RunResp.LogResp.value],[ValidationMsg.Warning.value])] for k, v in result.items()}
             rawWarnMsg = [b for a in rawWarnMsgByScen.values() for b in a]
         else:
-            rawWarnMsg = [f"⚠[bold yellow]{_['contents']}" for _ in filter_by_tags(result[RunResp.LogResp.value], [ValidationMsg.Warning.value])]
+            rawWarnMsg = [f"⚠{MsgColor.Warning.value}{_['contents']}" for _ in filter_by_tags(result[RunResp.LogResp.value], [ValidationMsg.Warning.value])]
 
         if rawWarnMsg and showWarning:
             rich.print("Warning Message from server:\n"+"\n".join(rawWarnMsg))
@@ -175,7 +189,7 @@ def read_single(pool_resp):
                 else:
                     result = pd.DataFrame([_['contents'][-1]+_['contents'][-1] for _ in pool_flow], columns=flow_header)
             except ValueError as e:
-                console.print(f"❌[bold red]Failed to match header:{flow_header} with {result[0]['contents']}")
+                console.print(f"❌{MsgColor.Error.value}Failed to match header:{flow_header} with {result[0]['contents']}")
                 console.print(f"error:{e}")
 
             result = result.set_index(idx)
@@ -240,34 +254,34 @@ def readResult(x):
             return result
 
     def loginLibrary(self, user, pw, **q):
-        deal_library_url = q['deal_library']+"/token"
+        deal_library_url = q['deal_library']+f"/{LibraryEndpoints.Token.value}"
         cred = {"user": user, "password": pw}
         r = self._send_req(json.dumps(cred), deal_library_url)
         if 'token' in r:
-            console.print(f"✅[bold green] login successfully,{r['msg']}")
+            console.print(f"✅{MsgColor.Success.value} login successfully,{r['msg']}")
             self.token = r['token']
         else:
             if hasattr(self, 'token'):
                 delattr(self, 'token')
-            console.print(f"❌[bold red]Failed to login,{r['msg']}")
+            console.print(f"❌{MsgColor.Error.value} Failed to login,{r['msg']}")
             return None
 
     def safeLogin(self, user, **q):
         try:
             pw = getpass.getpass()
             self.loginLibrary(user, pw, **q)
         except Exception as e:
-            console.print(f"❌[bold red]{e}")
+            console.print(f"❌{MsgColor.Error.value}{e}")
 
     def queryLibrary(self, ks, **q):
         if not hasattr(self, "token"):
-            console.print(f"❌[bold red] No token found , please call loginLibrary() to login")
+            console.print(f"❌{MsgColor.Error.value} No token found , please call loginLibrary() to login")
             return 
-        deal_library_url = q['deal_library']+"/query"
+        deal_library_url = q['deal_library']+f"/{LibraryEndpoints.Query.value}"
         d = {"bond_id": [k for k in ks]}
         q = {"read": True} | q
         result = self._send_req(json.dumps(d|q), deal_library_url, headers={"Authorization": f"Bearer {self.token}"})
-        console.print(f"✅[bold green] query success")
+        console.print(f"✅{MsgColor.Success.value} query success")
         if q['read']:
             if 'data' in result:
                 return pd.DataFrame(result['data'], columns=result['header'])
@@ -277,23 +291,23 @@ def queryLibrary(self, ks, **q):
             return result
 
     def listLibrary(self, **q):
-        deal_library_url = q['deal_library']+"/list"
+        deal_library_url = q['deal_library']+f"/{LibraryEndpoints.List.value}"
         result = self._send_req(json.dumps(q), deal_library_url)
-        console.print(f"✅[bold green]list success")
+        console.print(f"✅{MsgColor.Success.value}list success")
         if ('read' in q) and (q['read'] == True):
             return pd.DataFrame(result['data'], columns=result['header'])
         else:
             return result
 
     def runLibrary(self, _id, **p):
-        deal_library_url = p['deal_library']+"/run"
+        deal_library_url = p['deal_library']+f"/{LibraryEndpoints.Run.value}"
         read = p.get("read", True)
         pricingAssump = p.get("pricing", None)
         dealAssump = p.get("assump", None)
-        prod_flag = {"production":p.get("production", True)}
+        prod_flag = {"production": p.get("production", True)}
         runReq = mergeStrWithDict(self.build_req(_id, dealAssump, pricingAssump), prod_flag)
         if not hasattr(self, "token"):
-            console.print(f"❌[bold red] No token found , please call loginLibrary() to login")
+            console.print(f"❌{MsgColor.Error.value} No token found , please call loginLibrary() to login")
             return 
         result = self._send_req(runReq, deal_library_url, headers={"Authorization": f"Bearer {self.token}"})
 
@@ -308,9 +322,9 @@ def lookupReader(x):
         try:
             ri = result['run_info']
             result = result['run_result']
-            console.print(f"✅[bold green]run success with deal id={ri['deal_id']}/report num={ri['report_num']},doc_id={ri['doc_id']}")
+            console.print(f"✅{MsgColor.Success.value}run success with deal id={ri['deal_id']}/report num={ri['report_num']},doc_id={ri['doc_id']}")
         except Exception as e:
-            console.print(f"❌[bold red]message from API server:{result}")
+            console.print(f"❌{MsgColor.Error.value} message from API server:{result},\n,{e}")
             return None
         try:       
             classReader = lookupReader(p['reader'])
@@ -321,7 +335,7 @@ def lookupReader(x):
             else:
                 return result
         except Exception as e:
-            console.print(f"❌[bold red]{e}")
+            console.print(f"❌{MsgColor.Error.value}: Failed to read result with error = {e}")
             return None
 
     def _send_req(self, _req, _url: str, timeout=10, headers={})->dict:
@@ -333,10 +347,10 @@ def _send_req(self, _req, _url: str, timeout=10, headers={})->dict:
                 hdrs = self.hdrs | headers
                 r = self.session.post(_url, data=_req.encode('utf-8'), headers=hdrs, verify=False, timeout=timeout)
             except (ConnectionRefusedError, ConnectionError):
-                console.print(f"❌[bold red] Failed to talk to server {_url}")
+                console.print(f"❌{MsgColor.Error.value} Failed to talk to server {_url}")
                 return None
             except ReadTimeout:
-                console.print(f"❌[bold red] Failed to get response from server")
+                console.print(f"❌{MsgColor.Error.value} Failed to get response from server")
                 return None
             if r.status_code != 200:
                 console.print_json(_req)
@@ -346,5 +360,4 @@ def _send_req(self, _req, _url: str, timeout=10, headers={})->dict:
                 return json.loads(r.text)
             except JSONDecodeError as e:
                 console.print(e)
-                return None
-
+                return None

absbox/local/base.py

@@ -127,6 +127,7 @@
 
 cfIndexMap = {'cn':"日期",'en':"Date","english":"Date","chinese":"日期"}
 
+
 class DC(enum.Enum):  # TODO need to check with HS code
     DC_30E_360 = "DC_30E_360"
     DC_30Ep_360 = "DC_30Ep_360"
@@ -143,6 +144,14 @@ class DC(enum.Enum):  # TODO need to check with HS code
 #valid inspection tags
 inspectTags = ["InspectBal", "InspectBool", "InspectRate", "InspectInt"]
 
+
+class InspectTags(str, enum.Enum):
+    InspectBal = "InspectBal"
+    InspectBool = "InspectBool"
+    InspectRate = "InspectRate"
+    InspectInt = "InspectInt"
+
+
 class ValidationMsg(str, enum.Enum):
     Warning = "WarningMsg"
-    Error = "ErrorMsg"
+    Error = "ErrorMsg"

absbox/local/component.py

@@ -1222,7 +1222,6 @@ def identify_deal_type(x):
     if query(x, ["pool","tag"])=='SoloPool':
         y = x["pool"]['contents']
     elif x["pool"]['tag']=='MultiPool':
-        #print(list(x["pool"]['contents'].values())[0])
         assetTags = flat(query(x,["pool","contents",S.MVALS,S.ALL,"assets",S.ALL,"tag"]))
         if len(set(assetTags))>1:
             return "UDeal"
@@ -1670,7 +1669,8 @@ def readRunSummary(x, locale) -> dict:
         return None
 
     bndStatus = {'cn': ["本金违约", "利息违约", "起算余额"]
-                ,'en': ["Balance Defaults", "Interest Defaults", "Original Balance"]}
+                ,'en': ["Balance Defaults", "Interest Defaults", "Original Balance"]
+                }
     bond_defaults = [(_['contents'][0], _['tag'], _['contents'][1], _['contents'][2])
                      for _ in x if _['tag'] in set(['BondOutstanding', 'BondOutstandingInt'])]
     _fmap = {"cn": {'BondOutstanding': "本金违约", "BondOutstandingInt": "利息违约"}

absbox/local/util.py

@@ -414,4 +414,9 @@ def strFromLens(x) -> str:
     elif isinstance(x, optics.traversals.GetZoomAttrTraversal):
         return x.name   
     else:
-        return str(x)
+        return str(x)
+
+
+def enumVals(e):
+    ''' return a list of enum values '''
+    return [_.value for _ in [*e]]