# -*- coding: utf-8 -*-
"""
CSV
===
Interface to read and write csv files.
Note
----
The module utilizes the csv file writer/reader from numpy.
"""
__author__ = 'Christoph Kirst <christoph.kirst.ck@gmail.com>'
__license__ = 'GPLv3 - GNU General Pulic License v3 (see LICENSE.txt)'
__copyright__ = 'Copyright © 2020 by Christoph Kirst'
__webpage__ = 'http://idisco.info'
__download__ = 'http://www.github.com/ChristophKirst/ClearMap2'
import numpy as np
import ClearMap.IO.Source as src
import ClearMap.IO.Slice as slc
###############################################################################
### Source classe
###############################################################################
[docs]class Source(src.Source):
"""CSV array source."""
def __init__(self, location):
"""CSV source class construtor.
Arguments
---------
location : str
The filename of the csv source.
"""
self._location = location;
@property
def name(self):
return "Csv-Source";
@property
def location(self):
return self._location;
@location.setter
def location(self, value):
if value != self.location:
self._location = value;
@property
def array(self):
"""The underlying data array.
Returns
-------
array : array
The underlying data array of this source.
"""
return _array(self.location);
@array.setter
def array(self, value):
_write(self.location, value);
@property
def shape(self):
"""The shape of the source.
Returns
-------
shape : tuple
The shape of the source.
"""
return self.array.shape;
@shape.setter
def shape(self, value):
raise NotImplementedError('Cannot set shape of csv file');
@property
def dtype(self):
"""The data type of the source.
Returns
-------
dtype : dtype
The data type of the source.
"""
return self.array.dtype;
@dtype.setter
def dtype(self, value):
raise NotImplementedError('Cannot set dtype of csv file');
@property
def order(self):
"""The order of how the data is stored in the source.
Returns
-------
order : str
Returns 'C' for C contigous and 'F' for fortran contigous, None otherwise.
"""
return self.array.order;
@order.setter
def order(self, value):
raise NotImplementedError('Cannot set order of csv file');
@property
def element_strides(self):
"""The strides of the array elements.
Returns
-------
strides : tuple
Strides of the array elements.
Note
----
The strides of the elements module itemsize instead of bytes.
"""
array = self.array;
return tuple(s // array.itemsize for s in array.strides)
@property
def offset(self):
"""The offset of the memory map in the file.
Returns
-------
offset : int
Offset of the memeory map in the file.
"""
return 0;
### Data
def __getitem__(self, *args):
array = _array(self.location);
return array.__getitem__(*args);
def __setitem__(self, *args):
array = _array(self.location);
array.__setitem__(*args);
_write(self.location, array);
[docs] def as_memmap(self):
raise NotImplementedError('Memmap creation not implemented yet!')
[docs] def as_virtual(self):
return VirtualSource(source=self);
[docs] def as_real(self):
return self;
[docs] def as_buffer(self):
return self.array;
### Formatting
def __str__(self):
try:
name = self.name;
name = '%s' % name if name is not None else '';
except:
name ='';
try:
array = self.array;
except:
array = None;
try:
shape = array.shape
shape ='%r' % ((shape,)) if shape is not None else '';
except:
shape = '';
try:
dtype = array.dtype;
dtype = '[%s]' % dtype if dtype is not None else '';
except:
dtype = '';
try:
order = array.order;
order = '|%s|' % order if order is not None else '';
except:
order = '';
try:
location = self.location;
location = '%s' % location if location is not None else '';
if len(location) > 100:
location = location[:50] + '...' + location[-50:]
if len(location) > 0:
location = '{%s}' % location;
except:
location = '';
return name + shape + dtype + order + location
[docs]class VirtualSource(src.VirtualSource):
def __init__(self, source = None, shape = None, dtype = None, order = None, location = None, name = None):
super(VirtualSource, self).__init__(source=source, shape=shape, dtype=dtype, order=order, location=location, name=name);
if isinstance(source, Source):
self.location = source.location;
@property
def name(self):
return 'Virtual-Csv-Source';
[docs] def as_virtual(self):
return self;
[docs] def as_real(self):
return Source(location=self.location);
[docs] def as_buffer(self):
return self.as_real().as_buffer();
###############################################################################
### IO Interface
###############################################################################
[docs]def is_csv(source):
"""Checks if this source is a CSV source"""
if isinstance(source, Source):
return True;
if isinstance(source, str) and len(source) >= 3 and source[-3:] == 'csv':
return True;
return False;
[docs]def read(source, slicing = None, as_source = None, **kwargs):
"""Read data from a csv file.
Arguments
---------
source : str
The name of the CSV file.
slicing : slice, Slice or None
An optional sub-slice to consider.
as_source : bool
If True, return results as a source.
Returns
-------
array : array
The data in the csv file as a buffer or source.
"""
if not isinstance(source, Source):
source = Source(source);
if slicing is None:
if as_source:
return source;
else:
return source.array
else:
if as_source:
return slc.Slice(source, slicing=slicing);
else:
return source.__getitem__(slicing);
[docs]def write(sink, data, slicing = None, **kwargs):
"""Write data to a csv file.
Arguments
---------
sink : str
The name of the CSV file.
data : array
The data to write into the CSV file.
slicing : slice, Slice or None
An optional sub-slice to consider.
Returns
-------
sink : array or source
The sink csv file.
"""
if not isinstance(sink, Source):
sink = Source(sink);
if slicing is not None:
array = sink.array;
array[slicing]= data;
else:
array = data;
return _write(sink, array);
[docs]def create(location = None, shape = None, dtype = None, order = None, mode = None, array = None, as_source = True, **kwargs):
raise NotImplementedError('Creating CSV files not implemented yet!')
###############################################################################
### Helpers
###############################################################################
def _write(filename, points, **args):
"""Write point data to csv file
"""
np.savetxt(filename, points, delimiter=',', newline='\n', fmt='%.5e')
return filename
def _array(location, delimeter = ',', **args):
"""Read data from csv file.
Arguments
---------
location : str
Location of the csv array data.
delimteter : char
The delimater between subsequent array entries.
Returns
-------
array : array
The data as a numpy array.
"""
points = np.loadtxt(location, delimiter=delimeter);
return points;
###############################################################################
### Tests
###############################################################################
[docs]def test():
"""Test CSV module"""
import os
import numpy as np
import ClearMap.IO.CSV as csv
location = 'test.csv';
points = np.random.rand(5,3);
s = csv.Source(location);
print(s)
s.array = points;
print(s)
os.remove(location)
Source
