Scheduled service maintenance on November 22


On Friday, November 22, 2024, between 06:00 CET and 18:00 CET, GIN services will undergo planned maintenance. Extended service interruptions should be expected. We will try to keep downtimes to a minimum, but recommend that users avoid critical tasks, large data uploads, or DOI requests during this time.

We apologize for any inconvenience.

baseio.py 7.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199
  1. # -*- coding: utf-8 -*-
  2. """
  3. baseio
  4. ======
  5. Classes
  6. -------
  7. BaseIO - abstract class which should be overridden, managing how a
  8. file will load/write its data
  9. If you want a model for developing a new IO start from exampleIO.
  10. """
  11. import collections
  12. import logging
  13. from neo import logging_handler
  14. from neo.core import (AnalogSignal, Block,
  15. Epoch, Event,
  16. IrregularlySampledSignal,
  17. ChannelIndex,
  18. Segment, SpikeTrain, Unit)
  19. read_error = "This type is not supported by this file format for reading"
  20. write_error = "This type is not supported by this file format for writing"
  21. class BaseIO(object):
  22. """
  23. Generic class to handle all the file read/write methods for the key objects
  24. of the core class. This template is file-reading/writing oriented but it
  25. can also handle data read from/written to a database such as TDT sytem
  26. tanks or SQLite files.
  27. This is an abstract class that will be subclassed for each format
  28. The key methods of the class are:
  29. - ``read()`` - Read the whole object structure, return a list of Block
  30. objects
  31. - ``read_block(lazy=True, cascade=True, **params)`` - Read Block object
  32. from file with some parameters
  33. - ``read_segment(lazy=True, cascade=True, **params)`` - Read Segment
  34. object from file with some parameters
  35. - ``read_spiketrainlist(lazy=True, cascade=True, **params)`` - Read
  36. SpikeTrainList object from file with some parameters
  37. - ``write()`` - Write the whole object structure
  38. - ``write_block(**params)`` - Write Block object to file with some
  39. parameters
  40. - ``write_segment(**params)`` - Write Segment object to file with
  41. some parameters
  42. - ``write_spiketrainlist(**params)`` - Write SpikeTrainList object to
  43. file with some parameters
  44. The class can also implement these methods:
  45. - ``read_XXX(lazy=True, cascade=True, **params)``
  46. - ``write_XXX(**params)``
  47. where XXX could be one of the objects supported by the IO
  48. Each class is able to declare what can be accessed or written directly
  49. discribed by **readable_objects** and **readable_objects**.
  50. The object types can be one of the classes defined in neo.core
  51. (Block, Segment, AnalogSignal, ...)
  52. Each class does not necessary support all the whole neo hierarchy but part
  53. of it.
  54. This is described with **supported_objects**.
  55. All IOs must support at least Block with a read_block()
  56. ** start a new IO **
  57. If you want to implement your own file format, you should create a class
  58. that will inherit from this BaseFile class and implement the previous
  59. methods.
  60. See ExampleIO in exampleio.py
  61. """
  62. is_readable = False
  63. is_writable = False
  64. supported_objects = []
  65. readable_objects = []
  66. writeable_objects = []
  67. has_header = False
  68. is_streameable = False
  69. read_params = {}
  70. write_params = {}
  71. name = 'BaseIO'
  72. description = ''
  73. extensions = []
  74. mode = 'file' # or 'fake' or 'dir' or 'database'
  75. def __init__(self, filename=None, **kargs):
  76. self.filename = filename
  77. # create a logger for the IO class
  78. fullname = self.__class__.__module__ + '.' + self.__class__.__name__
  79. self.logger = logging.getLogger(fullname)
  80. # create a logger for 'neo' and add a handler to it if it doesn't
  81. # have one already.
  82. # (it will also not add one if the root logger has a handler)
  83. corename = self.__class__.__module__.split('.')[0]
  84. corelogger = logging.getLogger(corename)
  85. rootlogger = logging.getLogger()
  86. if not corelogger.handlers and not rootlogger.handlers:
  87. corelogger.addHandler(logging_handler)
  88. ######## General read/write methods #######################
  89. def read(self, lazy=False, cascade=True, **kargs):
  90. if Block in self.readable_objects:
  91. if (hasattr(self, 'read_all_blocks') and
  92. callable(getattr(self, 'read_all_blocks'))):
  93. return self.read_all_blocks(lazy=lazy, cascade=cascade,
  94. **kargs)
  95. return [self.read_block(lazy=lazy, cascade=cascade, **kargs)]
  96. elif Segment in self.readable_objects:
  97. bl = Block(name='One segment only')
  98. if not cascade:
  99. return bl
  100. seg = self.read_segment(lazy=lazy, cascade=cascade, **kargs)
  101. bl.segments.append(seg)
  102. bl.create_many_to_one_relationship()
  103. return [bl]
  104. else:
  105. raise NotImplementedError
  106. def write(self, bl, **kargs):
  107. if Block in self.writeable_objects:
  108. if isinstance(bl, collections.Sequence):
  109. assert hasattr(self, 'write_all_blocks'), \
  110. '%s does not offer to store a sequence of blocks' % \
  111. self.__class__.__name__
  112. self.write_all_blocks(bl, **kargs)
  113. else:
  114. self.write_block(bl, **kargs)
  115. elif Segment in self.writeable_objects:
  116. assert len(bl.segments) == 1, \
  117. '%s is based on segment so if you try to write a block it ' + \
  118. 'must contain only one Segment' % self.__class__.__name__
  119. self.write_segment(bl.segments[0], **kargs)
  120. else:
  121. raise NotImplementedError
  122. ######## All individual read methods #######################
  123. def read_block(self, **kargs):
  124. assert(Block in self.readable_objects), read_error
  125. def read_segment(self, **kargs):
  126. assert(Segment in self.readable_objects), read_error
  127. def read_unit(self, **kargs):
  128. assert(Unit in self.readable_objects), read_error
  129. def read_spiketrain(self, **kargs):
  130. assert(SpikeTrain in self.readable_objects), read_error
  131. def read_analogsignal(self, **kargs):
  132. assert(AnalogSignal in self.readable_objects), read_error
  133. def read_irregularlysampledsignal(self, **kargs):
  134. assert(IrregularlySampledSignal in self.readable_objects), read_error
  135. def read_channelindex(self, **kargs):
  136. assert(ChannelIndex in self.readable_objects), read_error
  137. def read_event(self, **kargs):
  138. assert(Event in self.readable_objects), read_error
  139. def read_epoch(self, **kargs):
  140. assert(Epoch in self.readable_objects), read_error
  141. ######## All individual write methods #######################
  142. def write_block(self, bl, **kargs):
  143. assert(Block in self.writeable_objects), write_error
  144. def write_segment(self, seg, **kargs):
  145. assert(Segment in self.writeable_objects), write_error
  146. def write_unit(self, ut, **kargs):
  147. assert(Unit in self.writeable_objects), write_error
  148. def write_spiketrain(self, sptr, **kargs):
  149. assert(SpikeTrain in self.writeable_objects), write_error
  150. def write_analogsignal(self, anasig, **kargs):
  151. assert(AnalogSignal in self.writeable_objects), write_error
  152. def write_irregularlysampledsignal(self, irsig, **kargs):
  153. assert(IrregularlySampledSignal in self.writeable_objects), write_error
  154. def write_channelindex(self, chx, **kargs):
  155. assert(ChannelIndex in self.writeable_objects), write_error
  156. def write_event(self, ev, **kargs):
  157. assert(Event in self.writeable_objects), write_error
  158. def write_epoch(self, ep, **kargs):
  159. assert(Epoch in self.writeable_objects), write_error