working transcoder fixed issue with threading and DB

Author: David Linko

anms-core/anms/init_adms.py

@@ -31,6 +31,7 @@
 from anms.routes.adms.adm import handle_adm
 from anms.shared.config_utils import ConfigBuilder
 from anms.shared.opensearch_logger import OpenSearchLogger
+from anms.shared.transmogrifier import TRANSMORGIFIER
 
 
 logger = OpenSearchLogger(__name__).logger
@@ -67,17 +68,10 @@ async def import_adms():
             logger.error('ADM %s handling failed: %s', adm_file.norm_name, err)
             logger.debug('%s', traceback.format_exc())
 
+     
     # Notify the aricodec of startup
-    config = ConfigBuilder.get_config()
-    host = config.get('MQTT_HOST')
-    port = config.get('MQTT_PORT')
-
-    logger.info('Connecting to MQTT broker %s to notify aricodec', host)
-    client = mqtt.client.Client()
-    client.connect(host, port)
-    msg = client.publish('aricodec/reload', b'')
-    msg.wait_for_publish()
-    client.disconnect()
+    TRANSMORGIFIER.reload()
+    
 
     logger.info('Startup finished')
 

anms-core/anms/routes/adms/adm.py

@@ -29,20 +29,17 @@
 from pydantic import BaseModel
 import io
 import traceback
-from typing import TextIO
 
 # Internal modules
-from sqlalchemy import delete, select, and_
-from sqlalchemy.engine import Result
-from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import delete,  and_
 
 from anms.models.relational.adms import (adm_data, data_model_view)
-from anms.models.relational.adms.data_model_view import DataModel as ADM
 
 from anms.routes.adms.adm_compare import (AdmCompare)
 from anms.shared.opensearch_logger import OpenSearchLogger
-from anms.shared.mqtt_client import MQTT_CLIENT
-from anms.models.relational import get_async_session, get_session
+
+from anms.shared.transmogrifier import TRANSMORGIFIER
+from anms.models.relational import get_async_session
 from anms.components.schemas.adm import DataModelSchema
 import ace
 from camp.generators import (create_sql)
@@ -296,7 +293,7 @@ async def update_adm(file: UploadFile, request: Request):
                 if error_message:
                     raise Exception(error_message)
                 # Notify the transcoder
-                MQTT_CLIENT.publish('aricodec/reload', adm_file.norm_name)
+                TRANSMORGIFIER.reload(adm_file.norm_name)
                 logger.info(f"{info_message} adm file:  {file.filename} successfully")
             except Exception as err:
                 logger.error(f"{sql_dialect} execution error: {err.args}")

anms-core/anms/routes/transcoder.py

@@ -96,22 +96,27 @@ async def transcoder_put_cbor_await(cbor: str):
 
 def _transcoder_put_cbor(input_cbor):
     transcoder_log_id = None
+    send_to_transcode = False
     with get_session() as session:
         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=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"
-            TRANSMORGIFIER.transcode(input_cbor)
+            send_to_transcode = True
         else:
             # the input_ari has already been submitted
             status = "ARI previously submitted, check log"
             transcoder_log_id = curr_uri.transcoder_log_id
-    
+        
+    if(send_to_transcode):
+        status = "Submitted ARI to transcoder"
+        TRANSMORGIFIER.transcode(input_cbor)
+
     return {"id": transcoder_log_id, "status": status}
 
 
@@ -142,6 +147,7 @@ def transcoder_incoming_str(input_ari: str):
 def _transcoder_put_str(input_ari: str):
     input_ari = input_ari.strip()
     transcoder_log_id = None
+    send_to_transcode = False
     with get_session() as session:
         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:
@@ -151,17 +157,19 @@ def _transcoder_put_str(input_ari: str):
             session.refresh(c1)
             transcoder_log_id = c1.transcoder_log_id
             session.commit()
-            status = "Submitted ARI to transcoder"
-            TRANSMORGIFIER.transcode(input_ari)
+            send_to_transcode = True
         else:
             # the input_ari has already been submitted
             status = "ARI previously submitted, check log"
             transcoder_log_id = curr_uri.transcoder_log_id
+    
+    if(send_to_transcode):
+        status = "Submitted ARI to transcoder"
+        TRANSMORGIFIER.transcode(input_ari)
 
     return {"id": transcoder_log_id, "status": status}
 
 
-
 # PUT 	/ui/incoming_send/str 	Body is str ARI to send to transcoder
 @router.put("/ui/incoming_send/str", status_code=status.HTTP_200_OK)
 async def transcoder_send_ari_str(eid: str, ari: str):

anms-core/anms/shared/mqtt_client.py

@@ -41,16 +41,17 @@ def __init__(self, config):
             return
 
         # Create MQTT Client
-        client = mqtt.client.Client("Core_MQTT_Client", clean_session=False)
-        client.on_connect = self._on_connect
-        client.on_message = self._on_message
-        logger.info('Connecting to MQTT broker at %s:%s', host, port)
-        client.connect_async(host, port, keepalive=60)
-        self.client = client
-        checking_child = Thread(target=self._check_pending)
-        checking_child.daemon = True
-        checking_child.start()
-        client.loop_start()
+        if config.Transcoder != "Internal":
+            client = mqtt.client.Client("Core_MQTT_Client", clean_session=False)
+            client.on_connect = self._on_connect
+            client.on_message = self._on_message
+            logger.info('Connecting to MQTT broker at %s:%s', host, port)
+            client.connect_async(host, port, keepalive=60)
+            self.client = client
+            checking_child = Thread(target=self._check_pending)
+            checking_child.daemon = True
+            checking_child.start()
+            client.loop_start()
 
     def publish(self, *args, **kwargs):
         ''' If connected, pass through a publish request. '''

anms-core/anms/shared/transmogrifier.py

@@ -24,7 +24,7 @@
 
 from anms.shared.config import ConfigBuilder
 from anms.shared.logger import Logger
-from anms.shared.mqtt_client import MQTT_CLIENT
+import anms.shared.mqtt_client
 from anms.shared.opensearch_logger import OpenSearchLogger
 from anms.models.relational import get_session
 from anms.models.relational.transcoder_log import TranscoderLog
@@ -36,44 +36,57 @@
 import sqlalchemy
 
 config = ConfigBuilder.get_config()
-LOGGER = OpenSearchLogger(__name__, log_console=True)
+LOGGER = OpenSearchLogger(__name__, log_console=True).logger
 
 
 # depending on what the config is for core will either use a MQTT server to send off commands or 
 # use an internal 
 class Transmorgifier:
 
     ''' The Transmogifier that can be configured to use an external or internal translator. '''
+    # args = config 
     def __init__(self, args):
         # if the transcoding in internal to core
-        if config.Transcoder == "internal":
-            LOGGER.info('Connecting to SQL DB at %s', args.db_uri)
-            self._dbeng = sqlalchemy.create_engine(args.db_uri)
-            self._adms = ace.AdmSet(cache_dir=False)
+        LOGGER.info(config.Transcoder)
+        if config.Transcoder == "Internal":
+            db_uri = f"postgresql://{config.DB_USER}:{config.DB_PASS}@{config.DB_HOST}/{config.DB_CHROOT}"
+            LOGGER.info(f'Connecting to SQL DB at {db_uri}')
+            self._dbeng = sqlalchemy.create_engine(db_uri)
+            
             self.transcode = self._transcode_internal
             self.reload = self._reload_internal
             self._adm_reload(None)
         else:
             # setting up the MQTT server instead
+            self.MQTT_CLIENT = anms.shared.mqtt_client.MQTT_CLIENT
             self.transcode = self._transcode_mqtt
             self.reload = self._reload_mqtt 
 
     def _transcode_mqtt(self, input):
         msg = json.dumps({'uri': input})
-        LOGGER.info('PUBLISH to transcode/CoreFacing/Outgoing, msg = %s' % msg)
-        MQTT_CLIENT.publish("transcode/CoreFacing/Outgoing", msg)
+        LOGGER.info(f'PUBLISH to transcode/CoreFacing/Outgoing, msg = {msg}')
+        self.MQTT_CLIENT.publish("transcode/CoreFacing/Outgoing", msg)
 
     def _transcode_internal(self, input):
+        self._ace_transcode(input)
+        
+        # picking up any stray items that didnt get translated
+        pending_uris = TranscoderLog.query.filter_by(parsed_as='pending').all()
+        for entrys in pending_uris:
+            try:
+                self._ace_transcode(entrys.input_string)
+            except Exception as err:
+                LOGGER.error('Failed to process pending entry: %s', err)
+
+    def _ace_transcode(self, input):
         # result object to fill in
         res_obj = {}
         res_obj['uri'] = ""
         res_obj['cbor'] = ""
-        
+        adms = ace.AdmSet()
         try:
-            req_obj = input
-            LOGGER.info('Request %s', req_obj)
-
-            in_text = req_obj['uri'].strip()
+            LOGGER.info(f'Request {input}')
+            in_text = input.strip()
             res_obj['inputString'] = in_text
             in_lower = in_text.casefold()
             if in_lower.startswith('ari:0x') or in_lower.startswith('0x'):
@@ -87,9 +100,9 @@ def _transcode_internal(self, input):
                     in_bytes = ace.cborutil.from_hexstr(in_text)
                     dec = ace.ari_cbor.Decoder()
                     ari = dec.decode(io.BytesIO(in_bytes))
-                    LOGGER.debug('decoded as ARI %s', ari)
-                    # ace.nickname.Converter(ace.nickname.Mode.FROM_NN, self._admsSession(self._dbeng), True)(ari)
-                    ari = ace.nickname.Converter(ace.nickname.Mode.FROM_NN, self._adms.db_session(), False)(ari)
+                    LOGGER.debug(f'decoded as ARI {ari}')
+                    # ace.nickname.Converter(ace.nickname.Mode.FROM_NN, admsSession(self._dbeng), True)(ari)
+                    ari = ace.nickname.Converter(ace.nickname.Mode.FROM_NN, adms.db_session(), False)(ari)
 
                 except Exception as err:
                     raise RuntimeError(f"Error decoding from `{in_text}`: {err}") from err
@@ -104,7 +117,7 @@ def _transcode_internal(self, input):
                     out_text = buf.getvalue()
                     if not out_text.startswith('ari:'):
                         out_text = 'ari:' + out_text
-                    LOGGER.debug('encoded as text %s', out_text)
+                    LOGGER.debug(f'encoded as text {out_text}')
                 except Exception as err:
                     raise RuntimeError(f"Error encoding from {ari}: {err}") from err
                 res_obj['uri'] = out_text
@@ -116,8 +129,8 @@ def _transcode_internal(self, input):
                 try:
                     dec = ace.ari_text.Decoder()
                     ari = dec.decode(io.StringIO(in_text))
-                    LOGGER.debug('decoded as ARI %s', ari)
-                    ari = ace.nickname.Converter(ace.nickname.Mode.FROM_NN, self._adms.db_session(), False)(ari)
+                    LOGGER.debug(f'decoded as ARI {ari}')
+                    ari = ace.nickname.Converter(ace.nickname.Mode.FROM_NN, adms.db_session(), False)(ari)
                 except Exception as err:
                     raise RuntimeError(f"Error decoding from `{in_text}`: {err}") from err
 
@@ -130,7 +143,7 @@ def _transcode_internal(self, input):
                     out_text = buf.getvalue()
                     if not out_text.startswith('ari:'):
                         out_text = 'ari:' + out_text
-                    LOGGER.debug('encoded as text %s', out_text)
+                    LOGGER.debug(f'encoded as text {out_text}')
                 except Exception as err:
                     raise RuntimeError(f"Error encoding from {ari}: {err}") from err
 
@@ -143,37 +156,47 @@ def _transcode_internal(self, input):
                     enc.encode(ari, buf)
 
                     hex_str = ace.cborutil.to_hexstr(buf.getvalue())
-                    LOGGER.info('encoded as binary %s', hex_str)
+                    LOGGER.info(f'encoded as binary {hex_str}')
                 except Exception as err:
                     raise RuntimeError(f"Error encoding from {ari}: {err}") from err
                 res_obj['cbor'] = hex_str
-
         except Exception as err:
             res_obj['ari'] = f'Failed to process: {err}'
             res_obj['parsedAs'] = 'ERROR'
-            LOGGER.error('Failed to process: %s', err)
-            LOGGER.info('Traceback:\n%s', traceback.format_exc())
-
-        LOGGER.info('Response %s', res_obj)
-        
-        # client.publish('transcode/CodexFacing/Outgoing', json.dumps(res_obj))
-        # just  log it back into the database
-
+            LOGGER.error(f'Failed to process: {err}')
+            LOGGER.info(f'Traceback:\n{traceback.format_exc()}')
+            
         # store in transcoder database
         with get_session() as session:
-            session.query(TranscoderLog).filter(TranscoderLog.input_string == res_obj['inputString']). \
-                update({
+                session.query(TranscoderLog).filter(TranscoderLog.input_string == input).update({
                     'parsed_as': res_obj['parsedAs'],
                     'ari': json.dumps(res_obj['ari']),
                     'cbor': res_obj['cbor'],
                     'uri':  res_obj['uri']
                 })
-            session.commit()
+                session.commit()
+        LOGGER.info(f'Response {res_obj}')
+        
+        # client.publish('transcode/CodexFacing/Outgoing', json.dumps(res_obj))
+        # just  log it back into the database
+
+       
+
+    def _reload_mqtt(self,adm_name=None):
+        config = ConfigBuilder.get_config()
+        host = config.get('MQTT_HOST')
+        port = config.get('MQTT_PORT')
 
-    def _reload_mqtt(self,adm_name):
+        LOGGER.info('Connecting to MQTT broker %s to notify aricodec' % host)
+        
+        msg = self.MQTT_CLIENT.publish('aricodec/reload', b'')
+        if adm_name:
+            msg = self.MQTT_CLIENT.publish('aricodec/reload', f'{adm_name}')
+        msg.wait_for_publish()
+        
         return
 
-    def _reload_internal(self, adm_name):
+    def _reload_internal(self, adm_name=None):
         try:
             self._adm_reload(adm_name)
         except Exception as err:
@@ -204,17 +227,18 @@ def _adm_reload(self, adm_name):
                 for row in curs.all():
                     self._handle_adm(*row)
 
-        LOGGER.info('ADMS present for: %s', self._adms.names())
+        
 
     def _handle_adm(self, adm_name, timestamp, data): 
-        LOGGER.info('Handling ADM: %s', adm_name)
+        LOGGER.info(f'Handling ADM:{adm_name}')
         LOGGER.info(type(data))
         # LOGGER.info(data.tos())
 
         io_buffer = io.StringIO(data.tobytes().decode('utf-8'))
-
-        self._adms.load_from_data(io_buffer)
+        adms = ace.AdmSet()
+        adms.load_from_data(io_buffer)
         LOGGER.info('Handling finished')
+        LOGGER.info('ADMS present for: %s', adms.names())
 
 
 # SIGNALTON transmorgifier