diff --git a/.github/workflows/django.yml b/.github/workflows/django.yml
deleted file mode 100644
index eb6bdcf..0000000
--- a/.github/workflows/django.yml
+++ /dev/null
@@ -1,32 +0,0 @@
-name: Django CI
-
-on:
- push:
- branches: [ master ]
- pull_request:
- branches: [ master ]
-
-jobs:
- build:
-
- runs-on: ubuntu-latest
- strategy:
- max-parallel: 4
- matrix:
- python-version: [3.6, 3.7, 3.8]
-
- steps:
- - uses: actions/checkout@v2
- - name: Set up Python ${{ matrix.python-version }}
- uses: actions/setup-python@v2
- with:
- python-version: ${{ matrix.python-version }}
- - name: Install Dependencies
- working-directory: api
- run: |
- python -m pip install --upgrade pip
- pip install -r requirements.txt
- - name: Run Tests
- working-directory: api
- run: |
- python manage.py test
diff --git a/api/neoview/views.py b/api/neoview/views.py
index e455a9f..df8b93a 100644
--- a/api/neoview/views.py
+++ b/api/neoview/views.py
@@ -1,134 +1,88 @@
# -*- coding: utf-8 -*-
from __future__ import unicode_literals
import os.path
-import hashlib
-import logging
-from time import sleep
-from urllib.request import urlopen, urlretrieve, HTTPError
-from urllib.parse import urlparse, urlunparse
-
from django.http import JsonResponse
-from django.utils.datastructures import MultiValueDictKeyError
-from django.conf import settings
from rest_framework.views import APIView
-from rest_framework import status
-
from neo.io import get_io
-import neo
-
-
-logger = logging.getLogger(__name__)
-
-
-def custom_get_io(filename):
- try:
- io = get_io(filename)
- except AssertionError as err:
- if "try_signal_grouping" in str(err):
- io = neo.io.Spike2IO(filename, try_signal_grouping=False)
- else:
- raise
- return io
-
+from neo import io
+from rest_framework import status
+from os.path import basename
+try:
+ from urllib import urlretrieve, HTTPError
+except ImportError:
+ from urllib.request import urlretrieve, HTTPError
+try:
+ from urllib.request import urlopen
+except ImportError:
+ from urllib2 import urlopen
+try:
+ unicode
+except NameError:
+ unicode = str
+# import logging
+from time import sleep
-def _get_cache_path(url):
- """
- For caching, we store files in a flat directory structure, where the directory name is
- based on the URL, but files in the same directory on the original server end up in the
- same directory in our cache.
- """
- url_parts = urlparse(url)
- base_url = urlunparse((url_parts.scheme, url_parts.netloc, os.path.dirname(url_parts.path), "", "", ""))
- dir_name = hashlib.sha1(base_url.encode('utf-8')).hexdigest()
- dir_path = os.path.join(getattr(settings, "DOWNLOADED_FILE_CACHE_DIR", ""),
- dir_name)
- os.makedirs(dir_path, exist_ok=True)
- return os.path.join(dir_path, os.path.basename(url_parts.path))
+# logger = logging.getLogger(__name__)
def _get_file_from_url(request):
url = request.GET.get('url')
-
- # we first open the url to resolve any redirects
- response = urlopen(url)
- resolved_url = response.geturl()
-
- filename = _get_cache_path(resolved_url)
- if not os.path.isfile(filename):
- urlretrieve(resolved_url, filename)
- # todo: wrap previous line in try..except so we can return a 404 if the file is not found
- # or a 500 if the local disk is full
-
- # if we have a text file, try to download the accompanying json file
- name, ext = os.path.splitext(filename)
- if ext[1:] in neo.io.AsciiSignalIO.extensions: # ext has a leading '.'
- metadata_filename = filename.replace(ext, "_about.json")
- metadata_url = resolved_url.replace(ext, "_about.json")
- try:
- urlretrieve(metadata_url, metadata_filename)
- except HTTPError:
- pass
-
- return filename
+ if url:
+ response = urlopen(url)
+ filename = basename(response.url)
+ if not os.path.isfile(filename):
+ urlretrieve(url, filename)
+ # todo: wrap previous line in try..except so we can return a 404 if the file is not found
+ # or a 500 if the local disk is full
+
+ # if we have a text file, try to download the accompanying json file
+ name, ext = os.path.splitext(filename)
+ if ext[1:] in io.AsciiSignalIO.extensions: # ext has a leading '.'
+ metadata_filename = filename.replace(ext, "_about.json")
+ metadata_url = url.replace(ext, "_about.json")
+ try:
+ urlretrieve(metadata_url, metadata_filename)
+ except HTTPError:
+ pass
+
+ return filename
+ else:
+ return JsonResponse({'error': 'URL parameter is missing', 'message': ''},
+ status=status.HTTP_400_BAD_REQUEST)
def _handle_dict(ob):
- return {k: str(v) for k, v in ob.items()}
-
-
-class NeoViewError(Exception):
-
- def __init__(self, message, status, detail=""):
- self.message = message
- self.status = status
- self.detail = detail
-
- def __str__(self):
- return f"{self.__class__.name}: {self.message} {self.detail} status={self.status}"
-
-
-def get_block(request):
- if not request.GET.get('url'):
- raise NeoViewError('URL parameter is missing', status.HTTP_400_BAD_REQUEST)
-
- lazy = False
- na_file = _get_file_from_url(request)
-
- if 'type' in request.GET and request.GET.get('type'):
- iotype = request.GET.get('type')
- method = getattr(neo.io, iotype)
- r = method(filename=na_file)
- if r.support_lazy:
- block = r.read_block(lazy=True)
- lazy = True
- else:
- block = r.read_block()
- else:
- neo_io = custom_get_io(na_file)
- try:
- if neo_io.support_lazy:
- block = neo_io.read_block(lazy=True)
- lazy = True
- else:
- block = neo_io.read_block()
-
- except IOError as err:
- # todo: need to be more fine grained. There could be other reasons
- # for an IOError
- raise NeoViewError('incorrect file type',
- status.HTTP_415_UNSUPPORTED_MEDIA_TYPE,
- str(err))
- return block, na_file, lazy
+ return {k: unicode(v) for k, v in ob.items()}
class Block(APIView):
def get(self, request, format=None, **kwargs):
- try:
- block, na_file, lazy = get_block(request)
- except NeoViewError as err:
- return JsonResponse({'error': err.message, 'message': err.detail},
- status=err.status)
+ lazy = False
+ na_file = _get_file_from_url(request)
+ neo_io = get_io(na_file)
+ if 'type' in request.GET and request.GET.get('type'):
+ iotype = request.GET.get('type')
+ method = getattr(io, iotype)
+ r = method(filename=na_file)
+ if r.support_lazy:
+ block = r.read_block(lazy=True)
+ lazy = True
+ else:
+ block = r.read_block()
+ else:
+ try:
+ if neo_io.support_lazy:
+ block = neo_io.read_block(lazy=True)
+ lazy = True
+ else:
+ block = neo_io.read_block()
+
+ except IOError as err:
+ # todo: need to be more fine grained. There could be other reasons
+ # for an IOError
+ return JsonResponse({'error': 'incorrect file type', 'message': str(err)},
+ status=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE)
block_data = {'block': [{
'annotations': _handle_dict(block.annotations),
@@ -196,29 +150,18 @@ def get(self, request, format=None, **kwargs):
class Segment(APIView):
def get(self, request, format=None, **kwargs):
-
- # parameter for segment
- # url --- string
- # semgent_id --- int
-
- try:
- block, na_file, lazy = get_block(request)
- except NeoViewError as err:
- return JsonResponse({'error': err.message, 'message': err.detail},
- status=err.status)
-
- # check for missing semgent_id parameter
- try:
- id_segment = int(request.GET['segment_id'])
- except MultiValueDictKeyError:
- return JsonResponse({'error': 'segment_id parameter is missing', 'message': ''},
- status=status.HTTP_400_BAD_REQUEST)
- # check for indexerror on segment_id
- try:
- segment = block.segments[id_segment]
- except IndexError:
- return JsonResponse({'error': 'IndexError on segment_id', 'message': ''},
- status=status.HTTP_400_BAD_REQUEST)
+ lazy = False
+ na_file = _get_file_from_url(request)
+ neo_io = get_io(na_file)
+ if neo_io.support_lazy:
+ block = neo_io.read_block(lazy=True)
+ lazy = True
+ else:
+ block = neo_io.read_block()
+ id_segment = int(request.GET['segment_id'])
+ # todo, catch MultiValueDictKeyError in case segment_id isn't given, and return a 400 Bad Request response
+ segment = block.segments[id_segment]
+ # todo, catch IndexError, and return a 404 response
seg_data_test = {
'name': "segment 1",
@@ -269,42 +212,25 @@ def get(self, request, format=None, **kwargs):
class AnalogSignal(APIView):
def get(self, request, format=None, **kwargs):
- # parameter for analogsignal
- # url --- string
- # check for missing segment_id p
- # analog_signal_id --- int
-
- try:
- block, na_file, lazy = get_block(request)
- except NeoViewError as err:
- return JsonResponse({'error': err.message, 'message': err.detail},
- status=err.status)
-
- # check for missing segment_id parameter
- try:
- id_segment = int(request.GET['segment_id'])
- except MultiValueDictKeyError:
- return JsonResponse({'error': 'segment_id parameter is missing', 'message': ''},
- status=status.HTTP_400_BAD_REQUEST)
-
- # check for missing analog_signal_id parameter
- try:
- id_analog_signal = int(request.GET['analog_signal_id'])
- except MultiValueDictKeyError:
- return JsonResponse({'error': 'analog_signal_id parameter is missing', 'message': ''},
- status=status.HTTP_400_BAD_REQUEST)
-
- # check for index error on segment_id
- try:
- segment = block.segments[id_segment]
- except IndexError:
- return JsonResponse({'error': 'IndexError on segment_id' , 'message': ''},
- status=status.HTTP_400_BAD_REQUEST)
-
- try:
- down_sample_factor = int(request.GET.get('down_sample_factor', 1))
- except ValueError:
- down_sample_factor = 1
+ lazy = False
+ na_file = _get_file_from_url(request)
+ while True: # todo, find better solution
+ try:
+ neo_io = get_io(na_file)
+ break
+ except OSError:
+ sleep(5)
+
+ if neo_io.support_lazy:
+ block = neo_io.read_block(lazy=True)
+ lazy = True
+ else:
+ block = neo_io.read_block()
+ id_segment = int(request.GET['segment_id'])
+ id_analog_signal = int(request.GET['analog_signal_id'])
+ # todo, catch MultiValueDictKeyError in case segment_id or analog_signal_id aren't given, and return a 400 Bad Request response
+ segment = block.segments[id_segment]
+
graph_data = {}
analogsignal = None
if len(segment.analogsignals) > 0:
@@ -314,9 +240,8 @@ def get(self, request, format=None, **kwargs):
analogsignal = segment.analogsignals[id_analog_signal]
graph_data["t_start"] = analogsignal.t_start.item()
graph_data["t_stop"] = analogsignal.t_stop.item()
-
- if down_sample_factor > 1:
- graph_data["sampling_period"] = analogsignal.sampling_period.item() * down_sample_factor
+ if request.GET['down_sample_factor']:
+ graph_data["sampling_period"] = analogsignal.sampling_period.item() * int(request.GET['down_sample_factor'])
else:
graph_data["sampling_period"] = analogsignal.sampling_period.item()
elif len(segment.irregularlysampledsignals) > 0:
@@ -328,22 +253,22 @@ def get(self, request, format=None, **kwargs):
# todo, catch any IndexErrors, and return a 404 response
- if analogsignal is None:
- return JsonResponse({})
-
analog_signal_values = []
+
if analogsignal.shape[1] > 1:
# multiple channels
- if not len(segment.irregularlysampledsignals) > 0 and down_sample_factor > 1:
+ if not len(segment.irregularlysampledsignals) > 0 and request.GET['down_sample_factor']:
+ dsf = int(request.GET['down_sample_factor'])
for i in range(0, len(analogsignal[0])):
- analog_signal_values.append(analogsignal[::down_sample_factor, i].magnitude[:, 0].tolist())
+ analog_signal_values.append(analogsignal[::dsf, i].magnitude[:, 0].tolist())
else:
for i in range(0, len(analogsignal[0])):
analog_signal_values.append(analogsignal[::, i].magnitude[:, 0].tolist())
else:
# single channel
- if not len(segment.irregularlysampledsignals) > 0 and down_sample_factor > 1:
- analog_signal_values = analogsignal[::down_sample_factor, 0].magnitude[:, 0].tolist()
+ if not len(segment.irregularlysampledsignals) > 0 and request.GET['down_sample_factor']:
+ dsf = int(request.GET['down_sample_factor'])
+ analog_signal_values = analogsignal[::dsf, 0].magnitude[:, 0].tolist()
else:
analog_signal_values = analogsignal.magnitude[:, 0].tolist()
@@ -358,28 +283,25 @@ def get(self, request, format=None, **kwargs):
class SpikeTrain(APIView):
def get(self, request, format=None, **kwargs):
+ lazy = False
+ na_file = _get_file_from_url(request)
+ while True:
+ try:
+ neo_io = get_io(na_file)
+ break
+ except OSError:
+ sleep(5)
- try:
- block, na_file, lazy = get_block(request)
- except NeoViewError as err:
- return JsonResponse({'error': err.message, 'message': err.detail},
- status=err.status)
-
- # check for missing segment_id parameter
- try:
- id_segment = int(request.GET['segment_id'])
- except MultiValueDictKeyError:
- return JsonResponse({'error': 'segment_id parameter is missing', 'message': ''},
- status=status.HTTP_400_BAD_REQUEST)
- #check for index error
- try:
- segment = block.segments[id_segment]
- except IndexError:
- return JsonResponse({'error': 'IndexError on segment_id', 'message': ''},
- status=status.HTTP_400_BAD_REQUEST)
+ if neo_io.support_lazy:
+ block = neo_io.read_block(lazy=True)
+ lazy = True
+ else:
+ block = neo_io.read_block()
+ id_segment = int(request.GET['segment_id'])
+ segment = block.segments[id_segment]
if lazy:
- spiketrains = [st.load() for st in segment.spiketrains]
+ spiketrains = segment.spiketrains.load()
else:
spiketrains = segment.spiketrains
diff --git a/api/neural_activity_app/settings.py b/api/neural_activity_app/settings.py
index 15e6deb..81ee70c 100644
--- a/api/neural_activity_app/settings.py
+++ b/api/neural_activity_app/settings.py
@@ -156,11 +156,6 @@
'level': 'DEBUG',
'propagate': True,
},
- 'django': {
- 'handlers': ['file'],
- 'level': 'INFO',
- },
},
}
-DOWNLOADED_FILE_CACHE_DIR = os.path.join(BASE_DIR, "download_cache")
diff --git a/api/neural_activity_app/templates/hbp.html b/api/neural_activity_app/templates/hbp.html
index c4b3c25..231a09b 100644
--- a/api/neural_activity_app/templates/hbp.html
+++ b/api/neural_activity_app/templates/hbp.html
@@ -11,8 +11,8 @@
-
-
+
+
@@ -118,7 +118,31 @@ Example with dynamic source
Example with spike trains
+
+
+ Example with tag attributes
+ downsamplefactor for down sampling the signal by specified positive integral value (default = 1, i.e. no down sampling);
+ segmentid for auto displaying specified segment; signalid for auto displaying specified signal; spiketrainselect
+ for displaying spike trains of segment.
+
+<visualizer-view
+ source="https://drive.humanbrainproject.eu/f/01927e7e3cff4010a98a/?dl=1"
+ downsamplefactor=4
+ segmentid="1"
+ signalid="0"
+ spiketrainselect="true"
+ height=300>
+</visualizer-view>
+
+
+
diff --git a/api/requirements.txt b/api/requirements.txt
index 8c1f48a..297bb67 100644
--- a/api/requirements.txt
+++ b/api/requirements.txt
@@ -2,13 +2,12 @@ Django==2.2.3
django-sslserver==0.20
djangorestframework==3.10.1
django-cors-headers==3.0.2
-numpy==1.19.2
-h5py==3.1.0
-nixio==1.5.0b6
+numpy==1.16.4
+h5py==2.9.0
+nixio==1.5.0b3
igor==0.3
-scipy==1.5.4
+scipy==1.3.0
klusta==3.0.16
-pynwb==1.4.0
#stfio
-#neo==0.9.0
-git+https://github.com/NeuralEnsemble/python-neo@neoview#egg=neo
+neo==0.8.0
+#git+https://github.com/NeuralEnsemble/python-neo@master#egg=neo
diff --git a/api/test/__init__.py b/api/test/__init__.py
deleted file mode 100644
index e69de29..0000000
diff --git a/api/test/testneoview.py b/api/test/testneoview.py
deleted file mode 100644
index 3c826f3..0000000
--- a/api/test/testneoview.py
+++ /dev/null
@@ -1,281 +0,0 @@
-from django.test import TestCase
-from neoview.views import Block, AnalogSignal, Segment, SpikeTrain
-from rest_framework.test import APIRequestFactory
-import json
-import os
-
-
-url_full_test_file = "https://github.com/teogale/test_file_api/raw/master/96711008.abf"
-factory = APIRequestFactory()
-
-
-class test_views(TestCase):
-
- def metadata_test(self,obj=None):
- # obj is a block or segment from neo
- # this method test common test for segment and block
-
- if obj==None:
- raise("obj parameter is none")
-
- else:
- # testing annotation data
- self.assertEqual(obj['annotations'],{'abf_version': '2.4'})
-
- #testing description
- self.assertEqual(obj['description'],'')
-
- # testing file_origin
- self.assertEqual(os.path.basename(obj['file_origin']),'96711008.abf')
-
- # testing name
- self.assertEqual(obj['name'],'')
-
- # testing rec_datetime only for block
- # self.assertEqual(obj['rec_datetime'],'1996-07-11T17:03:40')
-
- # testing file_name only for block
- # self.assertEqual(obj['file_name'],'96711008.abf')
-
-
-
-
- def test_get_block(self):
- # block made to test jsonresponse when reading a file with full given information
- # url parameter is required for block().get()
-
- # block number - 1
- # segment number - 16
- # contain only analog signal
-
-
- # block data --- type
- # annotation --- dictionnary
- # description --- string
- # file_origin --- string
- # name --- string
- # rec_datetime --- string
- # file_name --- string
-
- # segment data --- type
- # annotation --- dictionnary
- # description --- string
- # file_origin --- string
- # rec_datetime --- string
- # name --- string
-
- # irregularlysampledsignal --- list
- # analogsignals --- list
- # spiketrains --- list
-
-
- request = factory.get('blockdata/',{"url":url_full_test_file}, format='json')
- block = Block().get(request)
- json_data = json.loads(block.content)
-
-
-
- # testing the number of block
- number_of_block = len(json_data["block"])
- self.assertEqual(number_of_block, 1)
-
- # testing the number of segment
- number_of_segment = len(json_data["block"][0]['segments'])
- self.assertEqual(number_of_segment, 16)
-
- # testing rec_datetime
- self.assertEqual(json_data["block"][0]['rec_datetime'],'1996-07-11T17:03:40')
-
- # testing file_name
- self.assertEqual(os.path.basename(json_data["block"][0]['file_name']),'96711008.abf')
-
- # common test for block and segment
- self.metadata_test(obj=json_data['block'][0])
- for i in range(16):
- self.metadata_test(obj=json_data['block'][0]['segments'][i])
-
- # specific test for segment
- # test if irregularlysampledsignal is an empty list
- self.assertEqual(json_data["block"][0]['segments'][i]['irregularlysampledsignals'],[])
- # test if analogsignals is an empty list
- self.assertEqual(json_data["block"][0]['segments'][i]['analogsignals'],[])
- # test if spiketrains is an empty list
- self.assertEqual(json_data["block"][0]['segments'][i]['spiketrains'],[])
-
- def test_segment(self):
-
- # url && segment_id required for Segment().get()
- # segment data --- type
- # annotation --- dictionnary
- # description --- string
- # file_origin --- string
- # rec_datetime --- string
- # name --- string
-
- # irregularlysampledsignal --- list
- # analogsignals --- list
- # spiketrains --- list
-
- # segment composed of 2 analogsignal
-
- request = factory.get('segmentdata/',{"url":url_full_test_file, "segment_id":1}, format='json')
- segment = Segment().get(request)
- json_data = json.loads(segment.content)
-
-
- self.metadata_test(json_data)
- # test if irregularlysampledsignals is empty list
- self.assertEqual(json_data['irregularlysampledsignals'],[])
- # test if analogsignals contain 2 analogsignal
- self.assertEqual(json_data['analogsignals'],[{},{}])
- # test if spiketrains is an empty list
- self.assertEqual(json_data['spiketrains'],[])
-
-
- def test_get_analogsignal(self):
-
- # analogsignal require segment_id, analog_signal_id, down_sample_factor
-
- # analogsignal data --- type
- # t_start --- float
- # t_stop --- float
- # sampling_period --- float
- # values --- list
- # name --- string
- # time_dimensionality --- string
- # values_units --- string
-
- name = ['Chan0mV','AO#0']
- valueunit = ['mV','nA']
- for analogsignal_id in range(2):
- request = factory.get('analogsignaldata/',{"url":url_full_test_file,"segment_id":1,"analog_signal_id":analogsignal_id},format='json')
- analog_signal = AnalogSignal().get(request)
- json_data = json.loads(analog_signal.content)
-
-
- self.assertEqual(json_data['t_start'],4.0)
- self.assertEqual(json_data['t_stop'],4.7168)
- self.assertEqual(json_data['sampling_period'],0.0001)
- # not tested values
- self.assertEqual(json_data['name'],name[analogsignal_id])
- self.assertEqual(json_data['times_dimensionality'],'s')
-
- self.assertEqual(json_data['values_units'],valueunit[analogsignal_id])
-
- def test_spriketrain(self):
-
- # compose of 1 segment
- # 19 spiketrain
- # units --- float
- # t_stop --- float
- # times --- list
- # to test spiketrain an other file is needed as 96711008.abf doesnt contain one
-
- test_file = "https://drive.humanbrainproject.eu/f/076de9c19e0c4447a95e/?dl=1"
-
- request = factory.get('spiketraindata/',{"url":test_file,'segment_id':0, 'type': 'NixIO'}, format='json')
- spiketrain = SpikeTrain().get(request)
- json_data = json.loads(spiketrain.content)
- container = ["units","t_stop","times"]
-
- for i in range(20):
- for x in container:
- self.assertEqual(x in json_data[str(i)],True)
-
-
-
-
-
- def test_get_block_missing_param(self):
-
- # block made to test jsonresponse when missing requirement
- # url parameter(string) is required for block().get()
-
- # missing url test
- request = factory.get('blockdata/',{}, format='json')
- block = Block().get(request)
- json_data = json.loads(block.content)
- self.assertEqual(json_data,{'error': 'URL parameter is missing','message': ''})
-
-
- def test_get_segment_missing_param(self):
- # test for missing requirement in segment
- # url parameter(string) is requirement for segment().get()
- # segment_id --- int
-
- # missing url test
- request = factory.get('segmentdata/',{}, format='json')
- segment = Segment().get(request)
- json_data = json.loads(segment.content)
- self.assertEqual(json_data,{'error': 'URL parameter is missing','message': ''})
-
- # missing segment_id
- request = factory.get('segmentdata/',{"url":url_full_test_file}, format='json')
- segment = Segment().get(request)
- json_data = json.loads(segment.content)
- self.assertEqual(json_data,{'error': 'segment_id parameter is missing','message': ''})
-
- # index_error segment_id
- request = factory.get('segmentdata/',{"url":url_full_test_file,"segment_id":18}, format='json')
- segment = Segment().get(request)
- json_data = json.loads(segment.content)
- self.assertEqual(json_data,{'error': 'IndexError on segment_id','message': ''})
-
-
-
- def test_get_analogsignal_missing_param(self):
-
- # test for missing requirement in analogsignal
- # url parameter(string) is requirement for analogsignal().get()
- # segment_id --- int
- # analog_signal_id --- int
- # downsamplingfactor --- int
-
- # missing url test
- request = factory.get('analogsignaldata/',{}, format='json')
- analogsignal = AnalogSignal().get(request)
- json_data = json.loads(analogsignal.content)
- self.assertEqual(json_data,{'error': 'URL parameter is missing','message': ''})
-
- # missing segment_id
- request = factory.get('analogsignaldata/',{"url":url_full_test_file}, format='json')
- analogsignal = AnalogSignal().get(request)
- json_data = json.loads(analogsignal.content)
- self.assertEqual(json_data,{'error': 'segment_id parameter is missing','message': ''})
-
- # missing analog_signal_id
- request = factory.get('analogsignaldata/',{"url":url_full_test_file,"segment_id":1}, format='json')
- analogsignal = AnalogSignal().get(request)
- json_data = json.loads(analogsignal.content)
- self.assertEqual(json_data,{'error': 'analog_signal_id parameter is missing','message': ''})
-
- # index_error segment_id
- request = factory.get('analogsignaldata/',{"url":url_full_test_file,"segment_id":18,"analog_signal_id":1}, format='json')
- analogsignal = AnalogSignal().get(request)
- json_data = json.loads(analogsignal.content)
-
- self.assertEqual(json_data,{'error': 'IndexError on segment_id','message': ''})
-
-
- def test_get_spriketrain_missing_param(self):
-
- test_file = "https://drive.humanbrainproject.eu/f/076de9c19e0c4447a95e/?dl=1"
-
- # test for missing url
- request = factory.get('spiketraindata/',{}, format='json')
- spiketrain = SpikeTrain().get(request)
- json_data = json.loads(spiketrain.content)
- self.assertEqual(json_data,{'error': 'URL parameter is missing','message': ''})
-
-
- #test for missing segment_id
- request = factory.get('spiketraindata/',{'url':test_file, 'type': 'NixIO'}, format='json')
- spiketrain = SpikeTrain().get(request)
- json_data = json.loads(spiketrain.content)
- self.assertEqual(json_data,{'error': 'segment_id parameter is missing','message': ''})
-
- #test for indexerror on segment_id
- request = factory.get('spiketraindata/',{'url':test_file,'segment_id':2, 'type': 'NixIO'}, format='json')
- spiketrain = SpikeTrain().get(request)
- json_data = json.loads(spiketrain.content)
- self.assertEqual(json_data,{'error': 'IndexError on segment_id','message': ''})