Fixes to naming to make basic lint checks pass
This commit is contained in:
parent
e4cfcc76a7
commit
4eeff35e2d
4 changed files with 41 additions and 31 deletions
|
|
@ -88,8 +88,8 @@ class BackgroundDetectThing(Thing):
|
||||||
current_image = np.array(Image.open(current_image.open()))
|
current_image = np.array(Image.open(current_image.open()))
|
||||||
|
|
||||||
# we're working in the LUV colourspace as it collect colours together in a human-intuitive way
|
# we're working in the LUV colourspace as it collect colours together in a human-intuitive way
|
||||||
current_image_LUV = cv2.cvtColor(current_image, cv2.COLOR_RGB2LUV)
|
current_image_luv = cv2.cvtColor(current_image, cv2.COLOR_RGB2LUV)
|
||||||
mask = self.background_mask(current_image_LUV)
|
mask = self.background_mask(current_image_luv)
|
||||||
return np.count_nonzero(mask) / np.prod(mask.shape) * 100
|
return np.count_nonzero(mask) / np.prod(mask.shape) * 100
|
||||||
|
|
||||||
@thing_action
|
@thing_action
|
||||||
|
|
@ -114,11 +114,11 @@ class BackgroundDetectThing(Thing):
|
||||||
background = np.array(Image.open(background.open()))
|
background = np.array(Image.open(background.open()))
|
||||||
|
|
||||||
# we're working in the LUV colourspace as it collect colours together in a human-intuitive way
|
# we're working in the LUV colourspace as it collect colours together in a human-intuitive way
|
||||||
background_LUV = cv2.cvtColor(background, cv2.COLOR_RGB2LUV)
|
background_luv = cv2.cvtColor(background, cv2.COLOR_RGB2LUV)
|
||||||
|
|
||||||
ch1 = (background_LUV.T[0]).flatten()
|
ch1 = (background_luv.T[0]).flatten()
|
||||||
ch2 = (background_LUV.T[1]).flatten()
|
ch2 = (background_luv.T[1]).flatten()
|
||||||
ch3 = (background_LUV.T[2]).flatten()
|
ch3 = (background_luv.T[2]).flatten()
|
||||||
|
|
||||||
points = np.array([np.asarray(ch1), np.asarray(ch2), np.asarray(ch3)]).T
|
points = np.array([np.asarray(ch1), np.asarray(ch2), np.asarray(ch3)]).T
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -71,18 +71,18 @@ class SimulatedCamera(BaseCamera):
|
||||||
sprite[rr < i] = 255
|
sprite[rr < i] = 255
|
||||||
self.sprites.append(sprite)
|
self.sprites.append(sprite)
|
||||||
|
|
||||||
def generate_blobs(self, N: int = 1000):
|
def generate_blobs(self, n_blobs: int = 1000):
|
||||||
"""Generate coordinates of blobs
|
"""Generate coordinates of blobs
|
||||||
|
|
||||||
Blobs are characterised by X, Y, sprite
|
Blobs are characterised by X, Y, sprite
|
||||||
We also generate a KD tree to rapidly find blobs in an image
|
We also generate a KD tree to rapidly find blobs in an image
|
||||||
"""
|
"""
|
||||||
self.blobs = np.zeros((N, 3))
|
self.blobs = np.zeros((n_blobs, 3))
|
||||||
rng = np.random.default_rng()
|
rng = np.random.default_rng()
|
||||||
w = np.max(self.glyph_shape)
|
w = np.max(self.glyph_shape)
|
||||||
self.blobs[:, 0] = rng.uniform(w / 2, self.canvas_shape[0] - w / 2, N)
|
self.blobs[:, 0] = rng.uniform(w / 2, self.canvas_shape[0] - w / 2, n_blobs)
|
||||||
self.blobs[:, 1] = rng.uniform(w / 2, self.canvas_shape[1] - w / 2, N)
|
self.blobs[:, 1] = rng.uniform(w / 2, self.canvas_shape[1] - w / 2, n_blobs)
|
||||||
self.blobs[:, 2] = rng.choice(len(self.sprites), N)
|
self.blobs[:, 2] = rng.choice(len(self.sprites), n_blobs)
|
||||||
|
|
||||||
def generate_canvas(self):
|
def generate_canvas(self):
|
||||||
"""Generate a blank canvas"""
|
"""Generate a blank canvas"""
|
||||||
|
|
|
||||||
|
|
@ -296,7 +296,9 @@ class CameraStageMapper(Thing):
|
||||||
def assert_calibrated(self):
|
def assert_calibrated(self):
|
||||||
"""Raise an exception if the image_to_stage_displacement matrix is not set"""
|
"""Raise an exception if the image_to_stage_displacement matrix is not set"""
|
||||||
if self.image_to_stage_displacement_matrix is None:
|
if self.image_to_stage_displacement_matrix is None:
|
||||||
raise CSMUncalibratedError()
|
# Disable check of no message in raised exception as the message is explicitly
|
||||||
|
# added by CSMUncalibratedError
|
||||||
|
raise CSMUncalibratedError() # noqa: RSE102
|
||||||
|
|
||||||
@thing_property
|
@thing_property
|
||||||
def last_calibration(self) -> Optional[Dict]:
|
def last_calibration(self) -> Optional[Dict]:
|
||||||
|
|
|
||||||
|
|
@ -73,7 +73,7 @@ def unpack_autofocus(scan_data):
|
||||||
"""
|
"""
|
||||||
jpeg_times = scan_data["jpeg_times"]
|
jpeg_times = scan_data["jpeg_times"]
|
||||||
jpeg_sizes = scan_data["jpeg_sizes"]
|
jpeg_sizes = scan_data["jpeg_sizes"]
|
||||||
jpeg_sizes_MB = [x / 10**3 for x in jpeg_sizes]
|
jpeg_sizes_mb = [x / 10**3 for x in jpeg_sizes]
|
||||||
stage_times = scan_data["stage_times"]
|
stage_times = scan_data["stage_times"]
|
||||||
stage_positions = scan_data["stage_positions"]
|
stage_positions = scan_data["stage_positions"]
|
||||||
stage_height = [pos[2] for pos in stage_positions]
|
stage_height = [pos[2] for pos in stage_positions]
|
||||||
|
|
@ -86,7 +86,7 @@ def unpack_autofocus(scan_data):
|
||||||
|
|
||||||
turning = np.where(turningpoints(jpeg_heights))[0] + 1
|
turning = np.where(turningpoints(jpeg_heights))[0] + 1
|
||||||
|
|
||||||
return jpeg_heights[turning[0] : turning[1]], jpeg_sizes_MB[turning[0] : turning[1]]
|
return jpeg_heights[turning[0] : turning[1]], jpeg_sizes_mb[turning[0] : turning[1]]
|
||||||
|
|
||||||
|
|
||||||
def limit_focus_change(prev_pos, prev_z, new_pos, new_z, limit):
|
def limit_focus_change(prev_pos, prev_z, new_pos, new_z, limit):
|
||||||
|
|
@ -115,10 +115,12 @@ def steps_from_centre(current_loc, starting_loc, dx, dy):
|
||||||
return np.max(np.abs(np.divide(np.subtract(current_loc, starting_loc), step_size)))
|
return np.max(np.abs(np.divide(np.subtract(current_loc, starting_loc), step_size)))
|
||||||
|
|
||||||
|
|
||||||
def distance_to_site(current, next):
|
def distance_to_site(current_pos, next_pos):
|
||||||
next = np.array(next, dtype="float64")
|
next_pos = np.array(next_pos, dtype="float64")
|
||||||
current = np.array(current, dtype="float64")
|
current_pos = np.array(current_pos, dtype="float64")
|
||||||
return np.sqrt((next[1] - current[1]) ** 2 + (next[0] - current[0]) ** 2)
|
return np.sqrt(
|
||||||
|
(next_pos[1] - current_pos[1]) ** 2 + (next_pos[0] - current_pos[0]) ** 2
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
class NotEnoughFreeSpaceError(IOError):
|
class NotEnoughFreeSpaceError(IOError):
|
||||||
|
|
@ -439,13 +441,21 @@ class SmartScanThing(Thing):
|
||||||
# TODO: Consider using CSM calibration size instead
|
# TODO: Consider using CSM calibration size instead
|
||||||
# TODO: generalise to have 2D displacements for x and y (as the
|
# TODO: generalise to have 2D displacements for x and y (as the
|
||||||
# camera and stage may not be aligned).
|
# camera and stage may not be aligned).
|
||||||
CSM = self._csm.image_to_stage_displacement_matrix
|
csm_disp_matrix = self._csm.image_to_stage_displacement_matrix
|
||||||
|
|
||||||
dx = int(
|
dx = int(
|
||||||
np.abs(np.dot(np.array([0, arr.shape[1] * (1 - overlap)]), CSM)[0])
|
np.abs(
|
||||||
|
np.dot(
|
||||||
|
np.array([0, arr.shape[1] * (1 - overlap)]), csm_disp_matrix
|
||||||
|
)[0]
|
||||||
|
)
|
||||||
)
|
)
|
||||||
dy = int(
|
dy = int(
|
||||||
np.abs(np.dot(np.array([arr.shape[0] * (1 - overlap), 0]), CSM)[1])
|
np.abs(
|
||||||
|
np.dot(
|
||||||
|
np.array([arr.shape[0] * (1 - overlap), 0]), csm_disp_matrix
|
||||||
|
)[1]
|
||||||
|
)
|
||||||
)
|
)
|
||||||
|
|
||||||
self._scan_logger.info(
|
self._scan_logger.info(
|
||||||
|
|
@ -1144,7 +1154,7 @@ class SmartScanThing(Thing):
|
||||||
# as some of them should only be added at the end (as we can't overwrite)
|
# as some of them should only be added at the end (as we can't overwrite)
|
||||||
# them once they change
|
# them once they change
|
||||||
if not os.path.isfile(zip_fname):
|
if not os.path.isfile(zip_fname):
|
||||||
with zipfile.ZipFile(zip_fname, mode="w") as zip:
|
with zipfile.ZipFile(zip_fname, mode="w") as scan_zip:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
# get a list of files in the existing zip
|
# get a list of files in the existing zip
|
||||||
|
|
@ -1167,7 +1177,7 @@ class SmartScanThing(Thing):
|
||||||
]
|
]
|
||||||
tiff_name = ""
|
tiff_name = ""
|
||||||
|
|
||||||
with zipfile.ZipFile(zip_fname, mode="a") as zip:
|
with zipfile.ZipFile(zip_fname, mode="a") as scan_zip:
|
||||||
for file in files:
|
for file in files:
|
||||||
if "stitched.jp" in file:
|
if "stitched.jp" in file:
|
||||||
stitch_name = os.path.split(file)[1]
|
stitch_name = os.path.split(file)[1]
|
||||||
|
|
@ -1181,32 +1191,30 @@ class SmartScanThing(Thing):
|
||||||
pass
|
pass
|
||||||
else:
|
else:
|
||||||
logger.info(f"appending {file} to zip")
|
logger.info(f"appending {file} to zip")
|
||||||
zip.write(os.path.join(scan_folder, file), arcname=file)
|
scan_zip.write(os.path.join(scan_folder, file), arcname=file)
|
||||||
|
|
||||||
# Promote key files to the top level of the zip only at the end of the scan (when downloading)
|
# Promote key files to the top level of the zip only at the end of the scan (when downloading)
|
||||||
# and finally zip some of the final files
|
# and finally zip some of the final files
|
||||||
# TODO: if you download multiple times, you get duplicate files - is this a problem?
|
# TODO: if you download multiple times, you get duplicate files - is this a problem?
|
||||||
if download_zip:
|
if download_zip:
|
||||||
with zipfile.ZipFile(zip_fname, mode="a") as zip:
|
with zipfile.ZipFile(zip_fname, mode="a") as scan_zip:
|
||||||
for fname in ["stitched_from_stage.jpg", stitch_name, tiff_name]:
|
for fname in ["stitched_from_stage.jpg", stitch_name, tiff_name]:
|
||||||
fpath = os.path.join(images_folder, fname)
|
fpath = os.path.join(images_folder, fname)
|
||||||
if os.path.exists(fpath):
|
if os.path.exists(fpath):
|
||||||
logger.info(f"copying {fpath} to upper level")
|
logger.info(f"copying {fpath} to upper level")
|
||||||
zip.write(fpath, arcname=fname)
|
scan_zip.write(fpath, arcname=fname)
|
||||||
for file in files:
|
for file in files:
|
||||||
if any(banned_name in file for banned_name in files_to_delay):
|
if any(banned_name in file for banned_name in files_to_delay):
|
||||||
logger.info(f"we are finally adding {file} into zip")
|
logger.info(f"we are finally adding {file} into zip")
|
||||||
zip.write(os.path.join(scan_folder, file), arcname=file)
|
scan_zip.write(os.path.join(scan_folder, file), arcname=file)
|
||||||
logger.info("about to download zip")
|
logger.info("about to download zip")
|
||||||
return ZipBlob.from_file(zip_fname)
|
return ZipBlob.from_file(zip_fname)
|
||||||
|
|
||||||
@thing_action
|
@thing_action
|
||||||
def get_files_in_zip(self, zip_path):
|
def get_files_in_zip(self, zip_path):
|
||||||
"""List the relative paths of all files and folders in the zip folder specified"""
|
"""List the relative paths of all files and folders in the zip folder specified"""
|
||||||
zip = zipfile.ZipFile(zip_path)
|
scan_zip = zipfile.ZipFile(zip_path)
|
||||||
zip = [os.path.normpath(i) for i in zip.namelist()]
|
return [os.path.normpath(i) for i in scan_zip.namelist()]
|
||||||
|
|
||||||
return zip
|
|
||||||
|
|
||||||
|
|
||||||
class CaptureError(RuntimeError):
|
class CaptureError(RuntimeError):
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue