修改传递的消息为json格式{"obsid":"10109400638761","chipid":"12"}

L1/dockerfiles/csst-msc-l1-mbi-photmix/run.sh

 #!/bin/bash
 
-# obsid 10160000000 -  10160000136
-arr=($(echo $1 | tr "-" " "))
-obsid="${arr[0]}"
-detector="${arr[1]}"
-echo "obsid: "$obsid
-echo "detector: "$detector
+message=$1
+echo "message: "$message
 
 cd /pipeline/output/
-python /pipeline/app/run.py $obsid $detector
+python /pipeline/app/run.py $message
 
 exit_code=$?
 

L1/dockerfiles/csst-msc-l1-mbi/run.sh

 #!/bin/bash
 
-# obsid 10160000000 -  10160000136
-arr=($(echo $1 | tr "-" " "))
-obsid="${arr[0]}"
-detector="${arr[1]}"
-echo "obsid: "$obsid
-echo "detector: "$detector
+message=$1
+echo "message: "$message
 
 cd /pipeline/output/
-python /pipeline/app/run.py $obsid $detector
+python /pipeline/app/run.py $message
 
 exit_code=$?
 
@@ -39,7 +35,7 @@ fi
 
 if [ $exit_code -eq 0 ]; then  
   echo "finish mbi, start sink-job."  
-  scalebox task add --header dag_run_id=${dag_run_id} --upsert "$1"
+  scalebox task add --header dag_run_id=${dag_run_id} --upsert ${message}
   # scalebox task add $1
 else  
   echo "finish mbi, exit_code: $exit_code"  

L1/dockerfiles/csst-msc-l1-qc0/run.sh

 #!/bin/bash
 
-# obsid 10160000000 -  10160000136
-arr=($(echo $1 | tr "-" " "))
-obsid="${arr[0]}"
-detector="${arr[1]}"
-echo "obsid: "$obsid
-echo "detector: "$detector
+message=$1
+echo "message: "$message
 
 cd /pipeline/output/
-python /pipeline/app/run.py $obsid $detector
+python /pipeline/app/run.py $message
 
 exit_code=$?
 
@@ -39,7 +35,8 @@ else
 fi
 
 if [ $exit_code -eq 0 ]; then  
-    scalebox task add --header dag_run_id=${dag_run_id} --upsert "$1"
+    scalebox task add --header dag_run_id=${dag_run_id} --upsert ${message}
+
     echo "finish qc0, start sink-job."  
 else  
     echo "finish qc0, exit_code: $exit_code"  

L1/dockerfiles/redis-cli/config/appconfig.txt

-csst-msc-l1-mbi 9
\ No newline at end of file
+csst-msc-l1-mbi 1
\ No newline at end of file

L1/dockerfiles/redis-cli/run.sh

@@ -27,24 +27,16 @@ do
         else
             echo "$dag_id not found in appconfig.txt"
             code = 103
-            # exit 103
         fi
 
         # 提取dag_run_id
         dag_run_id=$(echo "$msg" | jq -r '.parameters.dag_run_id')
         echo "dag_run_id : "$dag_run_id
 
-        # 提取obsid
-        obsid=$(echo "$msg" | jq -r '.parameters.message.obsid')
-        echo "obsid :"$obsid
-
-        # 提取chipid
-        chipid=$(echo "$msg" | jq -r '.parameters.message.chipid')
-        echo "chipid :"$chipid
-
-        message=$obsid"-"$chipid
-        echo "message :"$message
-
+        message=$(echo "$msg" | jq -c '.parameters.message')
+        
+        # 提取message
+        echo "$message" >> "msgs.txt"
 
         #send-message $obsid
         job_name="message-router-csst"

L1/dockerfiles/redis-cli/test.sh

-redis-cli -h 10.3.10.28 -a 123456 lpush csst_data_list_zxl "{\"dag_id\": \"csst-msc-l1-mbi\", \"parameters\": {\"dag_run_id\": \"202411111841239406\", \"message\": {\"obsid\": \"11009101682009\", \"chipid\": \"16\"}}}"
-
-
-scalebox task add --app-id 16 --header dag_run_id="202411258348645214" --upsert --sink-job "message-router-csst" "11009101682009-09"
\ No newline at end of file
+redis-cli -h 10.3.10.28 -a 123456 lpush csst_data_list_zxl "{\"dag_id\": \"csst-msc-l1-mbi\", \"parameters\": {\"dag_run_id\": \"202411111841239406\", \"message\": {\"obsid\": \"11009101682009\", \"chipid\": \"16\"}}}"
\ No newline at end of file

L1/pipelines/csst-msc-l1-mbi/message-router/messageRoute.py

@@ -51,8 +51,6 @@ class messageRoute():
 
                 subkey = "task"+str(i+1)
                 upstream_tasks = config['tasks'][subkey].get('upstream_tasks')
-
-                # print("subkey "+subkey)
                 
                 # 不包含upstream_tasks的模块为流水线的第一个接收消息的模块
                 if upstream_tasks is None:
@@ -101,6 +99,8 @@ class messageRoute():
     def sendmsg(self,dag_run_id,job,message):
         #执行解包操作
         print("sendmsg dag_run_id is "+dag_run_id)
+        message = "'"+message+"'"
+
         command = f"scalebox task add --header dag_run_id={dag_run_id} --upsert --sink-job={job} {message}"
         print("command : "+command)
         result=subprocess.run(command, shell=True)