| """
|
| Using fastapis to call the core script for producing and
|
|
|
| """
|
| from dataclasses import asdict
|
| from typing import List
|
| from fastapi import FastAPI, HTTPException, Path
|
| import asyncio
|
| from pydantic import BaseModel
|
| import uvicorn
|
| from functools import partial
|
| import tifffile as tff
|
| import numpy as np
|
| from concurrent.futures import ProcessPoolExecutor, as_completed
|
| from src.spot_detection import SpotDetection, CountAgenCreationError, ImageFile
|
|
|
|
|
| class Item(BaseModel):
|
| """
|
| Base model for JSON request.
|
| """
|
| name: str
|
| channels: int
|
| z_stack: int
|
| x_dimension: int
|
| y_dimension: int
|
| previewUrl: str
|
|
|
|
|
| app = FastAPI()
|
| spot_detection = SpotDetection(None, None)
|
|
|
| @app.post("/files/run/")
|
| async def run_img_spotter(item: Item, threshold: float = 1.4):
|
| """
|
| Return the image created for the counted number of spots.
|
|
|
| Args:
|
|
|
| item (Item) : Item object containing
|
| - name: str
|
| - channels: int
|
| - z_stack: int
|
| - x_dimension: int
|
| - y_dimension: int
|
| - previewUrl: str
|
|
|
| threshold (float): Value of threshold. Default : 1.4
|
|
|
| Returns:
|
|
|
| dictionary containing image properties and link to generated graphs.
|
| """
|
| spot_detection.threshold = threshold
|
| image_object = ImageFile(name=item.name, channels=item.channels, x_dimension=item.x_dimension,
|
| y_dimension=item.y_dimension, z_stack=item.z_stack, previewUrl=item.previewUrl)
|
| kwargs = {
|
| "output_folder": spot_detection.output_img_dir,
|
| "plot_blob": True
|
| }
|
| pic = tff.imread(item.previewUrl)
|
| projected = np.max(pic, axis=item.z_stack)
|
| projected_norm = np.zeros(projected.shape)
|
| loop = asyncio.get_event_loop()
|
| urls = []
|
| futures = {}
|
| try:
|
| with ProcessPoolExecutor(max_workers=5) as pool:
|
| for index in range(projected.shape[0]):
|
| future = await loop.run_in_executor(pool, partial(spot_detection.wrapper, image_object, index, projected[index], **kwargs))
|
| futures.update({index: future})
|
| for fut in as_completed(future):
|
| try:
|
| urls.append(fut.result())
|
| except:
|
| raise
|
| except Exception as exc:
|
| raise HTTPException(
|
| status_code=500, detail=str(exc)) from exc
|
| return {"items": [{**asdict(image_object), "graphs": urls}]}
|
| class SpotDetection:
|
| """ Used to count the number of dot in a certain number of images."""
|
|
|
| threshold: float = 1.4
|
| graph_extension = 'png'
|
| status = {"items": []}
|
|
|
| def __init__(self, input_dir: str, output_dir: str) -> None:
|
| """ Initilizer.
|
| Args:
|
|
|
| input_dir (str): Input folder containing folders.
|
| output_dir (str): Output folder to place images.
|
| """
|
| logging.info("Class is called with in: %s, out: %s.",
|
| input_dir, output_dir)
|
| self.input_img_dir = input_dir
|
| self.output_img_dir = output_dir
|
|
|
| @staticmethod
|
| def wrapper(image_: ImageFile, index: int, project_channel_arr: Any, **kwargs):
|
| """ A wrapper function to return the spots and graphs.
|
|
|
| Args:
|
| image_ (ImageFile): Image file.
|
| index (int): Channel index.
|
| project_channel_arr: Projected channel.
|
| **kwargs: Keyword arguments.
|
|
|
| Returns:
|
|
|
| url (str): url to the generated images.
|
| """
|
| output_folder: str = kwargs.get('output_folder')
|
| extension_format: str = kwargs.get('extension_format', 'png')
|
| logging.info(
|
| "Arguments were: image_ = %s, index =%s, kwargs=%s", image_, index, kwargs)
|
| image, image_norm = SpotDetection.norm_img(
|
| project_channel_arr, **kwargs)
|
| spots_df = SpotDetection.spots(image, image_norm, **kwargs)
|
|
|
| if kwargs.get('plot_blob', False):
|
| logging.info(
|
| "Creating plot: image_ = %s, index =%s.", image_, index)
|
| url = SpotDetection.plot_results(
|
| index, image, image_name=image_.name,
|
| spots_=spots_df, output_folder=output_folder,
|
| extension_format=extension_format)
|
| return url
|
|
|
| @ staticmethod
|
| def spots(image_: T, image_norm_: int, **kwargs) -> Any:
|
| """ Returns a generator coontaint the number of spots on a image.
|
| Args:
|
|
|
| image_(T): np.array description of an image.
|
| image_norm_(int): normalized description.
|
| **kwargs: keywod variables containing threshold, max and min sigma.
|
|
|
| Yields:
|
|
|
| spot_data(int): number of spots.
|
|
|
| """
|
| max_sigma: int = (kwargs.get('max_radius', 5)**2)/2
|
| min_sigma: int = (kwargs.get('min_radius', 3)**2)/2
|
| sqrt_2: float = np.sqrt(2)
|
|
|
| spots_1 = blob_log(image_norm_, max_sigma=max_sigma, num_sigma=round(
|
| max_sigma - min_sigma), threshold=SpotDetection.threshold, min_sigma=min_sigma)
|
|
|
| y_inds = spots_1[:, 0].astype(int)
|
| x_inds = spots_1[:, 1].astype(int)
|
|
|
| radius = np.round(spots_1[:, 2]*sqrt_2)
|
| intensities = image_[tuple([y_inds, x_inds])]
|
|
|
| data_ = {'x': x_inds,
|
| 'y': y_inds,
|
| 'intensity': intensities,
|
| 'radius': radius, }
|
|
|
| spot_data = pd.DataFrame(data=data_)
|
| logging.info("Variable: threshold = %s, max=%s, min=%s, data=%s, spots=%s",
|
| SpotDetection.threshold, max_sigma, min_sigma, data_, spot_data)
|
| return spot_data
|
|
|
| @ staticmethod
|
| def plot_results(index: int, image_: T, **kwargs):
|
| """ Creates an image for the input images.
|
|
|
| Args:
|
|
|
| index(int): The channel number.
|
| image_(Array): The values to be plotted.
|
| **kwargs: Keyword variables.
|
|
|
| Raises:
|
|
|
| CountAgenCreationError
|
|
|
| """
|
| image_name: str = kwargs.get('image_name')
|
| extension_format: str = kwargs.get('extension_format')
|
| mult_factor: float = kwargs.get('mult_factor', 20)
|
| spots_: int = kwargs.get('spots_')
|
| cmap_im: int = kwargs.get('cmap_im', 'gray')
|
| cmap_spots: int = kwargs.get('cmap_im', 'Reds')
|
| alpha_spots: int = kwargs.get('alpha_spots', 0.4)
|
|
|
| try:
|
| with plt.rc_context({'figure.figsize': (40, 40), 'figure.dpi': 100}):
|
| plt.figure(index+1)
|
| plt.imshow(image_*mult_factor, cmap=cmap_im)
|
| s_c = plt.scatter(spots_['x'], spots_['y'],
|
| s=spots_['radius']*100,
|
| c=spots_['intensity'],
|
| cmap=cmap_spots,
|
| alpha=alpha_spots)
|
|
|
| output: str = kwargs.get('output_folder')
|
| outputed_img_dir = f"{output}/{image_name}"
|
| outputed_img_path = f"{outputed_img_dir}/{index+1}.{extension_format}"
|
|
|
| if not os.path.exists(outputed_img_dir):
|
| try:
|
| os.mkdir(outputed_img_dir)
|
| except Exception as exc:
|
| raise CountAgenCreationError from exc
|
| plt.colorbar(s_c, shrink=0.5)
|
| plt.savefig(outputed_img_path)
|
| except Exception as exc:
|
| logging.exception(exc, exc_info=True)
|
| raise CountAgenCreationError from exc
|
|
|
| return outputed_img_path
|
| INFO: 127.0.0.1:55739 - "POST /files/run/?threshold=1.4 HTTP/1.1" 500 Internal Server Error
|