add Zac's origin code for tab modules

PrelocatedCoordinatesModel.py
EmblModule.py
HelicalTable.py
This commit is contained in:
2022-07-18 09:12:26 +02:00
parent a4217ef1c8
commit b69b63f18b
3 changed files with 1404 additions and 0 deletions

443
EmblModule.py Executable file
View File

@@ -0,0 +1,443 @@
import logging
import numpy
import os
import random
import struct
import time
import yaml
#from scipy.misc import bytescale
import zmq
from os.path import join
from pathlib import Path
import qsingleton
from CoordinatesModel import CoordinatesModel
from app_config import settings, appsconf, option
EMBL_RESULTS_TIMEOUT = "embl/results_timeout"
EMBL_SAVE_FILES = "embl/save_files"
embl = appsconf["embl"]
#import imageio
import numpy as np
from PyQt5 import QtCore
from PyQt5.QtCore import pyqtSignal, Qt, QObject, pyqtSlot, QPoint
from PyQt5.QtGui import QStandardItemModel, QCursor
from PyQt5.QtWidgets import (
QWidget,
QVBoxLayout,
QHeaderView,
QPushButton,
QFormLayout,
QDoubleSpinBox,
QLabel,
QTableWidgetItem,
QLineEdit,
QTableWidget,
QTextEdit,
QProgressBar,
QHBoxLayout,
QSpinBox,
QTabWidget,
QTableView,
QMenu,
QProgressDialog,
)
import storage
from app_utils import assert_motor_positions
folders = storage.Folders()
import logging
logger = logging.getLogger(__name__)
class EmblTaskResultTimeoutException(Exception):
pass
ZPOS, FPATH = range(2)
# record separator
RS = 0x1e
class EmblMessage(QObject, metaclass=qsingleton.Singleton):
def __init__(self, parent=None, **kwargs):
super(EmblMessage, self).__init__(parent, **kwargs)
if not embl["enabled"]:
return
context = zmq.Context()
logger.info("Connecting to server...")
self.socket = context.socket(zmq.REQ)
uri = embl["uri"]
logger.info(f"EMBL alignment server at: {uri}")
self.socket.connect(uri)
def make_PAR(self, min: float, max: float, step: float, ppm: int, beam_xy: tuple):
puckid, sampleid = folders._prefix.split("_")
beam_x, beam_y = beam_xy
msg = (
f"PAR{RS:c}{min}{RS:c}{max}{RS:c}{step}{RS:c}{ppm}{RS:c}{beam_x:.0f}{RS:c}{beam_y:.0f}{RS:c}{puckid}{RS:c}{sampleid}"
)
logger.debug(f"PAR = {msg}")
self.socket.send_string(msg)
ans = self.socket.recv_string()
def make_IMG(self, pos: float, img):
width, height = img.shape
msg = f"IMG{RS:c}{pos:.3f}{RS:c}{width}{RS:c}{height}"
logger.info(f"sending img header: {msg}")
self.socket.send_string(msg)
self.socket.recv_string()
logger.info(
f"about to send img shape={img.shape} type={img.dtype} pos = {pos} mm"
)
self.socket.send(numpy.ascontiguousarray(img))
logger.info(f"waiting for recv")
ans = self.socket.recv_string()
logger.info(f"received!")
def finished(self):
self.make_STS("Finished")
def aborted(self):
self.make_STS("Aborted")
def make_STS(self, status):
msg = f"STS{RS:c}{status}"
self.socket.send_string(msg)
ans = self.socket.recv_string()
def make_RES(self):
n = 1
timeout = settings.value(EMBL_RESULTS_TIMEOUT, type=float) + time.time()
result = "failed"
while time.time() < timeout:
time.sleep(2.0)
logger.info(f"#{n}.requesting RES")
self.socket.send_string("RES")
result = self.socket.recv_string()
logger.info(f" RES = {result}")
if "Pending" != result:
break
n += 1
return result
embl_tube = EmblMessage()
class EmblWidget(QWidget):
def __init__(self, parent):
super(EmblWidget, self).__init__(parent)
parent.pixelsPerMillimeter.connect(self.save_ppm)
parent.beamCameraCoordinatesChanged.connect(self.save_beam_coordinates)
self.setLayout(QVBoxLayout())
w = QWidget()
self.layout().addWidget(w)
w.setLayout(QFormLayout())
w.layout().setLabelAlignment(Qt.AlignRight)
self._startZ = QDoubleSpinBox()
self._startZ.setValue(0.6)
self._startZ.setSuffix(" mm")
self._startZ.setRange(-5, 5)
self._startZ.setSingleStep(0.1)
self._startZ.setDecimals(3)
self._endZ = QDoubleSpinBox()
self._endZ.setValue(1.4)
self._endZ.setSuffix(" mm")
self._endZ.setRange(-5, 5)
self._endZ.setSingleStep(0.1)
self._endZ.setDecimals(3)
self._stepZ = QDoubleSpinBox()
self._stepZ.setValue(0.050)
self._stepZ.setSuffix(" mm")
self._stepZ.setRange(0.005, 1.000)
self._stepZ.setSingleStep(0.010)
self._stepZ.setDecimals(3)
self._resultsTimeout = QDoubleSpinBox()
self._resultsTimeout.setRange(1., 2000.)
self._resultsTimeout.setValue(60)
self._resultsTimeout.setSuffix(" seconds")
self._resultsTimeout.setDecimals(0)
self._factors = QLineEdit("1 1 1")
but = QPushButton("update results")
but.clicked.connect(self.apply_factors)
w.layout().addRow(QLabel("Start Z"), self._startZ)
w.layout().addRow(QLabel("End Z"), self._endZ)
w.layout().addRow(QLabel("Step"), self._stepZ)
w.layout().addRow(QLabel("Task Result Timeout"), self._resultsTimeout)
w.layout().addRow(QLabel("Factors"), self._factors)
w.layout().addRow(QLabel("Apply Factors"), but)
but = QPushButton("Scan Z")
but.clicked.connect(self.executeScan)
but.setAccessibleName("emblScanZ")
self.layout().addWidget(but)
tabs = QTabWidget(self)
self.layout().addWidget(tabs)
self._table = QTableWidget()
self._table.setColumnCount(2)
self._table.setHorizontalHeaderLabels(("Z (mm)", "File Path"))
self._table.horizontalHeader().setSectionResizeMode(QHeaderView.Stretch)
self.layout().addWidget(self._table)
tabs.addTab(self._table, "Z Scans")
tabs.setAccessibleName("embl_tabs")
tabs.setObjectName("embl_tabs")
view = self._results_view = QTableView(self)
view.setModel(CoordinatesModel())
view.verticalHeader().show()
view.horizontalHeader().show()
view.setContextMenuPolicy(Qt.CustomContextMenu)
view.customContextMenuRequested.connect(self.results_menu)
tabs.addTab(view, "Results")
self._tabs = tabs
def apply_factors(self):
self.populate_results(self._results)
@pyqtSlot(float)
def save_ppm(self, ppm: float):
self._ppm = int(ppm)
@pyqtSlot(float, float)
def save_beam_coordinates(self, bx: float, by: float):
logger.debug(f"saving beam_xy {bx} & {by}")
self._beam_xy = (int(bx), int(by))
@pyqtSlot(QPoint)
def results_menu(self, at_point):
table = self.sender()
model = table.model()
index = table.indexAt(at_point)
if not index.isValid():
return
coords = model.coords_at(index)
logger.debug(f"coord: {coords}")
menu = QMenu(table)
gotoAction = menu.addAction("Go here")
if menu.exec_(QCursor.pos()) is gotoAction:
logger.info(f"going to: {index.row()} => {coords}")
self.goto_position(coords)
def configure(self, motors, camera, zoom_device):
self._camera = camera
self._zoom_device = zoom_device
self.motors = motors
def executeScan(self):
fx, fy, cx, cz, omega = self.motors
x, y, z = fx.get_position(), fy.get_position(), cz.get_position()
logger.info(f"scan started at gonio: x,y,z = {x}, {y}, {z}")
self._table.setRowCount(0) # clear previous information
zs = self._startZ.value()
ze = self._endZ.value()
step = self._stepZ.value()
scan_config = {
"z-start": zs,
"z-end": ze + step, # inclusive last value
"z-step": step,
"ppm": self._ppm,
"beam_xy": self._beam_xy,
}
self._gonio_at_start = (x, y, ze + step)
base_Z = self.motors[3] # base_Z
camera = self._camera
zvals = np.arange(zs, ze, step)
numZs = len(zvals)
dlg = QProgressDialog(self)
dlg.setWindowModality(Qt.WindowModal)
dlg.setMinimumDuration(0)
dlg.setCancelButton(None)
dlg.setRange(0, 0)
dlg.setLabelText(f"<b>EMBL Z-Scan</b><br/>")
dlg.setAutoClose(True)
dlg.show()
dlg.setValue(random.randint(1, 20))
self._progress_dialog = dlg
self._acq_thread = AcquisitionThread(base_Z, camera, scan_config)
self._acq_thread.image_acquired.connect(self.addImage)
self._acq_thread.acquisition_finished.connect(self.acquisition_finished)
self._acq_thread.results_fetched.connect(self.populate_results)
self._acq_thread.message.connect(
lambda msg: dlg.setLabelText(f"<b>EMBL Z-Scan</b><br/>{msg}")
)
self._acq_thread.start()
def createModel(self, parent):
model = QStandardItemModel(0, 2, parent)
model.setHeaderData(ZPOS, QtCore.Qt.Horizontal, "Z (mm)")
model.setHeaderData(FPATH, QtCore.Qt.Horizontal, "Image Path")
self._model = model
return model
def addImage(self, idx, zpos, fpath):
logger.info("zpos = {}, fpath = {}".format(zpos, fpath))
rows = self._table.rowCount()
# table.setRowCount(rows + 1)
self._table.insertRow(rows)
item = QTableWidgetItem()
item.setText("{:.03f}".format(zpos))
self._table.setItem(rows, ZPOS, item)
item = QTableWidgetItem()
item.setText(str(fpath))
self._table.setItem(rows, FPATH, item)
self._table.resizeColumnsToContents()
def acquisition_finished(self):
self._progress_dialog.reset()
def goto_position(self, coords):
fx, fy, cx, cz, o = self.motors
tx, ty, bb, tz, to = coords
logger.info(f"moving gonio to: {tx}, {ty}, {tz}")
fx.move(tx, wait=True, ignore_limits=True)
fy.move(ty, wait=True, ignore_limits=True)
cz.move(tz, wait=True, ignore_limits=True)
@pyqtSlot(str)
def populate_results(self, results):
"""Results.txt
60.624%363.4026940025185%-768.32%-1052.52
angle between sample and camera, x, y, z
:return:
"""
if "no results" in results:
return
s1, s2, s3 = [int(c) for c in self._factors.text().split()]
self._results = results
self.coords = []
ox, oy, oz = self._gonio_at_start
oz = self._endZ.value()
for n, line in enumerate(results.split("\n")):
# o, z, y, x = [float(n) for n in line.split("%")]
omega, x, y, z = [float(n) for n in line.split("%")]
no, nx, ny, nz = [omega, ox + s1*x, oy + s2*y, oz + s3*z]
self.coords.append([nx, ny, 0, nz, no])
model = self._results_view.model()
model.update_coordinates(self.coords)
self._tabs.setCurrentIndex(1)
class AcquisitionThread(QtCore.QThread):
image_acquired = pyqtSignal(int, float, str)
results_fetched = pyqtSignal(str)
message = pyqtSignal(str)
acquisition_finished = pyqtSignal()
def __init__(self, z_motor, camera, scan_config):
QtCore.QThread.__init__(self)
self._scanconfig = scan_config
logger.debug(f"scanconfig: {scan_config}")
self.z_motor = z_motor
self.camera = camera
def save_params(self, params):
fname = Path(folders.res_folder) / "z_scan" / "parameters.yaml"
try:
with open(fname, "w") as fp:
yaml.dump(params, fp)
except:
logger.warning(f"failed to save z-scan parameter file: {fname}")
def save_image(self, n, z, img):
if not option(EMBL_SAVE_FILES):
return
zp = f"{n:02d}_Z{z:.3f}".replace(".", "p") + ".png"
fname = Path(folders.res_folder) / "z_scan" / zp
try:
imageio.imsave(fname, img)
except:
logger.error(f"failed to write: {fname}")
return f"failed to write image {n}"
return fname
def save_result(self, result):
if not option(EMBL_SAVE_FILES):
return
fname = Path(folders.res_folder) / "z_scan" / "result.txt"
with open(fname, "w") as fp:
fp.write(result)
def make_folders(self):
if not option(EMBL_SAVE_FILES):
return
fold = Path(folders.res_folder) / "z_scan"
fold.mkdir(parents=True, exist_ok=True)
def run(self):
self.make_folders()
cnf = self._scanconfig
zmin, zmax, zstep = cnf["z-start"], cnf["z-end"], cnf["z-step"]
zvals = np.arange(zmin, zmax, zstep)
num = len(zvals)
logger.debug(
f"Z-scan: min/max/step {zmin}/{zmax}/{zstep} ppm={cnf['ppm']} beam_xy={cnf['beam_xy']}"
)
logger.info(f"Z-vals: {', '.join([f'{z:.2f}' for z in zvals])}")
self.save_params(cnf)
embl_tube.make_PAR(zmin, zmax, zstep, cnf["ppm"], cnf["beam_xy"])
for n, z in enumerate(zvals):
idx = n + 1
self.message.emit(f"{idx}/{num} Z ➠ {z:.2f}")
self.z_motor.move(z, ignore_limits=True, wait=True)
assert_motor_positions([(self.z_motor, z, 0.1)], timeout=3.)
self.message.emit(f"{idx}/{num} acquiring")
img = self.camera.get_image()
self.message.emit(f"{idx}/{num} acquired")
img = np.rot90(img, 1)
img = bytescale(img)
fname = self.save_image(idx, z, img)
self.image_acquired.emit(idx, z, str(fname)) # just to populate table
self.message.emit(f"{idx}/{num} sending")
embl_tube.make_IMG(z, img)
self.message.emit(f"{idx}/{num} sent")
self.message.emit(f"acquisition finished")
embl_tube.finished()
self.message.emit(f"waiting for results")
result = embl_tube.make_RES()
self.save_result(result)
self.results_fetched.emit(result)
self.acquisition_finished.emit()

256
HelicalTable.py Normal file
View File

@@ -0,0 +1,256 @@
import sys
import os
import logging
import csv
from PyQt5 import QtCore, QtGui
import numpy as np
logger = logging.getLogger("helical")
from PyQt5.QtWidgets import (
QTableWidget,
QApplication,
QMainWindow,
QTableWidgetItem,
QFileDialog,
QHeaderView,
QAbstractItemView,
QMenu)
from PyQt5.QtCore import Qt, QFileInfo, pyqtSignal, QPoint, pyqtSlot
DATA_ITEM = 1
START_OMEGA_0 = 0
START_OMEGA_120 = 1
START_OMEGA_240 = 2
STOP_OMEGA_0 = 3
STOP_OMEGA_120 = 4
STOP_OMEGA_240 = 5
ROLE_XTAL_START = 1 + Qt.UserRole
ROLE_XTAL_END = 2 + Qt.UserRole
class HelicalTableWidget(QTableWidget):
gonioMoveRequest = pyqtSignal(float, float, float, float, float)
def __init__(self):
super().__init__()
self.check_change = True
self._scanAngularStep = 0.1
self._scanHorizontalCount = 5
self._scanVerticalCount = 10
self._current_xtal = None
self._start_angle = 0.0
self.init_ui()
def startAngle(self) -> float:
return self._start_angle
@pyqtSlot(float)
def setStartAngle(self, angle:float):
self._start_angle = angle
def init_ui(self):
self.setColumnCount(6)
self.setEditTriggers(QAbstractItemView.NoEditTriggers)
self.resizeColumnsToContents()
self.setSelectionMode(QAbstractItemView.SingleSelection)
self.setSelectionBehavior(QAbstractItemView.SelectItems)
labels = [
"0\nbx, bz\n(mm)",
"120\nbx, bz\n(mm)",
"240\nbx, bz\n(mm)",
"0\nbx, bz\n(mm)",
"120\nbx, bz\n(mm)",
"240\nbx, bz\n(mm)",
]
self.setHorizontalHeaderLabels(labels)
self.verticalHeader().resizeSections(QHeaderView.ResizeToContents)
self.horizontalHeader().setSectionResizeMode(QHeaderView.Stretch)
self.cellChanged.connect(self.c_current)
self.show()
def display_coords(self, x, y):
row = self.rowAt(y)
col = self.columnAt(x)
coords = self.gonio_coords_at(row, col)
omega = col % 3 * 120
data = coords[omega]
logger.info("gonio at {}, {}: {}".format(row, col, data))
def goto_gonio_position(self, x, y):
row = self.rowAt(y)
col = self.columnAt(x)
coords = self.gonio_coords_at(row, col)
omega = col % 3 * 120
data = coords[omega]
logger.info("move gonio to: {}".format(row, col, data))
self.gonioMoveRequest.emit(*data)
def remove_xtal(self, x, y):
row = self.rowAt(y)
col = self.columnAt(x)
logger.info("rowCount() = {}, removing row: {}".format(self.rowCount(), row))
self.removeRow(row)
if self.rowCount() == 0:
self._current_xtal = None
logger.info("rowCount() = {}".format(self.rowCount()))
def contextMenuEvent(self, event):
logger.info(event.pos())
x = event.pos().x()
y = event.pos().y()
menu = QMenu(self)
gotoAction = menu.addAction("Move Gonio Here")
gotoAction.triggered.connect(lambda b: self.goto_gonio_position(x, y))
removeAction = menu.addAction("Remove this line")
removeAction.triggered.connect(lambda b: self.remove_xtal(x, y))
menu.popup(QtGui.QCursor.pos())
def setScanHorizontalCount(self, count: int):
logger.debug("horizontal count: {}".format(count))
self._scanHorizontalCount = count
def scanHorizontalCount(self) -> int:
return self._scanHorizontalCount
def setScanVerticalCount(self, count: int):
logger.debug("vertical count: {}".format(count))
self._scanVerticalCount = count
def scanVerticalCount(self) -> int:
return self._scanVerticalCount
def setScanAngularStep(self, step: float):
self._scanAngularStep = step
def scanAngularStep(self) -> float:
return self._scanAngularStep
def scanTotalRange(self) -> float:
return self._scanVerticalCount * self._scanHorizontalCount * self._scanAngularStep
def gonio_coords_at(self, row, col):
role = ROLE_XTAL_START if col < 3 else ROLE_XTAL_END
coords = self.item(row, DATA_ITEM).data(role)
return coords
def add_xtal(self):
row = self.rowCount()
self.setRowCount(row + 1)
self._current_xtal = row
print(row)
for n in range(self.columnCount()):
self.setItem(row, n, QTableWidgetItem("unset"))
def set_xtal_start(self, data: list):
"""sets data to start position
data = [fx, fy, bx, bz, omega]
"""
self.set_data_point(ROLE_XTAL_START, data)
def set_xtal_end(self, data: list):
self.set_data_point(ROLE_XTAL_END, data)
def set_data_point(self, role: int, data: list):
if self._current_xtal is None:
self.add_xtal()
row, col = self._current_xtal, DATA_ITEM
try:
coords = self.item(row, DATA_ITEM).data(role)
except:
print("empty coords; initializing...")
coords = None
if coords is None:
coords = {}
fx, fy, bx, bz, omega = data
o = int(omega)
coords[o] = data
print("coords = {}".format(coords))
if role == ROLE_XTAL_END:
col = 3
else:
col = 0
for n in range(3):
omega = n * 120
try:
fx, fy, bx, bz, omega = coords[omega]
info = "{bx:.3f} mm\n{bz:.3f} mm".format(bx=bx, bz=bz)
except:
info = "unset"
# self.item(row, col + n).setData(Qt.DisplayRole, info)
self.item(row, col + n).setText(info)
self.item(row, col + n).setToolTip(info)
self.item(row, DATA_ITEM).setData(role, coords)
def c_current(self):
if self.check_change:
row = self.currentRow()
col = self.currentColumn()
value = self.item(row, col)
# value = value.text()
def load_datapoints(self):
self.check_change = False
path = QFileDialog.getOpenFileName(
self, "Open CSV", os.getenv("HOME"), "CSV(*.csv)"
)
if path[0] != "":
with open(path[0], newline="") as csv_file:
self.setRowCount(0)
self.setColumnCount(10)
my_file = csv.reader(csv_file, delimiter=",", quotechar="|")
for row_data in my_file:
row = self.rowCount()
self.insertRow(row)
if len(row_data) > 10:
self.setColumnCount(len(row_data))
for column, stuff in enumerate(row_data):
item = QTableWidgetItem(stuff)
self.setItem(row, column, item)
self.check_change = True
def get_data(self, as_numpy=False):
"""return a list of tuples with all defined data
[
(is_fiducial(boolean), x, y, gonio_x, gonio_y),...
]
"""
data = []
for row in range(self.rowCount()):
start = self.item(row, DATA_ITEM).data(ROLE_XTAL_START)
end = self.item(row, DATA_ITEM).data(ROLE_XTAL_END)
data.append([start, end])
# if as_numpy:
# data = np.asarray(data)
return data
if __name__ == "__main__":
class Sheet(QMainWindow):
def __init__(self):
super().__init__()
self.form_widget = HelicalTableWidget()
self.setCentralWidget(self.form_widget)
self.show()
app = QApplication(sys.argv)
sheet = Sheet()
sys.exit(app.exec_())

View File

@@ -0,0 +1,705 @@
# coding=utf-8
import logging
import os
from os.path import join
from typing import List
import numpy as np
import pandas as pd
import transformations as tfs
from PyQt5.QtCore import Qt, QFileInfo, pyqtSignal, pyqtSlot
from PyQt5.QtWidgets import (
QTableWidgetItem,
QFileDialog,
QGroupBox,
QWidget,
QGridLayout,
QTableWidget,
QAbstractItemView,
QItemDelegate,
QVBoxLayout,
QLabel,
QPushButton,
QApplication,
QMainWindow,
QHeaderView,
QLineEdit,
QSpinBox,
QMessageBox,
QAction,
QCheckBox)
logger = logging.getLogger(__name__)
# from coord_library import conversions
import conversions
from app_config import settings
from storage import Folders
folders = Folders()
XY_Coord = List[float]
DATA_ITEM = 1
ORIGINAL_X = 1
ORIGINAL_Y = 2
GONIO_X = 3
GONIO_Y = 4
RoleGoniometerCoord_X = 1 + Qt.UserRole
RoleGoniometerCoord_Y = 2 + Qt.UserRole
RoleCameraCoord_X = 3 + Qt.UserRole
RoleCameraCoord_Y = 4 + Qt.UserRole
RoleOriginalCoord_X = 5 + Qt.UserRole
RoleOriginalCoord_Y = 6 + Qt.UserRole
def sort_cmass_3d(coord):
"""sorts the 3D coordinates (np array 3 columns) with respect to their distane to the center of mass in 3d"""
cm = conversions.find_geo_center(coord)
# print cm
moved_coord = conversions.center_coord(coord, cm)
sorted_moved_coord = conversions.sort_center_3d(moved_coord)
minus_cm = np.multiply(cm, -1)
return conversions.center_coord(sorted_moved_coord, minus_cm)
def sort_cmass(coord):
"""sorts the 2D coordinates (np array 2 columns) with respect to their distane to the center of mass"""
cm = conversions.find_geo_center(coord)
# print cm
moved_coord = conversions.center_coord(coord, cm)
sorted_moved_coord = conversions.sort_center(moved_coord)
minus_cm = np.multiply(cm, -1)
return conversions.center_coord(sorted_moved_coord, minus_cm)
class MarkerDelegate(QItemDelegate):
def createEditor(self, parent, option, index):
comboBox = QLineEdit(parent)
comboBox.editingFinished.connect(self.emitCommitData)
return comboBox
def setEditorData(self, editor, index):
# pos = comboBox.findText(index.model().data(index), Qt.MatchExactly)
# comboBox.setCurrentIndex(pos)
editor.setText("{}".format(index.model().data(index)))
def setModelData(self, editor, model, index):
print("model {}".format(model))
print(
"delegate setData: {}x{} => {}".format(
index.row(), index.column(), editor.text()
)
)
model.setData(index, editor.text())
model.setData(index, float(editor.text()), Qt.UserRole)
def emitCommitData(self):
print("imagedelegate emitting")
self.commitData.emit(self.sender())
class PrelocatedCoordinates(QWidget):
prefixSelected = pyqtSignal(str)
dataFileLoaded = pyqtSignal(str)
prelocatedDataUpdated = pyqtSignal()
markersDeleted = pyqtSignal()
markerAdded = pyqtSignal(bool, list) # emits is_fiducial(boolean), [x, y, cx, cy]
selectedRowChanged = pyqtSignal(int)
moveFastStageRequest = pyqtSignal(float, float)
def __init__(self, parent=None):
super(PrelocatedCoordinates, self).__init__(parent)
self._xtals_transformed = True
self._current_row = None
layout = QVBoxLayout()
self.setLayout(layout)
frame = QWidget()
bl = QGridLayout()
frame.setLayout(bl)
self._label_prefix = QLabel("not set")
self._label_datafile = QLabel("not loaded")
self._label_datafile.setWordWrap(True)
bl.addWidget(QLabel("Prefix"), 0, 0)
bl.addWidget(self._label_prefix, 0, 1)
bl.addWidget(QLabel("Data File"), 1, 0)
bl.addWidget(self._label_datafile, 1, 1)
but1 = QPushButton("Load Datafile")
but1.clicked.connect(lambda: self.loadMarkers(None))
bl.addWidget(but1, 2, 0)
but1 = QPushButton("Save Datafile")
but1.clicked.connect(lambda: self.saveDataAs())
bl.addWidget(but1, 2, 1)
but2 = QPushButton("Clear Data")
but2.clicked.connect(self.clearMarkers)
bl.addWidget(but2, 3, 0)
# but = QPushButton("Random Data")
# but.clicked.connect(self.generate_random_data)
# bl.addWidget(but, 4, 0)
# self._num_random_points = QSpinBox()
# self._num_random_points.setMinimum(2)
# self._num_random_points.setMaximum(10000)
# self._num_random_points.setValue(10)
# self._num_random_points.setSuffix(" points")
# bl.addWidget(self._num_random_points, 4, 1)
but = QPushButton("dump numpy")
but.clicked.connect(self.dump_numpy)
bl.addWidget(but, 5, 0)
# but = QPushButton("Dump to console")
# but.clicked.connect(self.dump_data)
# bl.addWidget(but, 10, 0, 1, 1)
but = QCheckBox("collect fiducials")
but.setChecked(False)
but.setToolTip("Collect or not the fiducial positions.")
self._collect_fiducials = False
but.stateChanged.connect(self.set_collect_fiducials)
bl.addWidget(but, 10, 0, 1, 1)
but = QCheckBox("draw crystal marks")
but.setChecked(False)
self._draw_crystal_marks = False
but.stateChanged.connect(self.set_draw_crystal_marks)
bl.addWidget(but, 10, 1, 1, 1)
but = QPushButton("Transform")
but.clicked.connect(self.transform_non_fiducials_in_model)
bl.addWidget(but, 20, 0, 2, 2)
layout.addWidget(frame)
self.markersTable = QTableWidget()
self.markersTable.setSelectionMode(QAbstractItemView.SingleSelection)
self.markersTable.setSelectionBehavior(QAbstractItemView.SelectRows)
self.markersTable.setItemDelegate(MarkerDelegate(self))
# self.markersTable.horizontalHeader().setDefaultSectionSize(80)
self.markersTable.setColumnCount(5)
self.markersTable.setHorizontalHeaderLabels(
("Fiducial?", "orig X", "orig Y", "X", "Y")
)
self.markersTable.horizontalHeader().setSectionResizeMode(
0, QHeaderView.ResizeToContents
)
self.markersTable.horizontalHeader().setSectionResizeMode(
1, QHeaderView.ResizeToContents
)
self.markersTable.horizontalHeader().setSectionResizeMode(
2, QHeaderView.ResizeToContents
)
self.markersTable.horizontalHeader().setSectionResizeMode(
3, QHeaderView.ResizeToContents
)
self.markersTable.horizontalHeader().setSectionResizeMode(
4, QHeaderView.ResizeToContents
)
self.markersTable.horizontalHeader().setSectionResizeMode(QHeaderView.Stretch)
self.markersTable.itemSelectionChanged.connect(self.selection_changed)
self.markersTable.setContextMenuPolicy(Qt.ActionsContextMenu)
deleteRowAction = QAction("Delete this row", self)
deleteRowAction.triggered.connect(self.delete_selected)
self.markersTable.addAction(deleteRowAction)
moveRequestAction = QAction("Move Here", self)
moveRequestAction.triggered.connect(self.request_stage_movement)
self.markersTable.addAction(moveRequestAction)
layout.addWidget(self.markersTable, stretch=2)
self.connect_all_signals()
def delete_selected(self):
row = self._current_row
try:
row += 0
except:
logger.warning("select a row first")
QMessageBox.warning(
self,
"Select a marker first",
"You must select a marker first before updating its goniometer position",
)
return
self.markersTable.removeRow(row)
self.prelocatedDataUpdated.emit()
def selection_changed(self):
row = self.markersTable.currentRow()
if row < 0:
return
self._current_row = row
self.selectedRowChanged.emit(row)
logger.debug("selection changed: current row {}".format(row))
def connect_all_signals(self):
self.prefixSelected.connect(lambda t: self._label_prefix.setText(t))
self.dataFileLoaded.connect(lambda t: self._label_datafile.setText(t))
def markerItemChanged(self, item):
print(item.row(), item.column())
def set_fiducial_coords(self, camx, camy, gx, gy):
tbl = self.markersTable
row = self._current_row
try:
row += 0
except:
logger.warning("select a row first")
QMessageBox.warning(
self,
"Select a marker first",
"You must select a marker first before updating its goniometer position",
)
return
origx = tbl.item(row, ORIGINAL_X).data(RoleOriginalCoord_X)
origy = tbl.item(row, ORIGINAL_Y).data(RoleOriginalCoord_Y)
item = tbl.item(row, DATA_ITEM)
item.setData(RoleCameraCoord_X, camx)
item.setData(RoleCameraCoord_Y, camy)
item.setData(RoleGoniometerCoord_X, gx)
item.setData(RoleGoniometerCoord_Y, gy)
# item.setData(RoleOriginalCoord_X, origx)
# item.setData(RoleOriginalCoord_Y, origy)
# logger.debug(': [{}] = Original: {}, {} | Camera: {}, {} | Gonio: {}, {}'.format(1+row, origx, origy, camx, camy, gx, gy))
logger.debug(f": [{1+row}] = Original: {origx}, {origy} | Camera: {camx}, {camy} | Gonio: {gx}, {gy}")
tbl.item(row, GONIO_X).setData(Qt.DisplayRole, f"{gx:8.5f} mm\n{camx:8.1f} px")
tbl.item(row, GONIO_Y).setData(Qt.DisplayRole, f"{gy:8.5f} mm\n{camy:8.1f} px")
# mark row as a fiducial
tbl.item(row, 0).setCheckState(Qt.Checked)
tbl.item(row, 0).setData(Qt.UserRole, True)
tbl.selectRow(row + 1)
self.prelocatedDataUpdated.emit()
def set_selected_gonio_coords(self, xy: XY_Coord):
tbl = self.markersTable
row = self._current_row
try:
row += 0
except:
logger.warning("select a row first")
QMessageBox.warning(
self,
"Select a marker first",
"You must select a marker first before updating its goniometer position",
)
return
for n, v in enumerate(xy):
idx = 3 + n
tbl.setCurrentCell(row, idx) # gonio X
logger.debug("item: [{},{}] = {}".format(row, idx, v))
item = tbl.currentItem()
item.setData(Qt.EditRole, "{:.3f}".format(v))
item.setData(Qt.UserRole, v)
# mark row as a fiducial
tbl.setCurrentCell(row, 0)
tbl.currentItem().setCheckState(Qt.Checked)
tbl.currentItem().setData(Qt.UserRole, True)
def set_data_goniometer(self, row: int, xy: XY_Coord):
tbl = self.markersTable
row = self._current_row
try:
row += 0
except:
logger.warning("select a row first")
QMessageBox.warning(
self,
"Select a marker first",
"You must select a marker first before updating its goniometer position",
)
return
for n, v in enumerate(xy):
idx = 3 + n
tbl.setCurrentCell(row, idx) # gonio X
item = tbl.currentItem()
item.setData(Qt.EditRole, "{:.3f}".format(v))
item.setData(Qt.UserRole, v)
@pyqtSlot(int)
def set_draw_crystal_marks(self, val):
logger.info(f"{'' if val else 'not '}drawing crystal markers")
self._draw_crystal_marks = val
@pyqtSlot(int)
def set_collect_fiducials(self, val):
logger.info(f"{'' if val else 'not '}collecting fiducials")
self._collect_fiducials = val
def get_collection_targets(self):
return self.get_data(fiducials=self._collect_fiducials)
def get_data(self, fiducials=True, crystals=True, as_numpy=False):
"""return a list of tuples with all defined data
[
(is_fiducial(boolean), x, y, gonio_x, gonio_y),...
]
"""
data = []
item = self.markersTable.item
for row in range(self.markersTable.rowCount()):
is_fiducial = Qt.Checked == item(row, 0).checkState()
ditem = item(row, DATA_ITEM)
x = ditem.data(RoleGoniometerCoord_X)
y = ditem.data(RoleGoniometerCoord_Y)
cx = ditem.data(RoleCameraCoord_X)
cy = ditem.data(RoleCameraCoord_Y)
origx = ditem.data(RoleOriginalCoord_X)
origy = ditem.data(RoleOriginalCoord_Y)
if is_fiducial and fiducials:
data.append([is_fiducial, x, y, cx, cy, origx, origy])
if not is_fiducial and crystals:
data.append([is_fiducial, x, y, cx, cy, origx, origy])
if as_numpy:
data = np.asarray(data)
return data
def get_original_coordinates(self, fiducials=True):
"""return a numpy array with the original prelocated coordinates of the fiducial entries"""
data = self.get_data(fiducials=fiducials, crystals=not fiducials, as_numpy=True)
data = data[:, 5:]
zeros = np.zeros(data.shape)
zeros[:, 1] = 1
return np.concatenate((data, zeros), axis=1)
def get_goniometer_coordinates(self, fiducials=True):
"""return a numpy array with the goniometer coordinates of the fiducial entries"""
data = self.get_data(fiducials=fiducials, crystals=not fiducials, as_numpy=True)
data = data[:, 1:3]
zeros = np.zeros(data.shape)
zeros[:, 1] = 1
return np.concatenate((data, zeros), axis=1)
def get_camera_coordinates(self, fiducials=True):
"""return a numpy array with the goniometer coordinates of the fiducial entries"""
data = self.get_data(fiducials=fiducials, crystals=not fiducials, as_numpy=True)
data = data[:, 3:5]
zeros = np.zeros(data.shape)
zeros[:, 1] = 1
return np.concatenate((data, zeros), axis=1)
def dump_matrix(self, M):
scale, shear, angles, translate, perspective = tfs.decompose_matrix(M)
angles_deg = [a * 180 / np.pi for a in angles]
print("Transformation matrix Aerotech => SwissMX")
print(M)
print((" scale {:9.4f} {:9.4f} {:9.4f}".format(*scale)))
print((" shear {:9.4f} {:9.4f} {:9.4f}".format(*shear)))
print((" angles rad {:9.4f} {:9.4f} {:9.4f}".format(*angles)))
print((" angles deg {:9.4f} {:9.4f} {:9.4f}".format(*angles_deg)))
print((" translate {:9.4f} {:9.4f} {:9.4f}".format(*translate)))
print(("perspective {:9.4f} {:9.4f} {:9.4f}".format(*perspective)))
def transform_non_fiducials_in_model(self):
forg = self.get_original_coordinates(fiducials=True)
fgon = self.get_goniometer_coordinates(fiducials=True)
fcam = self.get_camera_coordinates(fiducials=True)
gmat = sort_cmass_3d(fgon)
omat = sort_cmass_3d(forg)
try:
M_org2gon = tfs.superimposition_matrix(omat.T, gmat.T, scale=True)
M_org2cam = tfs.superimposition_matrix(forg.T, fcam.T, scale=True)
except:
QMessageBox.warning(self.parent(), title="failed to find superimposition matrix", text="Failed to find superimposition matrix.\n\tPlease try again.")
return
# scale, shear, angles, translate, perspective = tfs.decompose_matrix(M_org2gon)
org_data = self.get_original_coordinates(fiducials=False)
gon_data = np.dot(M_org2gon, org_data.T)
cam_data = np.dot(M_org2cam, org_data.T)
tbl = self.markersTable
num_fiducials = forg.shape[0]
gon_data = (gon_data.T)[:, 0:2] # only X,Y matters
cam_data = (cam_data.T)[:, 0:2] # only X,Y matters
combined = np.concatenate((gon_data, cam_data), axis=1)
item = self.markersTable.item # function alias
for row, data in enumerate(
combined, num_fiducials
): # enumeration starts at *num_fiducials*
gx, gy, cx, cy = data
ditem = item(row, DATA_ITEM)
ditem.setData(RoleCameraCoord_X, cx)
ditem.setData(RoleCameraCoord_Y, cy)
ditem.setData(RoleGoniometerCoord_X, gx)
ditem.setData(RoleGoniometerCoord_Y, gy)
item(row, GONIO_X).setData(Qt.DisplayRole, f"{gx:8.5f} mm\n{cx:8.1f} px")
item(row, GONIO_Y).setData(Qt.DisplayRole, f"{gy:8.5f} mm\n{cy:8.1f} px")
self._xtals_transformed = True
self.prelocatedDataUpdated.emit()
def request_stage_movement(self):
logger = logging.getLogger("preloc.move_stage")
row = self._current_row
x = self.markersTable.item(row, DATA_ITEM).data(RoleGoniometerCoord_X)
y = self.markersTable.item(row, DATA_ITEM).data(RoleGoniometerCoord_Y)
logger.info(f"request move gonio to {x:.3f}, {y:.3f} mm")
self.moveFastStageRequest.emit(x, y)
def dump_numpy(self):
R = tfs.random_rotation_matrix(np.random.random(3))
d = self.get_goniometer_coordinates()
print("dumping")
print(d)
dR = np.dot(R, d)
print(dR.T)
def get_fiducials(self, as_numpy=False):
return self.get_data(crystals=False, as_numpy=as_numpy)
def get_crystals(self, as_numpy=False):
return self.get_data(fiducials=False, as_numpy=as_numpy)
def dump_data(self):
for ref, x, y, cx, cy, ox, oy in self.get_data():
print(f"fiducial:{ref} [{x}, {y}, {cx}, {cy}]")
def append_data(self, data: List):
"""append data (a list of values) to the model
len(data) == 2 => (X, Y) from prelocated coordinate
len(data) == 4 => (X, Y, GX, GY) plus gonio coordinates
len(data) == 5 => (fiducial?, X, Y, GX, GY) plus fiducial
"""
row = self.markersTable.rowCount()
self.markersTable.setRowCount(row + 1)
data = list(data)
if len(data) == 2:
data.extend([0, 0])
if len(data) == 4:
data.insert(0, False) # fiducial flag
self.addSingleMarker(row, data)
self.prelocatedDataUpdated.emit()
def clearMarkers(self):
self.markersTable.setRowCount(0)
self.markersDeleted.emit()
def generate_random_data(self):
import io
data = io.StringIO()
npoints = self._num_random_points.value()
for n in range(npoints):
x, y, a, b = (
np.random.randint(0, 2000),
np.random.randint(0, 2000),
np.random.randint(-4000, 4000) / 1000.,
np.random.randint(-4000, 4000) / 1000.,
)
data.write("{}\t{}\t{}\t{}\n".format(x, y, a, b))
data.seek(0)
data.name = "random.csv"
self.loadMarkers(data)
def saveDataAs(self, filename=None):
# filename = folders.get_file("prelocated-save.dat")
data_folder = settings.value("folders/last_prelocation_folder")
if filename is None:
filename, _ = QFileDialog.getSaveFileName(
parent=None,
caption="Open CSV data file",
directory=data_folder,
filter="CSV Data Files (*.csv);;All Files (*)",
)
if not filename:
return
settings.setValue("folders/last_prelocation_folder", QFileInfo(filename).canonicalPath())
logger.info("Saving data in {}".format(filename))
data = self.get_data(as_numpy=True)
df = pd.DataFrame(data)
df.to_csv(filename, float_format="%.6f")
def addSingleMarker(self, row, marker):
is_fiducial, origx, origy, gx, gy = marker
logger.debug(f": [{1+row}] | Original: {origx}, {origy} | Gonio: {gx}, {gy}")
item0 = QTableWidgetItem()
item0.setData(Qt.UserRole, is_fiducial)
item0.setFlags(item0.flags() & ~Qt.ItemIsEditable)
item0.setCheckState(Qt.Checked if is_fiducial else Qt.Unchecked)
item0.setTextAlignment(Qt.AlignCenter)
item1 = QTableWidgetItem("{:.1f}".format(origx))
item1.setTextAlignment(Qt.AlignRight)
item2 = QTableWidgetItem("{:.1f}".format(origy))
item2.setTextAlignment(Qt.AlignRight)
item3 = QTableWidgetItem("{:.3f} mm".format(gx))
item3.setFlags(item3.flags() & ~Qt.ItemIsEditable)
item3.setTextAlignment(Qt.AlignRight)
item4 = QTableWidgetItem("{:.3f} mm".format(gy))
item4.setFlags(item4.flags() & ~Qt.ItemIsEditable)
item4.setTextAlignment(Qt.AlignRight)
self.markersTable.setItem(row, 0, item0)
self.markersTable.setItem(row, ORIGINAL_X, item1)
self.markersTable.setItem(row, ORIGINAL_Y, item2)
self.markersTable.setItem(row, GONIO_X, item3)
self.markersTable.setItem(row, GONIO_Y, item4)
self.markerAdded.emit(is_fiducial, [origx, origy, gx, gy])
item = self.markersTable.item(row, DATA_ITEM)
item.setData(RoleCameraCoord_X, origx) # initially set to original
item.setData(RoleCameraCoord_Y, origy) # initially set to original
item.setData(RoleGoniometerCoord_X, gx)
item.setData(RoleGoniometerCoord_Y, gy)
item.setData(RoleOriginalCoord_X, origx)
item.setData(RoleOriginalCoord_Y, origy)
def loadMarkers(self, filename=None):
logger = logging.getLogger("preloc.loadMarkers")
def_folder = join(folders.pgroup_folder, "preloc_sheets")
data_folder = settings.value("folders/last_prelocation_folder", def_folder)
if filename is None:
filename, _ = QFileDialog.getOpenFileName(
self,
"Open CSV data file",
data_folder,
"Any file.. (*.txt);;CSV Data Files (*.csv);;All Files (*)",
)
# filename = folders.get_file("preloc-in.csv")
if not filename: # cancelled dialog
return
try:
settings.setValue("folders/last_prelocation_folder", QFileInfo(filename).canonicalPath())
settings.setValue("folders/last_prelocation_sheet", QFileInfo(filename).absolutePath())
except TypeError as e:
pass
self.clearMarkers()
logger.info(f"loading prelocated coords from {filename}")
df = pd.read_csv(filename, comment="#", header=None, delim_whitespace=False, delimiter="[\t,;]")
try:
prefix = QFileInfo(filename).baseName()
except:
prefix = filename.name
filename = prefix
logger.info(f"prefix => {prefix}")
gonio_coords_available = True
for data in df.as_matrix(): # FIXME FutureWarning: Method .as_matrix will be removed in a future version. Use .values instead.
row = self.markersTable.rowCount()
self.markersTable.setRowCount(row + 1)
original = np.copy(data)
data = list(data)
if len(data) in [2, 3]:
gonio_coords_available = False
data.extend([0, 0])
if len(data) == 4:
data.insert(0, False) # fiducial flag
elif len(data) == 5: # fiducial already there, convert to bool
data[0] = bool(data[0])
else:
QMessageBox.warning(
self,
"Wrong number of points in data file",
"I was expecting either 2, 3, 4 or 5 data points per line."
"\n\nFailed around a line with: {}".format(list(original)),
)
self.addSingleMarker(row, data)
self._xtals_transformed = False
self.prefixSelected.emit(prefix)
# only emit this signal if goniometer coordinates already read from file
if gonio_coords_available:
logger.debug(f"dataFileLoaded.emit => {filename}")
self.dataFileLoaded.emit(filename)
"""
Signals QTableWidget
void cellActivated(int row, int column)
void cellChanged(int row, int column)
void cellClicked(int row, int column)
void cellDoubleClicked(int row, int column)
void cellEntered(int row, int column)
void cellPressed(int row, int column)
void currentCellChanged(int currentRow, int currentColumn, int previousRow, int previousColumn)
void currentItemChanged(QTableWidgetItem *current, QTableWidgetItem *previous)
void itemActivated(QTableWidgetItem *item)
void itemChanged(QTableWidgetItem *item)
void itemClicked(QTableWidgetItem *item)
void itemDoubleClicked(QTableWidgetItem *item)
void itemEntered(QTableWidgetItem *item)
void itemPressed(QTableWidgetItem *item)
void itemSelectionChanged()
"""
class MainWindow(QMainWindow):
def __init__(self, parent=None):
super(MainWindow, self).__init__(parent)
self.centralWidget = QWidget()
self.markwi = PrelocatedCoordinates(title="Prelocated Coordinates", parent=self)
self.setCentralWidget(self.centralWidget)
mainLayout = QVBoxLayout()
mainLayout.addWidget(self.markwi)
self.centralWidget.setLayout(mainLayout)
if __name__ == "__main__":
import sys
app = QApplication(sys.argv)
mainWin = MainWindow()
mainWin.show()
sys.exit(app.exec_())