Source code for allensdk.internal.pipeline_modules.run_annotated_region_metrics
"""Run annotated region metrics calculations"""
import logging
import os
import h5py
from allensdk.internal.core.lims_utilities import get_input_json
from allensdk.internal.brain_observatory.annotated_region_metrics import get_metrics
from allensdk.internal.core.lims_pipeline_module import (PipelineModule,
run_module)
SDK_PATH = "/data/informatics/CAM/isi_metrics/allensdk"
SCRIPT_PATH = ("/data/informatics/CAM/isi_metrics/allensdk/allensdk/internal"
"/pipeline_modules/run_annotated_region_metrics.py")
[docs]def debug(region_id, storage_directory="./", local=True,
sdk_path=SDK_PATH, script_path=SCRIPT_PATH, lims_host="lims2"):
strategy_class = "AnnotatedRegionMetricsStrategy"
object_class = "AnnotatedRegion"
input_json = get_input_json(region_id, object_class, strategy_class,
lims_host)
exp_dir = os.path.join(storage_directory, str(region_id))
run_module(script_path,
input_json,
exp_dir,
sdk_path=sdk_path,
pbs=dict(vmem=4,
job_name="isi_metrics_{}".format(region_id),
walltime="1:00:00"),
local=local)
[docs]def load_arrays(h5_file):
with h5py.File(h5_file, "r") as f:
altitude_phase = f['retinotopy_altitude'][:]
azimuth_phase = f['retinotopy_azimuth'][:]
return altitude_phase, azimuth_phase
[docs]def main():
mod = PipelineModule()
data = mod.input_data()
h5_file = data["processed_h5"]
altitude_phase, azimuth_phase = load_arrays(h5_file)
del data["processed_h5"]
output_data = get_metrics(altitude_phase, azimuth_phase, **data)
mod.write_output_data(output_data)
if __name__ == "__main__": main()