38 anms fun bld 007 edds ari selection on build tab (#194)

* Rework to split amp-manager from ION containers

* Aligning ION config inputs, fixing proxy socket path

* Fixing container build after merge

* Fix compose config

* Cleaning up CI config

* Fixing access in CI

* Update checkout test port

* Fixing podman use and podman checkout CI job

* Stop in opposite order from start

* updating nm_host from ion-manger to anms-amp-manager-1. still having issues submitting reports

* updated name to amp-manager, removed debug log

* Renamed test containers file to testenv-compose to make its purpose more clear. Moved socket bind mount to an environment controlled path.

* Fixing puppet testenv disable behavior

* Fix hostname for the NM manager

* Separate anms network from testenv network

* Fix container name after merge

* updated to dtnma-camp

* updated to latest CAMP and DTNNA-TOOLS

* Updating uses of ACE, CAMP, and REFDM to latest development

* updated to latest dtnma libraries

* updating to latest sql tables

* handling ADM upload and report printing

* working translation and report handling

* updated for sending and viewing execsets

* added status msg to transcoder route

* added clearSearchOnSelect, added redundant transcoding check

* updated report processing to use ACE and enhanced transcoding

* removed un need nickname convertor

---------

Co-authored-by: Brian Sipos <brian.sipos@jhuapl.edu>
Co-authored-by: Brian Sipos <brian.sipos@gmail.com>
Co-authored-by: David Linko <david.linko@jhuapl.edu>

Author: 3 people

.env

@@ -53,4 +53,4 @@ ION_MGR_PORT=8089
 HTTP_PORT:80
 SOCKDIR=/var/tmp/nm
 
-ADM_PATH=deps/dtnma-adms
+ADM_PATH=deps/dtnma-adms

anms-core/anms/routes/ARIs/reports.py

@@ -30,17 +30,22 @@
 from sqlalchemy import select, and_
 from sqlalchemy.engine import Result
 from io import StringIO
+from urllib.parse import unquote
 
 from anms.components.schemas import ARIs
 from anms.models.relational import get_async_session, get_session
 
 from anms.models.relational.report import Report
 from anms.models.relational.execution_set import ExecutionSet
 from anms.shared.opensearch_logger import OpenSearchLogger
-
+import io
 
 import anms.routes.transcoder as transcoder
 
+# for handling report set and exec set  
+import ace
+import ace.models
+
 
 logger = OpenSearchLogger(__name__, log_console=True)
 
@@ -80,10 +85,11 @@ async def report_def_by_id(agent_id: str):
             ari_val = ""
             if(res):
                 ari_val = await transcoder.transcoder_put_cbor_await("ari:0x"+res.entries.hex())
-                ari_val =  ari_val['data']   
-            addition = {'exec_set': ari_val,'correlator_nonce':correlator_nonce}    
-            if addition not in final_res:
-                final_res.append(addition)
+                logger.info(ari_val)
+                ari_val =  ari_val['data']
+                addition = {'exec_set': ari_val,'correlator_nonce':correlator_nonce}    
+                if addition not in final_res:
+                    final_res.append(addition)
 
     return final_res
 
@@ -94,6 +100,8 @@ async def report_def_by_id(agent_id: str):
 async def report_ac(agent_id: str, correlator_nonce: int):
     agent_id = agent_id.strip()
     final_res = []
+    ari = None
+    # Load in adms 
     # get command that made the report as first entry 
     stmt = select(ExecutionSet).where(and_(ExecutionSet.agent_id == agent_id, ExecutionSet.correlator_nonce == correlator_nonce) )
     async with get_async_session() as session:
@@ -102,42 +110,78 @@ async def report_ac(agent_id: str, correlator_nonce: int):
         exec_set_entry=["time"]
         if result:
             exec_set = result.entries.hex()
-            exec_set = await transcoder.transcoder_put_cbor_await("ari:0x"+exec_set)
-            exec_set =  exec_set['data']  
-            # format 
-            # TODO HANDLE RPTT and split up multiple entries 
-            "ari:/EXECSET/n=12345;(//ietf/dtnma-agent/CTRL/inspect(//ietf/dtnma-agent/EDD/sw-version))"
-            execset_pattern = r"ari:/EXECSET/n=.+;\((.*)\)"
-            match = re.match(execset_pattern,exec_set)
-            if match:
-                exec_set_entry.extend(match.group(1).split(';'))
-            else:
-                exec_set_entry.extend([exec_set])
-        final_res.append(exec_set_entry)
-
+            # use ACE to handle report set decoding  
+            in_text = '0x'+exec_set
+            try:
+                in_bytes = ace.cborutil.from_hexstr(in_text)
+                dec = ace.ari_cbor.Decoder()
+                ari = dec.decode(io.BytesIO(in_bytes))
+                
+            except Exception as err:
+                logger.info(err)
+                
+    # current ARI should be  an exection set 
+    if ari:
+        logger.info(ari)
+        if type(ari.value) == ace.ari.ExecutionSet: 
+            try:
+                enc = ace.ari_text.Encoder()
+                buf = io.StringIO()
+                # run through targets and their parameters to get all things parts translated 
+                for targ in ari.value.targets: 
+                    if targ.params:
+                        for param in targ.params:
+                            enc.encode(param, buf)
+                            out_text = buf.getvalue()
+                            ari_val = await transcoder.transcoder_put_await_str(out_text)
+                            exec_set_entry.append(ari_val['data'])
+                    else:
+                        enc.encode(targ, buf)
+                        out_text = buf.getvalue()
+                        ari_val = await transcoder.transcoder_put_await_str(out_text)
+                        exec_set_entry.append(ari_val['data'])
+
+            except Exception as err:
+                logger.info(err)
+                
+                    
+    final_res.append(exec_set_entry)
+    ari = None
     stmt = select(Report).where(and_(Report.agent_id == agent_id, Report.correlator_nonce == correlator_nonce) )
     async with get_async_session() as session:
         result: Result = await session.scalars(stmt)
         for res in result.all():
-            # TODO translating the CBOR route might want to relook at currently cause large amount of transcoding vs just loading the string below              
-            # # translate the cbor
-            # rpt_set = res.report_list_cbor.hex()
-            # rpt_set = await transcoder.transcoder_put_cbor_await("ari:0x"+rpt_set)
-            # rpt_set =  rpt_set['data']   
-            rpt_set = res.report_list
-            # match 
-            # ari:/RPTSET/n=12345;r=/TP/20250611T114420.009992304Z;(t=/TD/PT0S;s=//1/1/CTRL/5(//1/1/EDD/1);(%220.0…
-            rptset_pattern = r"ari:/RPTSET/n=.+;r=.*;\(t=.*;s=.*;\((.*)\)\)"
-            match = re.match(rptset_pattern,rpt_set)
-            addition = [res.reference_time] 
-            if match:
-                # report entries 
-                rpt_entr = match.group(1)
-                addition.extend(rpt_entr.split(";"))
-            else:
-                addition.append(rpt_set)
+            # used to hold final report set 
+            addition = [res.reference_time]
+            rpt_set = res.report_list_cbor.hex()
+            # Using Ace to translate CBOR into ARI object to process individual parts  
+            in_text = '0x'+rpt_set
+            try:
+                in_bytes = ace.cborutil.from_hexstr(in_text)
+                dec = ace.ari_cbor.Decoder()
+                ari = dec.decode(io.BytesIO(in_bytes))
+
+            except Exception as err:
+                logger.error(err)
+                
+            # current ARI should be  an report set 
+            if ari:
+                logger.info(ari)
+                if type(ari.value) == ace.ari.ReportSet:                    
+                    for rpt in ari.value.reports:
+                        try:
+                            enc = ace.ari_text.Encoder()
+                            # running through and translating all parts of rptset
+                            for item in rpt.items:
+                                buf = io.StringIO()
+                                enc.encode(item, buf)
+                                out_text = buf.getvalue()
+                                ari_val = await transcoder.transcoder_put_await_str(out_text)
+                                addition.append(ari_val['data'])
+                        except Exception as err:
+                            logger.error(err)
 
-            if addition not in final_res:
-                final_res.append(addition)
+                if addition not in final_res:
+                    final_res.append(addition)
     return final_res
 

anms-core/anms/routes/transcoder.py

@@ -66,33 +66,39 @@ async def transcoder_log_by_id(id: str):
         return TranscoderLog.query.filter_by(transcoder_log_id=id).first()
 
 # PUT 	/ui/incoming/{cbor}/hex
-@router.put("/ui/incoming/{cbor}/hex", status_code=status.HTTP_200_OK)
-async def transcoder_put_cbor(cbor: str):
-    msg = json.dumps({'uri': cbor})
+@router.put("/ui/incoming/{input_cbor}/hex", status_code=status.HTTP_200_OK)
+async def transcoder_put_input_cbor(input_cbor: str):
+    msg = json.dumps({'uri': input_cbor})
     transcoder_log_id = None
     with get_session() as session:
-        curr_uri = TranscoderLog.query.filter_by(input_string=cbor).first()
+        curr_uri = TranscoderLog.query.filter(or_(TranscoderLog.input_string==input_cbor, TranscoderLog.cbor==input_cbor)).first()
         if curr_uri is None:
-            c1 = TranscoderLog(input_string=cbor, parsed_as='pending')
+            c1 = TranscoderLog(input_string=input_cbor, parsed_as='pending')
             session.add(c1)
             session.flush()
             session.refresh(c1)
             transcoder_log_id = c1.transcoder_log_id
             session.commit()
+            status = "Submitted ARI to transcoder"
+        else:
+            # the input_ari has already been submitted
+            status = "ARI previously submitted, check log"
+            transcoder_log_id = curr_uri.transcoder_log_id
 
     logger.info('PUBLISH to transcode/CoreFacing/Outgoing, msg = %s' % msg)
     MQTT_CLIENT.publish("transcode/CoreFacing/Outgoing", msg)
 
-    return  {"id": transcoder_log_id}
+    return {"id": transcoder_log_id, "status": status}
+
 
-# get 	/ui/incoming/{cbor}/hex
 @router.get("/ui/incoming/await/{cbor}/hex", status_code=status.HTTP_200_OK)
 async def transcoder_put_cbor_await(cbor: str):
     curr_uri = ""
     msg = json.dumps({'uri': cbor})
     transcoder_log_id = None
     with get_session() as session:
-        curr_uri = TranscoderLog.query.filter_by(input_string=cbor).first()
+        curr_uri = TranscoderLog.query.filter(or_(TranscoderLog.input_string==cbor, TranscoderLog.cbor==cbor)).first()
+
         if curr_uri is None:
             c1 = TranscoderLog(input_string=cbor, parsed_as='pending')
             session.add(c1)
@@ -115,27 +121,29 @@ async def transcoder_put_cbor_await(cbor: str):
     while True:
         with get_session() as session:
             curr_uri = TranscoderLog.query.filter_by(TranscoderLog.transcoder_log_id==transcoder_log_id).first()
-        if curr_uri.parsed_as == "CBOR":
-            curr_uri = curr_uri.uri
-            break
-        if curr_uri.parsed_as == "ERROR":
-            curr_uri = "ARI://BADARI"
-            break
+            if curr_uri.parsed_as != "pending":
+                if curr_uri.parsed_as == "ERROR":
+                    curr_uri = "ARI://BADARI"
+                else:
+                    curr_uri = curr_uri.uri
+                break
         time.sleep(1)
 
 
     return  {"data": curr_uri}
 
 # PUT 	/ui/incoming/str 	Body is str ARI to send to transcoder
-@router.get("/ui/incomin/await/str", status_code=status.HTTP_200_OK)
-def transcoder_put_await_str(ari: str):
-    ari = ari.strip()
-    msg = json.dumps({"uri": ari})
+@router.get("/ui/incoming/await/str", status_code=status.HTTP_200_OK)
+async def transcoder_put_await_str(input_ari: str):
+    input_ari = input_ari.strip()
+    msg = json.dumps({"uri": input_ari})
     transcoder_log_id = None
+    curr_uri = None
     with get_session() as session:
-        curr_uri = TranscoderLog.query.filter_by(input_string=ari).first()
+        curr_uri = TranscoderLog.query.filter(or_(TranscoderLog.input_string==input_ari,TranscoderLog.ari==input_ari, TranscoderLog.cbor==input_ari)).first()
+        
         if curr_uri is None:
-            c1 = TranscoderLog(input_string=ari, parsed_as='pending')
+            c1 = TranscoderLog(input_string=input_ari, parsed_as='pending')
             session.add(c1)
             session.flush()
             session.refresh(c1)
@@ -145,45 +153,55 @@ def transcoder_put_await_str(ari: str):
             MQTT_CLIENT.publish("transcode/CoreFacing/Outgoing", msg)
         else:
             transcoder_log_id = curr_uri.transcoder_log_id
+            if curr_uri.parsed_as != "pending":
+                if curr_uri.parsed_as == "ERROR":
+                    curr_uri = "ARI://BADARI"
+                else:
+                    curr_uri = curr_uri.uri
+                return  {"data": curr_uri}
+            
 
 
     while(True):
         with get_session() as session:
             curr_uri = TranscoderLog.query.filter_by(transcoder_log_id=transcoder_log_id).first()
-        if curr_uri.parsed_as == "URI":
-            curr_uri = curr_uri.uri
-            break
-        if curr_uri.parsed_as == "ERROR":
-            curr_uri = "ARI://BADARI"
-            break
-        time.sleep(1)
+            if curr_uri.parsed_as != "pending":
+                if curr_uri.parsed_as == "ERROR":
+                    curr_uri = "ARI://BADARI"
+                else:
+                    curr_uri = curr_uri.uri
+                break
+            time.sleep(1)
 
 
     return  {"data": curr_uri}
 
 
 # PUT 	/ui/incoming/str 	Body is str ARI to send to transcoder
 @router.put("/ui/incoming/str", status_code=status.HTTP_200_OK)
-def transcoder_put_str(ari: str):
-    ari = ari.strip()
-    msg = json.dumps({"uri": ari})
+async def transcoder_put_str(input_ari: str):
+    input_ari = input_ari.strip()
+    msg = json.dumps({"uri": input_ari})
     transcoder_log_id = None
     with get_session() as session:
-        curr_uri = TranscoderLog.query.filter_by(input_string=ari).first()
+        curr_uri = TranscoderLog.query.filter(or_(TranscoderLog.input_string==input_ari,TranscoderLog.ari==input_ari, TranscoderLog.cbor==input_ari)).first()
         if curr_uri is None:
-            c1 = TranscoderLog(input_string=ari, parsed_as='pending')
+            c1 = TranscoderLog(input_string=input_ari, parsed_as='pending')
             session.add(c1)
             session.flush()
             session.refresh(c1)
             transcoder_log_id = c1.transcoder_log_id
             session.commit()
+            status = "Submitted ARI to transcoder"
         else:
+            # the input_ari has already been submitted
+            status = "ARI previously submitted, check log"
             transcoder_log_id = curr_uri.transcoder_log_id
 
     logger.info('PUBLISH to transcode/CoreFacing/Outgoing, msg = %s' % msg)
     MQTT_CLIENT.publish("transcode/CoreFacing/Outgoing", msg)
 
-    return {"id": transcoder_log_id}
+    return {"id": transcoder_log_id, "status": status}
 
 
 

anms-core/pyproject.toml

@@ -31,8 +31,6 @@ dependencies = [
   "email-validator ~=1.3",
   "fastapi ~=0.86.0",
   "fastapi-pagination ~=0.9.1",
-  "grpcio ~=1.50.0",
-  "grpcio-tools ~=1.50.0",
   "gunicorn ~=23.0.0",
   "httpx ~=0.24.0",
   "itsdangerous ~=2.1.2",

anms-ui/public/app/components/management/agents/Agents.vue

@@ -106,6 +106,7 @@ import { mapGetters, mapActions } from "vuex";
 import api from "../../../shared/api.js";
 import AgentModal from "./AgentModal.vue";
 import AgentsManageModal from "./AgentsManageModal.vue";
+import toastr from "toastr";
 
 export default {
   name: "Agents",
@@ -215,9 +216,15 @@ export default {
       nodeList.forEach((node) => {
         api.methods
           .apiPostAgent(node.trim())
-          .then((response) => (this.results = response.status + " " + response.statusText))
+          .then((response) => {
+            this.results = response.status + " " + response.statusText;
+            toastr.success(this.results);
+            
+          })
           .catch((error) => {
             console.error(error);
+            toastr.error(error);
+
           });
       });
     },

anms-ui/public/app/components/management/agents/AgentsManageModal.vue

@@ -34,7 +34,7 @@
             </b-button>
           </b-col>
           <b-col cols="9">
-            <build :cbor="cbor" @updateResult="updateResults($event)"></build>
+            <build :cbor="cbor" :agentModal=true @updateResult="updateResults($event)"></build>
           </b-col>
         </b-row>
       </div>
@@ -126,7 +126,15 @@ export default {
         .then((response) => {
           if (response.data.parsed_as == "pending") {
             setTimeout(() => this.queryTranscoderLog(), 8000);
-          } else {
+          } else if(response.data.parsed_as == "ERROR") {
+            console.log(`Error translating transcoder log ID: ${this.transcoderLogId}! See transcoder log table for details`);
+            toastr.error(`Error translating transcoder log ID: ${this.transcoderLogId}! See transcoder log table for details`);
+            this.loading = false;
+            this.closeModal();
+            this.ariCBOR = null;
+            this.ariString=null;
+           } 
+          else {
             this.ariCBOR = response.data.cbor;
             this.submitRawCommand2Agents();
           }

anms-ui/public/app/components/management/agents/reports.vue

@@ -16,7 +16,7 @@
         value="-1">-- Select Sent Reports --</b-form-select-option>
       <b-form-select-option v-for="rpt, index in rptts"
         :key="index"
-        :value="index">{{ decodeURI(rpt) }}</b-form-select-option>
+        :value="index">{{ rpt }}</b-form-select-option>
     </b-form-select>
     <b-table sticky-header
       hover