코딩도상국

안녕하세요. 오늘은 파이썬으로 CSV형식으로 이루어진 파일구조를 알아보겠습니다. 

csv(comma-separated values) : 쉼표로 분리된 데이터형식을 말합니다. 

예를 들어, 

5x5의 데이터가 있다고 가정합니다. 이를 csv형식으로 변환하면.

열은 (,)로 표현하고, 행은 (\n : new line : 띄어쓰기)로 표현합니다.

당연히, 우리들이 많이 사용하는 엑셀로 csv파일을 표현할 수 있습니다. 

평소처럼 엑셀을 사용하고, 저장할 때 csv로 저장해주면 됩니다. 

엑셀에 기본 형태에 알아 보았으니, 이제 파이썬으로 코딩을 하겠습니다. 

아주 편하게, 파이썬은 csv파일을 표현하기 위한 라이브러리가 존재합니다. 

import csv

f = open('example.csv', 'w', encoding='utf-8') # file open 
wr = csv.writer(f)  # csv file write

wr.writerow(["a", "A"])
wr.writerow(["b", "B"])
wr.writerow(["c", "C"])
wr.writerow(["d", "D"])
wr.writerow(["e", "E"])
wr.writerow(["f", "F"])

f.close()  # file close

1. 우선 csv파일을 만들기 위해선 생성할 파일을 열어줍니다.

우선 우리는 파일을 작성하는 예제이기 때문에 'w'를 사용합니다. 

'w'  : write (쓰기), 'r' : read(읽기)

2. 열린 파일을 csv.wrtie()에  넣는다.

3. wr.write( )를 통해 csv파일에 넣고자 데이터를 작성한다. 

4. 당연히 파일을 열었으니, 파일을 닫아 작업을 완료한다. 

위 csv 라이브러리를 사용하면 간편하게 csv구조를 만들 수 있습니다. 

우리들은 csv파일 형식에 대해 알게 되었습니다. csv 라이브러리를 사용하지 않는다면 표현할 수 없는걸까요?

이번에는 csv파일을 사용하지 않고, 직접 우리들이 csv파일 형식으로 만들어 줍시다. 

f = open("example.csv", 'w')
list_i = ["a", "b", "c", "d", "e", "f"]

for i in list_i:
    f.write(i)
    f.write(",")
    f.write(i.upper())
    f.write("\n")
f.close()

1. 우선 csv파일을 만들기 위해선 생성할 파일을 열어줍니다.

우선 우리는 파일을 작성하는 예제이기 때문에 'w'를 사용합니다. 

'w'  : write (쓰기), 'r' : read(읽기)

2. write를 통해 자신이 쓰고자 하는 데이터를 입력한다. 

열을 표현하기 위해 쉼표(, : comma)를 직접 입력한다. 

행을 표현하기 위해 띄어쓰기(\n : new line)를 직접 입력한다. 

3. 파일을 열었으니, 파일을 닫아 작업을 완료한다.

당연히 코딩할 때, csv를 사용하나, 안사용하나 결과의 차이는 없습니다. 

하지만 좀 더 로우레벨로 코딩을 하고 싶거나, 파일 입출력에 공부를 하고 싶다면 이런식으로 접근하는 것도 나쁘지 않다고 생각합니다. 

큐(Queue) : 가장 먼저 넣은 데이터가 가장 먼저 나오는 구조

FIFO : First in First out

대게 Buffer에 많이 사용됩니다. 

Python의 장점 중 하나는 자료구조를 힘들게 구현할 필요가 없다는 점입니다. 

Class형으로 Queue를 표현할 수 있지만, 저는 최대한 단순하게 가는 코딩을 좋아하기 때문에 list를 이용해서 

큐 구조를 사용해보겠습니다..

우선, 재귀함수를 이용해서 FIFO 구조를 구현하겠습니다.

def stack(start, end):
    if start <= end:
        print('Put :', start)
        stack(start + 1, end)

        print("Get  :", end-start)
    else:
        print("--------")


stack(start=0, end=5)

#결과
Put : 0
Put : 1
Put : 2
Put : 3
Put : 4
Put : 5
--------
Get  : 0
Get  : 1
Get  : 2
Get  : 3
Get  : 4
Get  : 5

https://coding-yoon.tistory.com/43?category=830017

위 링크를 타 보시면 stack구조와 다른 내용은 거의 없습니다. 

0, 1, 2, 3, 4, 5 순으로 들어갔다면 0, 1, 2, 3, 4, 5으로 다시 나오는 구조입니다. 

import queue
queue_list = queue.Queue()

for i in range(10):
    queue_list.put(i)
    print("Put :", i)

print("------------------")

for i in range(10):
    print("Get :", queue_list.get())
    

# 결과
Put : 0
Put : 1
Put : 2
Put : 3
Put : 4
Put : 5
Put : 6
Put : 7
Put : 8
Put : 9
------------------
Get : 0
Get : 1
Get : 2
Get : 3
Get : 4
Get : 5
Get : 6
Get : 7
Get : 8
Get : 9

import queue를 사용하시면 간단하게 FIFO구조를 구현하실 수 있습니다. 

https://coding-yoon.tistory.com/23?category=830190

파이썬에는 장점 중 하나는 여러가지 라이브러리가 구현되어 있고, 손쉽게 이용할 수 있습니다.

위 글에 들어가보시면, Scipy 라이브러리에는 Low Pass Filter, High Pass Filter 등등 필터를 구현되어 있습니다. 

그렇다면 우리들 또한 라이브러리를 만들 수 있고, 배포할 수 있습니다.

이번 글은 Low Pass Filter, High Pass Filter, Band Pass Filter를 Scipy 라이브러리를 사용하지 않고 모듈화시키고 구현하겠습니다.

우선, 구현된 모습을 보겠습니다.

위 그림은 제가 Scipy 라이브러리를 사용하지 않았고, 제가 모듈화를 시켜서 돌린 것입니다.

파이썬에서 모듈화를 하는 방법은 굉장히 간단합니다.

예를 들어 filter_module의 이름을 가진 라이브러리라면, filter_module.py로 파이썬 파일을 만들어줍니다.

그리고 자신이 원하는 부분을 구현하면 됩니다.

단, 조건이 있다면 같은 프로젝트 안에 있어야 합니다. 아무리 같은 폴더 안에 있다 하더라도 같은 프로젝트로 공유하고 있지 않다면 인식을 할 수 없습니다. 

# filter_module.py
import numpy as np


# Low Pass Filter
def lpf_module(raw_data, time, f_cut_lpf, order, sr):
    """
    :param raw_data: Low Pass Filter 통과할 데이터
    :param time: 시간
    :param f_cut_lpf: 차단 주파수
    :param order: 차수
    :param sr: sample rate
    :param current_list: 진행도를 보여줄 list
    :return result: Low Pass Filter 데이터
    """
    sample_time = 1 / sr
    w_cut_lpf = 2 * np.pi * f_cut_lpf
    w_cut_period = 1 / w_cut_lpf
    raw_data_temp = raw_data.copy()
    previous_low_result = 0
    low_result = []
    low_list_result = []

    step = 0
    # low pass filter
    for k in range(order, 0, -1):
        for j in range(0, len(raw_data)):
            for i in range(0, len(time)):
                y = (w_cut_period * previous_low_result + sample_time * raw_data_temp[j][i]) / (
                        w_cut_period + sample_time)
                previous_low_result = y
                low_result.append(y)
            if k > 1:
                raw_data_temp[j] = low_result
                low_result = []
            if k == 1:
                low_list_result = np.append(low_list_result, low_result)
            low_result = []

    result = low_list_result.reshape(len(raw_data_temp), len(time))
    return result


# High Pass Filter
def hpf_module(raw_data, time, f_cut_hpf, order, sr):
    """
    :param raw_data: High Pass Filter 통과할 데이터
    :param time: 시간
    :param f_cut_hpf: 차단 주파수
    :param order: 차수
    :param sr: sample rate
    :param current_list: 진행도를 보여줄 list
    :return result: High Pass Filter 데이터
    """
    w_cut_hpf = 2 * np.pi * f_cut_hpf
    w_cut_period = 1 / w_cut_hpf
    sample_time = 1/sr
    """
    raw_data_temp = self.raw_data  # 주소를 복사
    raw_data_temp = self.raw_data.copy  # 값을 복사
    """
    raw_data_temp = raw_data.copy()  # raw data
    previous_high_result = 0  # previous result
    previous_raw_data = 0  # previous raw data
    high_result = []  # result
    high_list_result = []
    step = 0

    # high pass filter
    for k in range(order, 0, -1):
        for j in range(0, len(raw_data_temp)):
            for i in range(0, len(time)):
                y = w_cut_period / (w_cut_period + sample_time) * \
                    previous_high_result + w_cut_period / \
                    (w_cut_period + sample_time) * (
                            raw_data_temp[j][i] - previous_raw_data)
                previous_high_result = y
                previous_raw_data = raw_data_temp[j][i]
                high_result = np.append(high_result, y)
            if k > 1:
                raw_data_temp[j] = high_result
                high_result = []
            if k == 1:
                high_list_result = np.append(high_list_result, high_result)
            high_result = []

    # filter data reshape ( np.append 를 사용하면 일차배열로 쌓임 )
    result = high_list_result.reshape(len(raw_data_temp), len(time))
    return result


# Band Pass Filter
def bpf_module(raw_data, time, f_cut_hpf, f_cut_lpf, order, sr, current_list=None):
    """
        :param raw_data: Band Pass Filter 통과할 데이터
        :param time: 시간
        :param f_cut_lpf: 차단 주파수
        :param f_cut_hpf: 차단 주파수
        :param order: 차수
        :param sr: sample rate
        :param current_list: 진행도를 보여줄 list
        :return result: Band Pass Filter 데이터
        """
    sample_time = sr
    raw_data_temp = raw_data.copy()  # raw data
    w_cut_lpf = 2 * np.pi * f_cut_lpf
    w_cut_lpf_period = 1 / w_cut_lpf

    previous_low_result = 0  # previous raw data
    low_result = []  # result
    low_list_result = []
    previous_raw_data = 0  # previous raw data
    step = 0

    # low pass filter
    for k in range(order, 0, -1):
        for j in range(0, len(raw_data_temp)):
            for i in range(0, len(time)):
                y = (w_cut_lpf_period * previous_low_result + sample_time * raw_data_temp[j][i]) / (
                        w_cut_lpf_period + sample_time)
                previous_low_result = y
                low_result = np.append(low_result, y)
            if k > 1:
                raw_data_temp[j] = low_result
                low_result = []
            if k == 1:
                low_list_result = np.append(low_list_result, low_result)
            low_result = []

    raw_data_temp = low_list_result.reshape(len(raw_data_temp), len(time))
    previous_high_result = 0  # previous result
    w_cut_hpf = 2 * np.pi * f_cut_hpf
    w_cut_period = 1 / w_cut_hpf
    high_result = []  # result
    high_list_result = []
    step = 0
    # high pass filter
    for k in range(order, 0, -1):
        for j in range(0, len(raw_data_temp)):
            for i in range(0, len(time)):
                y = w_cut_period / (w_cut_period + sample_time) * \
                    previous_high_result + w_cut_period / \
                    (w_cut_period + sample_time) * (
                            raw_data_temp[j][i] - previous_raw_data)
                previous_high_result = y
                previous_raw_data = raw_data_temp[j][i]
                high_result = np.append(high_result, y)
            if k > 1:
                raw_data_temp[j] = high_result
                high_result = []
            if k == 1:
                high_list_result = np.append(high_list_result, high_result)
            high_result = []
    # filter data reshape ( np.append 를 사용하면 일차배열로 쌓임 )
    result = high_list_result.reshape(len(raw_data_temp), len(time))
    return result

이렇게 Low Pass Filter, High Pass Filter, Band Pass Filter를 구현된 저만의 라이브러리를 만들었습니다. 

그렇다면 어떻게 사용할까요? 사용방법도 굉장히 간단합니다. 평소 우리들이 다운받아 사용하는 라이브러리처럼

import 해서 불러오면 됩니다. from에는 파일명이 들어갑니다. 

from filter_module import *
"""
from filter_module import lpf_module
from filter_module import hpf_module
from filter_module import bpf_module
"""

raw_data = "데이터가 담겨 있는 이차원 배열형식"
time = "데이터의 시간영역"
cut_off_frequency = "차단주파수"
order = "차수"
sr = "sample rate"

# 저역통과필터
lpf_result = lpf_module(raw_data, time, cut_off_frequency, order, sr)
# 고역통과필터
hpf_result = hpf_module(raw_data, time, cut_off_frequency, order, sr)

이런식으로 모듈화를 시켜 프로그래밍을 진행하게 되면, 나중에 팀프로젝트할 때도 상당히 수월하며, 몇 달이 지나 수정

하고 싶다고 해도 비교적 쉽게 할 수 있습니다. 그리고 모듈화의 큰 장점은 재사용성입니다. 

나중에 필터기능이 필요할 때, 위 코딩처럼 주석과 모듈화를 시켜놓으면 filter_module.py만 불러오기만 하면 됩니다. 

처음엔 복잡할지라도 점점 코딩이 이뻐지기 시작하면서 기분도 좋아집니다@@

python List 메모리에 대해 알아 보겠습니다.

list_a = [1,2,3,4]
list_b = list_a

print("list_a :", list_a)
print("list_b :", list_b)

print("list_a 메모리 :", id(list_a))
print("list_b 메모리 :", id(list_b))

"""
결과
list_a : [1, 2, 3, 4]
list_b : [1, 2, 3, 4]
list_a 메모리 : 1305532916296
list_b 메모리 : 1305532916296
"""

list_a의 리스트를 list_b에 대입연산자( = )를 사용하게 되면,

list_b에 값이 저장되는 것이 아니라, list_a의 메모리 주소를 저장하게 되어 list_a와 list_b가 서로 메모리 주소를

공유합니다. 

만약, list_a에 변수를 추가하게 되면 어떻게 될까요?

list_a = [1,2,3,4]
list_b = list_a

#list_a에 5를 추가
list_a.append(5)

print("list_a :", list_a)
print("list_b :", list_b)

print("list_a 메모리 :", id(list_a))
print("list_b 메모리 :", id(list_b))

"""
결과
list_a : [1, 2, 3, 4, 5]
list_b : [1, 2, 3, 4, 5]
list_a 메모리 : 1866016776776
list_b 메모리 : 1866016776776
"""

결과는 list_a에 5가 추가된다면, list_b에도 5가 추가됩니다.

사실상 List명만 다를 뿐, 메모리를 서로 공유하기 때문에 같은 List로 볼 수 있습니다. 

그렇다면, list_a와 list_b가 서로 다른 주소를 가지게 하려면 어떻게 해야할까요?

list_a = [1,2,3,4]
# .copy를 사용하여 변수 저장
list_b = list_a.copy()

#list_a에 5를 추가
list_a.append(5)

print("list_a :", list_a)
print("list_b :", list_b)

print("list_a 메모리 :", id(list_a))
print("list_b 메모리 :", id(list_b))

"""
결과
list_a : [1, 2, 3, 4, 5]
list_b : [1, 2, 3, 4]
list_a 메모리 : 2337630282312
list_b 메모리 : 2337630282376
"""

copy를 사용하면 메모리의 주소가 아닌, List안의 값들이 저장됩니다.

list_a와 list_b의 메모리 주소가 다름을 알 수 있습니다. 

서로 메모리의 주소가 다르기 때문에 서로 다른 변수가 됩니다.