[关闭]
@liruiyi962464 2025-12-12T08:34:05.000000Z 字数 10435 阅读 124

LTE新设备python文件

Linux服务器部署

不用管 测试用

  1. minicom -D /dev/ttyUSB0
  2. ps | grep python
  3. cd /
  4. vi serial_util.py
  5. python3 start.py
  6. rm *.py
  8. cat /etc/init.d/Start
  9. rm /etc/init.d/Start /etc/rc.d/S92Start
  10. vi /etc/init.d/Start
  12. chmod +x /etc/init.d/Start
  13. /etc/init.d/Start enable 
  14. /etc/init.d/Start start
  15. /etc/init.d/Start stop
  17. /etc/init.d/Start stop 
  18. rm /at.py
  19. vi /at.py
  20. /etc/init.d/Start stop 
  21. /etc/init.d/Start start
  22. reboot

Start 配置自启服务

  1. #!/bin/sh /etc/rc.common
  3. USE_PROCD=1
  4. START=92
  5. STOP=15
  7. start_service() {
  8.     procd_open_instance
  9.     procd_set_param command python3 start.py
  10.     procd_set_param respawn  
  11.     procd_set_param stdout 1 
  12.     procd_set_param stderr 1
  13.     procd_close_instance
  14. }
  16. stop_service() {
  17.     killall python3 
  18. }
  20. restart() {
  21.     stop
  22.     start
  23. }

vi serial_util.py

  1. import serial
  2. import time
  3. import fcntl
  6. tty_com = "/dev/ttyUSB1"
  7. baudrate = 115200
  9. ip_addr = "172.0.0.11"
  10. deviceIp = '172.0.0.161'
  12. def open_serial():
  13.     try:
  14.         ser = serial.Serial(
  15.             tty_com, baudrate,
  16.             bytesize=serial.EIGHTBITS,
  17.             parity=serial.PARITY_NONE,
  18.             stopbits=serial.STOPBITS_ONE,
  19.             timeout=3600
  20.         )
  21.         fcntl.flock(ser.fileno(),fcntl.LOCK_EX)
  22.         return ser
  23.     except Exception as e:
  24.         print(f"fail to open ser:{str(e)}")
  25.         return None
  27. def send_AT(ser, cmd):
  28.     try:
  29.         if not ser.is_open:
  30.             ser.open()
  31.         ser.write(cmd.encode())
  32.         data = ser.read(1)
  33.         time.sleep(0.1)
  34.         data += ser.read(ser.inWaiting())
  35.         return data.decode()
  36.     except Exception as e:：：
  37.         print(f"AT fail: {str(e)}")
  38.         return "error"

vi api.py

  1. from flask import Flask, request
  2. import os,time
  3. import subprocess
  4. from serial_util import open_serial, send_AT,deviceIp
  6. app = Flask(__name__)
  7. app.secret_key = 'asdfghjkl'
  10. @app.route('/ping/<ip>/<num>', methods=['POST', 'GET'])
  11. def ping(ip, num):
  12.     """test for ping"""
  13.     try:
  14.         result = subprocess.run(
  15.             ['ping', '-c', num, ip],
  16.             capture_output=True, text=True, timeout=10
  17.         )
  18.         return result.stdout.split('\n')
  19.     except Exception as e:
  20.         print(f"fail to Ping: {str(e)}")
  21.         return "error"
  23. @app.route('/init', methods=['POST', 'GET'])
  24. def init():
  25.     state = request.args.get('state', '')
  26.     if not state.isdigit():
  27.         return "Invalid state", 400
  29.     try:
  30.         os.system("kill $(pgrep -f 'python3 at.py')")
  31.         ser = open_serial()
  32.         if not ser:
  33.             return "ser can not use", 500
  35.         commands = {
  36.             '0': "AT+CFUN=1\r\n",
  37.             '1': "AT+CFUN=0\r\nAT+ZEMCI=0\r\n",
  38.             '2': "AT+CFUN=1\r\n",
  39.             '3': "AT+CGATT=0\r\nAT+CGATT=1\r\n",
  40.             '4': "AT+CEREG=1\r\nAT+CGDCONT=1,\"IP\"\r\nAT+CGACT=1,1\r\nAT+ZGACT=1,1\r\n"
  41.         }
  42.         if state in commands:
  43.             send_AT(ser, commands[state])
  44.             time.sleep(1)
  45.         return state
  46.     except Exception as e:
  47.         print(f"fail to init: {str(e)}")
  48.         return "error", 500
  49.     finally:
  50.         os.system("nohup python3 at.py >> /dev/null 2>&1 &")
  51.         if ser:
  52.             ser.close()
  54. if __name__ == '__main__':
  55.     app.run(host=deviceIp, port=3652, debug=True)

vi at.py

  1. import os
  2. import re
  3. import json
  4. import subprocess
  5. import requests
  6. import time
  7. from serial_util import open_serial, send_AT, ip_addr, deviceIp
  10. def zemci():
  11.     try:
  12.         ser = open_serial()
  13.         if not ser:
  14.             return "no response"
  16.         send_AT(ser, "AT+ZEMCI=1\r\n")
  17.         time.sleep(1.2)
  18.         datas = send_AT(ser, "AT+ZEMCI=1\r\n").split('\r\n')
  19.         if len(datas) < 2 or 'ERROR' in datas[1] or datas[1] == 'OK':
  20.             return "no response"
  22.         zemci_data = datas[1].split(':')[1].split(',')
  23.         parameter = {
  24.             "deviceIp": deviceIp,
  25.             "cellId": hex(int(zemci_data[0]))[2:],
  26.             "tac": zemci_data[2],
  27.             "band1": zemci_data[6],
  28.             "rsrp": int(zemci_data[10]) - 140,
  29.             "rsrq": int(zemci_data[11]) * 0.5 - 20,
  30.             "pci": zemci_data[14][:4],
  31.             "earfcn": int(zemci_data[14][4:], 16),
  32.             "csq": zemci_data[10]
  33.         }
  34.         print("Cell info:", parameter)
  35.         return parameter
  36.     except Exception as e:
  37.         print("fail to get zemci:", str(e))
  38.         return "error"
  39.     finally:
  40.         if 'ser' in locals() and ser:
  41.             ser.close()
  44. def loop():
  45.     data = zemci()
  46.     if isinstance(data, dict) and data.get("deviceIp") == deviceIp:
  47.         try:
  48.             response = requests.post(
  49.                 f"http://{ip_addr}:8091/jeecg-boot/onlineStateDetection/deviceNetworkstatus/add",
  50.                 json=data,
  51.                 headers={"Content-Type": "application/json"},
  52.                 timeout=5
  53.             )
  54.             print("Upload cell quality success, code:", response.status_code)
  55.         except Exception as e:
  56.             print("fail to update network status:", str(e))
  57.     return data
  60. def attach():
  61.     ser = open_serial()
  62.     if ser:
  63.         send_AT(ser, "AT+ZGACT=1,1\r\n")
  64.         time.sleep(1)
  65.         ser.close()
  66.     return 0
  69. def state():
  70.     ser = open_serial()
  71.     if not ser:
  72.         return {
  73.             "ip": deviceIp,
  74.             "deviceIp": deviceIp,
  75.             "moduleResponding": 0,
  76.             "moudelState": 0,
  77.             "simState": 0
  78.         }
  80.     try:
  81.         resp = send_AT(ser, "AT\r\n")
  82.         module_responding = 1 if "OK" in resp else 0
  84.         if not module_responding:
  85.             return {
  86.                 "ip": deviceIp,
  87.                 "deviceIp": deviceIp,
  88.                 "moduleResponding": 0,
  89.                 "moudelState": 0,
  90.                 "simState": 0
  91.             }
  93.         resp = send_AT(ser, "AT+CFUN?\r\n")
  94.         moudelState = 1 if "+CFUN: 1" in resp else 0
  96.         resp = send_AT(ser, "AT+CPIN?\r\n")
  97.         simState = 1 if "+CPIN: READY" in resp else 0
  99.         return {
  100.             "ip": deviceIp,
  101.             "deviceIp": deviceIp,
  102.             "moduleResponding": module_responding,
  103.             "moudelState": moudelState,
  104.             "simState": simState
  105.         }
  107.     except Exception as e:
  108.         print("state error:", e)
  109.         return {
  110.             "ip": deviceIp,
  111.             "deviceIp": deviceIp,
  112.             "moduleResponding": 0,
  113.             "moudelState": 0,
  114.             "simState": 0
  115.         }
  116.     finally:
  117.         if ser and ser.is_open:
  118.             ser.close()
  121. def cfun():
  122.     ser = None
  123.     try:
  124.         ser = open_serial()
  125.         if not ser:
  126.             print("Serial open failed")
  127.             return -1
  129.         print("Restarting module...")
  130.         send_AT(ser, "AT+CFUN=0\r\n")
  131.         time.sleep(2)
  132.         send_AT(ser, "AT+ZEMCI=0\r\n")
  133.         time.sleep(1)
  135.         send_AT(ser, "AT+CFUN=1\r\n")
  136.         time.sleep(5)
  138.         send_AT(ser, "AT+CGACT=1,1\r\n")
  139.         time.sleep(3)
  140.         send_AT(ser, "AT+ZGACT=1,1\r\n")
  141.         time.sleep(3)
  143.         send_AT(ser, "AT+ZEMCI=1\r\n")
  144.         send_AT(ser, "AT+CGPADDR=1\r\n")
  145.         print("Module restart completed")
  146.         return 0
  147.     except Exception as e:
  148.         print("cfun failed:", e)
  149.         return -1
  150.     finally:
  151.         if ser:
  152.             ser.close()
  155. def activate_radio():
  156.     ser = None
  157.     try:
  158.         ser = open_serial()
  159.         if not ser:
  160.             print("Serial open failed, cannot activate radio")
  161.             return -1
  163.         print("Activating radio (AT+CFUN=1)...")
  164.         send_AT(ser, "AT+CFUN=1\r\n")
  165.         time.sleep(5)
  166.         print("Radio activation sent, waiting for network registration...")
  167.         return 0
  168.     except Exception as e:
  169.         print("Radio activation failed:", e)
  170.         return -1
  171.     finally:
  172.         if ser:
  173.             ser.close()
  176. if __name__ == "__main__":
  177.     print("Initializing module...")
  178.     cfun()
  180.     consecutive_restarts = 0
  181.     MAX_RESTARTS = 5
  182.     COOLDOWN_TIME = 600
  184.     while True:
  185.         try:
  186.             data = state()
  187.         except Exception as e:
  188.             print("Get state exception:", e)
  189.             data = None
  191.         ping_ok = os.system(f'ping -c 1 {ip_addr} > /dev/null 2>&1') == 0
  193.         module_responding = data.get('moduleResponding', 0) if isinstance(data, dict) else 0
  194.         module_ok = data.get('moudelState', 0) if isinstance(data, dict) else 0
  195.         sim_ok = data.get('simState', 0) if isinstance(data, dict) else 0
  197.         print("Status check: ping=", ping_ok, ", moduleResponding=", module_responding, ", module=", module_ok, ", sim=", sim_ok)
  199.         if not ping_ok or not module_responding:
  200.             print("Module unresponsive or network down, preparing to report...")
  202.             death_data = {
  203.                 "ip": deviceIp,
  204.                 "deviceIp": deviceIp,
  205.                 "moudelState": 0,
  206.                 "simState": 0
  207.             }
  209.             try:
  210.                 url = f"http://{ip_addr}:8091/jeecg-boot/system/deviceInfo/updateDeviceSIMStateAndMoudleState?ip={death_data['ip']}&moudelState={death_data['moudelState']}&simState={death_data['simState']}"
  211.                 response = requests.get(url, timeout=5)
  212.                 print("Report module dead status success, code:", response.status_code)
  213.             except Exception as e:
  214.                 print("Report dead status failed:", e)
  216.             if consecutive_restarts >= MAX_RESTARTS:
  217.                 print("Max restarts reached, cooling down for", COOLDOWN_TIME//60, "minutes...")
  218.                 time.sleep(COOLDOWN_TIME)
  219.                 consecutive_restarts = 0
  220.                 continue
  222.             print("Performing full restart... (attempt", consecutive_restarts + 1, ")")
  223.             cfun()
  224.             consecutive_restarts += 1
  225.             time.sleep(15)
  226.             continue
  228.         if not module_ok:
  229.             print("Radio off, activating radio...")
  230.             activate_radio()
  231.             time.sleep(8)
  232.             continue
  234.         try:
  235.             if consecutive_restarts > 0:
  236.                 print("Device recovered, reset restart counter from", consecutive_restarts)
  237.                 consecutive_restarts = 0
  239.             cell_data = loop()
  240.             if isinstance(cell_data, str):
  241.                 print("Cell quality get failed:", cell_data)
  242.             time.sleep(2)
  244.             fallback_data = {
  245.                 "ip": deviceIp,
  246.                 "deviceIp": deviceIp,
  247.                 "moudelState": 1,
  248.                 "simState": 0
  249.             }
  251.             report_data = data if isinstance(data, dict) else fallback_data
  253.             url = f"http://{ip_addr}:8091/jeecg-boot/system/deviceInfo/updateDeviceSIMStateAndMoudleState?ip={report_data['ip']}&moudelState={report_data['moudelState']}&simState={report_data['simState']}"
  254.             response = requests.get(url, timeout=5)
  255.             print("Report normal status success, code:", response.status_code)
  257.             time.sleep(5)
  259.         except Exception as e:
  260.             print("Report process error:", e)
  261.             time.sleep(5)

vi attach_api.py

  1. from flask import Flask, jsonify
  2. from serial_util import open_serial, send_AT,deviceIp
  3. import time
  5. app = Flask(__name__)
  7. def cfun():
  8.     ser = open_serial()
  9.     send_AT(ser,"AT+CFUN=0\r\n")
  10.     send_AT(ser,"AT+ZEMCI=0\r\n")
  11.     time.sleep(0.5)
  12.     send_AT(ser,"AT+CFUN=1\r\n&AT+CGACT=1,1\r\n&AT+ZGACT=1,1\r\n")
  13.     time.sleep(2)
  14.     send_AT(ser,"AT+CGACT=1,1\r\n")
  15.     print("dadsqweqweqfas")
  16.     time.sleep(2)
  17.     send_AT(ser,"AT+ZGACT=1,1\r\n")
  18.     time.sleep(2)
  19.     send_AT(ser,"AT+ZEMCI=1\r\n")
  20.     send_AT(ser,"AT+CGPADDR=1\r\n")
  21.     ser.close()
  22.     return 0
  24. @app.route('/attach/<num>', methods=['POST', 'GET'])
  25. def attach(num):
  26.     if not num.isdigit():
  27.         return jsonify({"error": "Invalid parameter"}), 500
  28.     result = []
  29.     try:
  30.         ser = open_serial()
  31.         if not ser:
  32.             return jsonify({"error": "ser can not use"}), 500
  33.         for _ in range(int(num)):
  34.             send_AT(ser, "AT+CGATT=0\r\n")
  35.             time.sleep(1)
  36.             start_time = time.time()
  37.             response = send_AT(ser, "AT+CGATT=1\r\n")
  38.             datas=response.split('\r\n')[1]
  39.             if datas != 'ok':
  40.                 result.append(int((time.time() - start_time) * 1000))
  41.             else:
  42.                 result.append('error')
  43.         return jsonify(result)
  44.     except Exception as e:
  45.         print(f"fail to attach: {str(e)}")
  46.         return jsonify({"error": "Internal error"}), 500
  47.     finally:
  48.         if ser:
  49.             ser.close()
  50.         cfun()
  51. if __name__ == '__main__':
  52.     app.run(host=deviceIp, port=3653, debug=True)

vi start.py

cd /
rm -rf service_runner.log
rm -rf start.py
vi start.py

  1. iimport threading
  2. import os
  3. import sys
  4. import subprocess
  5. from typing import Callable, Dict
  6. import time
  8. class ServiceManager:
  9.     def __init__(self):
  10.         self.services = ["at.py", "api.py", "attach_api.py"]
  11.         self.threads: Dict[str, threading.Thread] = {}
  12.         self.processes: Dict[str, subprocess.Popen] = {}
  13.         self.lock = threading.Lock()
  15.     def run_script(self, script_name: str) -> None:
  16.         try:
  17.             process = subprocess.Popen(
  18.                 [sys.executable, script_name],
  19.                 stdout=subprocess.PIPE,
  20.                 stderr=subprocess.STDOUT,
  21.                 text=True
  22.             )
  23.             with self.lock:
  24.                 self.processes[script_name] = process
  26.             while True:
  27.                 output = process.stdout.readline()  # type: ignore
  28.                 if output == '' and process.poll() is not None:
  29.                     break
  30.                 # if output:
  31.                 #     print(f"[{script_name}] {output.strip()}")
  33.             exit_code = process.poll()
  34.             # if exit_code != 0:
  35.             #     print(f"Error: {script_name} exited with code {exit_code}")
  36.         except Exception as e:
  37.             # print(f"Failed to start {script_name}: {str(e)}")
  38.             pass
  39.         finally:
  40.             with self.lock:
  41.                 if script_name in self.processes:
  42.                     del self.processes[script_name]
  44.     def create_daemon_thread(self, script_name: str) -> threading.Thread:
  45.         thread = threading.Thread(target=lambda: self.run_script(script_name))
  46.         thread.daemon = True
  47.         return thread
  49.     def restart_service(self, script_name: str):
  50.         with self.lock:
  51.             if script_name in self.threads and self.threads[script_name].is_alive():
  52.                 return
  53.             # print(f"Restarting {script_name}...")
  54.             thread = self.create_daemon_thread(script_name)
  55.             thread.start()
  56.             self.threads[script_name] = thread
  58.     def monitor(self):
  59.         try:
  60.             while True:
  61.                 with self.lock:
  62.                     for script in self.services.copy():
  63.                         thread = self.threads.get(script)
  64.                         if not thread or not thread.is_alive():
  65.                             # print(f"Warning: Service {script} died, restarting...")
  66.                             self.restart_service(script)
  67.                 time.sleep(5)
  68.         except KeyboardInterrupt:
  69.             # print("Received keyboard interrupt, shutting down...")
  70.             sys.exit(0)
  72. if __name__ == "__main__":
  73.     manager = ServiceManager()
  75.     for script in manager.services:
  76.         manager.restart_service(script)
  78.     monitor_thread = threading.Thread(target=manager.monitor)
  79.     monitor_thread.start()
  80.     monitor_thread.join()

/etc/init.d/Start start
ps | grep python3

添加新批注
在作者公开此批注前，只有你和作者可见。
回复批注