深度學習_1_Tensorflow_2_數據_文件讀取

tensorflow 數據讀取

• 隊列和線程

• 文件讀取,

• 圖片處理

問題:大文件讀取,讀取速度,

在tensorflow中真正的多線程

子線程讀取數據 向隊列放數據(如每次100個),主線程學習,不用全部數據讀取后,開始學習

隊列與對壘管理器,線程與協調器

tf.FIFOQueue(capacity=None,dtypes=None,name="fifo_queue") # 先進先出隊列

• dequeue() 出隊方法

• enqueue(vals,name=None) 入隊方法

• enqueue_many(vals,name=None) 多個入隊

• size(name=None) 還有對少數據

tf.RandomShuffleQueue() # 隨機出隊隊列

tf.train.QueueRunner(queue=None,enqueue_ops=None)

• queue:隊列

• enqueue_ops:添加線程的隊列操作列表,[op1,op2]*2 2表示指定兩個線程

• create_threads(sess,coord=None,start=False) 創建線程運行給定的入隊操作

• start:若為True線程自動啟動,Fasle需手動start()

• coord:線程協調器,后面線程管理器用到

• return:返回線程實例

線程協調器

tf.train.Coordinator()

• request_stop()

• should_stop() 檢查是否要求停止

• join(threads=None,stop_grace_period_secs=120) 等待線程終止

• return 返回線程協調器實例

同步,異步案例

import  tensorflow as tf
def sync():
    # 模擬一下同步  先處理數據,然后取數據訓練
    # tensorflow中,運行操作有依賴性
    # 1,首先定義隊列
    Q = tf.FIFOQueue(3,tf.float32)
    # 放入數據 tensorflow 會把傳入的數據當做張量 a=y 此時y不會當做列表  a=[x,] 此時a是list
    enq_many = Q.enqueue_many([[0.1,0.2,0.3],])
    # 2,定義讀取數據,取數據過程, 入隊,出隊
    out_q = Q.dequeue() # 雖然是個op 但取出來是數據
    data = out_q+1
    en_q = Q.enqueue(data)
    with tf.Session() as sess:
        # 初始化隊列
        sess.run(enq_many)
        # 處理數據
        for i in range(100):
            sess.run(en_q) # 有依賴性,會運行out_q,data,qn_q
        # 訓練數據
        for i in range(Q.size().eval()):
            print(sess.run(Q.dequeue())) # 33.2 33.3 34.1

模擬異步

import  tensorflow as tf
def asyn():
    # 1,定義隊列
    Q = tf.FIFOQueue(1000,tf.float32)
    # 2,定義子線程 操作
    var = tf.Variable(0.0)
    # 實現自增 tf.assign_add() # 主要作用是江var的值改變了
    data = tf.assign_add(var,tf.constant(1.0))
    en_q = Q.enqueue(data)
    # 3,定義隊列管理器op,指定子線程該做什么
    qr = tf.train.QueueRunner(Q,enqueue_ops=[en_q]*2)
    # 初始化變量op
    init = tf.global_variables_initializer()
    with tf.Session() as sess:
        sess.run(init)
        # 開啟線程管理器
        coord = tf.train.Coordinator()
        # 開啟子線程
        threads = qr.create_threads(sess,start=True,coord=coord) # coord指定管理器
        # 主線程,不斷取數據訓練
        for i in range(300): # 300 次后主線程結束,sess釋放,單子線程還在,會報錯,需要回收子線程
            print(sess.run(Q.dequeue()))
        coord.request_stop() # 請求回收
        coord.join(threads) # 等待回收

文件讀取

• 構造文件列表,
• 構造文件隊列,
• 閱讀器 讀取內容一行,
• 解碼,
• 批處理

文件隊列

tf.train.string_input_producer(string_tensor=None,shuffle=True) # 獎輸出字符串輸入到管道隊列

• string_tensor 含有文件名的1階張量

• num_epochs:過幾遍數據,默認無限

• return:具有字符串的隊列

文件閱讀器

• tf.TextLineReader # csv文件
• tf.FixedLengthRecordReader(record_bytes=1024) # 二進制
• tf.TFRecordReader # TfRecords 自帶的文件格式

共同方法 read(file_queue)

返回tensors元祖(key文件名字,value 默認的內容,一行,或字節)

解碼

tf.decode_csv(records=None,record_defaults=None,field_delim=",",name=None)

• recodes:上邊的value

• field_delim:默認分隔符

• record_defaults:指定每列類型,指定默認值

• 參數決定了張量的類型,並設置一個值,在字符串中缺少使用默認值

tf.decode_raw(bytes,out_type=None,little_endian=None,name=None)

將字節轉換為一個數字向量,字節為一字符類型的張量,與函數

tf.FixLengthRecordReader

搭配使用,二進制讀取為uint8格式

管道批處理

tf.train.batch()

• tensors:包含張量的列表

• batch_size:從隊列匯總讀取的批處理的大小

• num_threads:進入隊列的線程數

• capacity:整數,隊列的最大數量

案例

import  tensorflow as tf
def readcsv(filelist):
    """
    :param filelist:文件路徑列表
    :return:
    """
    import os
    path = r"E:\Workspace\PythonWorkSpace\Machine-Learning\asstes\temp"
    file_name = os.listdir(path)
    file_list = [os.path.join(path, file) for file in file_name]
    # 1,構造文件隊列
    file_queue = tf.train.string_input_producer(file_list)
    # 2,構造csv閱讀器 讀取隊列數據
    reader = tf.TextLineReader()
    key,value=reader.read(file_queue)
    # 3,對每行內容進行解碼
    records = [["None"],["None"]] # 指定第一列為字符串,默認值為None, 或指定為[1.0] 數字
    # 單行讀取
    first,second = tf.decode_csv(value,record_defaults=records) # 返回就是n列,就是n個值接收
    # 批量讀取,每次取出的大小只與batch_size:有關 若大於總數,會重復取
    first_batch,second_batch = tf.train.batch([first,second],batch_size=6,num_threads=1,capacity=6)
    with tf.Session() as sess:
        # 定義線程協調器
        coord = tf.train.Coordinator()
        threads = tf.train.start_queue_runners(sess,coord=coord)
        #print(sess.run([first,second]))
        print(sess.run([first_batch,second_batch]))
    coord.request_stop()
    coord.join(threads)
    return None

圖像讀取

特征值:像素

黑白:單通道[0-255] 像素數

彩色:三通道:RGB 像素數*3

圖片數字化

三要素:長度,寬度,通道數

3D張量 [200,200,1] [200,200,3]

圖片特征要一樣:像素要一樣-->大小相同 縮放圖片

圖片文件讀取

API:

圖像讀取器:tf.WholeFileReader

• ​ 將文件的全部內容作為值輸出的讀取器

• ​ 返回讀取實例

• ​ read(file_queuse) 輸出時一個文件名和該文件的值

圖像解碼器:tf.image.decode_jpeg(contents)

• ​ 將jpeg編碼為的圖像解碼為uint8張量

• ​ 返回uint8張量 3D形狀 [height,width,channels]

• ​ tf.image.decode_png(contents)

• ​ 解碼為uint8或uint16

• ​ 返回3D形狀

def picread():
    import os
    import os,tensorflow as tf
    path = r"D:\圖片\壁紙\手機壁紙\people"
    file_name = os.listdir(path)
    file_list = [os.path.join(path, file) for file in file_name]
    # 1, 構造文件隊列
    file_queue=tf.train.string_input_producer(file_list)
    # 2, 構造文件讀取器讀取圖片內容(默認一張)
    reader = tf.WholeFileReader()
    key,value=reader.read(file_queue)  # value tensor(shape=(),dtype=string)
    # 3, 對讀取的圖片數據進行解碼
    image = tf.image.decode_jpeg(value)  # tensor(shape=(?,?,?),dtype=uint8)
    # 4, 處理圖片大小統一
    image_resize = tf.image.resize_images(image,[1080,2160])  # shape(1080,2160,?)
    # 設置通道數,在批處理時要求形狀必須固定
    image_resize.set_shape([1080,2160,3])   # 此時 shape(1080,2160,3)
    # 5, 進行批處理
    image_batch = tf.train.batch([image_resize],batch_size=20,num_threads=4,capacity=20) # shape(20,1080,2160,3)

    with tf.Session() as sess:
        # 定義線程協調器
        coord = tf.train.Coordinator()
        threads = tf.train.start_queue_runners(sess, coord=coord)
        # print(sess.run([image_resize])) #
        print(sess.run([image_batch])) #
        coord.request_stop()
        coord.join(threads)
    return None

二進制文件處理

案例分析:https://www.cs.toronto.edu/~kriz/cifar.html

CIFAR-10數據集:6000張32*32彩色圖片 分為訓練集文件,測試集文件

文件中每一行為(標簽 像素值) 即為0-9 1024紅色通道,1024綠色,1024藍色 每行3072個字節 目標值+特征值

TFRecords 分析,存取

• ​ Tensorflow內置文件格式,二進制文件, 更好的利用內存,速度快,更好的賦值和移動
• ​ 為了將二進制數據和標簽 存儲在同一個文件中
• ​ 文件格式*.tfrecords
• ​ 寫入文件內容:Example協議塊 ----->類字典格式
• ​ 每個example樣本:{image:xxx,label:yyy}

API

tf.python_io.TFRecordsWriter(path)

• ​ path:TFrecords文件路徑
• ​ return:文件寫入器
• ​ write(record):向文件中寫入一個字符串記錄.,就是一個example
• ​ close():關閉文件寫入器

一個樣本--->example

tf.train.Example(features=None)

• ​ 寫入tfrecords文件
• ​ feature:tf.train.Features類型的特征實例
• ​ return:example協議塊

tf.train.Features(feature=None)

• ​ 構建每個樣本的信息鍵值對
• ​ feature:字典數據,key為名字,value為tf.train.Feature 實例
• ​ return:Features類型

tf.train.Feature(**options)

​ **options:例如

​ bytes_list=tf.train.BytesList(value=[Bytes])

​ int64_list=tf.train.Int64List(value=[Value])

​ float_list=tf.train.FloatList(value=[Value])

TfRecords讀取

• 同文件閱讀器流程,中間需要解析過程

• ​ 解析Tfrecords的example協議塊

• ​ tf.parse_single_example(seralized,features=None,name=None)

  • ​ 解析一個單一的example原型
  • ​ seralized:標量字符串Tensor,一個序列化的Example
  • ​ Features:dcit字典數據,鍵為讀取的名字,值為 Fixed LenFeature
  • ​ return:一個鍵值對組成的字典,鍵為讀取的名字

• ​ tf.FixedLengthFeature(shape.dtype)

  • ​ shape:輸入數據的形狀,一般不指定,為空列表
  • ​ dtype:輸入數據的類型,與存儲金文件的類型一致,只能為float32,int64,string

import tensorflow as tf
class CifarRead(object):
    """
    完成讀取二進制文件,寫進tfrecords,讀取tfrecords
    """
    def __init__(self,filelist,tfrecords_path):
        # tfrecords文件路徑
        self.tfrecords_path = tfrecords_path
        # 文件列表(文件的目錄)
        self.filelist = filelist
        #
        # 定義讀取圖片的屬性
        self.height = 32
        self.weight = 32
        self.channel = 3
        self.label_bytes = 1
        self.image_bytes= self.height*self.weight*self.channel
        self.bytes = self.image_bytes+self.label_bytes
    def read_and_decodes(self):
        # 1, 構造文件隊列
        file_queue = tf.train.string_input_producer(self.filelist)
        # 2, 構造文件讀取器,讀取內容,每個人樣本字節數
        reader = tf.FixedLengthRecordReader(self.bytes)
        key,value = reader.read(file_queue) # 此時shape() dtype=string
        # 3, 解碼內容
        label_image = tf.decode_raw(value,tf.uint8)  # shape(?,) dtype=uint8 一維 圖片與,目標值在一起
        # 4,分割圖片與標簽數據 特征值和目標值 (本例數據中,目標值在前邊,1,3072)
        label = tf.slice(label_image,[0],[self.label_bytes]) # 切割出label ,slice(value,從哪讀,讀多少)
        image = tf.slice(label_image,[self.label_bytes],[self.image_bytes])
        label = tf.cast(label,tf.int32)
        # image = tf.cast(image,tf.float32) # 需要計算的話轉為float32
        # 5, 對圖片數據進行形狀改變,[3072]--->[32,32,3]
        image_reshape = tf.reshape(image,[self.height,self.weight,self.channel])
        # 6, 批處理數據
        image_batch,label_batch = tf.train.batch([image_reshape,label],batch_size=10,num_threads=1,capacity=10)  # shape(10,32,32,3) shape(10,1)
        return image_batch,label_batch
    def write_to_tfrecords(self,image_batch,label_batch):
        """
        將圖片的特征值和目標值 存進tfrecords
        :param image_batch: n張圖片的目標值
        :param label_batch: n張圖片的目標值
        :return: None
        """
        # 1, 構造一個tfrecords文件,使用Tfrecords存儲器
        writer = tf.python_io.TFRecordWriter(self.tfrecords_path)
        # 2, 循環寫入,每張圖片都早example
        for i in range(image_batch.shape[0]):
            # 取出第i個圖片的特征值和目標值
            image = image_batch[i] # (32,32,3)
            label = label_batch[i] # (10,1)
            # 要轉化為張量,張量值轉化為字符串
            image = image.eval().tostring()
            label = int(label.eval()[0])
            # 構造一個樣本的example
            example = tf.train.Example(features=tf.train.Features(feature={
                "image":tf.train.Feature(bytes_list=tf.train.BytesList(value=[image])),
                "label":tf.train.Feature(int64_list=tf.train.Int64List(value=[label])),
            }))
            # 寫入單獨的樣本
            writer.write(example.SerializeToString())
        # 關閉
        writer.close()
        return None
    def read_form_tfrecords(self):
        # 1, 構造文件隊列
        file_queue = tf.train.string_input_producer([r"I:\人工智能數據\6000_32_32_圖片集\cifar-10-batches-bin\all.tfrecords"])
        # 2, 構造文件閱讀器,讀取內容 example value 為一個樣本序列化的example
        reader = tf.TFRecordReader()
        key,value = reader.read(file_queue)
        # 3, 解析example
        features = tf.parse_single_example(value,features={
            "image":tf.FixedLenFeature([],tf.string),
            "label":tf.FixedLenFeature([],tf.int64),
        })
            # features["image"] Tensor(shape=(),dtype=string)
            # features["label"] Tensor(shape=(),dtype=64)
        # 4,解碼內容 ,若內容時string 需要解碼,是int64,string32 不需要解碼
        image = tf.decode_raw(features["image"],tf.uint8)  # Tensor(shape(?,) uint8)
        # 固定形狀
        image_reshape = tf.reshape(image,[self.height,self.weight,self.channel])
        label = features["label"]  # Tensor(shape=(),int64)
        # 5, 進行批處理
        image_batch,label_batch = tf.train.batch([image_reshape,label],batch_size=20,num_threads=1,capacity=10)
        return image_batch,label_batch
def readBinary():
    import os
    # 數據地址 https://www.cs.toronto.edu/~kriz/cifar.html
    path = r"I:\人工智能數據\6000_32_32_圖片集\cifar-10-batches-bin"
    tfrecords = os.path.join(path,"all.tfrecords")
    file_name = os.listdir(path)
    # 獲取文件列表,並過濾 非bin文件
    filelist = [os.path.join(path,file) for file in file_name if file[-3:]=="bin"]
    with tf.Session() as sess:
        c = CifarRead(filelist, tfrecords)
        # 讀取數據並存入tfrecords
        image_batch, label_batch = c.read_and_decodes()
        c.write_to_tfrecords(image_batch=image_batch, label_batch=label_batch)
        # 從tfrercords中讀取
        # image_batch, label_batch = c.read_form_tfrecords()
        coord = tf.train.Coordinator()
        threads = tf.train.start_queue_runners(sess, coord=coord)
        print(sess.run([image_batch,label_batch]))
        coord.request_stop()
        coord.join(threads)
    return None