使用Python进行数据流读写的实验

Python 数据流 实验

在本实验中主要对已经生成的ASCII文本大文件进行读写实验，目的是完成对图像转换成的ASCII文件进行类似栅格图像的坐标指定存取。最终形成一个文件读取类，可以将ASCII文件读写封装为如同栅格文件对象一样。
主要处理步骤：
1.序列化ASCII
将ArcMap转换成的ASCII码文件序列化为一列，$x×y$行的文本文件或.dat文件。方法为
file_stream():

def file_stream(inpath, outpath, offset=6):
    """
    :param inpath:
    :param outpath:
    :param offset:
    :type offset: int
    :rtype : None
    """
    f = open(inpath, 'r')
    s = open(outpath, 'a')
    try:
        for i in xrange(offset):
            f.next()
        while True:
            ff = f.next()
            fi = iter(ff)
            try:
                while True:
                    fn = fi.next()
                    if fn != "\n":
                        if fn == " ":
                            s.write("\n")
                        else:
                            s.write(fn)
            except StopIteration:
                pass
    except StopIteration:
        pass

2.处理ASCII码文件的一些方法

__author__ = 'Administrator'
import arcpy
from arcpy import Raster
out_tmp_path = r'C:\Temp'
def RasterToASCII(raster_path, out_ascii_name):
    try:
        start = Raster(raster_path)
        arcpy.RasterToASCII_conversion(start, out_tmp_path + out_ascii_name)
    except SystemError:
        print 'Conversion Failed!'
    return open(out_tmp_path + out_ascii_name, 'r')
def ASCIIToRaster(ascii_path, origin_raster_path, out_raster_name):
    write_in = open(ascii_path, 'r')
    out_ascii = open(out_tmp_path + '\\' + out_raster_name + '.txt', 'w+')
    raster_props = RasterProperties(origin_raster_path)
    out_ascii.write('ncols' + ' ' * 9 + str(raster_props['width']) + '\n')
    out_ascii.write('nrows' + ' ' * 9 + str(raster_props['height']) + '\n')
    out_ascii.write('xllcorner' + ' ' * 5 + str(raster_props['extent'].XMin) + '\n')
    out_ascii.write('yllcorner' + ' ' * 5 + str(raster_props['extent'].YMin) + '\n')
    out_ascii.write('cellsize' + ' ' * 6 + str(raster_props['meanCellHeight']) + '\n')
    out_ascii.write('NODATA_value' + ' ' * 2 + str(raster_props['noDataValue']) + '\n')
    del raster_props
    for line in write_in:
        in_word = int(line)
        out_ascii.write(str(in_word) + ' ')
    del write_in
    out_ascii.close()
    try:
        tmp_ascii_path = out_tmp_path + '\\' + out_raster_name + '.txt'
        out_raster = out_tmp_path + '\\' + out_raster_name + '.tif'
        print 'out_raster:', out_raster
        print 'tmp_ascii_path', tmp_ascii_path
        arcpy.ASCIIToRaster_conversion(tmp_ascii_path, out_raster, "INTEGER")
    except:
        print "Conversion Failed!"
def file_stream(inpath, outpath, offset=6):
    """
    :param inpath:
    :param outpath:
    :param offset:
    :type offset: int
    :rtype : None
    """
    f = open(inpath, 'r')
    s = open(outpath, 'a')
    try:
        for i in xrange(offset):
            f.next()
        while True:
            ff = f.next()
            fi = iter(ff)
            try:
                while True:
                    fn = fi.next()
                    if fn != "\n":
                        if fn == " ":
                            s.write("\n")
                        else:
                            s.write(fn)
            except StopIteration:
                pass
    except StopIteration:
        pass
    s.close()
    r = open(outpath,'r')
    return r
def RasterProperties(raster_path):
    raster = Raster(raster_path)
    raster_props = {
    'bandCount': raster.bandCount, \
    'catalogPath': raster.catalogPath, \
    'compressionType': raster.compressionType, \
    'extent': raster.extent,\
    'format': raster.format, \
    'hasRAT': raster.hasRAT, \
    'height': raster.height, \
    'isInteger': raster.isInteger, \
    'maximum': raster.maximum, \
    'mean': raster.mean, \
    'meanCellHeight': raster.meanCellHeight,\
    'meanCellWidth': raster.meanCellWidth, \
    'minimum': raster.minimum, \
    'name': raster.name, \
    'noDataValue': raster.noDataValue, \
    'path': raster.path, \
    'pixelType': raster.pixelType, \
    'spatialReference': raster.spatialReference,\
    'standardDeviation': raster.standardDeviation,\
    'uncompressedSize': raster.uncompressedSize, \
    'width': raster.width}
    return raster_props

3.任意读取某行数据。暂行使用方法get_line()，这个方法将在后期作为一个ASCII码文件类的方法封装。

def get_line(file_path, x, y):
    """
    :type y: int
    :type x: int
    :type file_path: string
    :rtype line: int
    """
    f = open(file_path, 'r')
    line_num = x * y
    for i in xrange(line_num - 1):
        f.next()
    line = int(f.next())
    return line

4.ASCII码文件类。
分析：
一个ASCII码文件类应包括

# coding=utf-8
__author__ = 'Administrator'
filename = r'I:\data\testdata\modify.txt'
openmode='r'
class Data(object):
    def __init__(self, file_name, open_mode, *relate_raster):
        """
        :type self: class
        """
        self.file_name = file_name
        self.open_mode = open_mode
        if relate_raster is not None:
            self.raster = relate_raster
        else:
            self.raster = None
        self.file = open(self.file_name, self.open_mode)
        f = self.file
        self.domain = self.cal_domain(f)
        self.line_length = self.cal_length(f)
    @staticmethod
    def cal_length(f):
        """
        :rtype : int
        """
        line1 = len(f.readline())
        for line in f:
            if len(line) > line1:
                line1 = int(len(line))
        f.seek(0,0)
        return line1
    @staticmethod
    def cal_domain(f):
        """
        :rtype : tuple
        """
        min = 0
        max = 0
        for line in f:
            if int(line) < min:
                min = int(line)
            if int(line) > max:
                max = int(line)
        domain = (min, max)
        f.seek(0,0)
        return domain
    def get_line(self, x, y):
        """
        # 应该实现指针移动访问，这里实现了文件迭代访问，需要多次迭代，耗费大量时间。
        :type self: class
        """
        assert isinstance(self.line_length, int)
        line_num = self.line_length * x + y + 1
        for i in xrange(line_num - 1):
            self.file.next()
        one_line = int(self.file.next())
        assert isinstance(one_line, int)
        return one_line
'''
if __name__ == '__main__':
    a = Data(filename, openmode)
    print a.file_name
    print a.line_length
    print a.domain
'''