1. プロジェクト概要
本プロジェクトは、ZWO ASIカメラ、Astromechanics EFレンズコントローラー、および Acuter Traverse マウントを統合制御し、AIを用いた高度な自動追尾と自動録画を行うシステムである。 遠方の微小なターゲットをOpenCVの動体検知で捕捉し、接近した際にYOLOv8の物体認識へシームレスに移行する「ハイブリッド・トラッキング」を実現している。
2. システム構成（3ファイル構成）
複雑化を避け、保守性を高めるために以下の3ファイルに分割している。
device_controllers.py: ハードウェア制御層（Acuterマウントのシリアル通信、EFレンズのフォーカス/絞り制御）。camera_yolo.py: 画像処理層（ZWO ASIからのフレーム取得、YOLOv8による推論、OpenCVによる動体検知、動画保存ワーカー）。main.py: UIおよび統合制御層（PyQt6によるGUI、PID制御（P制御）によるトラッキングアルゴリズム、描画オーバーレイ）。
3. 開発の経緯と主要なトラブルシューティング
開発過程において、以下の技術的課題を解決しシステムを安定化させた。
- 名前空間の衝突解決 (v2.33)
- [課題] 標準ライブラリの
hardwareとファイル名が衝突しインポートエラーが発生。 - [解決] ファイル名を
device_controllers.pyに変更して分離。
- [課題] 標準ライブラリの
- マウントのオーバーシュート・行き過ぎ問題の解決 (v2.34 – v2.35)
- [課題] 追尾時にターゲットが中心（デッドバンド）に入ってもマウントが止まらず行き過ぎる。
- [解決] 毎秒20回のコマンド送信によるシリアルバッファの渋滞が原因と判明。通信頻度を毎秒10回に落とし、不要な読み返し（
:J）を削除。また、ブレーキ時にはバッファを強制破棄し、Acuterの瞬時停止コマンド（:L）を叩き込む設計に変更。
- Auto Engage（自動検知）と録画の細切れ問題 (v2.36)
- [課題] 対象を少しでも見失うと別ターゲットと認識され、動画が細切れになる。対象外をクリックしてもAuto Engageが即座に再ロックしてしまう。
- [解決] ロスト猶予期間を大幅に延長（最大150フレーム/約5秒）。距離制限を撤廃し、画面内に同じラベルが再出現した場合は同一ターゲットとしてリカバリーする処理を追加。空クリック時はAuto Engageをオフにして強制キャンセルするよう仕様変更。
- ハイブリッド・トラッキングの実装 (v2.38)
- [課題] YOLOが認識できない遠方の小さなターゲットを追尾できない。
- [解決] OpenCVの輪郭抽出を用いた動体検知（Motion Detect）を追加。動体を追尾中にYOLOが同じ座標で物体を認識した場合、YOLOのトラッキング（ID保持）に自動でアップグレードする連携機構を実装。
- Slewモード（手動/追尾回転）の動作不能問題 (v2.39)
- [課題] 矢印キーや追尾時にマウントが回転しない。
- [解決] Sky-Watcher（Acuter）の通信プロトコル仕様に基づき、Slewモードでも必須となる実行コマンド（
:J）を正しく再実装。また、逆回転を指示された際はモーター保護のため必ず停止コマンド（:L）を挟む処理を追加。
/Users/mars/acuter/main.py
# CodeName: ZWO_ASI_Acuter_EF_YOLOv8_ContinuousTrack_AutoEngage_UI_Clean
# Version: 2.39.0
import sys
import os
import time
import cv2
import numpy as np
import zwoasi as asi
from PyQt6.QtCore import Qt, QTimer
from PyQt6.QtGui import QImage, QPixmap
from PyQt6.QtWidgets import (
QApplication, QCheckBox, QFormLayout, QGroupBox, QHBoxLayout, QLabel,
QMainWindow, QScrollArea, QSpinBox, QDoubleSpinBox, QStatusBar,
QVBoxLayout, QWidget, QLineEdit, QPushButton, QMessageBox, QComboBox,
QGridLayout, QTabWidget
)
from device_controllers import AcuterController, AstromechanicsEFController, PositionWorker, MoveWorker
from camera_yolo import download_model_files, YOLO_MODEL_FILE, VideoLabel, CaptureThread, ControlRow, VideoSaveWorker
class MainWindow(QMainWindow):
def __init__(self, sdk_path):
super().__init__()
self.setWindowTitle('ZWO ASI Camera + EF Lens + Acuter + YOLOv8 Complete (v2.39.0)')
self.resize(1300, 900)
asi.init(sdk_path)
if asi.get_num_cameras() == 0:
raise RuntimeError('ZWOカメラが見つかりません')
self.camera = asi.Camera(0)
self.camera.set_image_type(asi.ASI_IMG_RGB24)
cam_prop = self.camera.get_camera_property()
supported_bins = cam_prop.get('SupportedBins', [1])
self.current_bins = 2 if 2 in supported_bins else 1
current_w = cam_prop['MaxWidth'] // self.current_bins
current_h = cam_prop['MaxHeight'] // self.current_bins
self.camera.set_roi(start_x=0, start_y=0, width=current_w, height=current_h, bins=self.current_bins)
self.current_start_x = 0
self.current_start_y = 0
self.target_id = None
self.last_target_pos = None
self.last_target_name = None
self.current_detections = []
self.last_frame_size = (current_w, current_h)
self.latest_frame = None
self.last_motion_box = None
self.pixel_size = 0.0038
self.next_track_time = 0.0
self.lost_counter = 0
self.is_recording_event = False
self.video_buffer = []
self.home_deg = {1: None, 2: None}
self.debug_info_dxdy = "Target diff: N/A"
self.debug_info_azalt = "Cmd: N/A"
self.debug_info_status = "Status: Idle"
self.calib_state = 0
self.calib_timer = QTimer(self)
self.calib_timer.timeout.connect(self.calib_step)
self.calib_pts_prev = None
self.calib_gray_prev = None
self.acuter = AcuterController()
self.acuter_poll_timer = QTimer(self)
self.acuter_poll_timer.timeout.connect(self.poll_acuter_position)
self.ef_controller = AstromechanicsEFController()
self.capture_thread = None
self._build_ui()
self._update_ef_ui_state(False)
self._frame_count = 0
self._start_capture()
QTimer.singleShot(500, self.auto_connect_devices)
def auto_connect_devices(self):
port_ef = self.ef_port_edit.text().strip()
if port_ef:
self.ef_controller.port = port_ef
self.statusBar().showMessage("EFレンズ自動接続試行中...")
try:
if self.ef_controller.connect():
self._update_ef_ui_state(True)
self.btn_ef_connect.setText("接続済")
self.btn_ef_connect.setStyleSheet("background-color: #28a745; color: white;")
self.on_ef_refresh_position()
except Exception:
self.ef_controller.disconnect()
port_acuter = self.acuter_port_input.text().strip()
if port_acuter:
self.statusBar().showMessage("Acuterマウント自動接続試行中...")
try:
if self.acuter.connect(port_acuter):
self.btn_acuter_connect.setText("切断")
self.btn_acuter_connect.setStyleSheet("background-color: #28a745; color: white;")
self.lbl_acuter_status.setText("接続済み (GoTo/Slew有効)")
self.lbl_acuter_status.setStyleSheet("color: #00FF00; font-weight: bold;")
self._set_acuter_controls_state(True)
self.acuter_poll_timer.start(500)
except Exception:
self.statusBar().showMessage("自動接続に失敗しました。後で手動接続してください。")
def _start_capture(self):
if self.capture_thread and self.capture_thread.isRunning():
self.capture_thread.stop()
self.capture_thread = CaptureThread(self.camera, YOLO_MODEL_FILE)
self.capture_thread.target_classes = self.get_current_target_classes()
self.capture_thread.detection_enabled = self.cb_detection.isChecked()
self.capture_thread.motion_enabled = self.cb_motion.isChecked()
self.capture_thread.frame_ready.connect(self._on_frame)
self.capture_thread.error.connect(lambda msg: self.statusBar().showMessage(f"取得エラー: {msg}"))
self.capture_thread.start()
def _build_ui(self):
central = QWidget()
self.setCentralWidget(central)
root = QHBoxLayout(central)
root.setContentsMargins(15, 15, 15, 15)
root.setSpacing(15)
# ========== 左ペイン ==========
left_widget = QWidget()
left_layout = QVBoxLayout(left_widget)
left_layout.setContentsMargins(0, 0, 0, 0)
left_layout.setSpacing(15)
self.preview_label = VideoLabel('starting...')
self.preview_label.setMinimumSize(640, 480)
self.preview_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
self.preview_label.setStyleSheet("background-color: #000000; color: #FFFFFF; border-radius: 8px;")
self.preview_label.clicked.connect(self._on_preview_clicked)
self.preview_label.roi_selected.connect(self._on_roi_dragged)
left_layout.addWidget(self.preview_label, stretch=1)
self.ef_group = QGroupBox("Canon EF Lens (Astromechanics)")
ef_layout = QVBoxLayout(self.ef_group)
ef_conn_layout = QHBoxLayout()
self.ef_port_edit = QLineEdit("/dev/tty.usbserial-AK06UIRD")
ef_conn_layout.addWidget(QLabel("ポート:"))
ef_conn_layout.addWidget(self.ef_port_edit, stretch=1)
self.btn_ef_connect = QPushButton("接続")
self.btn_ef_connect.clicked.connect(self.on_ef_connect)
self.btn_ef_disconnect = QPushButton("切断")
self.btn_ef_disconnect.clicked.connect(self.on_ef_disconnect)
ef_conn_layout.addWidget(self.btn_ef_connect)
ef_conn_layout.addWidget(self.btn_ef_disconnect)
ef_layout.addLayout(ef_conn_layout)
ef_pos_move_row = QHBoxLayout()
self.lbl_ef_position = QLabel("—")
self.lbl_ef_position.setStyleSheet("color: white; background-color: #333; padding: 4px; border-radius: 4px;")
self.btn_ef_refresh = QPushButton("更新")
self.btn_ef_refresh.clicked.connect(self.on_ef_refresh_position)
self.spin_ef_target = QSpinBox()
self.spin_ef_target.setRange(0, 32767)
self.spin_ef_target.setValue(5000)
self.spin_ef_target.setSingleStep(100)
self.btn_ef_goto = QPushButton("移動")
self.btn_ef_goto.clicked.connect(self.on_ef_goto)
ef_pos_move_row.addWidget(QLabel("現在位置:"))
ef_pos_move_row.addWidget(self.lbl_ef_position)
ef_pos_move_row.addWidget(self.btn_ef_refresh)
ef_pos_move_row.addSpacing(20)
ef_pos_move_row.addWidget(QLabel("目標位置:"))
ef_pos_move_row.addWidget(self.spin_ef_target)
ef_pos_move_row.addWidget(self.btn_ef_goto)
ef_layout.addLayout(ef_pos_move_row)
ef_rel_ap_row = QHBoxLayout()
self.spin_ef_rel = QSpinBox()
self.spin_ef_rel.setRange(1, 5000)
self.spin_ef_rel.setValue(100)
self.spin_ef_rel.setSingleStep(50)
self.btn_ef_in = QPushButton("← IN (−)")
self.btn_ef_in.clicked.connect(lambda: self.on_ef_relative(-1))
self.btn_ef_out = QPushButton("OUT (+) →")
self.btn_ef_out.clicked.connect(lambda: self.on_ef_relative(+1))
self.spin_ef_aperture = QSpinBox()
self.spin_ef_aperture.setRange(0, 30)
self.spin_ef_aperture.setValue(0)
self.btn_ef_aperture = QPushButton("設定")
self.btn_ef_aperture.clicked.connect(self.on_ef_set_aperture)
ef_rel_ap_row.addWidget(QLabel("相対:"))
ef_rel_ap_row.addWidget(self.spin_ef_rel)
ef_rel_ap_row.addWidget(self.btn_ef_in)
ef_rel_ap_row.addWidget(self.btn_ef_out)
ef_rel_ap_row.addSpacing(20)
ef_rel_ap_row.addWidget(QLabel("絞り(0=開放):"))
ef_rel_ap_row.addWidget(self.spin_ef_aperture)
ef_rel_ap_row.addWidget(self.btn_ef_aperture)
ef_rel_ap_row.addStretch()
ef_layout.addLayout(ef_rel_ap_row)
left_layout.addWidget(self.ef_group)
root.addWidget(left_widget, stretch=3)
# ========== 右ペイン ==========
right_scroll = QScrollArea()
right_scroll.setWidgetResizable(True)
right_scroll.setMinimumWidth(430)
right_scroll.setFrameShape(QScrollArea.Shape.NoFrame)
right_panel = QWidget()
right_layout = QVBoxLayout(right_panel)
right_layout.setContentsMargins(0, 0, 10, 0)
self.ai_group = QGroupBox("AI Object Tracking")
ai_layout = QVBoxLayout(self.ai_group)
self.cb_detection = QCheckBox("YOLOv8 トラッキングを有効にする (ID保持)")
self.cb_detection.setChecked(False)
self.cb_detection.toggled.connect(self.on_detection_toggled)
ai_layout.addWidget(self.cb_detection)
self.cb_motion = QCheckBox("動体検知 (Motion Detect) [YOLO認識前の小目標用]")
self.cb_motion.setChecked(True)
self.cb_motion.setStyleSheet("color: #ffcc00; font-weight: bold;")
self.cb_motion.toggled.connect(self.on_motion_toggled)
ai_layout.addWidget(self.cb_motion)
info_label = QLabel("※クリックで「対象ロックオン」、ドラッグで「ROI切り出し」")
info_label.setStyleSheet("color: #aaaaaa; font-size: 11px;")
ai_layout.addWidget(info_label)
targets_layout = QHBoxLayout()
self.cb_person = QCheckBox("人")
self.cb_bicycle = QCheckBox("自転車")
self.cb_car = QCheckBox("車")
self.cb_airplane = QCheckBox("航空機")
self.cb_bird = QCheckBox("鳥")
for cb in [self.cb_person, self.cb_bicycle, self.cb_car, self.cb_airplane, self.cb_bird]:
cb.setChecked(True)
cb.toggled.connect(self.update_target_classes)
targets_layout.addWidget(cb)
ai_layout.addLayout(targets_layout)
self.cb_auto_engage = QCheckBox("選択対象の自動検知＆録画 (Auto Engage)")
self.cb_auto_engage.setChecked(False)
self.cb_auto_engage.setStyleSheet("color: #5c9eff; font-weight: bold;")
ai_layout.addWidget(self.cb_auto_engage)
right_layout.addWidget(self.ai_group)
self.acuter_group = QGroupBox("Acuter Traverse Control")
self.acuter_layout = QVBoxLayout(self.acuter_group)
acuter_conn_layout = QHBoxLayout()
self.acuter_port_input = QLineEdit("/dev/cu.usbmodem4E94509B34001")
self.btn_acuter_connect = QPushButton("接続")
self.btn_acuter_connect.clicked.connect(self.toggle_acuter_connection)
acuter_conn_layout.addWidget(self.acuter_port_input)
acuter_conn_layout.addWidget(self.btn_acuter_connect)
self.acuter_layout.addLayout(acuter_conn_layout)
self.lbl_acuter_status = QLabel("未接続")
self.lbl_acuter_status.setStyleSheet("color: red; font-weight: bold;")
self.acuter_layout.addWidget(self.lbl_acuter_status)
self.btn_auto_track = QPushButton("Auto Tracking (OFF / 待機)")
self.btn_auto_track.setCheckable(True)
self.btn_auto_track.setFixedHeight(60)
self.btn_auto_track.setStyleSheet("font-size: 18px; font-weight: bold; background-color: #555555; color: white;")
self.btn_auto_track.toggled.connect(self.on_auto_track_toggled)
self.acuter_layout.addWidget(self.btn_auto_track)
pos_layout = QHBoxLayout()
self.lbl_acuter_az = QLabel("Az : --.- °")
self.lbl_acuter_az.setStyleSheet("font-family: 'Menlo', 'Consolas', 'Courier New'; font-size: 16px; font-weight: bold; color: white; background-color: #333; padding: 4px; border-radius: 4px;")
self.lbl_acuter_alt = QLabel("Alt: --.- °")
self.lbl_acuter_alt.setStyleSheet("font-family: 'Menlo', 'Consolas', 'Courier New'; font-size: 16px; font-weight: bold; color: white; background-color: #333; padding: 4px; border-radius: 4px;")
pos_layout.addWidget(self.lbl_acuter_az)
pos_layout.addWidget(self.lbl_acuter_alt)
self.acuter_layout.addLayout(pos_layout)
speed_form = QFormLayout()
self.acuter_speed_combo = QComboBox()
self.acuter_speed_combo.addItems(["1.0", "5.0", "10.0", "15.0"])
self.acuter_speed_combo.setCurrentText("15.0")
speed_form.addRow("最大回転速度(度/秒):", self.acuter_speed_combo)
self.acuter_layout.addLayout(speed_form)
goto_layout = QHBoxLayout()
self.acuter_az_input = QLineEdit("10.0")
self.acuter_az_input.setMaximumWidth(50)
self.btn_az_goto = QPushButton("Az GoTo")
self.btn_az_goto.setEnabled(False)
self.btn_az_goto.clicked.connect(lambda: self.acuter.start_move(1, float(self.acuter_az_input.text()), exact_goto=True))
self.acuter_alt_input = QLineEdit("10.0")
self.acuter_alt_input.setMaximumWidth(50)
self.btn_alt_goto = QPushButton("Alt GoTo")
self.btn_alt_goto.setEnabled(False)
self.btn_alt_goto.clicked.connect(lambda: self.acuter.start_move(2, float(self.acuter_alt_input.text()), exact_goto=True))
goto_layout.addWidget(QLabel("Az:"))
goto_layout.addWidget(self.acuter_az_input)
goto_layout.addWidget(QLabel("°"))
goto_layout.addWidget(self.btn_az_goto)
goto_layout.addSpacing(15)
goto_layout.addWidget(QLabel("Alt:"))
goto_layout.addWidget(self.acuter_alt_input)
goto_layout.addWidget(QLabel("°"))
goto_layout.addWidget(self.btn_alt_goto)
goto_layout.addStretch()
self.acuter_layout.addLayout(goto_layout)
dpad_layout = QGridLayout()
self.btn_up = QPushButton("▲")
self.btn_left = QPushButton("◀")
self.btn_stop_center = QPushButton("■ STOP")
self.btn_stop_center.setStyleSheet("background-color: #d9534f; color: white; font-weight: bold;")
self.btn_right = QPushButton("▶")
self.btn_down = QPushButton("▼")
huge_angle = 1000.0
self.btn_up.pressed.connect(lambda: self.acuter.start_move(2, -huge_angle, speed_deg_sec=float(self.acuter_speed_combo.currentText())))
self.btn_down.pressed.connect(lambda: self.acuter.start_move(2, huge_angle, speed_deg_sec=float(self.acuter_speed_combo.currentText())))
self.btn_left.pressed.connect(lambda: self.acuter.start_move(1, -huge_angle, speed_deg_sec=float(self.acuter_speed_combo.currentText())))
self.btn_right.pressed.connect(lambda: self.acuter.start_move(1, huge_angle, speed_deg_sec=float(self.acuter_speed_combo.currentText())))
self.btn_up.released.connect(lambda: self.acuter.stop_axis(2))
self.btn_down.released.connect(lambda: self.acuter.stop_axis(2))
self.btn_left.released.connect(lambda: self.acuter.stop_axis(1))
self.btn_right.released.connect(lambda: self.acuter.stop_axis(1))
self.btn_stop_center.clicked.connect(self.acuter.emergency_stop)
dpad_layout.addWidget(self.btn_up, 0, 1)
dpad_layout.addWidget(self.btn_left, 1, 0)
dpad_layout.addWidget(self.btn_stop_center, 1, 1)
dpad_layout.addWidget(self.btn_right, 1, 2)
dpad_layout.addWidget(self.btn_down, 2, 1)
self.acuter_layout.addLayout(dpad_layout)
right_layout.addWidget(self.acuter_group)
# 3. ZWO ASI カメラ設定
asi_group = QGroupBox('ZWO ASI & Tracking Settings')
asi_layout = QVBoxLayout(asi_group)
self.cam_tabs = QTabWidget()
self.tab_main = QWidget()
self.tab_adv = QWidget()
form_main = QFormLayout(self.tab_main)
form_adv = QFormLayout(self.tab_adv)
# ★ Mainタブにはキャリブレーションボタンのみを配置
self.btn_auto_calib = QPushButton("Run Auto Calibration")
self.btn_auto_calib.clicked.connect(self.start_auto_calib)
self.btn_auto_calib.setStyleSheet("background-color: #28a745; color: white; font-weight: bold; font-size: 16px; padding: 12px;")
form_main.addRow(self.btn_auto_calib)
# ★ その他の設定はすべてAdvancedタブへ移動
track_calib_group = QGroupBox("Tracking Settings & Axis Mapping")
track_calib_layout = QFormLayout(track_calib_group)
self.focal_spin = QDoubleSpinBox()
self.focal_spin.setRange(1.0, 2000.0)
self.focal_spin.setValue(55.0) # 初期値を55.0に変更
self.kp_spin = QDoubleSpinBox()
self.kp_spin.setRange(0.01, 5.0)
self.kp_spin.setSingleStep(0.1)
self.kp_spin.setValue(0.8)
self.deadband_spin = QSpinBox()
self.deadband_spin.setRange(5, 200)
self.deadband_spin.setValue(40)
self.combo_x_axis = QComboBox()
self.combo_x_axis.addItems(["Az (Axis 1)", "Alt (Axis 2)"])
self.combo_x_axis.setCurrentIndex(0)
self.combo_y_axis = QComboBox()
self.combo_y_axis.addItems(["Alt (Axis 2)", "Az (Axis 1)"])
self.combo_y_axis.setCurrentIndex(0)
self.cb_inv_x = QCheckBox("Invert X-Axis (Reverse)")
self.cb_inv_y = QCheckBox("Invert Y-Axis (Reverse)")
self.cb_inv_y.setChecked(True)
test_move_widget = QWidget()
test_move_grid = QGridLayout(test_move_widget)
self.btn_test_alt_p = QPushButton("Alt +1.0°")
self.btn_test_alt_m = QPushButton("Alt -1.0°")
self.btn_test_az_m = QPushButton("Az -1.0°")
self.btn_test_az_p = QPushButton("Az +1.0°")
self.btn_test_alt_p.clicked.connect(lambda: self.acuter.start_move(2, 1.0, exact_goto=True, speed_deg_sec=2.0))
self.btn_test_alt_m.clicked.connect(lambda: self.acuter.start_move(2, -1.0, exact_goto=True, speed_deg_sec=2.0))
self.btn_test_az_m.clicked.connect(lambda: self.acuter.start_move(1, -1.0, exact_goto=True, speed_deg_sec=2.0))
self.btn_test_az_p.clicked.connect(lambda: self.acuter.start_move(1, 1.0, exact_goto=True, speed_deg_sec=2.0))
self.calib_buttons = [self.btn_test_alt_p, self.btn_test_alt_m, self.btn_test_az_m, self.btn_test_az_p, self.btn_auto_calib]
for btn in self.calib_buttons: btn.setEnabled(False)
test_move_grid.addWidget(self.btn_test_alt_p, 0, 1)
test_move_grid.addWidget(self.btn_test_az_m, 1, 0)
test_move_grid.addWidget(self.btn_test_alt_m, 1, 1)
test_move_grid.addWidget(self.btn_test_az_p, 1, 2)
track_calib_layout.addRow("Focal Length (mm):", self.focal_spin)
track_calib_layout.addRow("Tracking Gain (Kp):", self.kp_spin)
track_calib_layout.addRow("Deadband (px):", self.deadband_spin)
track_calib_layout.addRow("Camera X-Axis:", self.combo_x_axis)
track_calib_layout.addRow("", self.cb_inv_x)
track_calib_layout.addRow("Camera Y-Axis:", self.combo_y_axis)
track_calib_layout.addRow("", self.cb_inv_y)
track_calib_layout.addRow("Test Move:", test_move_widget)
form_adv.addRow(track_calib_group)
cam_prop = self.camera.get_camera_property()
bin_group = QGroupBox("Sensor Binning")
bin_layout = QHBoxLayout(bin_group)
self.bin_combo = QComboBox()
for b in cam_prop['SupportedBins']:
if b != 0: self.bin_combo.addItem(f"Bin {b}x{b}", b)
self.bin_combo.setCurrentText(f"Bin {self.current_bins}x{self.current_bins}")
self.btn_apply_bin = QPushButton("Apply Binning")
self.btn_apply_bin.clicked.connect(self.on_apply_binning)
bin_layout.addWidget(QLabel("Binning:"))
bin_layout.addWidget(self.bin_combo)
bin_layout.addWidget(self.btn_apply_bin)
roi_group = QGroupBox("Sensor ROI (Resolution)")
roi_layout = QGridLayout(roi_group)
current_w = cam_prop['MaxWidth'] // self.current_bins
current_h = cam_prop['MaxHeight'] // self.current_bins
self.roi_w_spin = QSpinBox()
self.roi_w_spin.setRange(8, current_w)
self.roi_w_spin.setSingleStep(8)
self.roi_w_spin.setValue(current_w)
self.roi_h_spin = QSpinBox()
self.roi_h_spin.setRange(2, current_h)
self.roi_h_spin.setSingleStep(2)
self.roi_h_spin.setValue(current_h)
self.btn_apply_roi = QPushButton("Apply ROI (Center)")
self.btn_apply_roi.clicked.connect(self.on_apply_roi_center)
self.btn_reset_roi = QPushButton("Reset to Full Frame")
self.btn_reset_roi.clicked.connect(self.on_reset_roi)
self.btn_reset_roi.setStyleSheet("background-color: #555555;")
roi_layout.addWidget(QLabel("Width:"), 0, 0)
roi_layout.addWidget(self.roi_w_spin, 0, 1)
roi_layout.addWidget(QLabel("Height:"), 1, 0)
roi_layout.addWidget(self.roi_h_spin, 1, 1)
roi_layout.addWidget(self.btn_apply_roi, 2, 0, 1, 2)
roi_layout.addWidget(self.btn_reset_roi, 3, 0, 1, 2)
form_adv.addRow(roi_group)
form_adv.addRow(bin_group)
self.rows = []
for name, caps in sorted(self.camera.get_controls().items()):
row = ControlRow(self.camera, caps, lambda: self.statusBar().showMessage('ASIパラメータ更新', 1000))
self.rows.append(row)
# ★ カメラ設定も全て Advanced へ移動
form_adv.addRow(name, row)
self.cam_tabs.addTab(self.tab_main, "Main Controls")
self.cam_tabs.addTab(self.tab_adv, "Advanced")
asi_layout.addWidget(self.cam_tabs)
right_layout.addWidget(asi_group)
right_layout.addStretch()
right_scroll.setWidget(right_panel)
root.addWidget(right_scroll, stretch=2)
self.setStatusBar(QStatusBar())
def _update_ef_ui_state(self, connected: bool):
self.btn_ef_connect.setEnabled(not connected)
self.btn_ef_disconnect.setEnabled(connected)
self.btn_ef_goto.setEnabled(connected)
self.btn_ef_in.setEnabled(connected)
self.btn_ef_out.setEnabled(connected)
def on_ef_connect(self):
try:
self.ef_controller.port = self.ef_port_edit.text()
if self.ef_controller.connect():
self._update_ef_ui_state(True)
self.on_ef_refresh_position()
except: pass
def on_ef_disconnect(self):
self.ef_controller.disconnect()
self._update_ef_ui_state(False)
def on_ef_refresh_position(self):
if not self.ef_controller.is_connected: return
pos = self.ef_controller.get_position()
if pos is not None:
self.lbl_ef_position.setText(str(pos))
self.spin_ef_target.setValue(pos)
def on_ef_goto(self):
if self.ef_controller.is_connected:
self.ef_controller.move_absolute(self.spin_ef_target.value())
QTimer.singleShot(1000, self.on_ef_refresh_position)
def on_ef_relative(self, direction):
if self.ef_controller.is_connected:
target = self.spin_ef_target.value() + (self.spin_ef_rel.value() * direction)
self.ef_controller.move_absolute(target)
QTimer.singleShot(1000, self.on_ef_refresh_position)
def on_ef_set_aperture(self):
if self.ef_controller.is_connected:
self.ef_controller.set_aperture(self.spin_ef_aperture.value())
def toggle_acuter_connection(self):
if self.acuter.is_connected:
self.acuter.disconnect()
self.btn_acuter_connect.setText("接続")
self._set_acuter_controls_state(False)
else:
if self.acuter.connect(self.acuter_port_input.text()):
self.btn_acuter_connect.setText("切断")
self._set_acuter_controls_state(True)
self.acuter_poll_timer.start(500)
def _set_acuter_controls_state(self, state):
self.btn_az_goto.setEnabled(state)
self.btn_alt_goto.setEnabled(state)
self.btn_up.setEnabled(state)
self.btn_down.setEnabled(state)
self.btn_left.setEnabled(state)
self.btn_right.setEnabled(state)
self.btn_stop_center.setEnabled(state)
for btn in self.calib_buttons: btn.setEnabled(state)
def poll_acuter_position(self):
self.acuter.poll_position()
d1, d2 = self.acuter.current_deg.get(1), self.acuter.current_deg.get(2)
if d1 is not None: self.lbl_acuter_az.setText(f"Az : {d1:+8.3f} °")
if d2 is not None: self.lbl_acuter_alt.setText(f"Alt: {d2:+8.3f} °")
if self.capture_thread:
self.capture_thread.camera_is_moving = self.acuter.axis_moving.get(1, False) or self.acuter.axis_moving.get(2, False)
def on_auto_track_toggled(self, checked):
if checked:
self.btn_auto_track.setText("Tracking (ON)")
self.btn_auto_track.setStyleSheet("background-color: #d9534f; color: white; font-size: 18px; font-weight: bold;")
else:
self.btn_auto_track.setText("Tracking (OFF)")
self.btn_auto_track.setStyleSheet("background-color: #555; color: white; font-size: 18px; font-weight: bold;")
self.acuter.emergency_stop()
if getattr(self, 'is_recording_event', False):
self.stop_event_record_and_return()
def on_motion_toggled(self, checked):
if self.capture_thread:
self.capture_thread.motion_enabled = checked
if not checked:
self.capture_thread.cv2_tracker = None
def start_event_record_and_track(self):
self.home_deg = {1: self.acuter.current_deg.get(1), 2: self.acuter.current_deg.get(2)}
self.is_recording_event = True
self.video_buffer = []
if not self.btn_auto_track.isChecked(): self.btn_auto_track.setChecked(True)
def stop_event_record_and_return(self):
self.is_recording_event = False
if self.video_buffer:
save_dir = "/Users/mars/acuter/videos"
os.makedirs(save_dir, exist_ok=True)
filename = os.path.join(save_dir, time.strftime("track_%Y%m%d_%H%M%S.mp4"))
self.statusBar().showMessage(f"動画を保存中... {filename}", 5000)
self.video_saver = VideoSaveWorker(self.video_buffer, filename, 30.0)
self.video_saver.finished_ok.connect(lambda f: self.statusBar().showMessage(f"保存完了: {f}", 5000))
self.video_saver.error.connect(lambda err: self.statusBar().showMessage(f"保存エラー: {err}", 5000))
self.video_saver.start()
self.video_buffer = []
for axis in (1, 2):
h_deg = self.home_deg.get(axis)
c_deg = self.acuter.current_deg.get(axis)
if h_deg is not None and c_deg is not None:
diff = h_deg - c_deg
if diff > 180: diff -= 360
elif diff < -180: diff += 360
if abs(diff) > 0.05:
self.acuter.start_move(axis, diff, exact_goto=True, speed_deg_sec=15.0)
def get_current_target_classes(self):
targets = []
if self.cb_person.isChecked(): targets.append(0)
if self.cb_bicycle.isChecked(): targets.append(1)
if self.cb_car.isChecked(): targets.append(2)
if self.cb_airplane.isChecked(): targets.append(4)
if self.cb_bird.isChecked(): targets.append(14)
return targets
def update_target_classes(self):
if self.capture_thread:
self.capture_thread.target_classes = self.get_current_target_classes()
def on_detection_toggled(self, checked):
if self.capture_thread:
self.capture_thread.detection_enabled = checked
if not checked:
self.target_id = None
self.btn_auto_track.setChecked(False)
def _on_preview_clicked(self, lx, ly):
scale = min(self.preview_label.width() / self.last_frame_size[0], self.preview_label.height() / self.last_frame_size[1])
ox = (self.preview_label.width() - self.last_frame_size[0] * scale) / 2
oy = (self.preview_label.height() - self.last_frame_size[1] * scale) / 2
fx, fy = (lx - ox) / scale, (ly - oy) / scale
clicked_id = None
for d in self.current_detections:
if d[0] <= fx <= d[2] and d[1] <= fy <= d[3]:
clicked_id = d[6]
break
if clicked_id is not None:
self.target_id = clicked_id
if self.capture_thread: self.capture_thread.yolo_target_locked = True
self.lost_counter = 0
if not getattr(self, 'is_recording_event', False):
self.start_event_record_and_track()
if not self.btn_auto_track.isChecked():
self.btn_auto_track.setChecked(True)
return
if getattr(self, 'last_motion_box', None) is not None:
mx, my, mw, mh = self.last_motion_box
if mx <= fx <= mx+mw and my <= fy <= my+mh:
self.target_id = None
if self.capture_thread: self.capture_thread.yolo_target_locked = False
if not getattr(self, 'is_recording_event', False):
self.start_event_record_and_track()
if not self.btn_auto_track.isChecked():
self.btn_auto_track.setChecked(True)
return
self.target_id = None
if self.capture_thread: self.capture_thread.yolo_target_locked = False
self.cb_auto_engage.setChecked(False)
self.btn_auto_track.setChecked(False)
self.acuter.emergency_stop()
def _on_roi_dragged(self, lx, ly, lw, lh): pass
def on_reset_roi(self):
cam_prop = self.camera.get_camera_property()
max_w = cam_prop['MaxWidth'] // self.current_bins
max_h = cam_prop['MaxHeight'] // self.current_bins
self.camera.set_roi(start_x=0, start_y=0, width=max_w, height=max_h, bins=self.current_bins)
self._start_capture()
def on_apply_binning(self):
b = self.bin_combo.currentData()
self.camera.set_roi(bins=b)
self.current_bins = b
self._start_capture()
def on_apply_roi_center(self):
w, h = (self.roi_w_spin.value() // 8) * 8, (self.roi_h_spin.value() // 2) * 2
cam_prop = self.camera.get_camera_property()
sx = ((cam_prop['MaxWidth'] // self.current_bins - w) // 2 // 4) * 4
sy = ((cam_prop['MaxHeight'] // self.current_bins - h) // 2 // 2) * 2
self.camera.set_roi(start_x=sx, start_y=sy, width=w, height=h, bins=self.current_bins)
self._start_capture()
def _on_frame(self, frame: np.ndarray, detections: list, motion_box: object = None):
self._frame_count += 1
self.last_motion_box = motion_box
h, w = frame.shape[:2]
self.last_frame_size = (w, h)
self.current_detections = detections
self.latest_frame = frame.copy()
draw_frame = frame.copy()
cx, cy = w // 2, h // 2
db_px = self.deadband_spin.value()
cv2.circle(draw_frame, (cx, cy), db_px, (255, 255, 255), 1, cv2.LINE_AA)
cv2.line(draw_frame, (cx-20, cy), (cx+20, cy), (255,255,255), 1)
cv2.line(draw_frame, (cx, cy-20), (cx, cy+20), (255,255,255), 1)
motion_cx, motion_cy = None, None
if motion_box is not None:
mx, my, mw, mh = [int(v) for v in motion_box]
motion_cx, motion_cy = mx + mw//2, my + mh//2
cv2.rectangle(draw_frame, (mx, my), (mx+mw, my+mh), (0, 165, 255), 2)
cv2.putText(draw_frame, "MOTION", (mx, max(my-10, 10)), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 165, 255), 2)
if self.target_id is None and motion_box is not None and detections:
for d in detections:
dcx, dcy = (d[0]+d[2])//2, (d[1]+d[3])//2
if np.hypot(dcx - motion_cx, dcy - motion_cy) < 100:
self.target_id = d[6]
self.last_target_name = d[5]
if self.capture_thread: self.capture_thread.yolo_target_locked = True
self.statusBar().showMessage(f"Upgraded to YOLO Target: ID {self.target_id}", 3000)
break
if self.cb_auto_engage.isChecked() and self.target_id is None and detections:
best_det = max(detections, key=lambda d: d[4])
if best_det[6] is not None:
self.target_id = best_det[6]
self.last_target_name = best_det[5]
self.last_target_pos = ((best_det[0]+best_det[2])//2, (best_det[1]+best_det[3])//2)
self.lost_counter = 0
if self.capture_thread: self.capture_thread.yolo_target_locked = True
self.start_event_record_and_track()
if self.cb_auto_engage.isChecked() and self.target_id is None and motion_box is not None:
if not getattr(self, 'is_recording_event', False):
self.start_event_record_and_track()
is_tracking_active = False
tcx, tcy = None, None
tracking_color = (0, 255, 0)
if self.target_id is not None:
target_found = False
for d in detections:
if self.target_id == d[6]:
target_found = True
self.last_target_pos = ((d[0]+d[2])//2, (d[1]+d[3])//2)
break
if not target_found and self.last_target_pos:
best_d = None
min_dist = float('inf')
for d in detections:
if d[5] == self.last_target_name:
dist = np.hypot((((d[0]+d[2])//2) - self.last_target_pos[0]), (((d[1]+d[3])//2) - self.last_target_pos[1]))
if dist < min_dist:
min_dist = dist
best_d = d
if best_d is not None:
self.target_id = best_d[6]
target_found = True
self.lost_counter = 0
if target_found:
tcx, tcy = self.last_target_pos
is_tracking_active = True
tracking_color = (0, 0, 255)
self.lost_counter = 0
else:
self.lost_counter += 1
if self.last_target_pos:
gx, gy = self.last_target_pos
cv2.circle(draw_frame, (gx, gy), 15, (0,165,255), 2)
cv2.putText(draw_frame, "SEARCHING...", (gx+20, gy), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0,165,255), 2)
if self.lost_counter == 1:
self.acuter.emergency_stop()
self.debug_info_status = "Status: Target Lost (Braking)"
if self.lost_counter > 150:
self.debug_info_status = "Status: Target completely LOST! (Aborted)"
self.acuter.emergency_stop()
self.btn_auto_track.setChecked(False)
self.target_id = None
if self.capture_thread: self.capture_thread.yolo_target_locked = False
self.last_target_pos = None
self.lost_counter = 0
elif motion_box is not None:
tcx, tcy = motion_cx, motion_cy
is_tracking_active = True
tracking_color = (0, 165, 255)
for d in detections:
tid = d[6]
is_target = (self.target_id is not None and tid == self.target_id)
color = (0,0,255) if is_target else (0,255,0)
thickness = 4 if is_target else 2
cv2.rectangle(draw_frame, (d[0], d[1]), (d[2], d[3]), color, thickness)
id_str = f"ID:{tid} " if tid is not None else ""
label_text = f"{id_str}{d[5]} {d[4]*100:.1f}%"
if is_target:
label_text = "[LOCKED] " + label_text
cv2.putText(draw_frame, label_text, (d[0], max(d[1] - 10, 10)), cv2.FONT_HERSHEY_SIMPLEX, 0.7, color, 2)
if is_tracking_active and tcx is not None:
cv2.line(draw_frame, (int(tcx) - 15, int(tcy)), (int(tcx) + 15, int(tcy)), tracking_color, 2)
cv2.line(draw_frame, (int(tcx), int(tcy) - 15), (int(tcx), int(tcy) + 15), tracking_color, 2)
cv2.line(draw_frame, (int(tcx), int(tcy)), (cx, cy), (0, 255, 255), 1)
if self.btn_auto_track.isChecked() and self.calib_state == 0:
self._update_tracking(tcx, tcy, w, h)
if getattr(self, 'is_recording_event', False):
cv2.circle(draw_frame, (40, 40), 12, (0, 0, 255), -1)
cv2.putText(draw_frame, "REC", (60, 48), cv2.FONT_HERSHEY_SIMPLEX, 1.2, (0, 0, 255), 3)
self.video_buffer.append(draw_frame.copy())
cv2.putText(draw_frame, getattr(self, 'debug_info_dxdy', ''), (10, 90), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 255, 255), 2)
cv2.putText(draw_frame, getattr(self, 'debug_info_azalt', ''), (10, 120), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 255, 255), 2)
cv2.putText(draw_frame, getattr(self, 'debug_info_status', ''), (10, 150), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 255, 255), 2)
qimg = QImage(draw_frame.data, w, h, 3*w, QImage.Format.Format_BGR888)
self.preview_label.setPixmap(QPixmap.fromImage(qimg).scaled(self.preview_label.size(), Qt.AspectRatioMode.KeepAspectRatio))
def _update_tracking(self, tcx, tcy, w, h):
if not self.acuter.is_connected or time.time() < self.next_track_time: return
self.next_track_time = time.time() + 0.1
dx, dy = tcx - w/2.0, tcy - h/2.0
db = self.deadband_spin.value()
if abs(dx) < db and abs(dy) < db:
if self.acuter.axis_moving.get(1) or self.acuter.axis_moving.get(2):
self.acuter.emergency_stop()
self.debug_info_status = f"Status: Target Centered (<{db}px)"
return
deg_per_px = (self.pixel_size * self.current_bins / self.focal_spin.value()) * (180.0 / np.pi)
vx = dx * deg_per_px * self.kp_spin.value() * (-1.0 if self.cb_inv_x.isChecked() else 1.0)
vy = dy * deg_per_px * self.kp_spin.value() * (-1.0 if self.cb_inv_y.isChecked() else 1.0)
az_deg = vx if self.combo_x_axis.currentIndex() == 0 else vy
alt_deg = vx if self.combo_x_axis.currentIndex() == 1 else vy
spd = float(self.acuter_speed_combo.currentText())
db_deg = db * deg_per_px * self.kp_spin.value()
self.debug_info_dxdy = f"Target diff: dx={dx:.1f}px, dy={dy:.1f}px"
p_gain = 0.5
speed_az = max(0.1, min(spd, abs(az_deg) * p_gain))
speed_alt = max(0.1, min(spd, abs(alt_deg) * p_gain))
sent = False
if abs(az_deg) > db_deg:
self.acuter.start_move(1, az_deg, exact_goto=False, speed_deg_sec=speed_az)
sent = True
elif self.acuter.axis_moving.get(1):
self.acuter.stop_axis(1)
if abs(alt_deg) > db_deg:
self.acuter.start_move(2, alt_deg, exact_goto=False, speed_deg_sec=speed_alt)
sent = True
elif self.acuter.axis_moving.get(2):
self.acuter.stop_axis(2)
if sent:
self.debug_info_azalt = f"Speed: Az {speed_az:.1f} d/s, Alt {speed_alt:.1f} d/s"
self.debug_info_status = "Status: Continuous P-Control Tracking"
def start_auto_calib(self):
if not self.acuter.is_connected or self.latest_frame is None: return
self.btn_auto_track.setChecked(False)
self.calib_state = 1
self.calib_timer.start(1000)
def stop_calib(self, msg):
self.calib_state = 0
self.calib_timer.stop()
self.statusBar().showMessage(msg, 5000)
def calib_step(self):
if self.latest_frame is None: return
gray = cv2.cvtColor(self.latest_frame, cv2.COLOR_BGR2GRAY)
if self.calib_state == 1:
self.calib_pts_prev = cv2.goodFeaturesToTrack(gray, maxCorners=100, qualityLevel=0.01, minDistance=30)
if self.calib_pts_prev is None: return self.stop_calib("Failed")
self.calib_gray_prev = gray
self.acuter.start_move(1, 1.0, exact_goto=True, speed_deg_sec=2.0)
self.calib_wait = 3
self.calib_state = 2
elif self.calib_state == 2:
self.calib_wait -= 1
if self.calib_wait <= 0:
pts_next, status, _ = cv2.calcOpticalFlowPyrLK(self.calib_gray_prev, gray, self.calib_pts_prev, None)
diff = pts_next[status == 1] - self.calib_pts_prev[status == 1]
self.calib_dx1, self.calib_dy1 = np.median(diff[:, 0]), np.median(diff[:, 1])
self.acuter.start_move(1, -1.0, exact_goto=True, speed_deg_sec=2.0)
self.calib_wait = 3; self.calib_state = 3
elif self.calib_state == 3:
self.calib_wait -= 1
if self.calib_wait <= 0:
self.calib_pts_prev = cv2.goodFeaturesToTrack(gray, maxCorners=100, qualityLevel=0.01, minDistance=30)
self.calib_gray_prev = gray
self.acuter.start_move(2, 1.0, exact_goto=True, speed_deg_sec=2.0)
self.calib_wait = 3; self.calib_state = 4
elif self.calib_state == 4:
self.calib_wait -= 1
if self.calib_wait <= 0:
pts_next, status, _ = cv2.calcOpticalFlowPyrLK(self.calib_gray_prev, gray, self.calib_pts_prev, None)
diff = pts_next[status == 1] - self.calib_pts_prev[status == 1]
self.calib_dx2, self.calib_dy2 = np.median(diff[:, 0]), np.median(diff[:, 1])
self.acuter.start_move(2, -1.0, exact_goto=True, speed_deg_sec=2.0)
self.calib_wait = 3; self.calib_state = 5
elif self.calib_state == 5:
self.calib_wait -= 1
if self.calib_wait <= 0:
if abs(self.calib_dx1) > abs(self.calib_dy1):
self.combo_x_axis.setCurrentIndex(0); self.combo_y_axis.setCurrentIndex(0)
self.cb_inv_x.setChecked(bool(self.calib_dx1 > 0))
self.cb_inv_y.setChecked(bool(self.calib_dy2 > 0))
else:
self.combo_x_axis.setCurrentIndex(1); self.combo_y_axis.setCurrentIndex(1)
self.cb_inv_x.setChecked(bool(self.calib_dx2 > 0))
self.cb_inv_y.setChecked(bool(self.calib_dy1 > 0))
self.stop_calib("Success")
def closeEvent(self, event):
self.acuter.disconnect()
self.ef_controller.disconnect()
if self.capture_thread: self.capture_thread.stop()
self.camera.close()
super().closeEvent(event)
if __name__ == '__main__':
app = QApplication(sys.argv)
dark_stylesheet = """
QMainWindow { background-color: #2b2b2b; } QLabel { color: #e0e0e0; font-size: 13px; }
QGroupBox { color: #e0e0e0; border: 1px solid #555; border-radius: 6px; margin-top: 16px; padding-top: 15px; font-weight: bold; }
QGroupBox::title { subcontrol-origin: margin; subcontrol-position: top left; left: 10px; padding: 0 5px; }
QTabWidget::pane { border: 1px solid #555; } QTabBar::tab { background: #3c3c3c; color: white; padding: 8px 12px; }
QTabBar::tab:selected { background: #5c9eff; color: black; font-weight: bold; }
QComboBox, QLineEdit, QSpinBox, QDoubleSpinBox { background-color: #3c3c3c; color: white; border: 1px solid #555; padding: 4px; border-radius: 4px; }
QCheckBox { color: #e0e0e0; }
QPushButton { background-color: #3c3c3c; color: #ffffff; border: 1px solid #555; border-radius: 4px; padding: 6px; font-weight: bold; }
QPushButton:pressed { background-color: #5c9eff; color: #000; }
"""
app.setStyleSheet(dark_stylesheet)
window = MainWindow("/Users/mars/acuter/ASI_Camera_SDK/ASI_linux_mac_SDK_V1.41/lib/mac_arm64/libASICamera2.dylib")
window.show()
sys.exit(app.exec())/Users/mars/acuter/device_controllers.py
import time
import re
import serial
from typing import Optional
from PyQt6.QtCore import QThread, pyqtSignal
def hex_le_to_int(hex_str: str) -> int:
if len(hex_str) % 2 != 0:
hex_str = "0" + hex_str
reversed_hex = "".join([hex_str[i:i+2] for i in range(0, len(hex_str), 2)][::-1])
return int(reversed_hex, 16)
def int_to_hex_le(val: int, length_chars: int) -> str:
hex_str = f"{val:0{length_chars}X}"
reversed_hex = "".join([hex_str[i:i+2] for i in range(0, len(hex_str), 2)][::-1])
return reversed_hex
class AcuterController:
def __init__(self):
self.ser = None
self.cpr = {1: 1017435, 2: 1017435}
self.timer_freq = {1: 16000000, 2: 16000000}
self.current_deg = {1: None, 2: None}
self.axis_moving = {1: False, 2: False}
self.current_period = {1: None, 2: None}
self.current_dir = {1: None, 2: None}
@property
def is_connected(self):
return self.ser is not None and self.ser.is_open
def connect(self, port: str, baudrate: int = 115200) -> bool:
try:
self.ser = serial.Serial(port, baudrate, timeout=0.05)
for axis in (1, 2):
cpr_hex = self._send_cmd(f":a{axis}\r")
timer_hex = self._send_cmd(f":b{axis}\r")
if cpr_hex and "Error" not in cpr_hex:
self.cpr[axis] = hex_le_to_int(cpr_hex)
if timer_hex and "Error" not in timer_hex:
self.timer_freq[axis] = hex_le_to_int(timer_hex)
return True
except Exception as e:
self.disconnect()
raise e
def disconnect(self):
if self.is_connected:
self.emergency_stop()
self.ser.close()
self.ser = None
def _send_cmd(self, cmd: str) -> str:
if not self.is_connected: return ""
try:
self.ser.reset_input_buffer()
self.ser.write(cmd.encode('ascii'))
resp_bytes = self.ser.read_until(b'\r')
resp = resp_bytes.decode('ascii', errors='ignore').strip()
return resp[1:] if resp.startswith('=') else resp
except Exception:
return ""
def poll_position(self):
if not self.is_connected: return
for axis in (1, 2):
pos_hex = self._send_cmd(f":j{axis}\r")
if pos_hex and "Error" not in pos_hex and len(pos_hex) >= 5:
try:
current_pos = hex_le_to_int(pos_hex)
center = 0x800000 if len(pos_hex) <= 6 else 0x80000000
deg = ((current_pos - center) / float(self.cpr[axis])) * 360.0
self.current_deg[axis] = (deg + 180) % 360 - 180
except ValueError:
pass
def start_move(self, axis: int, angle_degrees: float, exact_goto: bool = False, speed_deg_sec: float = 10.0):
if not self.is_connected or speed_deg_sec <= 0 or angle_degrees == 0: return
try:
dir_char = "0" if angle_degrees >= 0 else "1"
steps_per_sec = (speed_deg_sec / 360.0) * self.cpr[axis]
period = max(1, int(self.timer_freq[axis] / steps_per_sec))
period_hex_le = int_to_hex_le(period, 6)
if exact_goto:
pos_hex = self._send_cmd(f":j{axis}\r")
if not pos_hex or "Error" in pos_hex: return
cur_pos = hex_le_to_int(pos_hex)
pos_len = len(pos_hex)
offset_steps = int(self.cpr[axis] * (angle_degrees / 360.0))
max_val = 1 << (pos_len * 4)
target_pos = (cur_pos + offset_steps) % max_val
target_hex_le = int_to_hex_le(target_pos, pos_len)
mode = "0" + dir_char
self._send_cmd(f":S{axis}{target_hex_le}\r")
self._send_cmd(f":I{axis}{period_hex_le}\r")
self._send_cmd(f":G{axis}{mode}\r")
self._send_cmd(f":J{axis}\r")
else:
if self.axis_moving.get(axis, False):
prev_dir = self.current_dir.get(axis)
prev_period = self.current_period.get(axis)
if prev_dir == dir_char and prev_period is not None:
if abs(period - prev_period) / float(prev_period) < 0.15:
return
elif prev_dir != dir_char:
self.ser.reset_output_buffer()
self._send_cmd(f":L{axis}\r")
time.sleep(0.05)
self.current_dir[axis] = dir_char
self.current_period[axis] = period
self.axis_moving[axis] = True
mode = "3" + dir_char
self._send_cmd(f":I{axis}{period_hex_le}\r")
self._send_cmd(f":G{axis}{mode}\r")
self._send_cmd(f":J{axis}\r")
except Exception:
pass
def stop_axis(self, axis: int):
if not self.is_connected: return
if self.axis_moving.get(axis, False):
self.ser.reset_output_buffer()
self._send_cmd(f":L{axis}\r")
self.axis_moving[axis] = False
self.current_period[axis] = None
self.current_dir[axis] = None
def emergency_stop(self):
if not self.is_connected: return
self.ser.reset_output_buffer()
self._send_cmd(":L1\r")
self._send_cmd(":L2\r")
self.axis_moving = {1: False, 2: False}
self.current_period = {1: None, 2: None}
self.current_dir = {1: None, 2: None}
class AstromechanicsEFController:
def __init__(self, port: str = "", baudrate: int = 38400, timeout: float = 1.0):
self.port = port
self.baudrate = baudrate
self.timeout = timeout
self.ser: Optional[serial.Serial] = None
def connect(self) -> bool:
try:
self.ser = serial.Serial(self.port, self.baudrate, timeout=self.timeout)
time.sleep(2)
self.ser.reset_input_buffer()
return self.get_position() is not None
except Exception as e:
self.ser = None
raise e
def disconnect(self):
if self.ser and self.ser.is_open:
self.ser.close()
self.ser = None
@property
def is_connected(self) -> bool:
return self.ser is not None and self.ser.is_open
def _send(self, cmd: str, expect_reply: bool = False) -> Optional[str]:
if not self.is_connected: return None
if not cmd.endswith("#"): cmd += "#"
self.ser.reset_input_buffer()
self.ser.write(cmd.encode("ascii"))
self.ser.flush()
if not expect_reply:
time.sleep(0.1)
return None
response = b""
start = time.time()
while time.time() - start < self.timeout:
if self.ser.in_waiting:
response += self.ser.read(self.ser.in_waiting)
if b"#" in response: break
time.sleep(0.1)
text = response.decode("ascii", errors="ignore").strip()
m = re.search(r"(\d+)#?", text)
return m.group(1) if m else None
def get_position(self) -> Optional[int]:
reply = self._send("P#", expect_reply=True)
return int(reply) if reply else None
def move_absolute(self, position: int):
self._send(f"M{position}#", expect_reply=False)
def set_aperture(self, index: int):
self._send(f"A{index:02d}#", expect_reply=False)
class PositionWorker(QThread):
position_ready = pyqtSignal(int)
error = pyqtSignal(str)
def __init__(self, controller):
super().__init__()
self.controller = controller
def run(self):
try:
pos = self.controller.get_position()
if pos is not None: self.position_ready.emit(pos)
else: self.error.emit("位置を取得できません")
except Exception as e:
self.error.emit(str(e))
class MoveWorker(QThread):
finished_ok = pyqtSignal(int)
error = pyqtSignal(str)
progress = pyqtSignal(int)
def __init__(self, controller, target, tolerance=5):
super().__init__()
self.controller = controller
self.target = target
self.tolerance = tolerance
def run(self):
try:
self.controller.move_absolute(self.target)
start = time.time()
while time.time() - start < 30.0:
pos = self.controller.get_position()
if pos is None:
time.sleep(0.2)
continue
self.progress.emit(pos)
if abs(pos - self.target) <= self.tolerance:
self.finished_ok.emit(pos)
return
time.sleep(0.25)
self.error.emit("移動タイムアウト")
except Exception as e:
self.error.emit(str(e))/Users/mars/acuter/camera_yolo.py
import os
import time
import urllib.request
import cv2
import numpy as np
import zwoasi as asi
from ultralytics import YOLO
from PyQt6.QtCore import Qt, QThread, pyqtSignal
from PyQt6.QtGui import QPainter, QPen, QColor
from PyQt6.QtWidgets import QLabel, QWidget, QHBoxLayout, QSlider, QSpinBox, QCheckBox
YOLO_MODEL_URL = "https://github.com/ultralytics/assets/releases/download/v8.4.0/yolov8n.pt"
YOLO_MODEL_FILE = "/Users/mars/acuter/yolov8n.pt"
def download_model_files():
if not os.path.exists(YOLO_MODEL_FILE):
print(f"Downloading {YOLO_MODEL_FILE} from GitHub. Please wait...")
try:
os.makedirs(os.path.dirname(YOLO_MODEL_FILE), exist_ok=True)
urllib.request.urlretrieve(YOLO_MODEL_URL, YOLO_MODEL_FILE)
print("Download completed successfully.")
except Exception as e:
print(f"Error downloading the model: {e}")
def create_tracker():
try:
return cv2.TrackerKCF_create()
except AttributeError:
try:
return cv2.TrackerMIL_create()
except AttributeError:
try:
return cv2.legacy.TrackerKCF_create()
except AttributeError:
return None
class VideoLabel(QLabel):
clicked = pyqtSignal(int, int)
roi_selected = pyqtSignal(int, int, int, int)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.start_point = None
self.end_point = None
self.is_drawing = False
def mousePressEvent(self, event):
if event.button() == Qt.MouseButton.LeftButton:
self.start_point = event.position().toPoint()
self.end_point = self.start_point
self.is_drawing = True
super().mousePressEvent(event)
def mouseMoveEvent(self, event):
if self.is_drawing:
self.end_point = event.position().toPoint()
self.update()
super().mouseMoveEvent(event)
def mouseReleaseEvent(self, event):
if event.button() == Qt.MouseButton.LeftButton and self.is_drawing:
self.end_point = event.position().toPoint()
self.is_drawing = False
self.update()
dist = (self.start_point.x() - self.end_point.x())**2 + (self.start_point.y() - self.end_point.y())**2
if dist < 25:
self.clicked.emit(self.start_point.x(), self.start_point.y())
else:
x1 = min(self.start_point.x(), self.end_point.x())
y1 = min(self.start_point.y(), self.end_point.y())
w = abs(self.start_point.x() - self.end_point.x())
h = abs(self.start_point.y() - self.end_point.y())
self.roi_selected.emit(x1, y1, w, h)
self.start_point = None
self.end_point = None
super().mouseReleaseEvent(event)
def paintEvent(self, event):
super().paintEvent(event)
if self.is_drawing and self.start_point and self.end_point:
painter = QPainter(self)
pen = QPen(QColor(0, 255, 255))
pen.setWidth(2)
pen.setStyle(Qt.PenStyle.DashLine)
painter.setPen(pen)
x = min(self.start_point.x(), self.end_point.x())
y = min(self.start_point.y(), self.end_point.y())
w = abs(self.start_point.x() - self.end_point.x())
h = abs(self.start_point.y() - self.end_point.y())
painter.drawRect(x, y, w, h)
painter.end()
class CaptureThread(QThread):
frame_ready = pyqtSignal(np.ndarray, list, object)
error = pyqtSignal(str)
def __init__(self, camera, model_file, parent=None):
super().__init__(parent)
self.camera = camera
self._running = False
self.model_file = model_file
self.yolo_model = None
self.target_classes = []
self.detection_enabled = False
self.motion_enabled = False
self.camera_is_moving = False
self.cv2_tracker = None
self.yolo_target_locked = False
self.prev_frame = None
def run(self):
self._running = True
try:
self.yolo_model = YOLO(self.model_file)
except Exception as e:
print(f"Failed to load YOLO model: {e}")
self.yolo_model = None
self.camera.start_video_capture()
try:
while self._running:
try:
frame = self.camera.capture_video_frame(timeout=2000)
except asi.ZWO_Error:
continue
except Exception as exc:
self.error.emit(str(exc))
continue
detections = []
if self.detection_enabled and self.yolo_model is not None and self.target_classes:
try:
results = self.yolo_model.track(
frame, imgsz=480, classes=self.target_classes,
persist=True, verbose=False, tracker="bytetrack.yaml"
)
for r in results:
for box in r.boxes:
x1, y1, x2, y2 = box.xyxy[0].cpu().numpy().astype(int)
conf = float(box.conf[0])
cls_id = int(box.cls[0])
name = self.yolo_model.names[cls_id]
track_id = int(box.id[0]) if box.id is not None else None
if conf > 0.25:
detections.append((x1, y1, x2, y2, conf, name, track_id))
except Exception:
pass
motion_box = None
if self.yolo_target_locked:
self.cv2_tracker = None
self.prev_frame = None
elif self.motion_enabled:
if self.cv2_tracker is not None:
success, bbox = self.cv2_tracker.update(frame)
if success:
motion_box = bbox
else:
self.cv2_tracker = None
if self.cv2_tracker is None:
gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
if self.prev_frame is None or self.camera_is_moving or self.prev_frame.shape != gray.shape:
self.prev_frame = gray.copy()
else:
diff = cv2.absdiff(self.prev_frame, gray)
self.prev_frame = gray.copy()
_, thresh = cv2.threshold(diff, 20, 255, cv2.THRESH_BINARY)
thresh = cv2.dilate(thresh, None, iterations=2)
contours, _ = cv2.findContours(thresh, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
best_area = 0
best_bbox = None
for c in contours:
area = cv2.contourArea(c)
if 15 < area < 10000:
if area > best_area:
best_area = area
best_bbox = cv2.boundingRect(c)
if best_bbox is not None:
self.cv2_tracker = create_tracker()
if self.cv2_tracker is not None:
try:
self.cv2_tracker.init(frame, best_bbox)
motion_box = best_bbox
except Exception:
self.cv2_tracker = None
self.frame_ready.emit(frame.copy(), detections, motion_box)
finally:
try:
self.camera.stop_video_capture()
except:
pass
def stop(self):
self._running = False
self.wait(2000)
class VideoSaveWorker(QThread):
finished_ok = pyqtSignal(str)
error = pyqtSignal(str)
def __init__(self, frames, filename, fps):
super().__init__()
self.frames = frames
self.filename = filename
self.fps = fps
def run(self):
try:
if not self.frames:
self.error.emit("保存するフレームがありません")
return
h, w = self.frames[0].shape[:2]
fourcc = cv2.VideoWriter_fourcc(*'mp4v')
out = cv2.VideoWriter(self.filename, fourcc, self.fps, (w, h))
for f in self.frames: out.write(f)
out.release()
self.finished_ok.emit(self.filename)
except Exception as e:
self.error.emit(str(e))
class ControlRow(QWidget):
def __init__(self, camera, caps, on_change, parent=None):
super().__init__(parent)
self.camera = camera
self.caps = caps
self.on_change = on_change
self._updating = False
control_type = caps['ControlType']
min_v, max_v, default_v = caps['MinValue'], caps['MaxValue'], caps['DefaultValue']
if control_type == 1:
max_v = 40000
if default_v > 40000: default_v = 40000
current_v, is_auto = camera.get_control_value(control_type)
if control_type == 1 and current_v > 40000: current_v = 40000
layout = QHBoxLayout(self)
layout.setContentsMargins(0, 0, 0, 0)
self.slider = QSlider(Qt.Orientation.Horizontal)
self.slider.setMinimum(min_v)
self.slider.setMaximum(max_v)
self.slider.setValue(current_v)
self.spin = QSpinBox()
self.spin.setMinimum(min_v)
self.spin.setMaximum(max_v)
self.spin.setMaximumWidth(120)
self.spin.setValue(current_v)
self.auto_box = QCheckBox('Auto')
self.auto_box.setChecked(is_auto)
self.auto_box.setEnabled(bool(caps['IsAutoSupported']))
layout.addWidget(self.slider, stretch=1)
layout.addWidget(self.spin)
layout.addWidget(self.auto_box)
self.slider.valueChanged.connect(self._on_slider)
self.spin.valueChanged.connect(self._on_spin)
self.auto_box.toggled.connect(self._on_auto)
if not caps['IsWritable']: self.setEnabled(False)
def _sync(self, value):
self._updating = True
self.slider.setValue(value)
self.spin.setValue(value)
self._updating = False
def _push(self, value, auto):
self.camera.set_control_value(self.caps['ControlType'], value, auto)
self.on_change()
def _on_slider(self, value):
if self._updating: return
self._sync(value)
self._push(value, self.auto_box.isChecked())
def _on_spin(self, value):
if self._updating: return
self._sync(value)
self._push(value, self.auto_box.isChecked())
def _on_auto(self, checked):
self._push(self.spin.value(), checked)
import sys
import os
import time
import cv2
import numpy as np
import zwoasi as asi
from PyQt6.QtCore import Qt, QTimer
from PyQt6.QtGui import QImage, QPixmap
from PyQt6.QtWidgets import (
QApplication, QCheckBox, QFormLayout, QGroupBox, QHBoxLayout, QLabel,
QMainWindow, QScrollArea, QSpinBox, QDoubleSpinBox, QStatusBar,
QVBoxLayout, QWidget, QLineEdit, QPushButton, QMessageBox, QComboBox,
QGridLayout, QTabWidget
)
from device_controllers import AcuterController, AstromechanicsEFController, PositionWorker, MoveWorker
from camera_yolo import download_model_files, YOLO_MODEL_FILE, VideoLabel, CaptureThread, ControlRow, VideoSaveWorker
class MainWindow(QMainWindow):
def __init__(self, sdk_path):
super().__init__()
self.setWindowTitle('ZWO ASI Camera + EF Lens + Acuter + YOLOv8 Complete (v2.39.0)')
self.resize(1300, 900)
asi.init(sdk_path)
if asi.get_num_cameras() == 0:
raise RuntimeError('ZWOカメラが見つかりません')
self.camera = asi.Camera(0)
self.camera.set_image_type(asi.ASI_IMG_RGB24)
cam_prop = self.camera.get_camera_property()
supported_bins = cam_prop.get('SupportedBins', [1])
self.current_bins = 2 if 2 in supported_bins else 1
current_w = cam_prop['MaxWidth'] // self.current_bins
current_h = cam_prop['MaxHeight'] // self.current_bins
self.camera.set_roi(start_x=0, start_y=0, width=current_w, height=current_h, bins=self.current_bins)
self.current_start_x = 0
self.current_start_y = 0
self.target_id = None
self.last_target_pos = None
self.last_target_name = None
self.current_detections = []
self.last_frame_size = (current_w, current_h)
self.latest_frame = None
self.last_motion_box = None
self.pixel_size = 0.0038
self.next_track_time = 0.0
self.lost_counter = 0
self.is_recording_event = False
self.video_buffer = []
self.home_deg = {1: None, 2: None}
self.debug_info_dxdy = "Target diff: N/A"
self.debug_info_azalt = "Cmd: N/A"
self.debug_info_status = "Status: Idle"
self.calib_state = 0
self.calib_timer = QTimer(self)
self.calib_timer.timeout.connect(self.calib_step)
self.calib_pts_prev = None
self.calib_gray_prev = None
self.acuter = AcuterController()
self.acuter_poll_timer = QTimer(self)
self.acuter_poll_timer.timeout.connect(self.poll_acuter_position)
self.ef_controller = AstromechanicsEFController()
self.capture_thread = None
self._build_ui()
self._update_ef_ui_state(False)
self._set_acuter_controls_state(False)
self._frame_count = 0
self._start_capture()
QTimer.singleShot(500, self.auto_connect_devices)
def auto_connect_devices(self):
port_ef = self.ef_port_edit.text().strip()
if port_ef:
self.ef_controller.port = port_ef
self.statusBar().showMessage("EFレンズ自動接続試行中...")
try:
if self.ef_controller.connect():
self._update_ef_ui_state(True)
self.btn_ef_connect.setText("接続済")
self.btn_ef_connect.setStyleSheet("background-color: #28a745; color: white;")
self.on_ef_refresh_position()
except Exception:
self.ef_controller.disconnect()
port_acuter = self.acuter_port_input.text().strip()
if port_acuter:
self.statusBar().showMessage("Acuterマウント自動接続試行中...")
try:
if self.acuter.connect(port_acuter):
self.btn_acuter_connect.setText("切断")
self.btn_acuter_connect.setStyleSheet("background-color: #28a745; color: white;")
self.lbl_acuter_status.setText("接続済み (GoTo/Slew有効)")
self.lbl_acuter_status.setStyleSheet("color: #00FF00; font-weight: bold;")
self._set_acuter_controls_state(True)
self.acuter_poll_timer.start(500)
except Exception:
self.statusBar().showMessage("自動接続に失敗しました。後で手動接続してください。")
def _start_capture(self):
if self.capture_thread and self.capture_thread.isRunning():
self.capture_thread.stop()
self.capture_thread = CaptureThread(self.camera, YOLO_MODEL_FILE)
self.capture_thread.target_classes = self.get_current_target_classes()
self.capture_thread.detection_enabled = self.cb_detection.isChecked()
self.capture_thread.motion_enabled = self.cb_motion.isChecked()
self.capture_thread.frame_ready.connect(self._on_frame)
self.capture_thread.error.connect(lambda msg: self.statusBar().showMessage(f"取得エラー: {msg}"))
self.capture_thread.start()
def _build_ui(self):
central = QWidget()
self.setCentralWidget(central)
root = QHBoxLayout(central)
root.setContentsMargins(15, 15, 15, 15)
root.setSpacing(15)
# ========== 左ペイン ==========
left_widget = QWidget()
left_layout = QVBoxLayout(left_widget)
left_layout.setContentsMargins(0, 0, 0, 0)
left_layout.setSpacing(15)
self.preview_label = VideoLabel('starting...')
self.preview_label.setMinimumSize(640, 480)
self.preview_label.setAlignment(Qt.AlignmentFlag.AlignCenter)
self.preview_label.setStyleSheet("background-color: #000000; color: #FFFFFF; border-radius: 8px;")
self.preview_label.clicked.connect(self._on_preview_clicked)
self.preview_label.roi_selected.connect(self._on_roi_dragged)
left_layout.addWidget(self.preview_label, stretch=1)
self.ef_group = QGroupBox("Canon EF Lens (Astromechanics)")
ef_layout = QVBoxLayout(self.ef_group)
ef_conn_layout = QHBoxLayout()
self.ef_port_edit = QLineEdit("/dev/tty.usbserial-AK06UIRD")
ef_conn_layout.addWidget(QLabel("ポート:"))
ef_conn_layout.addWidget(self.ef_port_edit, stretch=1)
self.btn_ef_connect = QPushButton("接続")
self.btn_ef_connect.clicked.connect(self.on_ef_connect)
self.btn_ef_disconnect = QPushButton("切断")
self.btn_ef_disconnect.clicked.connect(self.on_ef_disconnect)
ef_conn_layout.addWidget(self.btn_ef_connect)
ef_conn_layout.addWidget(self.btn_ef_disconnect)
ef_layout.addLayout(ef_conn_layout)
ef_pos_move_row = QHBoxLayout()
self.lbl_ef_position = QLabel("—")
self.lbl_ef_position.setStyleSheet("color: white; background-color: #333; padding: 4px; border-radius: 4px;")
self.btn_ef_refresh = QPushButton("更新")
self.btn_ef_refresh.clicked.connect(self.on_ef_refresh_position)
self.spin_ef_target = QSpinBox()
self.spin_ef_target.setRange(0, 32767)
self.spin_ef_target.setValue(5000)
self.spin_ef_target.setSingleStep(100)
self.btn_ef_goto = QPushButton("移動")
self.btn_ef_goto.clicked.connect(self.on_ef_goto)
ef_pos_move_row.addWidget(QLabel("現在位置:"))
ef_pos_move_row.addWidget(self.lbl_ef_position)
ef_pos_move_row.addWidget(self.btn_ef_refresh)
ef_pos_move_row.addSpacing(20)
ef_pos_move_row.addWidget(QLabel("目標位置:"))
ef_pos_move_row.addWidget(self.spin_ef_target)
ef_pos_move_row.addWidget(self.btn_ef_goto)
ef_layout.addLayout(ef_pos_move_row)
ef_rel_ap_row = QHBoxLayout()
self.spin_ef_rel = QSpinBox()
self.spin_ef_rel.setRange(1, 5000)
self.spin_ef_rel.setValue(100)
self.spin_ef_rel.setSingleStep(50)
self.btn_ef_in = QPushButton("← IN (−)")
self.btn_ef_in.clicked.connect(lambda: self.on_ef_relative(-1))
self.btn_ef_out = QPushButton("OUT (+) →")
self.btn_ef_out.clicked.connect(lambda: self.on_ef_relative(+1))
self.spin_ef_aperture = QSpinBox()
self.spin_ef_aperture.setRange(0, 30)
self.spin_ef_aperture.setValue(0)
self.btn_ef_aperture = QPushButton("設定")
self.btn_ef_aperture.clicked.connect(self.on_ef_set_aperture)
ef_rel_ap_row.addWidget(QLabel("相対:"))
ef_rel_ap_row.addWidget(self.spin_ef_rel)
ef_rel_ap_row.addWidget(self.btn_ef_in)
ef_rel_ap_row.addWidget(self.btn_ef_out)
ef_rel_ap_row.addSpacing(20)
ef_rel_ap_row.addWidget(QLabel("絞り(0=開放):"))
ef_rel_ap_row.addWidget(self.spin_ef_aperture)
ef_rel_ap_row.addWidget(self.btn_ef_aperture)
ef_rel_ap_row.addStretch()
ef_layout.addLayout(ef_rel_ap_row)
left_layout.addWidget(self.ef_group)
root.addWidget(left_widget, stretch=3)
# ========== 右ペイン ==========
right_scroll = QScrollArea()
right_scroll.setWidgetResizable(True)
right_scroll.setMinimumWidth(430)
right_scroll.setFrameShape(QScrollArea.Shape.NoFrame)
right_panel = QWidget()
right_layout = QVBoxLayout(right_panel)
right_layout.setContentsMargins(0, 0, 10, 0)
self.ai_group = QGroupBox("AI Object Tracking")
ai_layout = QVBoxLayout(self.ai_group)
self.cb_detection = QCheckBox("YOLOv8 トラッキングを有効にする (ID保持)")
self.cb_detection.setChecked(False)
self.cb_detection.toggled.connect(self.on_detection_toggled)
ai_layout.addWidget(self.cb_detection)
self.cb_motion = QCheckBox("動体検知 (Motion Detect) [YOLO認識前の小目標用]")
self.cb_motion.setChecked(True)
self.cb_motion.setStyleSheet("color: #ffcc00; font-weight: bold;")
self.cb_motion.toggled.connect(self.on_motion_toggled)
ai_layout.addWidget(self.cb_motion)
info_label = QLabel("※クリックで「対象ロックオン」、ドラッグで「ROI切り出し」")
info_label.setStyleSheet("color: #aaaaaa; font-size: 11px;")
ai_layout.addWidget(info_label)
targets_layout = QHBoxLayout()
self.cb_person = QCheckBox("人")
self.cb_bicycle = QCheckBox("自転車")
self.cb_car = QCheckBox("車")
self.cb_airplane = QCheckBox("航空機")
self.cb_bird = QCheckBox("鳥")
for cb in [self.cb_person, self.cb_bicycle, self.cb_car, self.cb_airplane, self.cb_bird]:
cb.setChecked(True)
cb.toggled.connect(self.update_target_classes)
targets_layout.addWidget(cb)
ai_layout.addLayout(targets_layout)
self.cb_auto_engage = QCheckBox("選択対象の自動検知＆録画 (Auto Engage)")
self.cb_auto_engage.setChecked(False)
self.cb_auto_engage.setStyleSheet("color: #5c9eff; font-weight: bold;")
ai_layout.addWidget(self.cb_auto_engage)
right_layout.addWidget(self.ai_group)
self.acuter_group = QGroupBox("Acuter Traverse Control")
self.acuter_layout = QVBoxLayout(self.acuter_group)
acuter_conn_layout = QHBoxLayout()
self.acuter_port_input = QLineEdit("/dev/cu.usbmodem4E94509B34001")
self.btn_acuter_connect = QPushButton("接続")
self.btn_acuter_connect.clicked.connect(self.toggle_acuter_connection)
acuter_conn_layout.addWidget(self.acuter_port_input)
acuter_conn_layout.addWidget(self.btn_acuter_connect)
self.acuter_layout.addLayout(acuter_conn_layout)
self.lbl_acuter_status = QLabel("未接続")
self.lbl_acuter_status.setStyleSheet("color: red; font-weight: bold;")
self.acuter_layout.addWidget(self.lbl_acuter_status)
self.btn_auto_track = QPushButton("Auto Tracking (OFF / 待機)")
self.btn_auto_track.setCheckable(True)
self.btn_auto_track.setFixedHeight(60)
self.btn_auto_track.setStyleSheet("font-size: 18px; font-weight: bold; background-color: #555555; color: white;")
self.btn_auto_track.toggled.connect(self.on_auto_track_toggled)
self.acuter_layout.addWidget(self.btn_auto_track)
pos_layout = QHBoxLayout()
self.lbl_acuter_az = QLabel("Az : --.- °")
self.lbl_acuter_az.setStyleSheet("font-family: 'Menlo', 'Consolas', 'Courier New'; font-size: 16px; font-weight: bold; color: white; background-color: #333; padding: 4px; border-radius: 4px;")
self.lbl_acuter_alt = QLabel("Alt: --.- °")
self.lbl_acuter_alt.setStyleSheet("font-family: 'Menlo', 'Consolas', 'Courier New'; font-size: 16px; font-weight: bold; color: white; background-color: #333; padding: 4px; border-radius: 4px;")
pos_layout.addWidget(self.lbl_acuter_az)
pos_layout.addWidget(self.lbl_acuter_alt)
self.acuter_layout.addLayout(pos_layout)
speed_form = QFormLayout()
self.acuter_speed_combo = QComboBox()
self.acuter_speed_combo.addItems(["1.0", "5.0", "10.0", "15.0"])
self.acuter_speed_combo.setCurrentText("15.0")
speed_form.addRow("最大回転速度(度/秒):", self.acuter_speed_combo)
self.acuter_layout.addLayout(speed_form)
goto_layout = QHBoxLayout()
self.acuter_az_input = QLineEdit("10.0")
self.acuter_az_input.setMaximumWidth(50)
self.btn_az_goto = QPushButton("Az GoTo")
self.btn_az_goto.setEnabled(False)
self.btn_az_goto.clicked.connect(lambda: self.acuter.start_move(1, float(self.acuter_az_input.text()), exact_goto=True))
self.acuter_alt_input = QLineEdit("10.0")
self.acuter_alt_input.setMaximumWidth(50)
self.btn_alt_goto = QPushButton("Alt GoTo")
self.btn_alt_goto.setEnabled(False)
self.btn_alt_goto.clicked.connect(lambda: self.acuter.start_move(2, float(self.acuter_alt_input.text()), exact_goto=True))
goto_layout.addWidget(QLabel("Az:"))
goto_layout.addWidget(self.acuter_az_input)
goto_layout.addWidget(QLabel("°"))
goto_layout.addWidget(self.btn_az_goto)
goto_layout.addSpacing(15)
goto_layout.addWidget(QLabel("Alt:"))
goto_layout.addWidget(self.acuter_alt_input)
goto_layout.addWidget(QLabel("°"))
goto_layout.addWidget(self.btn_alt_goto)
goto_layout.addStretch()
self.acuter_layout.addLayout(goto_layout)
dpad_layout = QGridLayout()
self.btn_up = QPushButton("▲")
self.btn_left = QPushButton("◀")
self.btn_stop_center = QPushButton("■ STOP")
self.btn_stop_center.setStyleSheet("background-color: #d9534f; color: white; font-weight: bold;")
self.btn_right = QPushButton("▶")
self.btn_down = QPushButton("▼")
huge_angle = 1000.0
self.btn_up.pressed.connect(lambda: self.acuter.start_move(2, -huge_angle, speed_deg_sec=float(self.acuter_speed_combo.currentText())))
self.btn_down.pressed.connect(lambda: self.acuter.start_move(2, huge_angle, speed_deg_sec=float(self.acuter_speed_combo.currentText())))
self.btn_left.pressed.connect(lambda: self.acuter.start_move(1, -huge_angle, speed_deg_sec=float(self.acuter_speed_combo.currentText())))
self.btn_right.pressed.connect(lambda: self.acuter.start_move(1, huge_angle, speed_deg_sec=float(self.acuter_speed_combo.currentText())))
self.btn_up.released.connect(lambda: self.acuter.stop_axis(2))
self.btn_down.released.connect(lambda: self.acuter.stop_axis(2))
self.btn_left.released.connect(lambda: self.acuter.stop_axis(1))
self.btn_right.released.connect(lambda: self.acuter.stop_axis(1))
self.btn_stop_center.clicked.connect(self.acuter.emergency_stop)
dpad_layout.addWidget(self.btn_up, 0, 1)
dpad_layout.addWidget(self.btn_left, 1, 0)
dpad_layout.addWidget(self.btn_stop_center, 1, 1)
dpad_layout.addWidget(self.btn_right, 1, 2)
dpad_layout.addWidget(self.btn_down, 2, 1)
self.acuter_layout.addLayout(dpad_layout)
right_layout.addWidget(self.acuter_group)
# 3. ZWO ASI カメラ設定
asi_group = QGroupBox('ZWO ASI & Tracking Settings')
asi_layout = QVBoxLayout(asi_group)
self.cam_tabs = QTabWidget()
self.tab_main = QWidget()
self.tab_adv = QWidget()
form_main = QFormLayout(self.tab_main)
form_adv = QFormLayout(self.tab_adv)
self.btn_auto_calib = QPushButton("Run Auto Calibration")
self.btn_auto_calib.clicked.connect(self.start_auto_calib)
self.btn_auto_calib.setStyleSheet("background-color: #28a745; color: white; font-weight: bold; font-size: 16px; padding: 12px;")
form_main.addRow(self.btn_auto_calib)
track_calib_group = QGroupBox("Tracking Settings & Axis Mapping")
track_calib_layout = QFormLayout(track_calib_group)
self.focal_spin = QDoubleSpinBox()
self.focal_spin.setRange(1.0, 2000.0)
self.focal_spin.setValue(55.0)
self.kp_spin = QDoubleSpinBox()
self.kp_spin.setRange(0.01, 5.0)
self.kp_spin.setSingleStep(0.1)
self.kp_spin.setValue(0.8)
self.deadband_spin = QSpinBox()
self.deadband_spin.setRange(5, 200)
self.deadband_spin.setValue(40)
self.combo_x_axis = QComboBox()
self.combo_x_axis.addItems(["Az (Axis 1)", "Alt (Axis 2)"])
self.combo_x_axis.setCurrentIndex(0)
self.combo_y_axis = QComboBox()
self.combo_y_axis.addItems(["Alt (Axis 2)", "Az (Axis 1)"])
self.combo_y_axis.setCurrentIndex(0)
self.cb_inv_x = QCheckBox("Invert X-Axis (Reverse)")
self.cb_inv_y = QCheckBox("Invert Y-Axis (Reverse)")
self.cb_inv_y.setChecked(True)
test_move_widget = QWidget()
test_move_grid = QGridLayout(test_move_widget)
self.btn_test_alt_p = QPushButton("Alt +1.0°")
self.btn_test_alt_m = QPushButton("Alt -1.0°")
self.btn_test_az_m = QPushButton("Az -1.0°")
self.btn_test_az_p = QPushButton("Az +1.0°")
self.btn_test_alt_p.clicked.connect(lambda: self.acuter.start_move(2, 1.0, exact_goto=True, speed_deg_sec=2.0))
self.btn_test_alt_m.clicked.connect(lambda: self.acuter.start_move(2, -1.0, exact_goto=True, speed_deg_sec=2.0))
self.btn_test_az_m.clicked.connect(lambda: self.acuter.start_move(1, -1.0, exact_goto=True, speed_deg_sec=2.0))
self.btn_test_az_p.clicked.connect(lambda: self.acuter.start_move(1, 1.0, exact_goto=True, speed_deg_sec=2.0))
self.calib_buttons = [self.btn_test_alt_p, self.btn_test_alt_m, self.btn_test_az_m, self.btn_test_az_p, self.btn_auto_calib]
for btn in self.calib_buttons: btn.setEnabled(False)
test_move_grid.addWidget(self.btn_test_alt_p, 0, 1)
test_move_grid.addWidget(self.btn_test_az_m, 1, 0)
test_move_grid.addWidget(self.btn_test_alt_m, 1, 1)
test_move_grid.addWidget(self.btn_test_az_p, 1, 2)
track_calib_layout.addRow("Focal Length (mm):", self.focal_spin)
track_calib_layout.addRow("Tracking Gain (Kp):", self.kp_spin)
track_calib_layout.addRow("Deadband (px):", self.deadband_spin)
track_calib_layout.addRow("Camera X-Axis:", self.combo_x_axis)
track_calib_layout.addRow("", self.cb_inv_x)
track_calib_layout.addRow("Camera Y-Axis:", self.combo_y_axis)
track_calib_layout.addRow("", self.cb_inv_y)
track_calib_layout.addRow("Test Move:", test_move_widget)
form_adv.addRow(track_calib_group)
cam_prop = self.camera.get_camera_property()
bin_group = QGroupBox("Sensor Binning")
bin_layout = QHBoxLayout(bin_group)
self.bin_combo = QComboBox()
for b in cam_prop['SupportedBins']:
if b != 0: self.bin_combo.addItem(f"Bin {b}x{b}", b)
self.bin_combo.setCurrentText(f"Bin {self.current_bins}x{self.current_bins}")
self.btn_apply_bin = QPushButton("Apply Binning")
self.btn_apply_bin.clicked.connect(self.on_apply_binning)
bin_layout.addWidget(QLabel("Binning:"))
bin_layout.addWidget(self.bin_combo)
bin_layout.addWidget(self.btn_apply_bin)
roi_group = QGroupBox("Sensor ROI (Resolution)")
roi_layout = QGridLayout(roi_group)
current_w = cam_prop['MaxWidth'] // self.current_bins
current_h = cam_prop['MaxHeight'] // self.current_bins
self.roi_w_spin = QSpinBox()
self.roi_w_spin.setRange(8, current_w)
self.roi_w_spin.setSingleStep(8)
self.roi_w_spin.setValue(current_w)
self.roi_h_spin = QSpinBox()
self.roi_h_spin.setRange(2, current_h)
self.roi_h_spin.setSingleStep(2)
self.roi_h_spin.setValue(current_h)
self.btn_apply_roi = QPushButton("Apply ROI (Center)")
self.btn_apply_roi.clicked.connect(self.on_apply_roi_center)
self.btn_reset_roi = QPushButton("Reset to Full Frame")
self.btn_reset_roi.clicked.connect(self.on_reset_roi)
self.btn_reset_roi.setStyleSheet("background-color: #555555;")
roi_layout.addWidget(QLabel("Width:"), 0, 0)
roi_layout.addWidget(self.roi_w_spin, 0, 1)
roi_layout.addWidget(QLabel("Height:"), 1, 0)
roi_layout.addWidget(self.roi_h_spin, 1, 1)
roi_layout.addWidget(self.btn_apply_roi, 2, 0, 1, 2)
roi_layout.addWidget(self.btn_reset_roi, 3, 0, 1, 2)
form_adv.addRow(roi_group)
form_adv.addRow(bin_group)
self.rows = []
for name, caps in sorted(self.camera.get_controls().items()):
row = ControlRow(self.camera, caps, lambda: self.statusBar().showMessage('ASIパラメータ更新', 1000))
self.rows.append(row)
form_adv.addRow(name, row)
self.cam_tabs.addTab(self.tab_main, "Main Controls")
self.cam_tabs.addTab(self.tab_adv, "Advanced")
asi_layout.addWidget(self.cam_tabs)
right_layout.addWidget(asi_group)
right_layout.addStretch()
right_scroll.setWidget(right_panel)
root.addWidget(right_scroll, stretch=2)
self.setStatusBar(QStatusBar())
def _update_ef_ui_state(self, connected: bool):
self.btn_ef_connect.setEnabled(not connected)
self.btn_ef_disconnect.setEnabled(connected)
self.btn_ef_goto.setEnabled(connected)
self.btn_ef_in.setEnabled(connected)
self.btn_ef_out.setEnabled(connected)
def on_ef_connect(self):
try:
self.ef_controller.port = self.ef_port_edit.text()
if self.ef_controller.connect():
self._update_ef_ui_state(True)
self.on_ef_refresh_position()
except: pass
def on_ef_disconnect(self):
self.ef_controller.disconnect()
self._update_ef_ui_state(False)
def on_ef_refresh_position(self):
if not self.ef_controller.is_connected: return
pos = self.ef_controller.get_position()
if pos is not None:
self.lbl_ef_position.setText(str(pos))
self.spin_ef_target.setValue(pos)
def on_ef_goto(self):
if self.ef_controller.is_connected:
self.ef_controller.move_absolute(self.spin_ef_target.value())
QTimer.singleShot(1000, self.on_ef_refresh_position)
def on_ef_relative(self, direction):
if self.ef_controller.is_connected:
target = self.spin_ef_target.value() + (self.spin_ef_rel.value() * direction)
self.ef_controller.move_absolute(target)
QTimer.singleShot(1000, self.on_ef_refresh_position)
def on_ef_set_aperture(self):
if self.ef_controller.is_connected:
self.ef_controller.set_aperture(self.spin_ef_aperture.value())
def toggle_acuter_connection(self):
if self.acuter.is_connected:
self.acuter.disconnect()
self.btn_acuter_connect.setText("接続")
self._set_acuter_controls_state(False)
else:
if self.acuter.connect(self.acuter_port_input.text()):
self.btn_acuter_connect.setText("切断")
self._set_acuter_controls_state(True)
self.acuter_poll_timer.start(500)
def _set_acuter_controls_state(self, state):
self.btn_az_goto.setEnabled(state)
self.btn_alt_goto.setEnabled(state)
self.btn_up.setEnabled(state)
self.btn_down.setEnabled(state)
self.btn_left.setEnabled(state)
self.btn_right.setEnabled(state)
self.btn_stop_center.setEnabled(state)
for btn in self.calib_buttons: btn.setEnabled(state)
def poll_acuter_position(self):
self.acuter.poll_position()
d1, d2 = self.acuter.current_deg.get(1), self.acuter.current_deg.get(2)
if d1 is not None: self.lbl_acuter_az.setText(f"Az : {d1:+8.3f} °")
if d2 is not None: self.lbl_acuter_alt.setText(f"Alt: {d2:+8.3f} °")
if self.capture_thread:
self.capture_thread.camera_is_moving = self.acuter.axis_moving.get(1, False) or self.acuter.axis_moving.get(2, False)
def on_auto_track_toggled(self, checked):
if checked:
self.btn_auto_track.setText("Tracking (ON)")
self.btn_auto_track.setStyleSheet("background-color: #d9534f; color: white; font-size: 18px; font-weight: bold;")
else:
self.btn_auto_track.setText("Tracking (OFF)")
self.btn_auto_track.setStyleSheet("background-color: #555; color: white; font-size: 18px; font-weight: bold;")
self.acuter.emergency_stop()
if getattr(self, 'is_recording_event', False):
self.stop_event_record_and_return()
def on_motion_toggled(self, checked):
if self.capture_thread:
self.capture_thread.motion_enabled = checked
if not checked:
self.capture_thread.cv2_tracker = None
def start_event_record_and_track(self):
self.home_deg = {1: self.acuter.current_deg.get(1), 2: self.acuter.current_deg.get(2)}
self.is_recording_event = True
self.video_buffer = []
if not self.btn_auto_track.isChecked(): self.btn_auto_track.setChecked(True)
def stop_event_record_and_return(self):
self.is_recording_event = False
if self.video_buffer:
save_dir = "/Users/mars/acuter/videos"
os.makedirs(save_dir, exist_ok=True)
filename = os.path.join(save_dir, time.strftime("track_%Y%m%d_%H%M%S.mp4"))
self.statusBar().showMessage(f"動画を保存中... {filename}", 5000)
self.video_saver = VideoSaveWorker(self.video_buffer, filename, 30.0)
self.video_saver.finished_ok.connect(lambda f: self.statusBar().showMessage(f"保存完了: {f}", 5000))
self.video_saver.error.connect(lambda err: self.statusBar().showMessage(f"保存エラー: {err}", 5000))
self.video_saver.start()
self.video_buffer = []
for axis in (1, 2):
h_deg = self.home_deg.get(axis)
c_deg = self.acuter.current_deg.get(axis)
if h_deg is not None and c_deg is not None:
diff = h_deg - c_deg
if diff > 180: diff -= 360
elif diff < -180: diff += 360
if abs(diff) > 0.05:
self.acuter.start_move(axis, diff, exact_goto=True, speed_deg_sec=15.0)
def get_current_target_classes(self):
targets = []
if self.cb_person.isChecked(): targets.append(0)
if self.cb_bicycle.isChecked(): targets.append(1)
if self.cb_car.isChecked(): targets.append(2)
if self.cb_airplane.isChecked(): targets.append(4)
if self.cb_bird.isChecked(): targets.append(14)
return targets
def update_target_classes(self):
if self.capture_thread:
self.capture_thread.target_classes = self.get_current_target_classes()
def on_detection_toggled(self, checked):
if self.capture_thread:
self.capture_thread.detection_enabled = checked
if not checked:
self.target_id = None
self.btn_auto_track.setChecked(False)
def _on_preview_clicked(self, lx, ly):
scale = min(self.preview_label.width() / self.last_frame_size[0], self.preview_label.height() / self.last_frame_size[1])
ox = (self.preview_label.width() - self.last_frame_size[0] * scale) / 2
oy = (self.preview_label.height() - self.last_frame_size[1] * scale) / 2
fx, fy = (lx - ox) / scale, (ly - oy) / scale
clicked_id = None
for d in self.current_detections:
if d[0] <= fx <= d[2] and d[1] <= fy <= d[3]:
clicked_id = d[6]
break
if clicked_id is not None:
self.target_id = clicked_id
if self.capture_thread: self.capture_thread.yolo_target_locked = True
self.lost_counter = 0
if not getattr(self, 'is_recording_event', False):
self.start_event_record_and_track()
if not self.btn_auto_track.isChecked():
self.btn_auto_track.setChecked(True)
return
if getattr(self, 'last_motion_box', None) is not None:
mx, my, mw, mh = self.last_motion_box
if mx <= fx <= mx+mw and my <= fy <= my+mh:
self.target_id = None
if self.capture_thread: self.capture_thread.yolo_target_locked = False
if not getattr(self, 'is_recording_event', False):
self.start_event_record_and_track()
if not self.btn_auto_track.isChecked():
self.btn_auto_track.setChecked(True)
return
self.target_id = None
if self.capture_thread: self.capture_thread.yolo_target_locked = False
self.cb_auto_engage.setChecked(False)
self.btn_auto_track.setChecked(False)
self.acuter.emergency_stop()
def _on_roi_dragged(self, lx, ly, lw, lh): pass
def on_reset_roi(self):
cam_prop = self.camera.get_camera_property()
max_w = cam_prop['MaxWidth'] // self.current_bins
max_h = cam_prop['MaxHeight'] // self.current_bins
self.camera.set_roi(start_x=0, start_y=0, width=max_w, height=max_h, bins=self.current_bins)
self._start_capture()
def on_apply_binning(self):
b = self.bin_combo.currentData()
self.camera.set_roi(bins=b)
self.current_bins = b
self._start_capture()
def on_apply_roi_center(self):
w, h = (self.roi_w_spin.value() // 8) * 8, (self.roi_h_spin.value() // 2) * 2
cam_prop = self.camera.get_camera_property()
sx = ((cam_prop['MaxWidth'] // self.current_bins - w) // 2 // 4) * 4
sy = ((cam_prop['MaxHeight'] // self.current_bins - h) // 2 // 2) * 2
self.camera.set_roi(start_x=sx, start_y=sy, width=w, height=h, bins=self.current_bins)
self._start_capture()
def _on_frame(self, frame: np.ndarray, detections: list, motion_box: object = None):
self._frame_count += 1
self.last_motion_box = motion_box
h, w = frame.shape[:2]
self.last_frame_size = (w, h)
self.current_detections = detections
self.latest_frame = frame.copy()
draw_frame = frame.copy()
cx, cy = w // 2, h // 2
db_px = self.deadband_spin.value()
cv2.circle(draw_frame, (cx, cy), db_px, (255, 255, 255), 1, cv2.LINE_AA)
cv2.line(draw_frame, (cx-20, cy), (cx+20, cy), (255,255,255), 1)
cv2.line(draw_frame, (cx, cy-20), (cx, cy+20), (255,255,255), 1)
motion_cx, motion_cy = None, None
if motion_box is not None:
mx, my, mw, mh = [int(v) for v in motion_box]
motion_cx, motion_cy = mx + mw//2, my + mh//2
cv2.rectangle(draw_frame, (mx, my), (mx+mw, my+mh), (0, 165, 255), 2)
cv2.putText(draw_frame, "MOTION", (mx, max(my-10, 10)), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 165, 255), 2)
if self.target_id is None and motion_box is not None and detections:
for d in detections:
dcx, dcy = (d[0]+d[2])//2, (d[1]+d[3])//2
if np.hypot(dcx - motion_cx, dcy - motion_cy) < 100:
self.target_id = d[6]
self.last_target_name = d[5]
if self.capture_thread: self.capture_thread.yolo_target_locked = True
self.statusBar().showMessage(f"Upgraded to YOLO Target: ID {self.target_id}", 3000)
break
if self.cb_auto_engage.isChecked() and self.target_id is None and detections:
best_det = max(detections, key=lambda d: d[4])
if best_det[6] is not None:
self.target_id = best_det[6]
self.last_target_name = best_det[5]
self.last_target_pos = ((best_det[0]+best_det[2])//2, (best_det[1]+best_det[3])//2)
self.lost_counter = 0
if self.capture_thread: self.capture_thread.yolo_target_locked = True
self.start_event_record_and_track()
if self.cb_auto_engage.isChecked() and self.target_id is None and motion_box is not None:
if not getattr(self, 'is_recording_event', False):
self.start_event_record_and_track()
is_tracking_active = False
tcx, tcy = None, None
tracking_color = (0, 255, 0)
if self.target_id is not None:
target_found = False
for d in detections:
if self.target_id == d[6]:
target_found = True
self.last_target_pos = ((d[0]+d[2])//2, (d[1]+d[3])//2)
break
if not target_found and self.last_target_pos:
best_d = None
min_dist = float('inf')
for d in detections:
if d[5] == self.last_target_name:
dist = np.hypot((((d[0]+d[2])//2) - self.last_target_pos[0]), (((d[1]+d[3])//2) - self.last_target_pos[1]))
if dist < min_dist:
min_dist = dist
best_d = d
if best_d is not None:
self.target_id = best_d[6]
target_found = True
self.lost_counter = 0
if target_found:
tcx, tcy = self.last_target_pos
is_tracking_active = True
tracking_color = (0, 0, 255)
self.lost_counter = 0
else:
self.lost_counter += 1
if self.last_target_pos:
gx, gy = self.last_target_pos
cv2.circle(draw_frame, (gx, gy), 15, (0,165,255), 2)
cv2.putText(draw_frame, "SEARCHING...", (gx+20, gy), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0,165,255), 2)
if self.lost_counter == 1:
self.acuter.emergency_stop()
self.debug_info_status = "Status: Target Lost (Braking)"
if self.lost_counter > 150:
self.debug_info_status = "Status: Target completely LOST! (Aborted)"
self.acuter.emergency_stop()
self.btn_auto_track.setChecked(False)
self.target_id = None
if self.capture_thread: self.capture_thread.yolo_target_locked = False
self.last_target_pos = None
self.lost_counter = 0
elif motion_box is not None:
tcx, tcy = motion_cx, motion_cy
is_tracking_active = True
tracking_color = (0, 165, 255)
for d in detections:
tid = d[6]
is_target = (self.target_id is not None and tid == self.target_id)
color = (0,0,255) if is_target else (0,255,0)
thickness = 4 if is_target else 2
cv2.rectangle(draw_frame, (d[0], d[1]), (d[2], d[3]), color, thickness)
id_str = f"ID:{tid} " if tid is not None else ""
label_text = f"{id_str}{d[5]} {d[4]*100:.1f}%"
if is_target:
label_text = "[LOCKED] " + label_text
cv2.putText(draw_frame, label_text, (d[0], max(d[1] - 10, 10)), cv2.FONT_HERSHEY_SIMPLEX, 0.7, color, 2)
if is_tracking_active and tcx is not None:
cv2.line(draw_frame, (int(tcx) - 15, int(tcy)), (int(tcx) + 15, int(tcy)), tracking_color, 2)
cv2.line(draw_frame, (int(tcx), int(tcy) - 15), (int(tcx), int(tcy) + 15), tracking_color, 2)
cv2.line(draw_frame, (int(tcx), int(tcy)), (cx, cy), (0, 255, 255), 1)
if self.btn_auto_track.isChecked() and self.calib_state == 0:
self._update_tracking(tcx, tcy, w, h)
if getattr(self, 'is_recording_event', False):
cv2.circle(draw_frame, (40, 40), 12, (0, 0, 255), -1)
cv2.putText(draw_frame, "REC", (60, 48), cv2.FONT_HERSHEY_SIMPLEX, 1.2, (0, 0, 255), 3)
self.video_buffer.append(draw_frame.copy())
cv2.putText(draw_frame, getattr(self, 'debug_info_dxdy', ''), (10, 90), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 255, 255), 2)
cv2.putText(draw_frame, getattr(self, 'debug_info_azalt', ''), (10, 120), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 255, 255), 2)
cv2.putText(draw_frame, getattr(self, 'debug_info_status', ''), (10, 150), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 255, 255), 2)
qimg = QImage(draw_frame.data, w, h, 3*w, QImage.Format.Format_BGR888)
self.preview_label.setPixmap(QPixmap.fromImage(qimg).scaled(self.preview_label.size(), Qt.AspectRatioMode.KeepAspectRatio))
def _update_tracking(self, tcx, tcy, w, h):
if not self.acuter.is_connected or time.time() < self.next_track_time: return
self.next_track_time = time.time() + 0.1
dx, dy = tcx - w/2.0, tcy - h/2.0
db = self.deadband_spin.value()
if abs(dx) < db and abs(dy) < db:
if self.acuter.axis_moving.get(1) or self.acuter.axis_moving.get(2):
self.acuter.emergency_stop()
self.debug_info_status = f"Status: Target Centered (<{db}px)"
return
deg_per_px = (self.pixel_size * self.current_bins / self.focal_spin.value()) * (180.0 / np.pi)
vx = dx * deg_per_px * self.kp_spin.value() * (-1.0 if self.cb_inv_x.isChecked() else 1.0)
vy = dy * deg_per_px * self.kp_spin.value() * (-1.0 if self.cb_inv_y.isChecked() else 1.0)
az_deg = vx if self.combo_x_axis.currentIndex() == 0 else vy
alt_deg = vx if self.combo_x_axis.currentIndex() == 1 else vy
spd = float(self.acuter_speed_combo.currentText())
db_deg = db * deg_per_px * self.kp_spin.value()
self.debug_info_dxdy = f"Target diff: dx={dx:.1f}px, dy={dy:.1f}px"
p_gain = 0.5
speed_az = max(0.1, min(spd, abs(az_deg) * p_gain))
speed_alt = max(0.1, min(spd, abs(alt_deg) * p_gain))
sent = False
if abs(az_deg) > db_deg:
self.acuter.start_move(1, az_deg, exact_goto=False, speed_deg_sec=speed_az)
sent = True
elif self.acuter.axis_moving.get(1):
self.acuter.stop_axis(1)
if abs(alt_deg) > db_deg:
self.acuter.start_move(2, alt_deg, exact_goto=False, speed_deg_sec=speed_alt)
sent = True
elif self.acuter.axis_moving.get(2):
self.acuter.stop_axis(2)
if sent:
self.debug_info_azalt = f"Speed: Az {speed_az:.1f} d/s, Alt {speed_alt:.1f} d/s"
self.debug_info_status = "Status: Continuous P-Control Tracking"
def start_auto_calib(self):
if not self.acuter.is_connected or self.latest_frame is None: return
self.btn_auto_track.setChecked(False)
self.calib_state = 1
self.calib_timer.start(1000)
def stop_calib(self, msg):
self.calib_state = 0
self.calib_timer.stop()
self.statusBar().showMessage(msg, 5000)
def calib_step(self):
if self.latest_frame is None: return
gray = cv2.cvtColor(self.latest_frame, cv2.COLOR_BGR2GRAY)
if self.calib_state == 1:
self.calib_pts_prev = cv2.goodFeaturesToTrack(gray, maxCorners=100, qualityLevel=0.01, minDistance=30)
if self.calib_pts_prev is None: return self.stop_calib("Failed")
self.calib_gray_prev = gray
self.acuter.start_move(1, 1.0, exact_goto=True, speed_deg_sec=2.0)
self.calib_wait = 3
self.calib_state = 2
elif self.calib_state == 2:
self.calib_wait -= 1
if self.calib_wait <= 0:
pts_next, status, _ = cv2.calcOpticalFlowPyrLK(self.calib_gray_prev, gray, self.calib_pts_prev, None)
diff = pts_next[status == 1] - self.calib_pts_prev[status == 1]
self.calib_dx1, self.calib_dy1 = np.median(diff[:, 0]), np.median(diff[:, 1])
self.acuter.start_move(1, -1.0, exact_goto=True, speed_deg_sec=2.0)
self.calib_wait = 3; self.calib_state = 3
elif self.calib_state == 3:
self.calib_wait -= 1
if self.calib_wait <= 0:
self.calib_pts_prev = cv2.goodFeaturesToTrack(gray, maxCorners=100, qualityLevel=0.01, minDistance=30)
self.calib_gray_prev = gray
self.acuter.start_move(2, 1.0, exact_goto=True, speed_deg_sec=2.0)
self.calib_wait = 3; self.calib_state = 4
elif self.calib_state == 4:
self.calib_wait -= 1
if self.calib_wait <= 0:
pts_next, status, _ = cv2.calcOpticalFlowPyrLK(self.calib_gray_prev, gray, self.calib_pts_prev, None)
diff = pts_next[status == 1] - self.calib_pts_prev[status == 1]
self.calib_dx2, self.calib_dy2 = np.median(diff[:, 0]), np.median(diff[:, 1])
self.acuter.start_move(2, -1.0, exact_goto=True, speed_deg_sec=2.0)
self.calib_wait = 3; self.calib_state = 5
elif self.calib_state == 5:
self.calib_wait -= 1
if self.calib_wait <= 0:
if abs(self.calib_dx1) > abs(self.calib_dy1):
self.combo_x_axis.setCurrentIndex(0); self.combo_y_axis.setCurrentIndex(0)
self.cb_inv_x.setChecked(bool(self.calib_dx1 > 0))
self.cb_inv_y.setChecked(bool(self.calib_dy2 > 0))
else:
self.combo_x_axis.setCurrentIndex(1); self.combo_y_axis.setCurrentIndex(1)
self.cb_inv_x.setChecked(bool(self.calib_dx2 > 0))
self.cb_inv_y.setChecked(bool(self.calib_dy1 > 0))
self.stop_calib("Success")
def closeEvent(self, event):
self.acuter.disconnect()
self.ef_controller.disconnect()
if self.capture_thread: self.capture_thread.stop()
self.camera.close()
super().closeEvent(event)
if __name__ == '__main__':
app = QApplication(sys.argv)
dark_stylesheet = """
QMainWindow { background-color: #2b2b2b; } QLabel { color: #e0e0e0; font-size: 13px; }
QGroupBox { color: #e0e0e0; border: 1px solid #555; border-radius: 6px; margin-top: 16px; padding-top: 15px; font-weight: bold; }
QGroupBox::title { subcontrol-origin: margin; subcontrol-position: top left; left: 10px; padding: 0 5px; }
QTabWidget::pane { border: 1px solid #555; } QTabBar::tab { background: #3c3c3c; color: white; padding: 8px 12px; }
QTabBar::tab:selected { background: #5c9eff; color: black; font-weight: bold; }
QComboBox, QLineEdit, QSpinBox, QDoubleSpinBox { background-color: #3c3c3c; color: white; border: 1px solid #555; padding: 4px; border-radius: 4px; }
QCheckBox { color: #e0e0e0; }
QPushButton { background-color: #3c3c3c; color: #ffffff; border: 1px solid #555; border-radius: 4px; padding: 6px; font-weight: bold; }
QPushButton:pressed { background-color: #5c9eff; color: #000; }
"""
app.setStyleSheet(dark_stylesheet)
window = MainWindow("/Users/mars/acuter/ASI_Camera_SDK/ASI_linux_mac_SDK_V1.41/lib/mac_arm64/libASICamera2.dylib")
window.show()
sys.exit(app.exec())