angelolab / mibi-bin-tools / 6

pending completion

Build # 6

Build Type

Pull #13

travis-ci-com

Committed by

web-flow

Commit Message

<a href="https://github.com/angelolab/mibi-bin-tools/commit/<a class=hub.com/angelolab/mibi-bin-tools/commit/8d86fb0f5da5616920e2f84f30dca3f3f4a62dfb">8d86fb0f5<a href="https://github.com/angelolab/mibi-bin-tools/commit/8d86fb0f5da5616920e2f84f30dca3f3f4a62dfb">">Merge </a><a class="double-link" href="https://github.com/angelolab/mibi-bin-tools/commit/<a class="double-link" href="https://github.com/angelolab/mibi-bin-tools/commit/a4caac3e31d641186ed49a536ac25f5c9a7c1be0">a4caac3e3</a>">a4caac3e3</a><a href="https://github.com/angelolab/mibi-bin-tools/commit/8d86fb0f5da5616920e2f84f30dca3f3f4a62dfb"> into 270263be8">270263be8</a>

Pull Request Pull Request #13: Bin file tests

Run Details

4 of 4 new or added lines in 1 file covered. (100.0%)

104 of 131 relevant lines covered (79.39%)

0.79 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

70.0

/mibi_bin_tools/bin_files.py

from typing import Any, Dict, List, Tuple, Union
import os
import json
import multiprocessing as mp

import numpy as np
import pandas as pd
import skimage.io as io

from mibi_bin_tools import io_utils, _extract_bin


def _mass2tof(masses_arr: np.ndarray, mass_offset: float, mass_gain: float,
              time_res: float) -> np.ndarray:
    """Convert array of m/z values to equivalent time of flight values

    Args:
        masses_arr (array_like):
            Array of m/z values
        mass_offset (float):
            Mass offset for parabolic transformation
        mass_gain (float):
            Mass gain for parabolic transformation
        time_res (float):
            Time resolution for scaling parabolic transformation

    Returns:
        array_like:
            Array of time of flight values; indicies paried to `masses_arr`
    """
    return (mass_gain * np.sqrt(masses_arr) + mass_offset) / time_res


def _set_tof_ranges(fov: Dict[str, Any], higher: np.ndarray, lower: np.ndarray,
                    time_res: float) -> None:
    """Converts and stores provided mass ranges as time of flight ranges within fov metadata

    Args:
        fov (Dict[str, Any]):
            Metadata for the fov.
        higher (array_like):
            Array of m/z values; upper bounds for integration
        lower (array_like):
            Array of m/z values; lower bounds for integration
        time_res (float):
            Time resolution for scaling parabolic transformation

    Returns:
        None:
            Fovs argument is modified in place
    """
    key_names = ('upper_tof_range', 'lower_tof_range')
    mass_ranges = (higher, lower)
    wrapping_functions = (np.ceil, np.floor)

    for key, masses, wrap in zip(key_names, mass_ranges, wrapping_functions):
        fov[key] = \
            wrap(
                _mass2tof(masses, fov['mass_offset'], fov['mass_gain'], time_res)
            ).astype(np.uint16)


def _write_out(img_data: np.ndarray, out_dir: str, fov_name: str, targets: List[str]) -> None:
    """Parses extracted data and writes out tifs

    Args:
        img_data (np.ndarray):
            Array containing the pulse counts, intensity, and intensity * width images
        out_dir (str | PathLike):
            Directory to save tifs
        fov_name (str):
            Name of the field of view
        targets (array_like):
            List of target names (i.e channels)
    """
    out_dirs = [
        os.path.join(out_dir, fov_name),
        os.path.join(out_dir, fov_name, 'intensities'),
        os.path.join(out_dir, fov_name, 'intensity_times_width')
    ]
    suffixes = [
        '',
        '_intensity',
        '_int_width'
    ]
    save_dtypes = [
        np.uint16,
        np.uint32,
        np.uint32,
    ]
    for i, (out_dir_i, suffix, save_dtype) in enumerate(zip(out_dirs, suffixes, save_dtypes)):
        if not os.path.exists(out_dir_i):
            os.makedirs(out_dir_i)
        for j, target in enumerate(targets):
            io.imsave(
                os.path.join(out_dir_i, f'{target}{suffix}.tiff'),
                img_data[i, :, :, j].astype(save_dtype),
                plugin='tifffile',
                check_contrast=False
            )


def _find_bin_files(data_dir: str,
                    include_fovs: Union[List[str], None] = None) -> Dict[str, Dict[str, str]]:
    """Locates paired bin/json files within the provided directory.

    Args:
        data_dir (str | PathLike):
            Directory containing bin/json files
        include_fovs (List | None):
            List of fovs to include. Includes all if None.

    Returns:
        Dict[str, Dict[str, str]]:
            Dictionary containing the names of the valid bin files
    """
    bin_files = io_utils.list_files(data_dir, substrs=['.bin'])
    json_files = io_utils.list_files(data_dir, substrs=['.json'])

    fov_names = io_utils.extract_delimited_names(bin_files, delimiter='.')

    fov_files = {
        fov_name: {
            'bin': fov_name + '.bin',
            'json': fov_name + '.json',
        }
        for fov_name in fov_names
        if fov_name + '.json' in json_files
    }

    if include_fovs is not None:
        fov_files = {fov_file: fov_files[fov_file] for fov_file in include_fovs}

    return fov_files


def _fill_fov_metadata(data_dir: str, fov: Dict[str, Any],
                       panel: Union[Tuple[float, float], pd.DataFrame],
                       intensities: Union[bool, List[str]], time_res: float,
                       channels: List[str] = None) -> None:
    """ Parses user input and mibiscope json to build extraction parameters

    Fills fov metadata with mass calibration parameters, builds panel, and sets intensity
    extraction flags.

    Args:
        data_dir (str):
            Directory containing bin files as well as accompanying json metadata files
        fov (Dict[str, Any]):
            Metadata for the fov.
        panel (tuple | pd.DataFrame):
            If a tuple, global integration range over all antibodies within json metadata.
            If a pd.DataFrame, specific peaks with custom integration ranges.  Column names must be
            'Mass' and 'Target' with integration ranges specified via 'Start' and 'Stop' columns.
        intensities (bool | List[str]):
            Whether or not to extract intensity and intensity * width images.  If a List, specific
            peaks can be extracted, ignoring the rest, which will only have pulse count images
            extracted.
        time_res (float):
            Time resolution for scaling parabolic transformation
        channels (List[str] | None):
            Filters panel for given channels.  All channels in panel extracted if None
    Returns:
        None:
            `fov` argument is modified in place
    """

    with open(os.path.join(data_dir, fov['json']), 'rb') as f:
        data = json.load(f)

    fov['mass_gain'] = data['fov']['fullTiming']['massCalibration']['massGain']
    fov['mass_offset'] = data['fov']['fullTiming']['massCalibration']['massOffset']

    if type(panel) is tuple:
        _parse_global_panel(data, fov, panel, time_res, channels)
    else:
        _parse_df_panel(fov, panel, time_res, channels)

    _parse_intensities(fov, intensities)


def _parse_global_panel(json_metadata: dict, fov: Dict[str, Any], panel: Tuple[float, float],
                        time_res: float, channels: List[str]) -> None:
    """Extracts panel contained in mibiscope json metadata

    Args:
        json_metadata (dict):
            metadata read via mibiscope json
        fov (Dict[str, Any]):
            Metadata for the fov.
        panel (tuple):
            Global integration range over all antibodies within json metadata.
            Column names must 'Mass' and 'Target' with integration ranges specified via 'Start' and
            'Stop' columns.
        time_res (float):
            Time resolution for scaling parabolic transformation
        channels (List[str] | None):
            Filters panel for given channels.  All channels in panel extracted if None
    Returns:
        None:
            `fov` argument is modified in place
    """
    rows = json_metadata['fov']['panel']['conjugates']
    fov['masses'], fov['targets'] = zip(*[
        (el['mass'], el['target'])
        for el in rows
        if channels is None or el['target'] in channels
    ])

    masses_arr = np.array(fov['masses'])
    _set_tof_ranges(fov, masses_arr + panel[1], masses_arr + panel[0], time_res)


def _parse_df_panel(fov: Dict[str, Any], panel: pd.DataFrame, time_res: float,
                    channels: List[str]) -> None:
    """Converts masses from panel into times for fov extraction-metadata structure

    Args:
        fov (Dict[str, Any]):
            Metadata for the fov.
        panel (pd.DataFrame):
            Specific peaks with custom integration ranges.  Column names must be 'Mass' and
            'Target' with integration ranges specified via 'Start' and 'Stop' columns.
        time_res (float):
            Time resolution for scaling parabolic transformation
        channels (List[str] | None):
            Filters panel for given channels.  All channels in panel extracted if None
    Returns:
        None:
            `fov` argument is modified in place
    """
    rows = panel.loc[panel['Target'].isin(panel['Target'] if channels is None else channels)]
    fov['masses'] = rows['Mass']
    fov['targets'] = rows['Target']

    _set_tof_ranges(fov, rows['Stop'].values, rows['Start'].values, time_res)


def _parse_intensities(fov: Dict[str, Any], intensities: Union[bool, List[str]]) -> None:
    """Sets intensity extraction flags within the extraction-metadata

    Args:
        fov (Dict[str, Any]):
            Metadata for the fov
        intensities (bool | List):
            Whether or not to extract intensity and intensity * width images.  If a List, specific
            peaks can be extracted, ignoring the rest, which will only have pulse count images
            extracted.
    Returns:
        None:
            `fov` argument is modified in place
    """

    if type(intensities) is list:
        fov['intensities'] = [target in intensities for target in fov['targets']]
    elif intensities is True:
        fov['intensities'] = fov['targets']

    # order the 'calc_intensity' bools
    if 'intensities' in fov.keys():
        fov['calc_intensity'] = [target in fov['intensities'] for target in fov['targets']]
    else:
        fov['calc_intensity'] = [False, ] * len(fov['targets'])


def extract_bin_files(data_dir: str, out_dir: str,
                      include_fovs: Union[List[str], None] = None,
                      panel: Union[Tuple[float, float], pd.DataFrame] = (-0.3, 0.0),
                      intensities: Union[bool, List[str]] = False, time_res: float = 500e-6,
                      write_parallel: bool = True):
    """Converts MibiScope bin files to pulse count, intensity, and intensity * width tiff images

    Args:
        data_dir (str | PathLike):
            Directory containing bin files as well as accompanying json metadata files
        out_dir (str | PathLike):
            Directory to save the tiffs in
        include_fovs (List | None):
            List of fovs to include.  Includes all if None.
        panel (tuple | pd.DataFrame):
            If a tuple, global integration range over all antibodies within json metadata.
            If a pd.DataFrame, specific peaks with custom integration ranges.  Column names must be
            'Mass' and 'Target' with integration ranges specified via 'Start' and 'Stop' columns.
        intensities (bool | List):
            Whether or not to extract intensity and intensity * width images.  If a List, specific
            peaks can be extracted, ignoring the rest, which will only have pulse count images
            extracted.
        time_res (float):
            Time resolution for scaling parabolic transformation
        write_parallel (bool):
            Try writing files out in parallel
    """
    fov_files = _find_bin_files(data_dir, include_fovs)

    for fov in fov_files.values():
        _fill_fov_metadata(data_dir, fov, panel, intensities, time_res)

    bin_files = \
        [(fov, os.path.join(data_dir, fov['bin'])) for fov in fov_files.values()]

    if write_parallel:
        with mp.Pool() as pool:
            for i, (fov, bf) in enumerate(bin_files):
                # call extraction cython here
                img_data = _extract_bin.c_extract_bin(
                    bytes(bf, 'utf-8'), fov['lower_tof_range'],
                    fov['upper_tof_range'], np.array(fov['calc_intensity'], dtype=np.uint8))
                pool.apply_async(
                    _write_out,
                    (img_data, out_dir, fov['bin'][:-4], fov['targets'])
                )
            pool.join()
            pool.close()
    else:
        for i, (fov, bf) in enumerate(bin_files):
            img_data = _extract_bin.c_extract_bin(
                bytes(bf, 'utf-8'), fov['lower_tof_range'],
                fov['upper_tof_range'], np.array(fov['calc_intensity'], dtype=np.uint8)
            )
            _write_out(img_data, out_dir, fov['bin'][:-4], fov['targets'])


def get_histograms_per_tof(data_dir: str, fov: str, channel: str, mass_range=(-0.3, 0.0),
                           time_res: float = 500e-6):
    """Generates histograms of pulse widths, pulse counts, and pulse intensities found within the
    given mass range

    Args:
        data_dir (str | PathLike):
            Directory containing bin files as well as accompanying json metadata files
        fov (str):
            Fov to generate histogram for
        channel (str):
            Channel to check widths for
        mass_range (tuple):
            Integration range
        time_res (float):
            Time resolution for scaling parabolic transformation
    """
    fov = _find_bin_files(data_dir, [fov])[fov]

    _fill_fov_metadata(data_dir, fov, mass_range, False, time_res, [channel])

    local_bin_file = os.path.join(data_dir, fov['bin'])

    widths, intensities, pulses = _extract_bin.c_extract_histograms(bytes(local_bin_file, 'utf-8'),
                                                                    fov['lower_tof_range'][0],
                                                                    fov['upper_tof_range'][0])
    return widths, intensities, pulses


def median_height_vs_mean_pp(data_dir: str, fov: str, channel: str,
                             panel: Union[Tuple[float, float], pd.DataFrame] = (-0.3, 0.0),
                             time_res: float = 500e-6):
    """Retrieves median pulse intensity and mean pulse count for a given channel

    Args:
        data_dir (str | PathLike):
            Directory containing bin files as well as accompanying json metadata files
        fov (str):
            Fov to generate histogram for
        channel (str):
            Channel to check widths for
        mass_range (tuple | pd.DataFrame):
            Integration range
        time_res (float):
            Time resolution for scaling parabolic transformation

    """

    fov = _find_bin_files(data_dir, [fov])[fov]
    _fill_fov_metadata(data_dir, fov, panel, False, time_res, [channel])

    local_bin_file = os.path.join(data_dir, fov['bin'])

    # TODO: fix median calculation in this call
    _, mean_pp = \
        _extract_bin.c_pulse_height_vs_positive_pixel(bytes(local_bin_file, 'utf-8'),
                                                      fov['lower_tof_range'][0],
                                                      fov['upper_tof_range'][0])

    _, intensities, _ = \
        _extract_bin.c_extract_histograms(bytes(local_bin_file, 'utf-8'),
                                          fov['lower_tof_range'][0],
                                          fov['upper_tof_range'][0])

    int_bin = np.cumsum(intensities) / intensities.sum()
    median_height = (np.abs(int_bin - 0.5)).argmin()

    return median_height, mean_pp

1	from typing import Any, Dict, List, Tuple, Union	1✔
2	import os	1✔
3	import json	1✔
4	import multiprocessing as mp	1✔
5
6	import numpy as np	1✔
7	import pandas as pd	1✔
8	import skimage.io as io	1✔
9
10	from mibi_bin_tools import io_utils, _extract_bin	1✔
11
12
13	def _mass2tof(masses_arr: np.ndarray, mass_offset: float, mass_gain: float,	1✔
14	time_res: float) -> np.ndarray:
15	"""Convert array of m/z values to equivalent time of flight values
16
17	Args:
18	masses_arr (array_like):
19	Array of m/z values
20	mass_offset (float):
21	Mass offset for parabolic transformation
22	mass_gain (float):
23	Mass gain for parabolic transformation
24	time_res (float):
25	Time resolution for scaling parabolic transformation
26
27	Returns:
28	array_like:
29	Array of time of flight values; indicies paried to `masses_arr`
30	"""
31	return (mass_gain * np.sqrt(masses_arr) + mass_offset) / time_res	1✔
32
33
34	def _set_tof_ranges(fov: Dict[str, Any], higher: np.ndarray, lower: np.ndarray,	1✔
35	time_res: float) -> None:
36	"""Converts and stores provided mass ranges as time of flight ranges within fov metadata
37
38	Args:
39	fov (Dict[str, Any]):
40	Metadata for the fov.
41	higher (array_like):
42	Array of m/z values; upper bounds for integration
43	lower (array_like):
44	Array of m/z values; lower bounds for integration
45	time_res (float):
46	Time resolution for scaling parabolic transformation
47
48	Returns:
49	None:
50	Fovs argument is modified in place
51	"""
52	key_names = ('upper_tof_range', 'lower_tof_range')	1✔
53	mass_ranges = (higher, lower)	1✔
54	wrapping_functions = (np.ceil, np.floor)	1✔
55
56	for key, masses, wrap in zip(key_names, mass_ranges, wrapping_functions):	1✔
57	fov[key] = \	1✔
58	wrap(
59	_mass2tof(masses, fov['mass_offset'], fov['mass_gain'], time_res)
60	).astype(np.uint16)
61
62
63	def _write_out(img_data: np.ndarray, out_dir: str, fov_name: str, targets: List[str]) -> None:	1✔
64	"""Parses extracted data and writes out tifs
65
66	Args:
67	img_data (np.ndarray):
68	Array containing the pulse counts, intensity, and intensity * width images
69	out_dir (str \| PathLike):
70	Directory to save tifs
71	fov_name (str):
72	Name of the field of view
73	targets (array_like):
74	List of target names (i.e channels)
75	"""
76	out_dirs = [	1✔
77	os.path.join(out_dir, fov_name),
78	os.path.join(out_dir, fov_name, 'intensities'),
79	os.path.join(out_dir, fov_name, 'intensity_times_width')
80	]
81	suffixes = [	1✔
82	'',
83	'_intensity',
84	'_int_width'
85	]
86	save_dtypes = [	1✔
87	np.uint16,
88	np.uint32,
89	np.uint32,
90	]
91	for i, (out_dir_i, suffix, save_dtype) in enumerate(zip(out_dirs, suffixes, save_dtypes)):	1✔
92	if not os.path.exists(out_dir_i):	1✔
93	os.makedirs(out_dir_i)	1✔
94	for j, target in enumerate(targets):	1✔
95	io.imsave(	1✔
96	os.path.join(out_dir_i, f'{target}{suffix}.tiff'),
97	img_data[i, :, :, j].astype(save_dtype),
98	plugin='tifffile',
99	check_contrast=False
100	)
101
102
103	def _find_bin_files(data_dir: str,	1✔
104	include_fovs: Union[List[str], None] = None) -> Dict[str, Dict[str, str]]:
105	"""Locates paired bin/json files within the provided directory.
106
107	Args:
108	data_dir (str \| PathLike):
109	Directory containing bin/json files
110	include_fovs (List \| None):
111	List of fovs to include. Includes all if None.
112
113	Returns:
114	Dict[str, Dict[str, str]]:
115	Dictionary containing the names of the valid bin files
116	"""
117	bin_files = io_utils.list_files(data_dir, substrs=['.bin'])	1✔
118	json_files = io_utils.list_files(data_dir, substrs=['.json'])	1✔
119
120	fov_names = io_utils.extract_delimited_names(bin_files, delimiter='.')	1✔
121
122	fov_files = {	1✔
123	fov_name: {
124	'bin': fov_name + '.bin',
125	'json': fov_name + '.json',
126	}
127	for fov_name in fov_names
128	if fov_name + '.json' in json_files
129	}
130
131	if include_fovs is not None:	1✔
132	fov_files = {fov_file: fov_files[fov_file] for fov_file in include_fovs}	1✔
133
134	return fov_files	1✔
135
136
137	def _fill_fov_metadata(data_dir: str, fov: Dict[str, Any],	1✔
138	panel: Union[Tuple[float, float], pd.DataFrame],
139	intensities: Union[bool, List[str]], time_res: float,
140	channels: List[str] = None) -> None:
141	""" Parses user input and mibiscope json to build extraction parameters
142
143	Fills fov metadata with mass calibration parameters, builds panel, and sets intensity
144	extraction flags.
145
146	Args:
147	data_dir (str):
148	Directory containing bin files as well as accompanying json metadata files
149	fov (Dict[str, Any]):
150	Metadata for the fov.
151	panel (tuple \| pd.DataFrame):
152	If a tuple, global integration range over all antibodies within json metadata.
153	If a pd.DataFrame, specific peaks with custom integration ranges. Column names must be
154	'Mass' and 'Target' with integration ranges specified via 'Start' and 'Stop' columns.
155	intensities (bool \| List[str]):
156	Whether or not to extract intensity and intensity * width images. If a List, specific
157	peaks can be extracted, ignoring the rest, which will only have pulse count images
158	extracted.
159	time_res (float):
160	Time resolution for scaling parabolic transformation
161	channels (List[str] \| None):
162	Filters panel for given channels. All channels in panel extracted if None
163	Returns:
164	None:
165	`fov` argument is modified in place
166	"""
167
168	with open(os.path.join(data_dir, fov['json']), 'rb') as f:	1✔
169	data = json.load(f)	1✔
170
171	fov['mass_gain'] = data['fov']['fullTiming']['massCalibration']['massGain']	1✔
172	fov['mass_offset'] = data['fov']['fullTiming']['massCalibration']['massOffset']	1✔
173
174	if type(panel) is tuple:	1✔
175	_parse_global_panel(data, fov, panel, time_res, channels)	1✔
176	else:
177	_parse_df_panel(fov, panel, time_res, channels)	1✔
178
179	_parse_intensities(fov, intensities)	1✔
180
181
182	def _parse_global_panel(json_metadata: dict, fov: Dict[str, Any], panel: Tuple[float, float],	1✔
183	time_res: float, channels: List[str]) -> None:
184	"""Extracts panel contained in mibiscope json metadata
185
186	Args:
187	json_metadata (dict):
188	metadata read via mibiscope json
189	fov (Dict[str, Any]):
190	Metadata for the fov.
191	panel (tuple):
192	Global integration range over all antibodies within json metadata.
193	Column names must 'Mass' and 'Target' with integration ranges specified via 'Start' and
194	'Stop' columns.
195	time_res (float):
196	Time resolution for scaling parabolic transformation
197	channels (List[str] \| None):
198	Filters panel for given channels. All channels in panel extracted if None
199	Returns:
200	None:
201	`fov` argument is modified in place
202	"""
203	rows = json_metadata['fov']['panel']['conjugates']	1✔
204	fov['masses'], fov['targets'] = zip(*[	1✔
205	(el['mass'], el['target'])
206	for el in rows
207	if channels is None or el['target'] in channels
208	])
209
210	masses_arr = np.array(fov['masses'])	1✔
211	_set_tof_ranges(fov, masses_arr + panel[1], masses_arr + panel[0], time_res)	1✔
212
213
214	def _parse_df_panel(fov: Dict[str, Any], panel: pd.DataFrame, time_res: float,	1✔
215	channels: List[str]) -> None:
216	"""Converts masses from panel into times for fov extraction-metadata structure
217
218	Args:
219	fov (Dict[str, Any]):
220	Metadata for the fov.
221	panel (pd.DataFrame):
222	Specific peaks with custom integration ranges. Column names must be 'Mass' and
223	'Target' with integration ranges specified via 'Start' and 'Stop' columns.
224	time_res (float):
225	Time resolution for scaling parabolic transformation
226	channels (List[str] \| None):
227	Filters panel for given channels. All channels in panel extracted if None
228	Returns:
229	None:
230	`fov` argument is modified in place
231	"""
232	rows = panel.loc[panel['Target'].isin(panel['Target'] if channels is None else channels)]	1✔
233	fov['masses'] = rows['Mass']	1✔
234	fov['targets'] = rows['Target']	1✔
235
236	_set_tof_ranges(fov, rows['Stop'].values, rows['Start'].values, time_res)	1✔
237
238
239	def _parse_intensities(fov: Dict[str, Any], intensities: Union[bool, List[str]]) -> None:	1✔
240	"""Sets intensity extraction flags within the extraction-metadata
241
242	Args:
243	fov (Dict[str, Any]):
244	Metadata for the fov
245	intensities (bool \| List):
246	Whether or not to extract intensity and intensity * width images. If a List, specific
247	peaks can be extracted, ignoring the rest, which will only have pulse count images
248	extracted.
249	Returns:
250	None:
251	`fov` argument is modified in place
252	"""
253
254	if type(intensities) is list:	1✔
255	fov['intensities'] = [target in intensities for target in fov['targets']]	1✔
256	elif intensities is True:	1✔
257	fov['intensities'] = fov['targets']	1✔
258
259	# order the 'calc_intensity' bools
260	if 'intensities' in fov.keys():	1✔
261	fov['calc_intensity'] = [target in fov['intensities'] for target in fov['targets']]	1✔
262	else:
263	fov['calc_intensity'] = [False, ] * len(fov['targets'])	1✔
264
265
266	def extract_bin_files(data_dir: str, out_dir: str,	1✔
267	include_fovs: Union[List[str], None] = None,
268	panel: Union[Tuple[float, float], pd.DataFrame] = (-0.3, 0.0),
269	intensities: Union[bool, List[str]] = False, time_res: float = 500e-6,
270	write_parallel: bool = True):
271	"""Converts MibiScope bin files to pulse count, intensity, and intensity * width tiff images
272
273	Args:
274	data_dir (str \| PathLike):
275	Directory containing bin files as well as accompanying json metadata files
276	out_dir (str \| PathLike):
277	Directory to save the tiffs in
278	include_fovs (List \| None):
279	List of fovs to include. Includes all if None.
280	panel (tuple \| pd.DataFrame):
281	If a tuple, global integration range over all antibodies within json metadata.
282	If a pd.DataFrame, specific peaks with custom integration ranges. Column names must be
283	'Mass' and 'Target' with integration ranges specified via 'Start' and 'Stop' columns.
284	intensities (bool \| List):
285	Whether or not to extract intensity and intensity * width images. If a List, specific
286	peaks can be extracted, ignoring the rest, which will only have pulse count images
287	extracted.
288	time_res (float):
289	Time resolution for scaling parabolic transformation
290	write_parallel (bool):
291	Try writing files out in parallel
292	"""
293	fov_files = _find_bin_files(data_dir, include_fovs)	×
294
295	for fov in fov_files.values():	×
296	_fill_fov_metadata(data_dir, fov, panel, intensities, time_res)	×
297
298	bin_files = \	×
299	[(fov, os.path.join(data_dir, fov['bin'])) for fov in fov_files.values()]
300
301	if write_parallel:	×
302	with mp.Pool() as pool:	×
303	for i, (fov, bf) in enumerate(bin_files):	×
304	# call extraction cython here
305	img_data = _extract_bin.c_extract_bin(	×
306	bytes(bf, 'utf-8'), fov['lower_tof_range'],
307	fov['upper_tof_range'], np.array(fov['calc_intensity'], dtype=np.uint8))
308	pool.apply_async(	×
309	_write_out,
310	(img_data, out_dir, fov['bin'][:-4], fov['targets'])
311	)
312	pool.join()	×
313	pool.close()	×
314	else:
315	for i, (fov, bf) in enumerate(bin_files):	×
316	img_data = _extract_bin.c_extract_bin(	×
317	bytes(bf, 'utf-8'), fov['lower_tof_range'],
318	fov['upper_tof_range'], np.array(fov['calc_intensity'], dtype=np.uint8)
319	)
320	_write_out(img_data, out_dir, fov['bin'][:-4], fov['targets'])	×
321
322
323	def get_histograms_per_tof(data_dir: str, fov: str, channel: str, mass_range=(-0.3, 0.0),	1✔
324	time_res: float = 500e-6):
325	"""Generates histograms of pulse widths, pulse counts, and pulse intensities found within the
326	given mass range
327
328	Args:
329	data_dir (str \| PathLike):
330	Directory containing bin files as well as accompanying json metadata files
331	fov (str):
332	Fov to generate histogram for
333	channel (str):
334	Channel to check widths for
335	mass_range (tuple):
336	Integration range
337	time_res (float):
338	Time resolution for scaling parabolic transformation
339	"""
340	fov = _find_bin_files(data_dir, [fov])[fov]	×
341
342	_fill_fov_metadata(data_dir, fov, mass_range, False, time_res, [channel])	×
343
344	local_bin_file = os.path.join(data_dir, fov['bin'])	×
345
346	widths, intensities, pulses = _extract_bin.c_extract_histograms(bytes(local_bin_file, 'utf-8'),	×
347	fov['lower_tof_range'][0],
348	fov['upper_tof_range'][0])
349	return widths, intensities, pulses	×
350
351
352	def median_height_vs_mean_pp(data_dir: str, fov: str, channel: str,	1✔
353	panel: Union[Tuple[float, float], pd.DataFrame] = (-0.3, 0.0),
354	time_res: float = 500e-6):
355	"""Retrieves median pulse intensity and mean pulse count for a given channel
356
357	Args:
358	data_dir (str \| PathLike):
359	Directory containing bin files as well as accompanying json metadata files
360	fov (str):
361	Fov to generate histogram for
362	channel (str):
363	Channel to check widths for
364	mass_range (tuple \| pd.DataFrame):
365	Integration range
366	time_res (float):
367	Time resolution for scaling parabolic transformation
368
369	"""
370
371	fov = _find_bin_files(data_dir, [fov])[fov]	×
372	_fill_fov_metadata(data_dir, fov, panel, False, time_res, [channel])	×
373
374	local_bin_file = os.path.join(data_dir, fov['bin'])	×
375
376	# TODO: fix median calculation in this call
377	_, mean_pp = \	×
378	_extract_bin.c_pulse_height_vs_positive_pixel(bytes(local_bin_file, 'utf-8'),
379	fov['lower_tof_range'][0],
380	fov['upper_tof_range'][0])
381
382	_, intensities, _ = \	×
383	_extract_bin.c_extract_histograms(bytes(local_bin_file, 'utf-8'),
384	fov['lower_tof_range'][0],
385	fov['upper_tof_range'][0])
386
387	int_bin = np.cumsum(intensities) / intensities.sum()	×
388	median_height = (np.abs(int_bin - 0.5)).argmin()	×
389
390	return median_height, mean_pp	×

angelolab / mibi-bin-tools / 6

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous