(temp) 整理檔案與註解
parent
b0f1bd56f5
commit
7ce094d211
@ -1,45 +0,0 @@
|
||||
|
||||
|
||||
|
||||
import rclpy
|
||||
from rclpy.node import Node
|
||||
from std_msgs.msg import String
|
||||
import time
|
||||
|
||||
# import mavros_msgs.srv
|
||||
|
||||
class TalkerNode(Node):
|
||||
def __init__(self):
|
||||
start_time = time.time()
|
||||
super().__init__('talker_node')
|
||||
end_time = time.time()
|
||||
print(f"Node initialization took {end_time - start_time:.2f} seconds")
|
||||
|
||||
self.publisher_ = self.create_publisher(String, 'hahatest/_1', 10)
|
||||
self.timer = self.create_timer(1.0, self.timer_callback) # 每秒執行一次
|
||||
self.get_logger().info('TalkerNode has been started.')
|
||||
|
||||
def timer_callback(self):
|
||||
msg = String()
|
||||
msg.data = 'Hello, ROS 2!'
|
||||
self.publisher_.publish(msg)
|
||||
self.get_logger().info(f'Published: "{msg.data}"')
|
||||
|
||||
def main(args=None):
|
||||
rclpy.init(args=args)
|
||||
node = TalkerNode()
|
||||
|
||||
print("Before sleep")
|
||||
time.sleep(5) # 等待 5 秒鐘
|
||||
print("After sleep")
|
||||
try:
|
||||
start_time = time.time()
|
||||
while time.time() - start_time < 10: # 持續 10 秒鐘
|
||||
rclpy.spin_once(node)
|
||||
time.sleep(1) # 每秒執行一次
|
||||
finally:
|
||||
node.destroy_node()
|
||||
rclpy.shutdown()
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@ -0,0 +1,46 @@
|
||||
from pymavlink.dialects.v20 import common as mavlink
|
||||
|
||||
class NullBuffer:
|
||||
def write(self, data): pass
|
||||
def seek(self, pos): pass
|
||||
def tell(self): return 0
|
||||
|
||||
class CapturingBuffer:
|
||||
def __init__(self):
|
||||
self.last_data = b''
|
||||
def write(self, data):
|
||||
self.last_data = bytes(data)
|
||||
def seek(self, pos): pass
|
||||
def tell(self): return 0
|
||||
|
||||
# 1. 初始化（僅作為編碼器）
|
||||
# file 參數可以是任何擁有 write() 方法的物件，這裡用 BytesIO 模擬
|
||||
# 初始化方法一：使用 BytesIO
|
||||
# import io
|
||||
# out_buf = io.BytesIO()
|
||||
# mav = mavlink.MAVLink(out_buf, srcSystem=1, srcComponent=191)
|
||||
|
||||
# 初始化方法二：使用自定義的 NullBuffer
|
||||
mav = mavlink.MAVLink(NullBuffer(), srcSystem=1, srcComponent=191)
|
||||
mav.seq = 254
|
||||
|
||||
# 2. 建立心跳包並取得二進制數據
|
||||
# 連續產出「不同」的訊息物件
|
||||
for i in range(3):
|
||||
msg = mav.heartbeat_encode(mavlink.MAV_TYPE_GCS, 0, 0, 0, 0)
|
||||
data = msg.pack(mav)
|
||||
mav.seq = (mav.seq + 1) & 0xFF
|
||||
# MAVLink 2 的序列號在第 3 個 Byte (Index 2)
|
||||
print(f"第 {i} 次發送, Seq: {data[4]}, Raw: {data.hex()}")
|
||||
|
||||
|
||||
|
||||
print("分隔線")
|
||||
|
||||
buf = CapturingBuffer()
|
||||
mav_buf2 = mavlink.MAVLink(buf, srcSystem=1, srcComponent=191)
|
||||
|
||||
for i in range(3):
|
||||
mav_buf2.heartbeat_send(mavlink.MAV_TYPE_GCS, 0, 0, 0, 0)
|
||||
data = buf.last_data
|
||||
print(f"第 {i} 次發送, Seq: {data[4]}, Raw: {data.hex()}")
|
||||
File diff suppressed because it is too large
Load Diff
@ -0,0 +1,71 @@
|
||||
import asyncio
|
||||
import serial_asyncio
|
||||
|
||||
SERIAL_PORT = '/dev/ttyUSB0' # 修改為你的 serial port
|
||||
SERIAL_BAUDRATE = 115200 # 修改為你的 baudrate
|
||||
UDP_REMOTE_IP = '192.168.1.100' # 修改為目標 IP
|
||||
UDP_REMOTE_PORT = 5005 # 修改為目標 port
|
||||
UDP_LOCAL_PORT = 5006 # 本地 UDP 監聽 port
|
||||
|
||||
class SerialToUDP(asyncio.Protocol):
|
||||
def __init__(self, udp_transport):
|
||||
self.udp_transport = udp_transport
|
||||
|
||||
def connection_made(self, transport):
|
||||
self.transport = transport
|
||||
|
||||
def data_received(self, data):
|
||||
# Serial 收到資料，轉發到 UDP
|
||||
self.udp_transport.sendto(data, (UDP_REMOTE_IP, UDP_REMOTE_PORT))
|
||||
|
||||
def write_to_serial(self, data):
|
||||
self.transport.write(data)
|
||||
|
||||
class UDPToSerial(asyncio.DatagramProtocol):
|
||||
def __init__(self, serial_proto):
|
||||
self.serial_proto = serial_proto
|
||||
|
||||
def datagram_received(self, data, addr):
|
||||
# UDP 收到資料，轉發到 Serial
|
||||
self.serial_proto.write_to_serial(data)
|
||||
|
||||
async def main():
|
||||
loop = asyncio.get_running_loop()
|
||||
|
||||
# 定義協議工廠函數
|
||||
def create_empty_protocol():
|
||||
return asyncio.DatagramProtocol()
|
||||
|
||||
# 建立 UDP1 傳輸
|
||||
udp_transport, _ = await loop.create_datagram_endpoint(
|
||||
create_empty_protocol,
|
||||
local_addr=('0.0.0.0', UDP_LOCAL_PORT)
|
||||
)
|
||||
|
||||
# 建立 Serial 傳輸
|
||||
serial_proto = SerialToUDP(udp_transport)
|
||||
def get_serial_protocol():
|
||||
return serial_proto
|
||||
|
||||
_, serial_transport = await serial_asyncio.create_serial_connection(
|
||||
loop, get_serial_protocol, SERIAL_PORT, baudrate=SERIAL_BAUDRATE
|
||||
)
|
||||
|
||||
# 建立 UDP2 監聽
|
||||
udp_proto = UDPToSerial(serial_proto)
|
||||
def get_udp_protocol():
|
||||
return udp_proto
|
||||
|
||||
udp_listen_transport, _ = await loop.create_datagram_endpoint(
|
||||
get_udp_protocol,
|
||||
local_addr=('0.0.0.0', UDP_LOCAL_PORT)
|
||||
)
|
||||
|
||||
# 保持運行
|
||||
try:
|
||||
await asyncio.Future()
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
|
||||
if __name__ == '__main__':
|
||||
asyncio.run(main())
|
||||
@ -1,27 +0,0 @@
|
||||
#!/bin/bash
|
||||
|
||||
# 網站清單
|
||||
DOMAINS=("google.com" "smarter.nchu.edu.tw")
|
||||
|
||||
echo "網站 SSL 憑證剩餘天數："
|
||||
echo "---------------------------"
|
||||
|
||||
for domain in "${DOMAINS[@]}"; do
|
||||
end_date=$(echo | openssl s_client -servername "$domain" -connect "$domain:443" 2>/dev/null |
|
||||
openssl x509 -noout -enddate | cut -d= -f2)
|
||||
|
||||
end_timestamp=$(date -d "$end_date" +%s)
|
||||
now_timestamp=$(date +%s)
|
||||
|
||||
remaining_days=$(( (end_timestamp - now_timestamp) / 86400 ))
|
||||
|
||||
if [ $remaining_days -lt 0 ]; then
|
||||
status="已過期 ❌"
|
||||
elif [ $remaining_days -lt 15 ]; then
|
||||
status="即將到期 ⚠️"
|
||||
else
|
||||
status="正常 ✅"
|
||||
fi
|
||||
|
||||
printf "%-20s 到期日：%-25s 剩餘天數：%3d 天 %s\n" "$domain" "$end_date" "$remaining_days" "$status"
|
||||
done
|
||||
@ -0,0 +1,40 @@
|
||||
import socket
|
||||
import sys
|
||||
import select
|
||||
|
||||
# 設定來源 IP 與 Port
|
||||
SRC_IP = '127.0.0.1' # 監聽所有介面
|
||||
SRC_PORT = 16661 # 請自行修改
|
||||
|
||||
# 建立 UDP 監聽 socket
|
||||
src_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
src_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||||
src_sock.bind((SRC_IP, SRC_PORT))
|
||||
print(f"Listening for UDP on {SRC_IP}:{SRC_PORT}")
|
||||
|
||||
# 設定目標 Unix socket 路徑
|
||||
UNIX_SOCKET_PATH = '/tmp/unix_socket_mavlink.sock' # 請自行修改
|
||||
|
||||
# 建立 Unix socket 連線
|
||||
unix_sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
||||
try:
|
||||
unix_sock.connect(UNIX_SOCKET_PATH)
|
||||
except Exception as e:
|
||||
print(f"Failed to connect to unix socket: {e}")
|
||||
sys.exit(1)
|
||||
|
||||
while True:
|
||||
# 使用 select 監聽兩個 socket
|
||||
readable, _, _ = select.select([src_sock, unix_sock], [], [])
|
||||
for sock in readable:
|
||||
if sock is src_sock:
|
||||
data, addr = src_sock.recvfrom(4096)
|
||||
if data:
|
||||
unix_sock.sendall(data)
|
||||
# print(f"Received UDP data from {addr}: {data}") # debug
|
||||
# break # debug
|
||||
elif sock is unix_sock:
|
||||
data = unix_sock.recv(4096)
|
||||
if data:
|
||||
# 回送到最近收到資料的 UDP client
|
||||
src_sock.sendto(data, addr)
|
||||
Loading…
Reference in New Issue