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()