Only grab a JPEG from the stream to get scanning working
This commit is contained in:
jaknapper
parent
d4247235e3
commit
574aa1c6c0
1 changed files with 49 additions and 77 deletions
|
|
@ -576,58 +576,8 @@ class SmartScanThing(Thing):
|
|||
) as f:
|
||||
json.dump(data, f, ensure_ascii=False, indent=4)
|
||||
|
||||
# We will capture images and process them with this function, defined once here.
|
||||
# Most of the variables it needs will be "baked in" so the arguments are just the ones
|
||||
# that change each iteration.
|
||||
# We also pre-calculate a normalisation image based on the LST and white balance
|
||||
raw_image = cam.capture_array(stream_name="raw")
|
||||
# TODO: assert the image is 10-bit packed, or deal with other formats!
|
||||
rgb = rggb2rgb(raw2rggb(raw_image))
|
||||
lst = dict(cam.lens_shading_tables)
|
||||
lum = np.array(lst["luminance"])
|
||||
Cr = np.array(lst["Cr"])
|
||||
Cb = np.array(lst["Cb"])
|
||||
gr, gb = cam.colour_gains
|
||||
G = 1 / lum
|
||||
R = (
|
||||
G / Cr / gr * np.min(Cr)
|
||||
) # The extra /np.max(Cr) emulates the quirky handling of Cr in
|
||||
B = G / Cb / gb * np.min(Cb) # the picamera2 pipeline
|
||||
white_norm_lores = np.stack([R, G, B], axis=2)
|
||||
zoom_factors = [
|
||||
i / n for i, n in zip(rgb[..., :3].shape, white_norm_lores.shape)
|
||||
]
|
||||
white_norm = zoom(white_norm_lores, zoom_factors, order=1)[
|
||||
: rgb.shape[0], : rgb.shape[1], :
|
||||
] # Could use some work
|
||||
colour_correction_matrix = np.array(cam.colour_correction_matrix).reshape(
|
||||
(3, 3)
|
||||
)
|
||||
contrast_algorithm = cam.tuning["algorithms"][9]["rpi.contrast"]
|
||||
gamma = np.array(contrast_algorithm["gamma_curve"]).reshape((-1, 2))
|
||||
gamma_8bit = interp1d(gamma[:, 0] / 255, gamma[:, 1] / 255)
|
||||
|
||||
def process_raw_image(img):
|
||||
normed = img / white_norm
|
||||
corrected = np.dot(
|
||||
colour_correction_matrix, normed.reshape((-1, 3)).T
|
||||
).T.reshape(normed.shape)
|
||||
corrected[corrected < 0] = 0
|
||||
corrected[corrected > 255] = 255
|
||||
return gamma_8bit(corrected)
|
||||
|
||||
logger.info(
|
||||
f"Generated normalisation image with shape {white_norm.shape}, "
|
||||
f"max {white_norm.max(axis=(0, 1))}, min {white_norm.min(axis=(0, 1))}"
|
||||
)
|
||||
norm_inputs = {
|
||||
"luminance": lum,
|
||||
"Cr": Cr,
|
||||
"Cb": Cb,
|
||||
"gain_red": gr,
|
||||
"gain_blue": gb,
|
||||
}
|
||||
|
||||
# This is the function we'll use to grab (or later capture) an image
|
||||
# and save it with metadata
|
||||
def capture_and_save(acquired: Event, name: str) -> None:
|
||||
"""Capture an image and save it to disk
|
||||
|
||||
|
|
@ -636,38 +586,60 @@ class SmartScanThing(Thing):
|
|||
"""
|
||||
try:
|
||||
capture_start = time.time()
|
||||
metadata = metadata_getter()
|
||||
raw_image = cam.capture_array(stream_name="raw")
|
||||
image, metadata = capture_image()
|
||||
acquired.set()
|
||||
acquisition_time = time.time()
|
||||
# Save the raw image
|
||||
np.savez(
|
||||
os.path.join(raw_images_folder, name + ".npz"),
|
||||
raw_image=raw_image,
|
||||
**norm_inputs,
|
||||
)
|
||||
# Process it into 8 bit RGB
|
||||
processed = process_raw_image(rggb2rgb(raw2rggb(raw_image)))
|
||||
processed[processed > 255] = 255
|
||||
processed[processed < 0] = 0
|
||||
img = Image.fromarray(processed.astype(np.uint8), mode="RGB")
|
||||
img.save(
|
||||
os.path.join(images_folder, name), quality=95, subsampling=0
|
||||
)
|
||||
exif_dict = piexif.load(os.path.join(images_folder, name))
|
||||
exif_dict["Exif"][piexif.ExifIFD.UserComment] = json.dumps(
|
||||
metadata
|
||||
).encode("utf-8")
|
||||
piexif.insert(
|
||||
piexif.dump(exif_dict), os.path.join(images_folder, name)
|
||||
)
|
||||
save_capture(name, image, metadata)
|
||||
save_time = time.time()
|
||||
acquisition_duration = round(acquisition_time - capture_start, 1)
|
||||
saving_duration = round(save_time - acquisition_time, 1)
|
||||
logger.info(
|
||||
f"Acquired {name} in {acquisition_time - capture_start:.1f}s then {save_time - acquisition_time:.1f}s saving to disk"
|
||||
"Acquired {} in {}s then {}s saving to disk".format(
|
||||
name, acquisition_duration, saving_duration
|
||||
)
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
f"An error occurred while saving {name}: {e}", exc_info=e
|
||||
"An error occurred while saving {}: {}".format(name, e),
|
||||
exc_info=e,
|
||||
)
|
||||
|
||||
def capture_image():
|
||||
"""Capture an image in memory and return it with metadata
|
||||
|
||||
This will set the event `acquired` once the image has been acquired, so
|
||||
that the stage may be moved while it's saved.
|
||||
"""
|
||||
try:
|
||||
metadata = metadata_getter()
|
||||
image = cam.capture_array()[..., :3]
|
||||
return image, metadata
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
"An error occurred while capturing: {}".format(e), exc_info=e
|
||||
)
|
||||
return 0, 0
|
||||
|
||||
def save_capture(name, image, metadata):
|
||||
try:
|
||||
jpeg_path = os.path.join(images_folder, name)
|
||||
PIL_image = Image.fromarray(image.astype("uint8"), "RGB").save(
|
||||
jpeg_path, quality=95, subsampling=0
|
||||
)
|
||||
try:
|
||||
exif_dict = piexif.load(os.path.join(images_folder, name))
|
||||
exif_dict["Exif"][piexif.ExifIFD.UserComment] = json.dumps(
|
||||
metadata
|
||||
).encode("utf-8")
|
||||
piexif.insert(
|
||||
piexif.dump(exif_dict), os.path.join(images_folder, name)
|
||||
)
|
||||
except:
|
||||
pass
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
"An error occurred while saving {}: {}".format(name, e),
|
||||
exc_info=e,
|
||||
)
|
||||
|
||||
# At the start of the loop, we simultaneously capture an image and move to the next scan point.
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue