260 lines
10 KiB
Python
260 lines
10 KiB
Python
run("imgproc/CoverDetection")
|
|
run("imgproc/CoverCorrelation")
|
|
|
|
|
|
class CoverDetectionConfig(RegisterConfig):
|
|
def __init__(self):
|
|
self.number_images = 1
|
|
self.continuous = False
|
|
self.modulo = 1
|
|
self.center_x = 820.0
|
|
self.center_y = 570.0
|
|
self.scale_x = 1.0
|
|
self.scale_y = 1.0
|
|
self.rotation_degrees = 0.0
|
|
self.correlation_threshold = 0.60
|
|
self.diameter_target=95
|
|
self.diameter_disk=670
|
|
self.diameter_dewar=1800
|
|
self.border_dewar=0.56
|
|
self.size_mask_disk=0.8
|
|
self.size_mask_dewar=0.98
|
|
self.target_time_tolerance = 20.0
|
|
self.max_detect_offset=10
|
|
self.open_strategy = "both"
|
|
self.threshold_method = "IsoData"
|
|
|
|
class CoverCorrelation(ReadonlyAsyncRegisterBase):
|
|
def __init__(self, name):
|
|
ReadonlyAsyncRegisterBase.__init__(self, name)
|
|
self.precision = 3
|
|
|
|
def set(self, value):
|
|
self.onReadout(value)
|
|
|
|
|
|
class CoverDetection(ReadonlyAsyncRegisterBase, ReadonlyRegisterArray):
|
|
|
|
def __init__(self, name, image):
|
|
ReadonlyAsyncRegisterBase.__init__(self, name, CoverDetectionConfig())
|
|
self.image = image
|
|
|
|
class ImgListener (ImageListener):
|
|
def onImage(s, origin, image, data):
|
|
self.append(image)
|
|
def onError(s, origin, ex):
|
|
print ex
|
|
|
|
self.listener = ImgListener()
|
|
self.enabled = False
|
|
self.images = []
|
|
self.processing_time = None
|
|
self.renderer = None
|
|
self.error = None
|
|
self.error_overlay=None
|
|
self.img_counter, self.proc_counter = 0, 0
|
|
self.offset_px = self.offset_um = None
|
|
#self.correlation = CoverCorrelation("cover_correlation")
|
|
self.correlation_error = False
|
|
self.disc_marker_ov = None
|
|
self.center_marker_ov = None
|
|
self.debug = False
|
|
|
|
def doInitialize(self):
|
|
try:
|
|
self.enable()
|
|
self.clear()
|
|
#self.grab_ref_image()
|
|
except:
|
|
traceback.print_exc()
|
|
raise
|
|
|
|
def doClose(self):
|
|
self.disable()
|
|
#self.correlation.close()
|
|
|
|
def set_renderer(self, renderer):
|
|
self.clear()
|
|
self.renderer = renderer
|
|
|
|
def enable(self):
|
|
if not self.enabled:
|
|
self.enabled = True
|
|
self.image.addListener(self.listener)
|
|
self.img_counter, self.proc_counter = 0, 0
|
|
self.offset_px = self.offset_um = None
|
|
|
|
def disable(self):
|
|
if self.enabled:
|
|
self.image.removeListener(self.listener)
|
|
self.enabled = False
|
|
self.onReadout(None)
|
|
self.clear()
|
|
|
|
def clear(self):
|
|
self.set_marker(None)
|
|
self.set_error_overlay(None)
|
|
#self.set_correlation(None)
|
|
|
|
def append(self, image):
|
|
self.img_counter += 1
|
|
if self.img_counter % self.config.modulo == 0:
|
|
ip = load_image(ImagingUtils.grayscale(image, None))
|
|
self.images.append(ip)
|
|
while len(self.images) > self.config.number_images:
|
|
self.images.pop(0)
|
|
if len(self.images) == self.config.number_images:
|
|
start = time.time()
|
|
if len(self.images) ==1:
|
|
ip = self.images[0]
|
|
else:
|
|
ip = integrate(self.images)
|
|
|
|
self.error = None
|
|
center = (self.config.center_x, self.config.center_y)
|
|
try:
|
|
x,y = detect_cover_position(ip, center, \
|
|
diam_target=self.config.diameter_target, diam_disk= self.config.diameter_disk, diam_dewar=self.config.diameter_dewar, \
|
|
mask_disk=self.config.size_mask_disk, mask_dewar=self.config.size_mask_dewar, border_dewar = self.config.border_dewar, \
|
|
offset_threshold=self.config.max_detect_offset , target_time_tolerance= self.config.target_time_tolerance, \
|
|
open_strategy= self.config.open_strategy, threshold_method=self.config.threshold_method, debug=self.debug)
|
|
|
|
except Exception as e:
|
|
self.error = e
|
|
x,y = -1, -1
|
|
try:
|
|
no_cover = detect_no_cover( ip, center, diam_dewar=self.config.diameter_dewar, border_dewar=self.config.border_dewar, \
|
|
threshold_method=self.config.threshold_method, debug=self.debug)
|
|
if no_cover:
|
|
self.error = "No cover not detected"
|
|
except Exception as e2:
|
|
print e2
|
|
|
|
|
|
self.processing_time = time.time()-start
|
|
self.proc_counter += 1
|
|
|
|
#try:
|
|
# corr = get_correlation(ip.getBufferedImage())
|
|
# self.correlation.set(corr)
|
|
# self.set_correlation(corr)
|
|
#except Exception as e:
|
|
# #traceback.print_exc()
|
|
# if self.error is None:
|
|
# self.error = e
|
|
self.set(x, y, self.processing_time * 1000, self.error)
|
|
|
|
#traceback.print_exc()
|
|
|
|
if not self.config.continuous:
|
|
self.images=[]
|
|
|
|
def calculate_factors(self, x1, y1, x2, y2, mx1, my1, mx2, my2):
|
|
dx = x2-x1
|
|
dy = y2-y1
|
|
dmx = mx2-mx1
|
|
dmy = my2-my1
|
|
sx = dmx/dx
|
|
sy = dmy / dy
|
|
tp=math.atan(dy/dx)
|
|
tm=math.atan(dmy/dmx)
|
|
t = tm - tp
|
|
return sx, sy, math.degrees(t)
|
|
|
|
def pixel_to_millis(self, x, y):
|
|
try:
|
|
offx, offy = x-self.config.center_x, y-self.config.center_y
|
|
if self.config.rotation_degrees > 0.1:
|
|
t = math.radians(self.config.rotation_degrees)
|
|
offx, offy = (offx * math.cos(t) + offy * math.sin(t)), (-offx * math.sin(t) + offy * math.cos(t))
|
|
mx, my = self.config.scale_x*offx, self.config.scale_y*offy
|
|
return round(mx,3), round(my,3)
|
|
except:
|
|
return float('nan'), float('nan')
|
|
|
|
def set(self, x, y, tm, error):
|
|
if self.enabled:
|
|
if error:
|
|
self.offset_px = self.offset_um = None
|
|
value = str(error)
|
|
else:
|
|
offx, offy = x-self.config.center_x, y-self.config.center_y
|
|
self.offset_px = opx, opy = round(offx), round(offy)
|
|
self.offset_um = omx, omy = self.pixel_to_millis(x,y)
|
|
value = to_array([x, y, opx, opy, omx, -omy, tm], 'i')
|
|
self.onReadout(value)
|
|
if self.renderer is not None:
|
|
if error is None:
|
|
marker = Overlays.Crosshairs(self.renderer.getPenProfile(), Point(int(x),int(y)), Dimension(-1, -1))
|
|
else:
|
|
marker = None
|
|
self.set_marker(marker);
|
|
self.set_error_overlay(error)
|
|
|
|
def set_marker(self, marker):
|
|
if self.renderer is not None:
|
|
self.renderer.setMarker(marker)
|
|
|
|
def set_error_overlay(self, error):
|
|
if self.renderer is not None:
|
|
if error is None:
|
|
error_overlay = None
|
|
else:
|
|
error_overlay = Overlays.Text(self.renderer.getPenErrorText(), str(error), Font("Verdana", Font.PLAIN, 12), Point(20, 20))
|
|
error_overlay.setFixed(True)
|
|
error_overlay.setAnchor(Overlay.ANCHOR_VIEWPORT_TOP_LEFT)
|
|
self.renderer.updateOverlays(error_overlay, self.error_overlay)
|
|
self.error_overlay = error_overlay
|
|
else:
|
|
self.error_overlay = None
|
|
|
|
def grab_ref_image(self):
|
|
if self.config.number_images>1:
|
|
images = []
|
|
for i in range(self.config.number_images):
|
|
ip = load_image(ImagingUtils.grayscale(self.image.output, None))
|
|
images.append(ip)
|
|
self.image.waitNext(-1)
|
|
ip = integrate(images)
|
|
image = ip.getBufferedImage()
|
|
else:
|
|
image = self.image.output
|
|
save_ref_img(image)
|
|
|
|
#def set_correlation(self, corr):
|
|
# self.corr = corr
|
|
# error = corr is not None and corr < self.config.correlation_threshold
|
|
# if error != self.correlation_error:
|
|
# self.correlation_error = error
|
|
# if error:
|
|
# print ("Image anomally detected")
|
|
# else:
|
|
# print ("Image ok")
|
|
# if error:
|
|
# raise Exception ("Image anomally detected")
|
|
|
|
def show_disk_marker(self):
|
|
size = int(DISK_SIZE_MM * (self.config.scale_x + self.config.scale_y)/2.0)
|
|
pn = show_panel(self.image)
|
|
ov=Overlays.Ellipse(Pen(Color.RED), Point(0,0), Dimension(661,661))
|
|
ov.movable=True
|
|
ov.solid=True
|
|
pn.updateOverlays(ov, self.disc_marker_ov)
|
|
self.disc_marker_ov = ov
|
|
|
|
def show_center_marker(self):
|
|
pn = show_panel(self.image)
|
|
ov=Overlays.Crosshairs(Pen(Color.RED), Point(int(self.config.center_x), int(self.config.center_y)), Dimension(-1,-1))
|
|
ov.movable=False
|
|
pn.updateOverlays(ov, self.center_marker_ov)
|
|
self.center_marker_ov = ov
|
|
|
|
|
|
|
|
add_device(CoverDetection("cover_detection", img), force = True)
|
|
#add_device(cover_detection.correlation, force = True)
|
|
#renderer=show_panel(img)
|
|
#cover_detection.set_renderer(renderer)
|
|
|
|
|