이 글의 풀이는 효율보다는 객체 지향적 코드 작성에 의의를 두고 작성한 풀이입니다. 실제 업무에서 유지보수 및 개선을 염두하는 느낌으로 작성했습니다. 그리고 문제에서 보장한 입력 조건에 대해선 검증을 하지 않습니다. 질문과 피드백은 언제나 환영합니다!

문제: 퍼즐 게임 챌린지

문제 링크: https://school.programmers.co.kr/learn/courses/30/lessons/340212 (문제는 간단한 참고용으로 복붙만 했습니다. 문제를 안보셨다면 링크타고 가독성좋은 곳에서 보시는걸 추천합니다)

문제 설명

당신은 순서대로 n개의 퍼즐을 제한 시간 내에 풀어야 하는 퍼즐 게임을 하고 있습니다. 각 퍼즐은 난이도와 소요 시간이 정해져 있습니다. 당신의 숙련도에 따라 퍼즐을 풀 때 틀리는 횟수가 바뀌게 됩니다. 현재 퍼즐의 난이도를 diff, 현재 퍼즐의 소요 시간을 time_cur, 이전 퍼즐의 소요 시간을 time_prev, 당신의 숙련도를 level이라 하면, 게임은 다음과 같이 진행됩니다.

diff ≤ level이면 퍼즐을 틀리지 않고 time_cur만큼의 시간을 사용하여 해결합니다. diff > level이면, 퍼즐을 총 diff - level번 틀립니다. 퍼즐을 틀릴 때마다, time_cur만큼의 시간을 사용하며, 추가로 time_prev만큼의 시간을 사용해 이전 퍼즐을 다시 풀고 와야 합니다. 이전 퍼즐을 다시 풀 때는 이전 퍼즐의 난이도에 상관없이 틀리지 않습니다. diff - level번 틀린 이후에 다시 퍼즐을 풀면 time_cur만큼의 시간을 사용하여 퍼즐을 해결합니다. 예를 들어 diff = 3, time_cur = 2, time_prev = 4인 경우, level에 따라 퍼즐을 푸는데 걸리는 시간은 다음과 같습니다.

level = 1이면, 퍼즐을 3 - 1 = 2번 틀립니다. 한 번 틀릴 때마다 2 + 4 = 6의 시간을 사용하고, 다시 퍼즐을 푸는 데 2의 시간을 사용하므로 총 6 × 2 + 2 = 14의 시간을 사용하게 됩니다. level = 2이면, 퍼즐을 3 - 2 = 1번 틀리므로, 6 + 2 = 8의 시간을 사용하게 됩니다. level ≥ 3이면 퍼즐을 틀리지 않으며, 2의 시간을 사용하게 됩니다. 퍼즐 게임에는 전체 제한 시간 limit가 정해져 있습니다. 제한 시간 내에 퍼즐을 모두 해결하기 위한 숙련도의 최솟값을 구하려고 합니다. 난이도, 소요 시간은 모두 양의 정수며, 숙련도도 양의 정수여야 합니다.

퍼즐의 난이도를 순서대로 담은 1차원 정수 배열 diffs, 퍼즐의 소요 시간을 순서대로 담은 1차원 정수 배열 times, 전체 제한 시간 limit이 매개변수로 주어집니다. 제한 시간 내에 퍼즐을 모두 해결하기 위한 숙련도의 최솟값을 정수로 return 하도록 solution 함수를 완성해 주세요.

제한사항

1 ≤ diffs의 길이 = times의 길이 = n ≤ 300,000 diffs[i]는 i번째 퍼즐의 난이도, times[i]는 i번째 퍼즐의 소요 시간입니다. diffs[0] = 1 1 ≤ diffs[i] ≤ 100,000 1 ≤ times[i] ≤ 10,000 1 ≤ limit ≤ 1015 제한 시간 내에 퍼즐을 모두 해결할 수 있는 경우만 입력으로 주어집니다.

Code

from typing import List


class Puzzle:

    def __init__(self, diff: int, time_cur: int):
        self.diff = diff
        self.solving_time = time_cur


class Player:

    def find_best_level(self, puzzles: List[Puzzle], limit: int) -> int:
        min_diff, max_diff = 0, max(p.diff for p in puzzles)
        level = (max_diff + min_diff + 1) // 2
        prev_level = -1

        while prev_level != level:
            solve_times = self.solve_puzzles(puzzles, level)
            if solve_times > limit:
                min_diff = level
            else:
                max_diff = level

            prev_level = level
            level = value = (max_diff + min_diff + 1) // 2

        return level

    def solve_puzzles(self, puzzles: List[Puzzle], level) -> int:
        def solve_time(_cur_p, _prev_p):
            return (
                _cur_p.solving_time
                + max(_cur_p.diff - level, 0) * (_cur_p.solving_time + _prev_p.solving_time)
            )

        solve_times = puzzles[0].solving_time
        for p_idx in range(1, len(puzzles)):
            cur_p, prev_p = puzzles[p_idx], puzzles[p_idx-1]
            solve_times += solve_time(cur_p, prev_p)

        return solve_times


def solution(diffs, times, limit):
    puzzles = [Puzzle(d, t) for d, t in zip(diffs, times)]
    player = Player()

    return player.find_best_level(puzzles, limit)

코드 설명

플레이어와 퍼즐 두 객체를 정의했습니다. 퍼즐 객체는 퍼즐에 대한 정보를 가지고 있습니다. 퍼즐을 푸는 주체는 플레이어이므로 퍼즐을 받아 해결하는건 플레이어의 역할로 둡니다. 그리고 퍼즐을 풀기 위해선 이전 퍼즐의 정보가 필요하기 때문에 퍼즐을 푸는 것을 퍼즐의 역할로 두기에는 애매해집니다.

class Puzzle:

    def __init__(self, diff: int, time_cur: int):
        self.diff = diff
        self.solving_time = time_cur

퍼즐 객체는 난이도(diff), 풀이 시간(solving_time) 두 가지의 정보만을 가지기 때문에 굳이 클래스로 두어야할까? 라는 생각을 예전에는 했었던 것 같습니다. 만약 딕셔너리나 리스트, 튜플로 정의한다면 해당 변수가 가진 정보를 보려면 정의한 단계를 다시 찾아가봐야 하는 불편함이 생깁니다.

# ex. list
puzzles = [(1, 2), (3, 2)]
puzzles[0][1]  # 1시간만 지나도 못알아볼 듯

# ex. dict
puzzles = [{"diff": 1, "solving_time": 3}, ...]
puzzles[0]["diff"]  # 첫 예시보다는 나아졌지만 어떤 키를 가지는지 확인하려면 정의 단계로 돌아가야함

플레이어 객체는 함수 외에 특별히 가지는 속성은 없기 때문에 클래스로 사용하지 않고 함수만 정의해서 사용해도 문제 풀이에는 이상이 없습니다. 하지만 추상화를 통해 객체 지향적으로 정의해두었으므로, 추후 퍼즐이 플레이어의 능력에 의존하게 된다면 수월한 개발을 기대할 수 있습니다.

class Player:
    def find_best_level(self, ...): ...

    def solve_puzzles(self, ...): ...

 # 만약 플레이어의 체력에 따라 영향을 받는 기능이 추가된다면
 class Player:
     def __init__(self, hp):
        self.hp = hp

    # 함수의 인자를 추가할 필요없이 self로 접근가능
    def find_best_level(self, ...): ...

물론 당장 구현할 필요가 없기 때문에 함수만 구현해도 됩니다. 다만 퍼즐을 푼다는 것은 플레이어의 능력에 의존할 가능성이 크기 때문에 개발 방향을 고려한다면 플레이어로 추상화 해두는 것이 개인적으로 좋아보입니다.

TODO

• 현재 퍼즐 객체의 풀이시간이 이전 퍼즐 객체에 의존성이 있음
  • 만약 퍼즐 객체들간 관계가 복잡해지기 시작한다면 Puzzles 객체를 통해 퍼즐리스트를 관리하는 방향이 필요해보임
• best_level의 의미가 불명확. 베스트의 기준을 담을 수 있는 구체적인 함수명이나 주석이 필요해보임
• find_best_level 내의 while문(이분탐색 알고리즘) 부분은 분리 가능성이 보임.

고민사항

• diff, times 길이가 최대 300,000 이라는 점에서 알고리즘 고려는 필수였다.
• find_best_level, solve_puzzles 함수를 좀 더 작은 단위로 쪼갤 수 있을지 생각을 꽤 했다.
  • 이전 퍼즐과 현재 퍼즐을 푸는 함수를 내부 함수로 둘지 인스턴스 함수로 둘지 고민했는데 함수가 복잡하지 않다는 점(테스트 불필요)과 다른 곳에서는 사용하지 않을 가능성이 크다는 점에서 내부 함수로 두었다.

이 글의 풀이는 효율보다는 객체 지향에 의의를 두고 작성한 풀이입니다. 실제 업무에서 유지보수 및 개선을 염두하는 느낌으로 작성했습니다. 그리고 문제에서 보장한 입력 조건에 대해선 검증을 하지 않습니다. 질문과 피드백은 언제나 환영합니다!

문제: 동영상 재생기

문제 링크: https://school.programmers.co.kr/learn/courses/30/lessons/340213 (문제는 간단한 참고용으로 복붙만 했습니다. 문제를 안보셨다면 링크타고 가독성좋은 곳에서 보시는걸 추천합니다)

문제 설명

당신은 동영상 재생기를 만들고 있습니다. 당신의 동영상 재생기는 10초 전으로 이동, 10초 후로 이동, 오프닝 건너뛰기 3가지 기능을 지원합니다. 각 기능이 수행하는 작업은 다음과 같습니다.

10초 전으로 이동: 사용자가 "prev" 명령을 입력할 경우 동영상의 재생 위치를 현재 위치에서 10초 전으로 이동합니다. 현재 위치가 10초 미만인 경우 영상의 처음 위치로 이동합니다. 영상의 처음 위치는 0분 0초입니다. 10초 후로 이동: 사용자가 "next" 명령을 입력할 경우 동영상의 재생 위치를 현재 위치에서 10초 후로 이동합니다. 동영상의 남은 시간이 10초 미만일 경우 영상의 마지막 위치로 이동합니다. 영상의 마지막 위치는 동영상의 길이와 같습니다. 오프닝 건너뛰기: 현재 재생 위치가 오프닝 구간(op_start ≤ 현재 재생 위치 ≤ op_end)인 경우 자동으로 오프닝이 끝나는 위치로 이동합니다. 동영상의 길이를 나타내는 문자열 video_len, 기능이 수행되기 직전의 재생위치를 나타내는 문자열 pos, 오프닝 시작 시각을 나타내는 문자열 op_start, 오프닝이 끝나는 시각을 나타내는 문자열 op_end, 사용자의 입력을 나타내는 1차원 문자열 배열 commands가 매개변수로 주어집니다. 이때 사용자의 입력이 모두 끝난 후 동영상의 위치를 "mm:ss" 형식으로 return 하도록 solution 함수를 완성해 주세요.

제한사항

• video_len의 길이 = pos의 길이 = op_start의 길이 = op_end의 길이 = 5
  • video_len, pos, op_start, op_end는 "mm:ss" 형식으로 mm분 ss초를 나타냅니다.
  • 0 ≤ mm ≤ 59
  • 0 ≤ ss ≤ 59
  • 분, 초가 한 자리일 경우 0을 붙여 두 자리로 나타냅니다.
  • 비디오의 현재 위치 혹은 오프닝이 끝나는 시각이 동영상의 범위 밖인 경우는 주어지지 않습니다.
  • 오프닝이 시작하는 시각은 항상 오프닝이 끝나는 시각보다 전입니다.
• 1 ≤ commands의 길이 ≤ 100
  • commands의 원소는 "prev" 혹은 "next"입니다.
  • "prev"는 10초 전으로 이동하는 명령입니다.
  • "next"는 10초 후로 이동하는 명령입니다.

Code

def convert_string_to_sec(time_str: str) -> int:
    m, s = time_str.split(":")
    return int(m) * 60 + int(s)


def convert_sec_to_string(s: int) -> int:
    return f"{str(s//60).zfill(2)}:{str(s%60).zfill(2)}"


class Video:
    def __init__(self, metadata: dict):
        self.len: int = convert_string_to_sec(metadata["length"])
        self.opening_start: int = convert_string_to_sec(metadata["opening_start"])
        self.opening_end: int = convert_string_to_sec(metadata["opening_end"])
        self.current_pos: int = convert_string_to_sec(metadata["current_pos"])
        self.metadata = metadata


class VideoPlayer:
    def __init__(self, video: Video):
        self._video = video
        self._adjust_cur_pos()

    def input_command(self, cmd: str):
        cmd_interpreter = {"next": 10, "prev": -10}
        self._video.current_pos += cmd_interpreter[cmd]
        self._adjust_cur_pos()

    def get_current_pos(self) -> str:
        return convert_sec_to_string(self._video.current_pos)

    def _adjust_cur_pos(self):
        if self._video.current_pos < 0:
            self._video.current_pos = 0
        elif self._video.current_pos > self._video.len:
            self._video.current_pos = self._video.len

        if self._video.opening_start <= self._video.current_pos < self._video.opening_end:
            self._video.current_pos = self._video.opening_end


def solution(video_len, pos, op_start, op_end, commands):
    video_metadata = {
        "length": video_len,
        "opening_start": op_start,
        "opening_end": op_end,
        "current_pos": pos,
    }
    video = Video(video_metadata)
    video_player = VideoPlayer(video)

    for cmd in commands:
        video_player.input_command(cmd)

    return video_player.get_current_pos()

코드 해석

문제에서 객체는 비디오와 비디오를 다루는 비디오플레이어 2가지입니다. 비디오 객체는 비디오가 가지는 데이터를 다루고, 비디오 플레이어 객체는 비디오를 컨트롤하는 역할을 가진다고 볼 수 있습니다.

비디오 객체는 값을 검증, 보관, 표현하는 책임을 가지고, 비디오플레이어 객체는 비디오의 값에 접근하여 읽기, 쓰기 등의 책임을 맡습니다.

Video

• 비디오가 가지는 원본 데이터를 메타데이터로 저장합니다.
  • 원본데이터는 문자열이지만 커맨드에 따라 값을 조정해야하기 때문에 각 시간값들은 초 단위로 가집니다.
• 각 메타데이터 값들을 초 단위의 int 타입의 속성값을 가집니다.
  • datetime을 사용하지 않는 이유는 재생 관련 시간이 날짜에 해당하지 않는 점, (해당 문제에서는) 초 단위에서 해결 가능하기 때문에 int값으로 문제를 해결합니다.

VideoPlayer

• 비디오가 가진 값을 조작하기 위한 객체로 봅니다.
• VideoPlayer객체를 통해 Video값을 다룹니다
• input_command(): 커맨드를 받아 현재 재생위치를 조정합니다.
• _adjust_cur_pos(): 문제에서 주어진 조건에 따라 재생 위치를 조정합니다.

convert_string_to_sec(), convert_sec_to_string()

• "%M:%S" 값과 초단위의 int값을 변환하는 함수입니다.
• 위의 두 클래스에 종속시켜도 무방해 보이지만, 함수의 역할이 비디오에 국한된게 아니라는 생각이 들어 유틸리티 함수 정도로 생각하여 모듈 레벨의 함수로 정의했습니다.

TODO

• 실제 업무로 맡아야했다면 비디오의 메타데이터 값을 검증하는 부분이 수행되어야 합니다.
• 줄여쓴 변수명이 있습니다. 통용되는 수준의 변수명인지, 해석의 여지가 다분한지 확인이 필요합니다.
• convert_string_to_sec(), convert_sec_to_string() 함수명이 명확하지 않습니다. 명확한 함수명으로 변경하거나 간단한 주석이 필요해 보입니다.
• 비디오 저장 과정이 구현되지 않았습니다. 필요하다면 Video 객체에서 현재 값을 메타데이터에 저장하거나 내보내는 부분이 구현되어야 합니다.

마무리

처음에는 비디오 데이터 중 불변값(재생 시간, 오프닝 시간대)을 비디오 객체 비디오플레이어 객체에다가 현재 재생 위치를 추가했다. 하지만 내가 보았던 영상들 대부분은 이전 재생 위치를 기억하고 그 때부터 재생이 된다는게 생각이나서 현재 재생 위치(아마도 마지막 재생 위치 정도일 듯 하다)를 메타데이터로 포함시켰다.

input_command() 가 아닌 input_commands()를 통해 커맨드 리스트를 그대로 넣을까도 생각했지만 단위테스트를 고려해서 조금 더 작은 단위로 구현했다. 하지만 커맨드가 항상 리스트로 들어온다면 input_commands()로 구현하는 것이 맞을 듯 하다.

분명 레벨1 문제인데 레벨2보다 더 고생해서 작성했다.. 역시 고통은 구현보단 머리싸매기에서 많이 오는것 같다.

목표

• 기존에 푸는데만 집중했던 문제들을 클린 코드, 객체 지향 방식으로 리팩토링
• 완벽한 효율보다는 다양한 스타일로 작성

문제

코테 문제) https://school.programmers.co.kr/learn/courses/30/lessons/17683

네오는 자신이 기억한 멜로디를 가지고 방금그곡을 이용해 음악을 찾는다. 그런데 라디오 방송에서는 한 음악을 반복해서 재생할 때도 있어서 네오가 기억하고 있는 멜로디는 음악 끝부분과 처음 부분이 이어서 재생된 멜로디일 수도 있다. 반대로, 한 음악을 중간에 끊을 경우 원본 음악에는 네오가 기억한 멜로디가 들어있다 해도 그 곡이 네오가 들은 곡이 아닐 수도 있다. 그렇기 때문에 네오는 기억한 멜로디를 재생 시간과 제공된 악보를 직접 보면서 비교하려고 한다. 다음과 같은 가정을 할 때 네오가 찾으려는 음악의 제목을 구하여라.

• 방금그곡 서비스에서는 음악 제목, 재생이 시작되고 끝난 시각, 악보를 제공한다.
• 네오가 기억한 멜로디와 악보에 사용되는 음은 C, C#, D, D#, E, F, F#, G, G#, A, A#, B 12개이다.
• 각 음은 1분에 1개씩 재생된다. 음악은 반드시 처음부터 재생되며 음악 길이보다 재생된 시간이 길 때는 음악이 끊김 없이 처음부터 반복해서 재생된다. 음악 길이보다 재생된 시간이 짧을 때는 처음부터 재생 시간만큼만 재생된다.
• 음악이 00:00를 넘겨서까지 재생되는 일은 없다.
• 조건이 일치하는 음악이 여러 개일 때에는 라디오에서 재생된 시간이 제일 긴 음악 제목을 반환한다. 재생된 시간도 같을 경우 먼저 입력된 음악 제목을 반환한다.
• 조건이 일치하는 음악이 없을 때에는 “(None)”을 반환한다.

입력 형식 입력으로 네오가 기억한 멜로디를 담은 문자열 m과 방송된 곡의 정보를 담고 있는 배열 musicinfos가 주어진다.

• m은 음 1개 이상 1439개 이하로 구성되어 있다.
• musicinfos는 100개 이하의 곡 정보를 담고 있는 배열로, 각각의 곡 정보는 음악이 시작한 시각, 끝난 시각, 음악 제목, 악보 정보가 ','로 구분된 문자열이다.
• 음악의 시작 시각과 끝난 시각은 24시간 HH:MM 형식이다.
• 음악 제목은 ',' 이외의 출력 가능한 문자로 표현된 길이 1 이상 64 이하의 문자열이다.
• 악보 정보는 음 1개 이상 1439개 이하로 구성되어 있다.

기존 코드

def rep(s):
    # '#'음정이 있는 음을 한 문자로 줄이기
    d = {'C#':'c','D#':'d','F#':'f','G#':'g','A#':'a'}
    # 딕셔너리를 돌며 모든 #문자를 축소
    for k,v in d.items():
        s = s.replace(k,v)
    return s

def solution(m, musicinfos):
    # 주어진 m 치환
    m = rep(m)
    # 제목 , 재생시간을 저장할 튜플
    answer = ('',0)
    for musicinfo in musicinfos:
        # 음악 시작, 끝, 제목, 음 으로 나눈다.
        s,e,title,music = musicinfo.split(',')
        # 재생시간을 구해준다
        run_time = int(e[:2])*60+int(e[3:]) - int(s[:2])*60 - int(s[3:])
        # 음에서 '#'음정 치환
        music = rep(music)

        # 음악 길이가 재생시간보다 길어질 때 까지 늘려준 다음 재생시간만큼 자른다.
        while len(music)<=run_time:
            music*=2
        music = music[:run_time]
        # 음악 안에 찾는 멜로디 m이 없으면 넘어가고, 있으면 재생시간이 큰 쪽을 answer에 저장
        if music.find(m)==-1:
            continue
        else:
            answer = (title,run_time) if answer[1] < run_time else answer
    # answer의 첫 원소가 초기화상태 그대로면, 없을 시 문구 리턴
    if answer[0] =='':
        return '(None)'

    # 있다면 title 리턴
    return answer[0]

수정코드

class TimeUtils:
    @staticmethod
    def time_to_minutes(time: str) -> int:
        hours, minutes = map(int, time.split(':'))
        return hours * 60 + minutes

    @staticmethod
    def calculate_playtime(start: str, end: str) -> int:
        return TimeUtils.time_to_minutes(end) - TimeUtils.time_to_minutes(start)


class Music:
    def __init__(self, info: list[str]):
        start, end, title, melody = info.split(',')

        self.playtime = TimeUtils.calculate_playtime(start, end)
        self.title = title
        self.melody = melody

    @property
    def played_melody(self) -> str:
        conv_melody: str = convert_sharp_to_lowercase(self.melody)
        repeat: int = self.playtime // len(conv_melody) + 1

        return (repeat * conv_melody)[:self.playtime]


class MusicFinder:
    def __init__(self):
        self.music_list = []

    def append(self, music: Music) -> None:
        self.music_list.append(music)

    def find(self, melody: str) -> Music:
        target_melody = convert_sharp_to_lowercase(melody)
        found_music = None
        max_playtime = 0

        for music in self.music_list:
            if (
                target_melody in music.played_melody
                and music.playtime > max_playtime
            ):
                found_music = music
                max_playtime = music.playtime

        return found_music


def convert_sharp_to_lowercase(s: str) -> str:
    SHARP = '#'
    converted_melody = []
    idx = len(s) - 1

    while idx >= 0:
        if s[idx] == SHARP:
            converted_melody.append(s[idx - 1].lower())
            idx -= 2
        else:
            converted_melody.append(s[idx])
            idx -= 1

    return ''.join(reversed(converted_melody))


def solution(m: str, musicinfos: list) -> str:
    music_finder: MusicFinder = MusicFinder()

    for music_info in musicinfos:
        music_finder.append(Music(music_info))

    found_music = music_finder.find(m)

    return found_music.title if found_music else "(None)"

개선점

• PEP8 가이드라인에 따라 작성된 코드

• 타입 힌트 추가

• 지나친 주석 제거

• 객체 지향적 설계를 통해 코드의 재사용성과 유지보수성 증가

  • TimeUtils, Music, MusicFinder 객체를 통한 역할 분리

• 길이 2인 음정 변환 로직 개선.

  • 기존 방식: 길이 2인 음정들을 반복 탐색하며 변환 (시간복잡도 O(N: 음정길이 * M: 길이 2인 음정갯수))
  • 개선 방식: 뒤에서부터 순차 탐색하며 #이 나오면 문자 압축 (시간복잡도 O(N))

• 명확한 변수 및 메서드 명명과 중복 코드 제거를 통해 가독성과 명확성 향상.

목표

• 기존에 문제를 푸는데만 집중해서 풀었던 문제들을 클린 코드, 객체 지향 방식으로 리팩토링
• 완벽한 효율보다는 다양한 스타일로 작성

문제

코테 문제) https://school.programmers.co.kr/learn/courses/30/lessons/131127

정현이가 원하는 제품을 나타내는 문자열 배열 want와 정현이가 원하는 제품의 수량을 나타내는 정수 배열 number, XYZ 마트에서 할인하는 제품을 나타내는 문자열 배열 discount가 주어졌을 때, 회원등록시 정현이가 원하는 제품을 모두 할인 받을 수 있는 회원등록 날짜의 총 일수를 return 하는 solution 함수를 완성하시오. 가능한 날이 없으면 0을 return 합니다.

제한사항

• 1 ≤ want의 길이 = number의 길이 ≤ 10
  • 1 ≤ number의 원소 ≤ 10
  • number[i]는 want[i]의 수량을 의미하며, number의 원소의 합은 10입니다.
• 10 ≤ discount의 길이 ≤ 100,000
• want와 discount의 원소들은 알파벳 소문자로 이루어진 문자열입니다.
  • 1 ≤ want의 원소의 길이, discount의 원소의 길이 ≤ 12

기존 코드

from collections import defaultdict


DAY_RANGE = 10


def check_list(want_list, dc_list):
    for want_item, cnt in want_list.items():
        if want_item not in dc_list:
            return False
        elif dc_list[want_item] < cnt:
            return False

    return True


def solution(want, number, discount):
    answer = 0
    dc_list = defaultdict(int)
    want_list = {item:cnt for item, cnt in zip(want, number)}

    for item in discount[:DAY_RANGE]:
        dc_list[item] += 1

    loop_cnt = len(discount) - DAY_RANGE + 1

    for day in range(loop_cnt):
        if check_list(want_list, dc_list):
            answer += 1

        if day + DAY_RANGE < len(discount):
            dc_list[discount[day]] -= 1
            dc_list[discount[day + DAY_RANGE]] += 1

    return answer

수정코드

from collections import Counter

class Mart:
    def __init__(self, discount: list) -> None:
        self.day: int = 0
        self.day_range: int = 10
        self.discounts: list = discount
        self.discounts_counter: Counter = Counter(discount[:self.day_range])

    def shift_discounts_to_next_day(self) -> None:
        if self.is_last_day():
            return
        self._remove_old_discount()
        self._add_new_discount()
        self.day += 1

    def is_last_day(self) -> bool: 
        return self.day + self.day_range == len(self.discounts)

    def _remove_old_discount(self) -> None:
        item_to_remove = self.discounts[self.day]
        self.discounts_counter[item_to_remove] -= 1
        if self.discounts_counter[item_to_remove] == 0:
            del self.discounts_counter[item_to_remove]

    def _add_new_discount(self) -> None:
        item_to_add = self.discounts[self.day + self.day_range]
        self.discounts_counter[item_to_add] += 1

    @property
    def discount_period(self) -> int:
        return len(self.discounts) - self.day_range + 1


class Visitor:
    def __init__(self, want: list, number: list) -> None:
        self.buying_counter: Counter = Counter(dict(zip(want, number)))

    def can_purchase_with_discount(self, discounts_counter: Counter) -> bool:
        for item, required_count in self.buying_counter.items():
            if discounts_counter[item] < required_count:
                return False
        return True


def solution(want: list, number: list, discount: list) -> int:
    answer: int = 0
    visitor: Visitor = Visitor(want, number)
    mart: Mart = Mart(discount)

    for _ in range(mart.discount_period):
        if visitor.can_purchase_with_discount(mart.discounts_counter):
            answer += 1
        mart.shift_discounts_to_next_day()

    return answer

개선점

• PEP8 가이드라인에 따라 작성된 코드
• 타입 힌트 추가
• defaultdict 대신 Counter 사용으로 코드 간소화
• 객체 지향적 설계를 통해 코드의 재사용성과 유지보수성 증가
  • Mart, Visitor 객체를 통해 역할 분리
• 메서드 분리와 프로퍼티 사용으로 코드의 모듈성을 높이고 가독성을 개선
  • 맹글링을 통해 내부 사용과 프라이빗 변수에 대해 표현
• 명확한 변수 및 메서드 명명.

목표

• 기존에 문제를 푸는데만 집중해서 풀었던 문제들을 클린 코드, 객체 지향 방식으로 리팩토링
• 완벽한 효율보다는 다양한 스타일로 작성

문제

코테 문제) https://school.programmers.co.kr/learn/courses/30/lessons/60057

압축할 문자열 s가 매개변수로 주어질 때, 위에 설명한 방법으로 1개 이상 단위로 문자열을 잘라 압축하여 표현한 문자열 중 가장 짧은 것의 길이를 return 하도록 solution 함수를 완성해주세요.

제한사항

• s의 길이는 1 이상 1,000 이하입니다.
• s는 알파벳 소문자로만 이루어져 있습니다.

기존 코드

def solution(s):
    l = len(s)
    if l==1:
        return 1
    answer = []

    for n in range(1,(l//2)+1):
        ss = ''
        cnt = 1
        for i in range(0,l,n):

            if s[i:i+n] == s[i+n:i+2*n]:
                cnt+=1

            else:
                if cnt==1:
                    ss+=s[i:i+n]
                else:
                    ss+=(str(cnt)+s[i-n:i])
                    cnt=1
        answer.append(len(ss))

    return min(answer)

수정코드

class StringCompressor:
    def __init__(self, string):
        self.string: str = string
        self.__best_compressed: str | None = None

    def _compress_with_slice_size(self, slice_size: int) -> str:
        compressed: list[str] = []
        count: int = 1

        for i in range(0, len(self.string), slice_size):
            current_slice: str = self.string[i:i + slice_size]
            next_slice: str = self.string[i + slice_size:i + 2 * slice_size]

            if current_slice == next_slice:
                count += 1
            elif count > 1:
                compressed.append(f"{count}{current_slice}")
                count = 1
            else:
                compressed.append(current_slice)

        return ''.join(compressed)

    def _find_best_compression(self) -> str:
        if len(self.string) == 1:
            return self.string

        best_compressed = self.string
        max_slice_size = len(self.string) // 2

        for slice_size in range(1, max_slice_size + 1):
            compressed = self._compress_with_slice_size(slice_size)
            if len(compressed) < len(best_compressed):
                best_compressed = compressed

        return best_compressed

    @property
    def best_compressed(self) -> str:
        if not self.__best_compressed:
            self.__best_compressed = self._find_best_compression()

        return self.__best_compressed


def solution(s:str) -> int:
    compressor = StringCompressor(s)

    return len(compressor.best_compressed)

개선점

• PEP8 가이드라인에 따라 작성된 코드

• 타입 힌트 추가

• 객체 지향적 설계를 통해 코드의 재사용성과 유지보수성 증가

  • StringCompressor 객체를 통해 문자열 압축에 관한 역할 부여

• 메서드 분리와 프로퍼티 사용으로 코드의 모듈성을 높이고 가독성을 개선

  • 맹글링을 통해 내부 사용과 프라이빗 변수에 대해 표현
  • 지연 계산으로 효율적인 동작 구현

• 리스트를 통한 문자열 조작으로 효율성을 향상.

  • 기존의 코드에서 사용한 문자열 += 문자열 방식에서 리스트에 문자열을 저장하는 방식으로 변경

    문자열은 불변 객체

    파이썬에서 문자열은 불변 객체이므로 새로운 문자열을 더할 때마다 새로운 객체를 생성하는 방식이 적용됩니다. 따라서 문자열이 계속해서 더해지는 상황에서는 차라리 리스트에 문자열을 담고 마지막에 하나의 문자열을 생성하는 것이 더 효율적인 방식이 될 수 있습니다.

• 명확한 변수 및 메서드 명명과 중복 코드 제거를 통해 가독성과 명확성 향상.

목표

• 기존에 문제를 푸는데만 집중해서 풀었던 문제들을 클린 코드, 객체 지향 방식으로 리팩토링
• 완벽한 효율보다는 다양한 스타일로 작성

문제

코테 문제) https://school.programmers.co.kr/learn/courses/30/lessons/142085

준호가 처음 가지고 있는 병사의 수 n, 사용 가능한 무적권의 횟수 k, 매 라운드마다 공격해오는 적의 수가 순서대로 담긴 정수 배열 enemy가 매개변수로 주어집니다. 준호가 몇 라운드까지 막을 수 있는지 return 하도록 solution 함수를 완성해주세요.

제한사항

• 1 ≤ n ≤ 1,000,000,000
• 1 ≤ k ≤ 500,000
• 1 ≤ enemy의 길이 ≤ 1,000,000
• 1 ≤ enemy[i] ≤ 1,000,000
• enemy[i]에는 i + 1 라운드에서 공격해오는 적의 수가 담겨있습니다.
• 모든 라운드를 막을 수 있는 경우에는 enemy[i]의 길이를 return 해주세요.

기존 코드

from heapq import heappush,heappop

def solution(n, k, enemy):
    answer = 0
    h = []
    for i,e_n in enumerate(enemy):
        heappush(h,-e_n)
        n-=e_n

        if n < 0:
            if k<=0:
                return i
            else:
                n -= heappop(h)
                k -= 1
    return len(enemy)

수정코드

from heapq import heappush, heappop


class Player:
    def __init__(self, soldiers: int, chances: int):
        self.__soldiers: int = soldiers
        self.__chances: int = chances
        self.__enemy_heap: list[int] = []

    def fight_enemy(self, enemy: int) -> None:
        self.__soldiers -= enemy
        heappush(self.__enemy_heap, -enemy)

    def use_chance(self) -> None:
        if self.__enemy_heap and self.__chances > 0:
            self.__soldiers -= heappop(self.__enemy_heap)
            self.__chances -= 1

    @property
    def is_defeated(self) -> bool:
        return self.__soldiers < 0 and self.__chances <= 0

    @property
    def current_soldiers(self) -> int:
        return self.__soldiers


def solution(n: int, k: int, enemy: list[int]) -> int:
    player = Player(n, k)

    for _round, enemy_cnt in enumerate(enemy, start=1):
        player.fight_enemy(enemy_cnt)

        if player.current_soldiers < 0:
            player.use_chance()

        if player.is_defeated:
            return _round - 1

    return len(enemy)

개선점

• PEP8 가이드라인에 따라 작성된 코드
• 타입 힌트 추가
• 객체 지향 프로그래밍 (OOP) 사용
  • Player 클래스를 정의해서 객체 지향 프로그래밍을 사용합니다. 코드의 재사용성, 확장성, 유지보수성을 높여줍니다
  • 클래스 내부에 병사 수와 기회 수를 관리하는 방식을 통해 관련된 로직을 메서드로 분리합니다. 기능을 명확히 하고 코드의 가독성을 향상시킵니다.
• 명확한 상태 관리
  • Player 클래스의 is_defeated 및 current_soldiers와 같은 프로퍼티를 통해 코드의 흐름을 더 직관적으로 표현합니다.
  • 병사 수와 기회 수를 클래스 내부에서 관리해서 전역 변수나 매개변수를 일일이 확인하지 않아도 됩니다.
• 역할 분리
  • Player 클래스가 전투와 관련된 로직을 담당하고 solution 함수는 전투의 흐름을 제어하는 역할을 하면서 역할 분리가 명확해집니다.
• 메서드 사용을 통한 로직 분리
  • 전투를 진행하는 로직 (fight_enemy)과 기회를 사용하는 로직 (use_chance)을 별도의 메서드로 분리하여 각 메서드가 하나의 책임만 가집니다. (단일 책임 원칙, Single Responsibility Principle)
• 가독성 향상
  • 명시적 메서드와 프로퍼티를 사용하여 가독성이 향상되었습니다. solution 함수에서 플레이어가 전투를 수행하고, 기회를 사용하는 과정이 더 직관적으로 이해됩니다.

프로그래머스의 컴파일 옵션이 3.8 버전이므로 typing을 사용해야 하지만 (list[int] -> List[int]) 파이썬 3.9 이상부터는 typing없이 타입 힌트가 가능하므로 최근의 버전에 맞춰 작성했습니다.

• 독립된 메모리(프로세스), 공유메모리(스레드)
• 많은 메모리 필요(프로세스), 적은 메모리(스레드)
• 좀비(데드)프로세스 생성 가능성, 좀비(데드) 스레드 생성 쉽지 않음
• 오버헤드 큼(프로세스), 오버헤드 작음(스레드)
• 생성/소멸 다소 느림(프로세스), 생성/소멸 빠름(스레드)
• 코드 작성 쉬움/디버깅 어려움(프로세스), 코드작성 어려움/디버깅 어려움(스레드)

기본 실행

from multiprocessing import Process
import time
import logging

def proc_func(name):

    print("Sub-Process {}: starting".format(name))
    """
    Process Task
    """
    print("Sub-Process {}: finishing".format(name))


def main():
    format = "%(asctime)s: %(message)s"
    logging.basicConfig(format=format, level=logging.INFO, datefmt="%H:%M:%S")

    p = Process(target=proc_func, args=("First",))

    logging.info("Main-Process : before creating Process")
    p.start()

    logging.info("Main-Process : During Process")

    # logging.info("Main-Process : Terminated Process")
    # p.terminate()

    logging.info("Main-Process : Joined Process")
    p.join()

    print(f"Process p is alive: {p.is_alive()}")


if __name__ == "__main__":
    main()

• 스레드와 마찬가지로 target 함수, args 인자를 가진다.
• terminate(): 강제 종료

프로세스 종료 관리

프로세스는 독립적이기 때문에 부모 프로세스가 종료했더라도 자식 프로세스는 그대로 남아있을 수 있다. 그 역할을 다했더라도 컴퓨팅 자원을 그대로 가진 채로 비효율을 유발할 수 있기 때문에 역할이 끝난다면 반드시 종료를 명시해주어야 한다.

from multiprocessing import Process, current_process
import os
import random
import time


# 실행 방법
def square(n):
    # 랜덤 sleep
    time.sleep(random.randint(1, 3))
    process_id = os.getpid()
    process_name = current_process().name
    # 제곱
    result = n * n
    # 정보 출력
    print(f"Process ID: {process_id}, Process Name: {process_name}")
    print(f"Result of {n} square : {result}")


if __name__ == "__main__":
    processes = list()

    for i in range(10):
        p = Process(name=str(i), target=square, args=(i,))

        processes.append(p)

        p.start()

    # Join
    for process in processes:
        process.join()

    # 종료
    print("Main-Processing Done!")

위 코드의 Join과 같이 실행한 프로세스를 리스트에 담아서 모든 리스트에 대해 join()을 해줘서 모든 프로세스가 안전하게 종료되고 좀비 프로세스가 없도록 관리한다.

ProcessPoolExcuter

from concurrent.futures import ProcessPoolExecutor, as_completed
import urllib.request

URLS = ['http://www.daum.net/',
        'http://www.cnn.com/',
        'http://europe.wsj.com/',
        'http://www.bbc.co.uk/',
        'http://some-made-up-domain.com/']

# 실행 함수
def load_url(url, timeout):
    with urllib.request.urlopen(url, timeout=timeout) as conn:
        return conn.read()

def main():
    # 프로세스풀 Context 영역
    with ProcessPoolExecutor(max_workers=5) as executor:
        # Future 로드(실행X)
        future_to_url = {executor.submit(load_url, url, 60): url for url in URLS}

        # 실행
        for future in as_completed(future_to_url): # timeout=1(테스트 추천)
            # Key값이 Future 객체
            url = future_to_url[future]
            try:
                data = future.result()
            except Exception as exc:
                print('%r generated an exception: %s' % (url, exc))
            else:
                print('%r page is %d bytes' % (url, len(data)))

# 메인 시작
if __name__ == '__main__':
    main()

ProcessPoolExcuter을 이용하여 여러 url에 대한 요청을 병렬적으로 처리하는 코드이다.

with문을 통해 ProcessPoolExcuter을 다루면 해당 with문이 끝날 때 ProcessPoolExcuter에 의해 실행된 멀티프로세스들이 함께 종료되어 더 안전한 코드 작성이 가능하다.

Memory Sharing

프로세스는 독립적이기 때문에 직접적인 메모리 공유가 되지 않는다. 공유 메모리 객체나 통신 방법을 이용한다.

Reference https://docs.python.org/3/library/multiprocessing.html#synchronization-between-processes

Value, Array

from multiprocessing import Process, current_process, Value, Array
import random
import os


def generate_update_number(v : int):
    for i in range(50):
        v.value += 1
    print(current_process().name, "data", v.value)

def main():
    # 부모 프로세스 아이디
    parent_process_id = os.getpid()
    # 출력
    print(f"Parent process ID {parent_process_id}")

    processes = list()

    share_value = Value('i', 0)
    for _ in range(1,10):
        # 생성
        p = Process(target=generate_update_number, args=(share_value,))
        # 배열에 담기
        processes.append(p)
        # 실행
        p.start()

    for p in processes:
        p.join()

    # 최종 프로세스 부모 변수 확인
    print("Final Data(share_value) in parent process",  share_value.value)

if __name__ == '__main__':
    main()

• Value, Array는 말 그대로 변수, 또는 리스트를 공유메모리 맵에 저장시킬수 있다.

  multiprocessing.Value(typecode_or_type, *args, lock=True)

• 공유 메모리에 할당된 ctypes객체를 반환

• 값에 대한 접근은 Value.value로 접근 가능

• typecode_or_type: 반환된 객체의 형을 결정

  • 

• lock: 액세스를 동기화하기 위한 Lock객체 생성. 동기화를 보장해준다.

multiprocessing.Array(typecode_or_type, size_or_initializer, *, lock=True)

• multiprocessing.Value와 거의 비슷하게 사용된다.
• size_or_initializer: 말 그대로 사이즈를 지정해주거나 초기화 객체를 입력한다.

Queue

from multiprocessing import Process, Queue, current_process
import time
import os

# 실행 함수
def worker(id, baseNum, q):

    process_id = os.getpid()
    process_name = current_process().name

    sub_total = 0

    for i in range(baseNum):
        sub_total += 1

    q.put(sub_total)

    print(f"Process ID: {process_id}, Process Name: {process_name}")
    print(f"Result : {sub_total}")


def main():
    processes = list()

    start_time = time.time()

    # Queue 선언
    q = Queue()

    for i in range(5):
        p = Process(name=str(i), target=worker, args=(1, 100000000, q))

        processes.append(p)
        p.start()

    # Join
    for process in processes:
        process.join()

    # 순수 계산 시간
    print("--- %s seconds ---" % (time.time() - start_time))

    # 종료 플래그
    q.put("exit")

    total = 0

    # 대기 상태
    while True:
        tmp = q.get()
        if tmp == "exit":
            break
        else:
            total += tmp

    print()

    print("Main-Processing Total_count={}".format(total))
    print("Main-Processing Done!")

if __name__ == '__main__':
    main()

• Queue는 Array와 비슷하지만 Queue 처리방식을 따른다.
• while문에서 대기 상태일 때, q.get()에서 멈춰있으므로 while문을 돌고있는 상태가 아니다. 덕분에 컴퓨팅 코스트의 낭비는 일어나지 않는다.
• queue.Queue의 클론에 가깝다.

Pipe

from multiprocessing import Process, Pipe, current_process
import time
import os

# 실행 함수
def worker(id, baseNum, conn):

    process_id = os.getpid()
    process_name = current_process().name

    sub_total = 0

    for _ in range(baseNum):
        sub_total += 1

    # Produce
    conn.send(sub_total)
    conn.close()

    # 정보 출력
    print(f"Result : {sub_total}")

def main():

    # 시작 시간
    start_time = time.time()

    # Pipe 선언
    parent_conn, child_conn = Pipe()

    p = Process(target=worker, args=(1, 100000000, child_conn))

    p.start()

    p.join()

    # 순수 계산 시간
    print("--- %s seconds ---" % (time.time() - start_time))

    print()

    print("Main-Processing : {}".format(parent_conn.recv()))
    print("Main-Processing Done!")

if __name__ == "__main__":
    main()

multiprocessing.Pipe([duplex])

• 파이프의 끝을 나타내는 Connection 객체 쌍 (conn1, conn2)를 반환

  Connection 객체 기본적으로 send(), recv()와 같이 송수신을 가진다. Reference: https://docs.python.org/ko/3/library/multiprocessing.html#multiprocessing.connection.Connection

• duplex가 True(default)면 양방향 통신이고, False이면 단방향 통신이다.
  • 단방향 통신이라면 conn1은 수신, conn2는 송신만 가능하다.

logging 활용

기본적으로 스레드는 디버깅이 어렵기 때문에 logging을 통해 실행 과정을 잘 표현하는 것이 중요하다.

import logging


# 기본 설정
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(levelname)s - %(message)s',
    filename='/path/to/logfile.log'
    datefmt='%Y-%m-%d %H:%M:%S'
    )

로그 레벨

DEBUG: 가장 낮은 레벨. 상세한 정보를 기록할 때 사용 INFO: 일반적인 정보를 기록할 때 사용 WARNING: 예상치 못한 일이 발생, 문제가 될만한 상황을 기록할 때 사용 ERROR: 프로그램의 일부 기능이 제대로 동작하지 않을 때 사용 CRITICAL: 프로그램 자체가 정상적으로 동작할 수 없을 때 사용

기본 실행

import logging
import threading

# 스레드 실행 함수
def thread_func(name):
    logging.info("Sub-Thread %s: starting", name)
    """
    Thread task code 
    """
    logging.info("Sub-Thread %s: finishing", name)


# 메인스레드 영역
if __name__ == "__main__":
    # Logging format 설정
    format = "%(asctime)s: %(message)s"
    logging.basicConfig(format=format, level=logging.INFO, datefmt="%H:%M:%S")
    logging.info("Main-Thread : before creating thread")

    # 함수 인자 확인
    x = threading.Thread(target=thread_func, args=("First",))

    logging.info("Main-Thread : before running thread")

    # 서브 스레드 시작
    x.start()

    logging.info("Main-Thread : wait for the thread to finish")

    # x.join()  # x 스레드 종료까지 메인스레드 대기

    logging.info("Main-Thread : all done")

Daemon Thread

메인 스레드 종료 시 서브스레드 함께 종료

import logging
import threading

# 스레드 실행 함수
def thread_func(name):
    logging.info("Sub-Thread %s: starting", name)
    """
    Thread task code 
    """
    logging.info("Sub-Thread %s: finishing", name)


# 메인스레드 영역
if __name__ == "__main__":
    # Logging format 설정
    format = "%(asctime)s: %(message)s"
    logging.basicConfig(format=format, level=logging.INFO, datefmt="%H:%M:%S")
    logging.info("Main-Thread : before creating thread")

    # 방법 1. 직접 daemon 옵션 추가
    x = threading.Thread(target=thread_func, args=("First",), daemon=True)
    # 방법 2. 속성 설정
    # x = .Thread(target=thread_func, args=("First",))
    # x.daemon = True

    logging.info("Main-Thread : before running thread")

    # 서브 스레드 시작
    x.start()

    logging.info("Main-Thread : wait for the thread to finish")

    logging.info("Main-Thread : all done")
    # 이 시점(메인스레드 종료)에서 서브스레드 강제 종료

ThreadPoolExcuter

import logging
from concurrent.futures import ThreadPoolExecutor

# 스레드 실행 함수
def task(name):
    logging.info("Sub-Thread %s: starting", name)
    """
    Thread task
    """
    logging.info("Sub-Thread %s: finishing result: %d", name, result)

    return task_result


# 메인 영역
def main():
    # Logging format 설정
    format = "%(asctime)s: %(message)s"
    logging.basicConfig(format=format, level=logging.INFO, datefmt="%H:%M:%S")

    logging.info("Main-Thread : before creating and running thread")

    """     방법1. 직접 스레드 추가     """
    # max_workers : 작업의 개수가 넘어가면 직접 설정이 유리
    executor = ThreadPoolExecutor(max_workers=3)

    task1 = executor.submit(task, ('First',))
    task2 = executor.submit(task, ('Second',))

    # 결과 값 있을 경우
    # print(task1.result())
    # print(task2.result())



    """     방법2. with 구문 사용     """
    with ThreadPoolExecutor(max_workers=3) as executor:
        # 곧바로 결과값 받아오기
        tasks = executor.map(task, ['First', 'Second'])

        # 결과값 리스트 확인
        # print(list(tasks))  

        # 스레드 객체 리스트 받기
        threads = [excuter.submit(task, idx) for idx in range(5)]

        # 스레드 객체를 통한 결과값 확인
        # print([t.result() for t in threads])

    logging.info("Main-Thread : all done")

if __name__ == '__main__':
    main()

ThreadPoolExecutor 내부 동작 과정

1. 작업 제출

• 사용자는 ThreadPoolExecutor의 submit 또는 map 메서드를 사용해 작업을 제출.
• 제출된 각 작업은 내부적으로 Future 객체로 변환.
  • Future 객체는 작업의 상태를 추적하고 결과를 저장
• Future 객체는 실행을 대기 중인 작업 목록을 관리하는 큐에 추가됨.

  2. 큐 관리

• ThreadPoolExecutor는 내부적으로 작업 대기열로서 큐를 사용*
• 스레드 풀의 스레드 중 하나가 큐에서 작업을 가져감. (이 과정은 스레드가 사용 가능할 때까지 자동)

  3. 작업 실행

• 스레드 풀의 스레드는 큐에서 가져온 작업을 실행.
• 작업이 완료되면 해당 작업에 연결된 Future 객체는 작업의 결과 또는 발생한 예외를 저장.

  4. 결과 반환

• 사용자는 Future 객체의 result 메서드를 호출하여 작업의 결과를 받음.
• 작업이 아직 완료되지 않았다면, result 메서드는 작업이 완료될 때까지 대기.

Discriptor Magic Method

디스크립터 클래스는 3개의 매직 메서드를 가지고 각 메서드는 공통적으로 self, obj 파라미터를 가진다.

• self: 디스크립터 클래스를 가리키는 파라미터
• obj: 디스크립터가 사용되는 객체(클래스)를 가리키는 파라미터

__get__(self, obj, type=None): 속성에 접근할 때 실행. type은 오브젝트의 타입을 말한다.

__set__(self, obj, value): 속성을 새 value 값으로 설정할 때 실행. __delete__(self, obj): 속성을 삭제할 때 실행.

class DescriptorEx1(object): 

    def __init__(self, name = 'Default'): 
        self.name = name 

    def __get__(self, obj, objtype): 
        return "Get method called. -> self : {}, obj : {}, objtype : {}, name : {}".format(self, obj, objtype, self.name) 

    def __set__(self, obj, name): 
        print('Set method called.')
        if isinstance(name, str): 
            self.name = name 
        else: 
            raise TypeError("Name should be string") 

    def __delete__(self, obj):
        print('Delete method called.')
        self.name = None

class Sample1(object): 
    name = DescriptorEx1()

s1 = Sample1() 

# __set__ 호출 
s1.name = "Descriptor Test1"

# s1.name = 7  # 예외 발생

# attr 확인
# __get__ 호출
print(s1.name)  # Get method called. -> self : <__main__.DescriptorEx1 object at 0x000001E0A5EC1390>, obj : <__main__.Sample1 object at 0x000001E0A5EC1BA8>, objtype : <class '__main__.Sample1'>, name : Descriptor Test1

# __delete__ 호출
del s1.name

# 재확인
# __get__ 호출
print(s1.name)  # Get method called. -> self : <__main__.DescriptorEx1 object at 0x000001E0A5EC1390>, obj : <__main__.Sample1 object at 0x000001E0A5EC1BA8>, objtype : <class '__main__.Sample1'>, name : None

property

property를 이용한 디스크립터 작성 방식도 있다.

참고

class property(fget=None, fset=None, fdel=None, doc=None)

class DescriptorEx2(object): 

    def __init__(self, value): 
        self._name = value 

    def getVal(self): 
        return "Get method called. -> self : {}, name : {}".format(self, self._name) 

    def setVal(self, value): 
        print('Set method called.')
        if isinstance(value, str): 
            self._name = value
        else: 
            raise TypeError("Name should be string") 

    def delVal(self):
        print('Delete method called.')
        self._name = None

    name = property(getVal, setVal, delVal, 'Property Method Example.')

메타클래스는 그 대상이 클래스가 된다고 생각하면 된다. 클래스가 어떻게 생성되고 동작할 지 정의하는 역할을 한다. 기본적으로 모든 파이썬 클래스는 type이라는 내장 메타클래스를 사용하여 생성됩니다.

즉, 메타클래스는 내가 의도하는 방향으로 클래스를 커스텀하는 것이다.

type

우리가 흔히 타입을 알기 위해 사용하는 type함수는 명확히 어떤 것을 반환하는 것인지 알고 넘어가자.

class SampleA(): # Class == Object
    pass

obj1 = SampleA() # 변수에 할당, 복사 가능, 새로운 속성, 함수의 인자로 넘기기 가능

# obj1 -> SampleA instance
# SampleA -> type metaclass
# type -> type metaclass
print(obj1.__class__)  # <class '__main__.SampleA'>
print(type(obj1))  # <class '__main__.SampleA'>
print('Ex1 > ', obj1.__class__ is type(obj1))  # True

코드를 보면 알 수 있듯이 type 함수가 반환하는 것은 해당 객체(인스턴스)의 __class__(본래의 원형)이다.

그럼 type함수에 본래의 클래스를 넣으면 어떻게 되는지 알면 메타클래스에 대해 알 수 있다.

print(type(SampleA))  # <class 'type'>

for cls in (int, str, float, dict, tuple):
    print(type(cls)))
"""
<class 'type'>
<class 'type'>
<class 'type'>
<class 'type'>
<class 'type'>
"""

위를 보면 알 수 있는 것이 모든 클래스의 원형은 type이라는 메타클래스인 것이다.

type을 이용한 클래스 동적 생성

tpye(Name(이름), Bases(상속), Dct(속성,메소드))

s1 = type('Sample1', (), {})

print('Ex1 > ', s1)  # '__main__.Sample1'
print('Ex1 > ', type(s1))  # <class 'type'>
print('Ex1 > ', s1.__base__)  # <class 'object'>
print('Ex1 > ', s1.__dict__)  # {'__module__': '__main__', '__dict__': <attribute '__dict__' of 'sample1' objects>, '__weakref__': <attribute '__weakref__' of 'sample1' objects>, '__doc__': None}

위 방식처럼 type을 이용하여 미리 정의된 클래스가 아니라 코드로 클래스가 동적으로 정의될 수 있다. type클래스의 인자 순서대로 클래스명, 상속받는 객체, 속성 및 메소드 를 받게된다.

특정 클래스를 상속

class Parent1:
    pass

s2 = type(
        'Sample2', 
        (Parent1,), 
        dict(attr1=100, attr2='hi')
    )

print(s2)  # <class '__main__.Sample2'>
print(type(s2))  # <class 'type'>
print(s2.__base__)  # <class '__main__.Parent1'>
print(s2.__dict__)  # {'attr1': 100, 'attr2': 'hi', '__module__': '__main__', '__doc__': None}
print(s2.attr1, s2.attr2)  # 100 hi

특정 메소드 할당

class SampleEx:  
    attr1 = 30
    attr2 = 100

    def add(self, m, n):
        return m + n

    def mul(self, m, n):
        return m * n


s3 = type(
        'Sample3', 
        (object, ), # 생략 가능
        dict(attr1=30, attr2=100, add=lambda x, y: x + y, mul=lambda x, y: x * y)
    )

위에서 s3와 SmapleEx의 인스턴스는 완전히 동일한 기능을 한다.

Type Checking

정적 타입 검사 (Static Type Checking)

특징

• 컴파일 시간에 타입 검사
• 프로그램이 실행되기 전에* 컴파일 시간에 변수의 타입 검사한다.
• 타입 선언 필요성 대부분의 정적 타입 언어에서는 변수를 선언할 때 타입을 명시해야 한다. (C++, Java, Swift 등)
• 타입 안정성 높은 타입 안정성을 제공한다. 타입 오류가 프로그램이 실행되기 전에 발견되어 실행 시간에 발생할 수 있는 예기치 않은 동작이나 충돌을 예방할 수 있다.
• 최적화와 성능 컴파일러는 타입 정보를 사용하여 코드 최적화를 수행할 수 있기 때문에 프로그램의 실행 속도가 대체로 빠르다.

장점

• 오류 감지 실행 전에 많은 오류를 감지할 수 있어 안정성이 높다.
• 성능 최적화 컴파일 시 타입 정보를 사용하여 최적화할 수 있다.

  단점

  유연성 부족, 초기 개발 속도가 느리다 정도가 있는데 사실 정적 타입 검사가 아니더라도 고려해야 하는 부분이라 그렇게 단점은 아니다.

동적 타입 검사 (Dynamic Type Checking)

특징

• 런타임 시 타입 검사 런타임(프로그램이 실행되는 동안)에 변수의 타입을 검사합니다. (Python, JavaScript, Ruby)
• 타입 선언의 유연성 변수를 선언할 때 타입을 명시할 필요가 없고 런타임에 할당된 값에 따라 타입이 결정된다.
• 런타임 오류 타입 오류는 프로그램 실행 중에만 감지될 수 있으므로 오류 발견이 늦어질 수 있다.
• 유연성과 편의성 코드 작성이 더 간결하고 유연해져서 빠른 개발과 프로토타이핑에 유리하다.

정적 타입 선언과 반대되는 장단점을 가지고 있다. 특히 런타임 에러가 나는건 정말 피곤한 단점이다.

오버라이딩이 부모클래스 상속과 관련된 개념을 다뤘다면 오버로딩은 부모클래스 상관없이 메소드 간 파라미터를 다룬다.

Overloading

같은 클래스 내에서 동일한 이름의 메소드를 여러 개 가지면서 각각의 메소드가 다른 매개변수를 갖도록 하는 기술이다. 이런 메소드들은 매개변수의 수나 타입에 따라 구별된다.

오버로딩은 정적 타입 체킹의 한 예시다. 컴파일러는 메소드를 호출할 때 제공된 인자에 기반하여 어떤 메소드를 실행할지 결정하게 된다.

하지만 파이썬에서는 기본적으로 오버로딩을 지원하지 않는다! 예를 들어,

class SampleA():
    def add(self, x, y):
        return x + y

    def add(self, x, y, z):
        return x + y + z

a = SampleA()
print(a.add(1,2))  # Error!

위 처럼 파이썬에서는 동일 이름의 메소드가 있을 때 가장 마지막으로 선언된 함수로 정의된다.

그래서 기본적으로는 디폴트 인자를 통해 인자 갯수에 대응하거나

def add(a, b, c=0):
    return a + b + c

가변 인자를 사용하여 가변 길이의 인자를 받을 수 있다.

def add(*args):
    return sum(args)

아니면 데이터 타입을 체킹하고 그에 따른 리턴을 만들 수도 있다.

def add(self, datatype, *args):
    if datatype =='int': 
        return sum(args)

    if datatype =='str': 
        return ''.join([x for x in args])

multipledispatch.dispatch

외부 라이브러리인 multipledispach를 통해 오버로딩을 편하게 도와주는 데코레이터 dispatch를 이용하면 가장 편하다.

from multipledispatch import dispatch

class SampleC():

    @dispatch(int,int) 
    def product(x, y): 
        return x * y 

    @dispatch(int,int,int) 
    def product(x, y, z): 
        return x * y * z

    @dispatch(float,float,float) 
    def product(x, y, z): 
        return x * y * z

c = SampleC()

print('Ex3 > ', c.product(5, 6))  # 30
print('Ex3 > ', c.product(5, 6, 7))  # 210
print('Ex3 > ', c.product(5.0, 6.0, 7.0))  # 210.0

기존에는 마지막 함수만이 정의되어 에러가 났지만 dispatch를 통해 같은 함수에 다양한 자료를 받을 수 있게 만들어 준다. 특히 데이터 타입에 의해 함수가 분류되는걸 확인할 수 있다는 점에서 가독성에도 좋다.

자식 클래스가 부모 클래스로부터 상속받은 메소드를 재정의(override)하는 것을 의미한다. 이 때 자식 클래스는 상속받은 메소드와 동일한 이름, 매개변수를 가지지만, 구체적인 실행 내용은 변경할 수 있다.

Method Overriding의 목적

• 확장성 (Extensibility): 자식 클래스는 부모 클래스의 기능을 확장하거나 수정할 수 있기 떄문에 코드 재사용성과 유지보수가 용이하다.
• 다형성 (Polymorphism): 같은 인터페이스나 클래스 계층 구조에 속하는 객체들이 다양한 방식으로 작동할 수 있다.

특징

• 메소드 시그니쳐: 오버라이딩 할 메소드는 부모 클래스의 메소드와 동일한 이름, 매개변수 리스트를 가져야 한다.
• 접근 제어: 오버라이딩된 메소드는 부모 클래스의 메소드보다 접근성이 더 제한적이면 안된다.
• 리턴 타입: 자식 클래스의 메소드가 부모 클래스의 메소드와 동일하거나 더 구체적인 리턴 타입을 가져야 한다.
• super 키워드: 자식 클래스에서는 super키워드를 사용해서 부모 클래스의 메소드를 호출할 수 있다.

상속 방식

부모 클래스로부터 상속을 받을 때 인스턴스화가 되는 시점에 부모로부터 상속을 받는다.

class ParentEx1():
    def __init__(self):
        self.value = 5

    def get_value(self):
        return self.value

class ChildEx1(ParentEx1):
    pass

c1 = ChildEx1()
p1 = ParentEx1()


# 부모 & 자식 모든 속성 출력
print('Ex1 > ', dir(ParentEx1))
print('Ex1 > ', dir(ChildEx1))
"""
Ex1 >  ['__class__', '__delattr__', '__dict__', '__dir__', '__doc__', '__eq__', '__format__', '__ge__', '__getattribute__', '__gt__', '__hash__', '__init__', '__init_subclass__', '__le__', '__lt__', '__module__', '__ne__', '__new__', '__reduce__', '__reduce_ex__', '__repr__', '__setattr__', '__sizeof__', '__str__', '__subclasshook__', '__weakref__', 'get_value', 'value']
Ex1 >  ['__class__', '__delattr__', '__dict__', '__dir__', '__doc__', '__eq__', '__format__', '__ge__', '__getattribute__', '__gt__', '__hash__', '__init__', '__init_subclass__', '__le__', '__lt__', '__module__', '__ne__', '__new__', '__reduce__', '__reduce_ex__', '__repr__', '__setattr__', '__sizeof__', '__str__', '__subclasshook__', '__weakref__', 'get_value']
"""


# 부모 & 자식 인스턴스 속성 출력
print('Ex1 > ', ParentEx1.__dict__)
print('Ex1 > ', ChildEx1.__dict__)
"""
Ex1 >  {'__module__': '__main__', '__init__': <function ParentEx1.__init__ at 0x0000028A15F81840>, 'get_value': <function ParentEx1.get_value at 0x0000028A15F818C8>, '__dict__': <attribute '__dict__' of 'ParentEx1' objects>, '__weakref__': <attribute '__weakref__' of 'ParentEx1' objects>, '__doc__': None}
Ex1 >  {'__module__': '__main__', '__doc__': None}
"""

• 부모 & 자식의 모든 속성을 출력하면 둘의 속성이 동일하게 출력되는 것이 확인된다.
  • 자세히 보면 자식클래스에는 value가 없다. value는 부모 클래스의 __init__ 메소드가 실행되면 생성되는 것이므로 메소드만 물려받은 자식 클래스에는 없는 것이 당연하다.
• 부모 & 자식의 인스턴스 속성을 출력하면 자식클래스의 인스턴스는 클래스라는 것 외에는 거의 가진것이없다. 이것은 자식클래스가 실제로 상속받는 것은 인스턴스화가 될 때 받는다는 것을 알 수 있다.

Overriding

class ParentEx2():
    def __init__(self):
        self.value = 5

    def get_value(self):
        return self.value

class ChildEx2(ParentEx1):
    def get_value(self):
        return self.value * 10


c2 = ChildEx2()

print('Ex2 > ', c2.get_value())  # 50

위와 같은 방식으로 부모클래스의 메소드를 재정의하여 같은 메소드로 다른 동작방식을 정의할 수 있다.

단! 위에 적었던 오버라이딩의 특징을 유념한 채 재정의 해야한다!

다형성(Polymorphism)

다형성에 대해 보기에 앞서 super()함수에 대해 알아야 한다.

super()

자식클래스에서 부모 클래스를 호출할 때 사용된다.

• 인자가 있을 때:

  super(자식클래스, 자식클래스의 인스턴스) 형태로 사용한다.
  • super(Child, self).__init__():

  • Child 클래스의, self 인스턴스의 부모클래스가 가진 __init__메소드를 호출

• 인자가 없을 때:

  자동으로 현재 클래스와 인스턴스를 참조
  • super().__init__:

• 현재 클래스(자식)의 인스턴스의 부모클래스가 가진 __init__메소드를 호출

예시

import datetime

class Logger(object):  # 로그메세지를 출력하는 부모클래스
    def log(self, msg):
        print(msg)

class TimestampLogger(Logger):  # 로그메세지와 시간까지 출력하는 자식클래스 
    def log(self, msg):
        message = "{ts} {msg}".format(ts=datetime.datetime.now(),
                                      msg=msg)
        # super().log(message)
        super(TimestampLogger, self).log(message)

class DateLogger(Logger):  # 더 정확한 시간표시를 해주는 자식클래스
    def log(self, msg):
        message = "{ts} {msg}".format(ts=datetime.datetime.now().strftime('%Y-%m-%d'),
                                      msg=msg)
        # super().log(message)
        super(DateLogger, self).log(message)

l = Logger()
t = TimestampLogger()
d = DateLogger()


l.log("Called logger.")
t.log("Called timestamp logger.")
d.log("Called date logger.")
"""
Called logger.
2024-01-21 10:16:52.064729 Called timestamp logger.
2024-01-21 Called date logger.
"""

underscore _은 파이썬에서 자주 볼 수 있고 다양하게 사용된다.

주로 인터프리터, 값 무시, 네이밍에 사용된다.

값 무시

값 무시는 unpacking을 할 때 무시하는 방식을 말한다.

x, _, y = (1, 2, 3)
print(x, y)  # 1, 3

a, *_, b = (1, 2, 3, 4, 5)
print(a, b)  # 1, 5

a, *i, b = (1, 2, 3, 4, 5)  # unpacking
print(a, b, i)  # 1, 5, [2, 3, 4]

for _ in range(10):
    pass

접근 지정자

일반적으로 선언하는 변수는 public하게 선언한다는 의미이다. 언제든 수정할 수 있다. 하지만 underscore을 어떻게 사용하여 변수를 선언하냐에 따라 그 의미가 달라진다.

_variable: 일반적으로 내부적으로 사용되는 변수나 함수를 나타내는 데 사용한다. 파이썬의 강제적인 규칙은 아니지만 널리 사용되는 약속이다.(사용하면 좋다는 뜻)

• _internal_function()이 있다면, 함수가 내부적으로만 사용되며 외부에서는 사용하지 않기를 권장한다는 의미. 즉, 사용할 순 있지만 강제적이진 않다

__variable__: 파이썬의 특수한 메소드나 속성을 나타낸다. "매직 메소드"라고 주로 부른다. 예를 들어, __init__, __str__, __len__ 등이 있고 context manager의 __enter__, __exit__도 있다.

__variable: 클래스의 속성을 이름 맹글링(name mangling)을 통해 해당 클래스 내에서만 사용할 수 있도록 하는 데 사용된다. 이는 클래스의 상속 구조에서 속성 이름 충돌을 방지하기 위해 사용된다.

Name mangling

파이썬은 underscore을 통해 public, private 속성을 나타내지만 실제로 접근을 완전히 제한하지 않고 수정하려면 수정할 수 있다. __variable이 제한하는 수준은 다음 정도이다.

class TestClass:
    def __init__(self):
        self.name = Son
        self.__age = 30

test = TestClass()
print(test.name)  # "Son"
print(test.__age)  # AttributeError: 'TestClass' object has no attribute '__age'

print(dir(test))
"""
['_TestClass__age', '__class__', '__delattr__', '__dict__', '__dir__', 
'__doc__', '__eq__', '__format__', '__ge__', '__getattribute__', '__gt__', 
'__hash__', '__init__', '__init_subclass__', '__le__', '__lt__', '__module__',
 '__ne__', '__new__', '__reduce__', '__reduce_ex__', '__repr__', '__setattr__',
 '__sizeof__', '__str__', '__subclasshook__', '__weakref__', 'name']
"""

위 결과처럼 __variable은 _ClassName__variable로 바뀌는 것이다. 따라서 접근만 어렵고 수정하려면 할 수 있다.

굳이 이렇게 하는 이유

1.캡슐화 강화: (완벽하진 않더라도) private 속성을 가지기 위함 2.상속 안전성: 클래스 간 상속 시 부모 클래스의 변수를 오버라이딩하는 것을 막아줌 (_ParentClassName__variable은 하나밖에 없을테니까)

다 뜻이 있다

Getter, Setter

getter와 setter는 객체의 속성에 대한 접근 및 수정 방법을 제어하는 메소드이다.

Getter 메소드

Getter는 객체의 특정 속성 값을 검색하는 데 사용된다. 일반적으로 해당 속성에 대한 값을 반환하고 추가적인 계산이나 처리를 수행할 수도 있습니다. @property 데코레이터를 사용하여 getter 메소드를 정의할 수 있다.

Setter 메소드

Setter는 객체의 속성 값을 설정하거나 수정하는 데 사용된다. 이 메소드는 새로운 값을 받아 해당 속성에 할당하기 전에 유효성 검사나 추가적인 처리를 수행할 수 있다. 파이썬에서는 @value.setter 데코레이터를 사용하여 setter 메소드를 정의한다.

class MyClass:
    def __init__(self, value):
        self._value = value

    @property
    def value(self):
        # Getter 메소드
        return self._value

    @value.setter
    def value(self, new_value):
        # Setter 메소드 - 여기서 추가적인 유효성 검사를 수행할 수 있음
        if new_value < 0:
            raise ValueError("Value cannot be negative")
        self._value = new_value

cls = MyClass()
print(cls.value)  # getter
cls.value = 10  # setter
cls.value = -5  # setter

위를 보면 일반적인 public을 선언하는것과 쓰임새가 달라지는 것이 거의 없어보인다. 실제로 과도한 getter, setter은 가독성을 해치기도 한다. 그래서 위와 같은 유효성 검사를 할 때 사용하는 것이 일반적인 사용방식이다.

간단하게는 다음과 같은 코드가 있다.

file = open('./test.txt', 'w')
try:
    file.write('test text\nend.')
finally:
    file.close()

이렇게 file.close()를 해서 자원을 돌려줘야 OS는 다시 자원을 다른 곳에 할당해줄 수 있다.

흔히 사용하는 다음 코드는 자동으로 자원을 반환한다.

with open('./test.txt', 'w') as file:
    file.write('test text\nend.')

Context Manager

컨텍스트 매니저는 리소스 관리를 위한 구조로, 주로 파일, 네트워크 연결, 데이터베이스 세션과 같은 리소스의 할당 및 해제를 자동화하는 데 사용된다. 위에서 볼 수 있듯이 with 문과 함께 사용되며, 이를 통해 코드의 가독성과 안정성을 높일 수 있다.

컨텍스트 매니저는 __enter__, __exit__ 두 개의 매직 메서드를 구현한다.

__enter__

with 문이 시작될 때 실행된다. 여기서 리소스를 할당하거나 초기화하는 코드를 작성할 수 있고, with 문과 함께 사용되는 변수에 할당될 값을 반환할 수 있다.

__exit__

with 문이 끝날 때 실행된다. 인자로 (예외 유형, 예외 값, 백트레이스) 세 개의 인자를 받는다. 여기서는 리소스의 정리 작업을 수행하고, 예외가 발생한 경우 이 메서드 내에서 처리할 수 있으며, 예외를 무시하려면 True를 반환하면 된다.

class MyFileWriter():
    def __init__(self, filename, method):
        print("init")
        self.file_obj = open(file_name, method)
    def __enter__(self):
        print("enter")
        return self.file_obj
    def __exit__(self, exc_type, value, trace_back):
        print("exit")
        if exc_type:
            print("exc!")
        self.file_obj.close()

# Custom
with MyFileWriter('./test.txt', 'w') as f:
    f.write('test text\nend.')

# Origin
with open('./test.txt', 'w') as file:
    file.write('test text\nend.')

위 코드에서 보듯이 context manager을 통해 커스텀한 방식으로 파일을 다룰 수 있다.

만약 어떤 함수의 실행시간을 측정하려 할 때, time 함수 대신 다음 context manager 객체를 만들 수 있다.

import time

class ExcuteTimer():
    def __init__(self, msg):
        self._msg = msg

    def __enter__(self):
        self._start = time.monotonic()
        return self._start

    def __exit(self, exc_type, exc_val, exc_traceback)
        if exc_type:
            print("exception!")
        else:
            print(f'{self._msg}: {time.monotonic() - self._start} sec'
        return True


with ExcuteTimer("function excute time") as v:
    print(v)  # __enter__ 메서드의 반환값
    # 시간을 측정하고 싶은 함수나 코드 . . .
    for i in range(10**8):
        pass

Context Manager Decorator

파이썬의 표준 라이브러리 중 하나이다. 위에서 매직 메서드를 이용하여 구현하던 부분을 데코레이터 형식으로 지원하여 좀 더 쉽게 context manager를 구현할 수 있다.

import contextlib
import time


@contextlib.contextmanager
def ExcuteTimerDe(msg):
    start = time.monotonic()
    try:
        yield start  # __enter__
    except BaseException as e:
        print(f"logging: {e}")
    else:  # __exit__
        print(f"{msg} : {time.monotonic() - start} sec")


with ExcuteTimerDe("function excute time") as v:
    print(v)
    for i in range(10**8):
        pass
    # raise ValueError("error msg")가 있고 이 부분이 발생한다면 ExcuteTimerDe의 에러문구로 출력된다.

이것은 위에서 클래스로 선언한 타이머와 완전히 동일한 기능을 한다.

contextlib.contextmanager을 데코레이터로 한 함수는 yield의 시작을 __enter__, 이후의 코드를 __exit__로 구분한다.

a = 100
b = a
b = 10
print(a)  # 100

a = [100]
b = a
b[0] = 10
pritn(a)  # [10]

a = (100,)
b = a
b = 10
print(a)  # (100)

위의 원리는 하나다. 좌변은 변수, 우변은 참조 객체이다. 참조 객체(a)가

1. 불변 객체(int, str, float, tuple 등)일 때 변수(b)가 새로운 객체를 참조받는다면 변수는 새로운 객체를 생성하고 참조한다.
2. 가변 객체(list, dict 등)일 때 변수(b)가 새로운 객체를 참조받는다면 변수는 동일한 객체의 참조를 공유한다.

오해하지 않아야 하는 점은 가변 객체는 내부의 요소를 변경할 수 있다라는 점이다.

a = [100]
b = a
b = 10
print(a)  # [100]

여기서 b = 10을 하는 순간 list 요소가 아닌 list자체를 바꾸는 것이므로 b는 10이라는 새로운 객체를 할당받게 되는 것이다.

이런 기본 원리를 알아야 copy에서의 실수가 없다.

Shallow Copy

해석 그대로 얕은 복사를 의미한다.

import copy

a = [10, 20, 30]
b = copy.copy(a)
print(id(a), id(b))  # 140456783782272, 140573327194048
b.append(40)
print(a)  # [10, 20, 30, 40]

언뜻보면 b = a를 했을 때와 같아 보이지만 주소값이 다른 것을 확인할 수 있다.

b = a 일 때는 두 주소값이 같았기에 이것이 의미하는건 변수명만 다르고 같은 주소를 참조하는 것이다. 반면 b = copy.copy(a)를 통해 shallow copy를 하면 a와 별개의 객체를 가리키는 b가 생성되지만 그 내부의 요소들은 같은 a,b가 공유되고 있다.

이 때문에 복사는 하지만 얕은 복사라 한다.

Deep Copy

깊은 복사를 의미하고 완전히 같은 형태의 새로운 객체를 복사한다. 앞의 복사와 달리 완전한 별개의 객체가 생성된다. 당연히 그만큼 메모리가 사용되므로 큰 사이즈의 객체를 다룰 때 조심히 사용해야 한다.

import copy

a = [10, 20, 30]
b = copy.deepcopy(a)
print(a, b)  # 140456783782272, 140573327194048
b.append(40)
print(a)  # [10, 20, 30]
print(b)  # [10, 20, 30, 40]

lambda

lambda는 일회성 함수로 여겨지고 재사용할 함수가 아니라면 선언하는 편이다. 힙 영역에서 사용 즉시 소멸되어 메모리를 절약할 수 있다. pythonic한 코드를 작성하는데도 도움을 주고 파이썬의 가비지 컬렉션에서 Count가 0으로 취급되어 이런 부분도 좋다.

다음과 같이 선언하는 방법도 있다.

f = lambda a, b, c: a + b * c

print(f(1,2,3))  # 7

Map

iterable 객체에서 각 원소에게 동일한 함수를 적용시킬 때 사용한다. map(원소마다 적용할 함수, iterable 데이터)이고 map object를 반환한다. 그리고 Lazy Evaluation을 사용하여 메모리 최적화와 성능 향상에 도움을 준다.

Lazy Evaluation(지연 계산)

계산 결과가 실제로 필요한 순간 전까지는 생성되지 않는 방식이다. 만약 즉시 계산된다면 해당 결과 데이터가 모두 메모리에 저장되므로 큰 데이터 집합을 다룰 때 메모리 문제가 발생할 수 있다.

arr = [i for i in range(1,11)]

result = map(lambda x: x**2, arr)  # result는 map 오브젝트
result2 = list(result)  # 리스트 형태
print(result2)  # [100, 400, 900, ..., 10000]

Filter

map()과 마찬가지로 적용할 함수, iterable 데이터를 입력받고, 이 때 적용할 함수는 bool값을 반환하도록 한다. True인 원소만을 반환하는 filter 객체를 반환한다.

arr = [i for i in range(1,11)]

result = map(lambda x: x%2==0, arr)  # result는 map 오브젝트
result2 = list(result)  # 리스트 형태
print(result2)  # [2, 4, 6, 8, 10]

이렇게 함수를 작성하다보면 변수 영역이 상당히 중요하다. 전역 변수인지, 로컬 변수인지, 데코레이터 형식에서 사용되는 로컬 변수는 어떻게 다룰지 등등..

중요한건 몰라도 잘 구현하지만 좋은 코드가 아니라는 점이다. 어떻게 보면 간단하고 명확하게 알지 않아도 구현하는데 어려움이 없는 개념이기에 지나치기 쉽고 안좋은 코딩 방식이 고착화될 가능성이 높다고 생각되는 부분이다.

때문에 변수의 영역에 대해 깊지 않지만 정확하게 짚기 위한 포스터를 작성해본다.

Variable scope

Global, Local Variable

전역 변수는 특정 함수나 클래스가 아닌 스크립트 영역에 선언된 변수로 어느 영역에서든 접근이 가능하다.

지역 변수는 당연히 해당 영역에서만 사용 가능하다. 다른 언어에서는 전역, 지역 변수의 이름이 겹친다면 지역 후 전역 변수를 사용하지만 파이썬에서는 global이라는 명확한

var = 10

print(var)  # 10

def fn():
    var = 20
    print(var)  # 20

def gn():
    print(var)  # Error

def foo():
    global var
    var += 100
    print(var)  # 110

global로 가져온 변수는 전역변수를 읽고 수정까지 할 수 있지만 pythonic한 코드도 아니고 권장되는 코드 스타일도 아니다. 전역 변수를 특정 부분에서 수정한다면 다른 곳에서 전역 변수를 사용할 때 영향을 받기 때문이다. 전역 변수는 주로 변하지 않는 고정 값에 사용하는 것이 옳다.

개인적으로 global을 쓴 경험이 있긴했다. FastAPI에서 딥러닝 모델을 사용하는데 GPU에 단 한 번만 올려놓아야 하는 상황이였다.

1. model=None을 전역변수로 선언하고
2. FastAPI가 실행될 때 model=DLModel()로 초기화
3. 다른 코드에서 해당 모델을 가져와서 사용

Closure

Closure 형식에서는 nonlocal이 사용될 수 있다. 스코프 내의 하위 스코프가 있을 때, 하위 스코프가 상위 스코프의 지역 변수를 사용할 수 있는 방법이다.

Closure

어떤 함수의 내부 함수가 외부 함수의 변수를 참조할 때, 외부 함수가 종료되어도 내부 함수가 외부 함수의 변수를 참조할 수 있도록 어딘가에 저장하는 함수

조건

• 어떤 함수의 내부 함수일 것
• 내부 함수가 외부 함수의 변수를 참조할 것
• 외부 함수가 내부 함수를 리턴할 것

def outer(number):
    a = number

    def inner():
        nonlocal a
        a += 200
        print(a)

    return inner

f = outer(100)
f()  # 300
f()  # 500
f()  # 700
f()  # 900

변수에 접근하려면 locals, globals가 있다. 각각 지역변수, 전역변수를 딕셔너리 형태로 저장하고 있고 key값은 str타입으로 저장되어 있다.

def fn(number):
    x = 10

    def printt():
        print("Inner func")
        print(locals())
    return printt

func = fn(100)

func()
# Inner func
# {'number': 100, 'x': 10, 'printt': <function fn.<locals>.printt at 0x00000193734C1840>}

여기서 알 수 있듯이 함수의 인자는 지역 변수로 들어간다.

globals도 마찬가지이고 파이썬 내부적으로 다음 두 코드는 같은 코드로 볼 수 있다.

a = 100

globals()['a'] = 100

머신러닝 모델을 사용하여 한 번에 대량의 데이터에 대해 예측을 수행하는 과정 데이터,모델 저장소에서 배치 서빙 파이프라인에 넣어주고 이를 통해 예측값을 얻는다.

• 데이터 저장소의 데이터는 주로 특정 시간 단위로 모아진 데이터와 같이 같은 종류, 다른 환경에서의 데이터들의 집합이다.

배치 서빙 외에도 대표적인 세 가지 서빙 파이프라인이 있다.

서빙 종류

기본 아키텍쳐

배치 서빙 코드

데이터 준비

# [make_batch_data.py]

from datetime import datetime
from sklearn.datasets import load_iris
from minio import Minio


#
# dump data
#
iris = load_iris(as_frame=True)
X = iris["data"]
X.sample(100).to_csv("batch.csv", index=None)

#
# minio client
#
url = "localhost:9000"
access_key = "minio"
secret_key = "miniostorage"
client = Minio(url, access_key=access_key, secret_key=secret_key, secure=False)

#
# upload data to minio
#
bucket_name = "not-predicted"
object_name = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

if not client.bucket_exists(bucket_name):
    client.make_bucket(bucket_name)

client.fput_object(bucket_name, object_name, "batch.csv")

• redis와 같은 캐시메모리를 사용하지 않고 간단하게 볼 예정이므로 버킷명(not_predicted)구분한다.
• 데이터 구분을 위해 데이터명은 생성 시간으로 구분

  윈도우에선 파일명에 :가 포함될 수 없으므로 시간을 저장할 때 -로 바꾸어야 한다

모델 불러오기

# [local_predict.py]
import os

import mlflow
import pandas as pd
from minio import Minio

os.environ["MLFLOW_S3_ENDPOINT_URL"] = "http://localhost:9000"
os.environ["MLFLOW_TRACKING_URI"] = "http://localhost:5001"
os.environ["AWS_ACCESS_KEY_ID"] = "minio"
os.environ["AWS_SECRET_ACCESS_KEY"] = "miniostorage"


def predict(run_id, model_name):
    #
    # load model: mlflow의 모델을 불러오기
    #
    clf = mlflow.pyfunc.load_model(f"runs:/{run_id}/{model_name}")

    #
    # minio client
    #
    url = "localhost:9000"
    access_key = "minio"
    secret_key = "miniostorage"
    client = Minio(url, access_key=access_key, secret_key=secret_key, secure=False)

    #
    # get data list to predict: 예측할 데이터 불러오기
    #
    if "predicted" not in client.list_buckets():
        # 최초 실행시 predicted bucket 생성
        client.make_bucket("predicted")
    # 추론이 안된 데이터 추출
    predicted_set = set(objects.object_name for objects in client.list_objects(bucket_name="predicted"))
    to_predict_list = [
        objects.object_name 
        for objects in client.list_objects(bucket_name="not-predicted") 
        if objects.object_name not in predicted_set
        ]
    print(to_predict_list)
    #
    # predict
    #
    for filename in to_predict_list:
        print("data to predict:", filename)
        # download and read data
        client.fget_object(bucket_name="not-predicted", object_name=filename, file_path=filename)
        data = pd.read_csv(filename)

        # predict
        pred = clf.predict(data)

        # save to minio prediction bucket
        pred_filename = f"pred_{filename}"
        pred.to_csv(pred_filename, index=None)
        client.fput_object(bucket_name="predicted", object_name=filename, file_path=pred_filename)


if __name__ == "__main__":
    from argparse import ArgumentParser

    parser = ArgumentParser() # 스크립트 실행 시 인자 받아오기
    parser.add_argument("--run-id", type=str)
    parser.add_argument("--model-name", type=str, default="my_model")
    args = parser.parse_args()

    #
    # predict
    #
    predict(args.run_id, args.model_name)

mlflow 모델 불러오기

로컬에서 설계한 모델이 아닌 mlflow 서버에 저장된(실제로는 minio와 같은 스토리지에 저장된) 모델 불러오기

1. 모델 다운로드

import os

import mlflow

os.environ["MLFLOW_S3_ENDPOINT_URL"] = "http://localhost:9000"
os.environ["MLFLOW_TRACKING_URI"] = "http://localhost:5001"
os.environ["AWS_ACCESS_KEY_ID"] = "minio"
os.environ["AWS_SECRET_ACCESS_KEY"] = "miniostorage"


if __name__ == "__main__":
    from argparse import ArgumentParser

    parser = ArgumentParser() # 스크립트 실행 시 인자 받아오기
    parser.add_argument("--run-id", type=str)
    parser.add_argument("--model-name", type=str, default="my_model")
    args = parser.parse_args()

    mlflow.artifacts.download_artifacts(run_id=args.run_id, artifact_path=args.model_name, dst_path="./downloads")

mlflow.artifacts.download_artifacts를 이용하여 모델을 원하는 경로로 다운로드

2. Dockerfile 작성 후 빌드

# [Dockerfile]

FROM amd64/python:3.9-slim

WORKDIR /usr/app/

#
# 모델을 불러오기 위한 패키지 다운로드
#
RUN pip install -U pip &&\
    pip install mlflow==2.3.2 minio==7.1.15

#
# 모델을 실행하기 위한 패키지 다운로드 (캐싱을 위한 분리)
#
COPY requirements.txt requirements.txt
RUN pip install -r requirements.txt

#
# 모델 다운로드
#
COPY downloads/ downloads/

COPY model_predict.py predict.py

#
# 도커 명령어로 파일을 실행하기 위한 요약
#
ENTRYPOINT [ "python", "predict.py", "--run-id" ]

[Build]

$ docker build -t <image_name> <dockerfile_path>

3. 컨테이너 실행

로컬에서 mlflow, minio가 docker-compose로 실행된 상태라면 해당 docker network 포함시킨다

$ docker run --network <network-name> <image_name> <run_id>

• 실험 관리+파일: 학습 데이터, 패키지, 파라미터 등을 함께 저장

MLflow 아키텍쳐

Backend Store (Remote host)

• 수치 데이터와 MLflow 서버의 정보들을 체계적으로 관리하기 위한 DB
• 저장 항목: 메타 데이터, 모델 정보, 학습 중 생기는 정보

Artifact Store(S3 remote host)

• 학습된 모델을 저장하는 Model Registry로써 이용하기 위한 스토리지 서버
• 기본적인 파일 시스템보다 체계적으로 관리할 수 있으며 외부에 있는 스토리지 서버도 사용할 수 있다는 장점이 있다.

docker-compose

version: "3"  
services:
  mlflow-artifact-store:
    image: minio/minio  # MinIO 공식 Docker 이미지
    ports:  
      - 9000:9000 
      - 9001:9001 
    environment: 
      MINIO_ROOT_USER: minio
      MINIO_ROOT_PASSWORD: miniostorage
    command: server /data/minio --console-address :9001  # MinIO 서버를 실행할 명령어. 데이터 저장 폴더와 콘솔 주소를 설정.
    healthcheck:  # 서비스의 상태를 확인하는 방법을 정의.
      test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"] 
      interval: 30s  # 헬스체크 수행 간격
      timeout: 20s  # 헬스체크 타임아웃 시간
      retries: 3  # 헬스체크 실패 시 재시도 횟수

  mlflow-backend-store:  # 두 번째 서비스: MLflow의 메타데이터 저장소로 사용되는 PostgreSQL 데이터베이스.
    image: postgres:14.0
    environment:
      POSTGRES_USER: mlflowuser
      POSTGRES_PASSWORD: mlflowpassword
      POSTGRES_DB: mlflowdatabase
    healthcheck:  # PostgreSQL 서비스의 상태를 확인하는 방법을 정의.
      test:
        ["CMD", "pg_isready", "-q", "-U", "mlflowuser", "-d", "mlflowdatabase"]  # 데이터베이스의 준비 상태를 확인하는 명령어.
      interval: 10s
      timeout: 5s 
      retries: 5 

  mlflow-server:  # 세 번째 서비스: MLflow 서버.
    build:
      context: .
      dockerfile: Dockerfile
    depends_on:  
      mlflow-artifact-store:
        condition: service_started
      mlflow-backend-store:
        condition: service_healthy
    ports:  
      - 5001:5000  
    environment:  
      AWS_ACCESS_KEY_ID: minio  # AWS 스타일의 S3 저장소에 접근하기 위한 액세스 키로, 여기서는 MinIO에서 사용.
      AWS_SECRET_ACCESS_KEY: miniostorage 
      MLFLOW_S3_ENDPOINT_URL: http://mlflow-artifact-store:9000  # MLflow에서 사용할 S3 엔드포인트의 URL
    command:  
      - /bin/sh
      - -c
      - |  # 여러 줄에 걸친 명령을 표시하기 위한 YAML 문법.
        mc config host add mlflowminio http://mlflow-artifact-store:9000 minio miniostorage &&  # MinIO 클라이언트를 설정
        mc mb --ignore-existing mlflowminio/mlflow  # 이미 존재하지 않는 경우에만 MinIO 버킷 생성.
        mlflow server \  # MLflow 서버 시작.
        --backend-store-uri postgresql://mlflowuser:mlflowpassword@mlflow-backend-store/mlflowdatabase \  # PostgreSQL을 백엔드 저장소로 사용.
        --default-artifact-root s3://mlflow \  # 아티팩트의 기본 저장 위치로 S3 버킷을 지정
        --host 0.0.0.0  # 모든 IP에서 서버에 접근할 수 있도록 설정.

코드

데이터 업로드

import pandas as pd
from sklearn.datasets import load_iris
from minio import Minio
from minio.versioningconfig import VersioningConfig, ENABLED

#
# dump data
#
iris = load_iris(as_frame=True)
X, y = iris["data"], iris["target"]
data = pd.concat([X, y], axis="columns")
data.sample(100).to_csv("iris.csv", index=None)

#
# minio client
#
url = "localhost:9000"
access_key = "minio"
secret_key = "miniostorage"
client = Minio(url, access_key=access_key, secret_key=secret_key, secure=False)

#
# upload data to minio
#
bucket_name = "raw-data"
object_name = "iris"
if not client.bucket_exists(bucket_name):
    client.make_bucket(bucket_name)
    config = client.set_bucket_versioning(bucket_name, VersioningConfig(ENABLED))

client.fput_object(bucket_name, object_name, "iris.csv")

모델 학습 및 저장

import os
import uuid

import optuna
import mlflow
import pandas as pd
from minio import Minio
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score

UNIQUE_PREFIX = str(uuid.uuid4())[:8]  # 유니크한 프리픽스를 생성하여 실행별로 구분
BUCKET_NAME = "raw-data"  # MinIO 버킷 이름 설정
OBJECT_NAME = "iris"  # MinIO 오브젝트 이름 설정

# MinIO 접속을 위한 환경변수 설정
os.environ["MLFLOW_S3_ENDPOINT_URL"] = "http://localhost:9000"
os.environ["MLFLOW_TRACKING_URI"] = "http://localhost:5001"
os.environ["AWS_ACCESS_KEY_ID"] = "minio"
os.environ["AWS_SECRET_ACCESS_KEY"] = "miniostorage"

def download_data():
    # MinIO 클라이언트 인스턴스 생성
    url = "localhost:9000"
    access_key = "minio"
    secret_key = "miniostorage"
    client = Minio(url, access_key=access_key, secret_key=secret_key, secure=False)

    # 데이터 다운로드
    object_stat = client.stat_object(BUCKET_NAME, OBJECT_NAME)
    data_version_id = object_stat.version_id
    client.fget_object(BUCKET_NAME, OBJECT_NAME, file_path="download_data.csv")
    return data_version_id

def load_data():
    # 다운로드된 데이터 로딩
    data_version_id = download_data()
    df = pd.read_csv("download_data.csv")
    X, y = df.drop(columns=["target"]), df["target"]
    data_dict = {"data": X, "target": y, "version_id": data_version_id}
    return data_dict

def objective(trial):
    # 새로운 파라미터 제안
    trial.suggest_int("n_estimators", 100, 1000, step=100)
    trial.suggest_int("max_depth", 3, 10)

    run_name = f"{UNIQUE_PREFIX}-{trial.number}"  # 실행 이름 설정
    with mlflow.start_run(run_name=run_name):
        # 제안된 파라미터를 로깅
        mlflow.log_params(trial.params)

        # 데이터 로딩
        data_dict = load_data()
        mlflow.log_param("bucket_name", BUCKET_NAME)
        mlflow.log_param("object_name", OBJECT_NAME)
        mlflow.log_param("version_id", data_dict["version_id"])
        X, y = data_dict["data"], data_dict["target"]
        X_train, X_valid, y_train, y_valid = train_test_split(X, y, test_size=0.3, random_state=2024)

        # 모델 학습
        clf = RandomForestClassifier(
            n_estimators=trial.params["n_estimators"], max_depth=trial.params["max_depth"], random_state=2024
        )
        clf.fit(X_train, y_train)

        # 학습된 모델 평가
        y_pred = clf.predict(X_valid)
        acc_score = accuracy_score(y_valid, y_pred)

        # 평가 결과 로깅
        mlflow.log_metric("accuracy", acc_score)
    return acc_score

def train_best_model(params):
    run_name = f"{UNIQUE_PREFIX}-best-model"  # 최적 모델 실행 이름 설정
    with mlflow.start_run(run_name=run_name):
        # 파라미터 로깅
        mlflow.log_params(params)

        # 데이터 로딩
        data_dict = load_data()
        mlflow.log_param("bucket_name", BUCKET_NAME)
        mlflow.log_param("object_name", OBJECT_NAME)
        mlflow.log_param("version_id", data_dict["version_id"])
        X, y = data_dict["data"], data_dict["target"]

        # 최적의 파라미터로 모델 학습
        clf = RandomForestClassifier(
            n_estimators=params["n_estimators"], max_depth=params["max_depth"], random_state=2024
        )
        clf.fit(X, y)

        # 학습된 모델 저장
        mlflow.sklearn.log_model(sk_model=clf, artifact_path="my_model")
        return clf

if __name__ == "__main__":
    # MLflow 실험 설정
    study_name = "hpo-tutorial"
    mlflow.set_experiment(study_name)

    # Optuna 연구 생성 및 설정
    sampler = optuna.samplers.RandomSampler(seed=2024)
    study = optuna.create_study(sampler=sampler, study_name=study_name, direction="maximize")

    # 최적화 실행
    study.optimize(objective, n_trials=5)

    # 최적의 파라미터로 최적 모델 학습 및 저장
    best_params = study.best_params
    best_clf = train_best_model(best_params)

모델 불러오기

import os

import mlflow
import pandas as pd
from minio import Minio


BUCKET_NAME = "raw-data"
OBJECT_NAME = "iris"

os.environ["MLFLOW_S3_ENDPOINT_URL"] = "http://localhost:9000"
os.environ["MLFLOW_TRACKING_URI"] = "http://localhost:5001"
os.environ["AWS_ACCESS_KEY_ID"] = "minio"
os.environ["AWS_SECRET_ACCESS_KEY"] = "miniostorage"


def download_data():
    # Minio 클라이언트 객체를 생성
    url = "localhost:9000"
    access_key = "minio"
    secret_key = "miniostorage"
    client = Minio(url, access_key=access_key, secret_key=secret_key, secure=False)

    # Minio 서버로부터 데이터를 다운로드
    # stat_object 메소드로 객체의 메타데이터를 얻어오고 버전 ID를 획득
    object_stat = client.stat_object(BUCKET_NAME, OBJECT_NAME)
    data_version_id = object_stat.version_id
    # fget_object 메소드로 데이터를 로컬 파일로 저장합니다.
    client.fget_object(BUCKET_NAME, OBJECT_NAME, file_path="download_data.csv")
    return data_version_id


def load_data():
    # 데이터를 다운로드하고 pandas DataFrame으로 로드
    data_version_id = download_data()
    df = pd.read_csv("download_data.csv")
    X, y = df.drop(columns=["target"]), df["target"]
    # 데이터와 메타데이터를 포함하는 딕셔너리를 반환
    data_dict = {"data": X, "target": y, "version_id": data_version_id}
    return data_dict


def load_sklearn_model(run_id, model_name):
    # MLflow를 사용하여 저장된 Scikit-Learn 모델을 로드
    clf = mlflow.sklearn.load_model(f"runs:/{run_id}/{model_name}")
    return clf


def load_pyfunc_model(run_id, model_name):
    # MLflow의 PyFunc 인터페이스를 통해 모델을 로드
    # Scikit-Learn 모델이라도 일관된 방식으로 사용
    clf = mlflow.pyfunc.load_model(f"runs:/{run_id}/{model_name}")
    return clf


if __name__ == "__main__":
    from argparse import ArgumentParser

    # 커맨드 라인 인자를 파싱하기 위한 ArgumentParser를 생성
    parser = ArgumentParser()
    parser.add_argument("--run-id", type=str)
    parser.add_argument("--model-name", type=str, default="my_model")
    args = parser.parse_args()

    # 데이터를 로드
    data_dict = load_data()
    X = data_dict["data"]

    # Scikit-Learn을 통해 모델을 로드하고 예측을 수행
    sklearn_clf = load_sklearn_model(args.run_id, args.model_name)
    sklearn_pred = sklearn_clf.predict(X)
    print("sklearn")
    print(sklearn_clf)
    print(sklearn_pred)

    # PyFunc를 통해 모델을 로드하고 예측을 수행
    pyfunc_clf = load_pyfunc_model(args.run_id, args.model_name)
    pyfunc_pred = pyfunc_clf.predict(X)
    print("pyfunc")
    print(pyfunc_clf)
    print(pyfunc_pred)

Custom Model

모델뿐만 아니라 모델의 전처리, 후처리 프로세스를 지난 후 결과값이 나올 때, 모든 프로세스를 묶어서 하나의 모델로 지정하는 것

이전 코드에서 모델의 예측결과는 각 아이리스의 종류에 매핑되어 있는 숫자로 표현되어 있다. 이러한 숫자를 다시 아이리스 종류로 매핑한 결과를 post process로 설정하려면,

1. 클래스로 선언 후
2. mlflow.pyfunc를 이용한다.

mlflow.pyfunc

MLflow 프레임워크에서 제공하는 기능 중 하나로, 다양한 머신러닝 라이브러리로 생성된 모델들을 파이썬 함수(pythonic function)처럼 다룰 수 있게 해주는 모듈이다.

• 사용자는 모델을 훨씬 쉽게 배포하고 호출할 수 있다
• MLflow 플랫폼 상에서 모델의 일관된 인터페이스를 가질 수 있다.
• PyFunc은 MLflow의 "Flavor" 중 하나로, 모델을 MLflow와 호환 가능한 형식으로 저장하고 로드하는 표준 방식을 제공한다.

mlflow.pyfunc의 주요 함수

mlflow.pyfunc.load_model()

• 저장된 PyFunc 모델을 로드하여 Python 함수로 사용.
• [설명] (https://mlflow.org/docs/latest/python_api/mlflow.pyfunc.html#mlflow.pyfunc.load_model)

mlflow.pyfunc.log_model()

• 현재 MLflow 실행(run)에 PyFunc 모델을 로깅하여 MLflow 서버에 모델을 저장
• [설명] (https://mlflow.org/docs/latest/python_api/mlflow.pyfunc.html#mlflow.pyfunc.log_model)

mlflow.pyfunc.save_model()

• PyFunc 모델을 파일 시스템에 저장.
• [설명] (https://mlflow.org/docs/latest/python_api/mlflow.pyfunc.html#mlflow.pyfunc.save_model)

커스텀 모델 선언

커스텀 모델 클래스 선언

class MyModel:
    def __init__(self, clf):
        self.clf = clf

    def predict(self, X):
        X_pred = self.clf.predict(X)
        X_pred_df = pd.Series(X_pred).map({0: "virginica", 1: "setosa", 2: "versicolor"})
        return X_pred_df

• map()을 통해 후처리 과정을 붙인다.

train_best_model(params) 커스텀 - 모델 저장

def train_best_model(params):
        .
        .
        #
        # my custom model: 커스텀 모델 클래스 불러오기
        #
        my_model = MyModel(clf)
        #
        # save model
        #
        with open("model.dill", "wb") as f:
            # 커스텀 모델을 'model.dill' 파일에 직렬화하여 저장 
            dill.dump(my_model, f)

        # '_load_pyfunc' 함수를 정의하는 새로운 'loader.py' 스크립트 파일을 작성
        # 저장된 모델을 로드하는 데 사용. textwrap.dedent로 앞부분의 공통 들여쓰기를 제거
        with open("loader.py", "w") as f:
            f.write(
                textwrap.dedent(
                    """
                    import os
                    import dill

                    def _load_pyfunc(path):
                        if os.path.isdir(path):
                            path = os.path.join(path, "model.dill")

                        with open(path, "rb") as f:
                            return dill.load(f)
                    """
                )
            )

        # MLflow를 사용하여 'my_model'이라는 아티팩트 경로에 모델을 로그 
        # 모델 데이터는 'model.dill'에 있으며, 'loader' 모듈은 모델을 로드하는 데 사용
        # 'loader.py'는 로더 모듈에 필요한 의존성을 포함하는 코드 파일
        mlflow.pyfunc.log_model(
            artifact_path="my_model",
            data_path="model.dill",
            loader_module="loader",
            code_path=["loader.py"],
        )
        return clf

커스텀 모델 불러오기

이전 모델의 출력 결과물

base) (mlops-py3.9) PS C:\Users\wlsgy\Desktop\MLOps\05_model_registry> python section1_load_model.py --run-id 90beba76af604fcfa98a90a20fe47d1f

Downloading artifacts: 100%|█████████████████████████████████████████████████████████████████████████████████████████████████████████████| 5/5 [00:00<00:00, 156.08it/s]
pyfunc
mlflow.pyfunc.loaded_model:
  artifact_path: my_model
  flavor: mlflow.sklearn
  run_id: 90beba76af604fcfa98a90a20fe47d1f

[0 1 1 1 1 1 1 0 0 2 1 2 0 2 0 1 0 0 0 1 0 2 1 2 0 1 2 2 2 0 0 0 1 0 2 1 1
 1 2 0 2 1 0 0 1 0 2 0 2 2 1 0 0 2 0 2 2 0 1 0 0 1 2 2 1 2 2 1 0 0 0 2 1 0
 1 2 1 2 2 2 2 2 1 2 2 1 2 0 2 1 0 1 0 2 1 1 0 2 2 0]

커스텀 모델의 출력 결과물

 (base) (mlops-py3.9) PS C:\Users\wlsgy\Desktop\MLOps\05_model_registry> python section2_load_model.py --run-id 184a78ab15094237b340403160005b1e
Downloading artifacts: 100%|████████████████████████████████████████████████████████████████████████████████████████████████████████████| 6/6 [00:00<00:00, 2803.99it/s]
pyfunc
mlflow.pyfunc.loaded_model:
  artifact_path: my_model
  flavor: loader
  run_id: 184a78ab15094237b340403160005b1e

0      virginica
1         setosa
2         setosa
3         setosa
4         setosa
         ...    
95        setosa
96     virginica
97    versicolor
98        setosa
99     virginica
Length: 100, dtype: object

즉, 커스텀 모델을 관리하려면

1. 클래스 선언을 통해 커스텀 모델 클래스를 생성한다.
2. mlflow.pyfunc와 dill등을 통해 커스텀 모델을 mlflow에 로깅하도록 한다.
3. 모델을 불러와서 추론의 결과값을 얻는다.

의 순서로 진행한다.

머신러닝을 학습시킬 때의 데이터와 현실에서의 데이터는 차이가 있다.

• 데이터를 학습하는 동안에도 현실의 데이터 분포가 바뀐다.
• 학습한 모델을 서비스하는 동안에도 현실의 데이터 분포가 바뀐다.

컨셉 드리프트(Concept Drift)

• 예측 모델이 학습한 대상 변수의 조건부 분포가 변경되는 경우
• 예를 들어, 소비자의 구매 패턴이 시간에 따라 변할 수 있으며, 이로 인해 과거에 학습된 패턴이 더 이상 유효하지 않게 될 수 있다.

데이터 분포 드리프트(Data Distribution Drift)

• 입력 데이터의 마진 분포가 변경되었지만, 조건부 분포는 동일하게 유지되는 경우
• 특정 입력 변수의 범위나 분포가 시간에 따라 변화하지만, 출력 변수와의 관계는 일정한 경우

라벨 드리프트(Label Drift)

• 출력 변수 자체의 분포가 변하는 현상
• 예를 들어, 질병의 발병률이나 고객의 선호도 같은 것이 시간에 따라 달라질 수 있다.

시즌성 드리프트(Seasonal Drift)

• 계절성 요인으로 인해 일시적으로 데이터의 분포가 변하는 경우
• 예를 들어, 날씨 변화에 따른 의류 판매량의 변동, 휴일 시즌에 따른 소비 패턴의 변화 등이 있다.

MINIO

• 고성능, 고가용성, 클라우드 네이티브 환경을 위해 설계된 오픈 소스 객체 스토리지 솔루션
• AWS S3(Simple Storage Service)의 클라우드 스토리지 서비스와 유사한 API를 제공.
• 분산 시스템 환경에서도 확장 가능하며, 개인 클라우드 또는 공개 클라우드 인프라에서의 배포를 지원.

특징

• 고성능: 멀티 코어 CPU에 최적화. 고성능의 스루풋과 낮은 지연 시간을 제공.
• 확장성: 클러스터링을 통해 확장할 수 있고 수십 페타바이트 규모의 데이터를 저장하고 관리 가능.
• 간편성: 간단한 설치와 설정으로 빠르게 배포가 가능. 컨테이너화하여 Docker, Kubernetes와 같은 오케스트레이션 시스템에서 운영할 수 있습니다.
• 호환성: AWS S3와 호환되는 API를 제공하여, S3를 사용하는 어플리케이션을 변경 없이 MinIO로 마이그레이션할 수 있는 경로를 제공.
• 다양한 워크로드 지원: AI/ML 워크로드, 데이터 분석, 백업 및 아카이빙, 웹사이트 호스팅 등 다양한 스토리지 요구 사항을 지원.
• 보안: 기본적으로 데이터 암호화를 지원. 클라이언트 측 암호화, 서버 측 암호화(SSE)를 모두 지원하며, TLS/SSL을 통한 데이터 전송 시 암호화도 지원합니다.

MinIO는 간단한 바이너리 파일로 배포되며 고가용성과 함께 객체 스토리지 솔루션을 제공하기 위해 설계된 라이브러리와 툴을 포함하고 있다.

기본 사용법

1. 도커 컨테이너로 실행

$ docker run -p 9000:9000 -p 9001:9001 --name minio1 \
  -e "MINIO_ROOT_USER=<username>" \
  -e "MINIO_ROOT_PASSWORD=<password>" \
  -v /mnt/data:/data \
  minio/minio server /data --console-address ":9001"

• -p 9000:9000: 호스트의 9000 포트를 컨테이너의 9000 포트에 매핑. MinIO는 기본적으로 9000 포트에서 서비스를 제공.
• -p 9001:9001: 호스트의 9001 포트를 컨테이너의 MinIO의 새로운 관리 콘솔 포트에 매핑. 이 콘솔은 최신 버전의 MinIO에서 관리 및 모니터링 인터페이스를 제공.
• --name minio1: 실행 중인 컨테이너의 이름을 minio1로 설정.
• -e "MINIO_ROOT_USER=<username>": MinIO의 접근 키(유저 ID) 환경 변수를 설정
• -e "MINIO_ROOT_PASSWORD=<password>": MinIO의 비밀 키(패스워드) 환경 변수를 설정.
• -v /mnt/data:/data: 호스트 시스템의 /mnt/data 디렉터리를 컨테이너의 /data 디렉터리에 볼륨으로 마운트. MinIO가 데이터를 저장하는 위치이다.
• minio/minio server /data: minio/minio 공식이미지를 통해 MinIO 서버를 시작하는 커맨드. /data 디렉터리를 사용하여 객체 데이터를 저장.
• --console-address ":9001": MinIO의 관리 콘솔을 포트 9001에서 사용할 수 있게 설정.

2. 키 발급

1. 컨테이너 url로 접속 후 설정한 아이디, 비밀번호로 로그인한다.
2. User/Access Keys 란에서 키 발급 후 json 파일 저장

   {
        "url":"http://localhost:9001/api/v1/service-account-credentials",
        "accessKey":"QDBS194oK7vbYR516knJ",
        "secretKey":"1bikmj3guo2z5Ej05NCsf9GZhhFAUQLXbjCrRZi1",
        "api":"s3v4",
        "path":"auto"
   }

   • Access Key: 외부에서 사용할 때는 아이디, 패스워드 유출을 막기 위해 임시적인 액세스 키를 통해 접근한다.

3. Object Browser

Object Browser에 들어가여 버킷을 생성하면 다음과 같은 UI가 뜬다. 옆의 설명을 해석해보면,

• MinIO는 버킷을 사용하여 개체를 구성하고 파일시스템의 디렉토리 구조와 유사하다.
• Versioning: 이 기능을 통해 동일 객체의 여러 버전을 관리 가능하다.
• Locking: 객체 삭제 방지.
• Quota: 버킷의 데이터 양 제한
• Resuming: 이정 기간동안 객체 삭제 방지를 하기 위한 규칙 지정.

AWS S3와 호환되는만큼 아주 유사한 구조이다.

4. 데이터 업로드, 다운로드

업로드 코드

import pandas as pd
from sklearn.datasets import load_iris
# minio 패키지에서 Minio 클래스를 임포트. MinIO 서버와 상호작용하기 위한 클라이언트 인터페이스를 제공.
from minio import Minio
# minio 패키지에서 버전 관리 설정을 위한 VersioningConfig 클래스와 상수 ENABLED를 임포트.
from minio.versioningconfig import VersioningConfig, ENABLED

#############      데이터 준비      #############
# Iris 데이터셋을 로드하여 pandas 데이터프레임으로 변환.
iris = load_iris(as_frame=True)
# 데이터 프레임에서 feature 값(X)과 타겟 값(y)을 추출.
X, y = iris["data"], iris["target"]
# Features 데이터프레임(X)과 target 시리즈(y)를 하나의 데이터프레임으로 결합.
data = pd.concat([X, y], axis="columns")
# 데이터프레임에서 무작위로 100개의 샘플을 선택하고 'iris.csv' 파일로 저장. 인덱스는 저장X.
data.sample(100).to_csv("iris.csv", index=None)


#############      minio클라이언트      #############
# MinIO 클라이언트 객체를 생성. MinIO 서버의 URL과 접근 키, 비밀 키를 설정하고, 보안 연결(HTTPS)을 사용하지 않음.
url = "0.0.0.0:9000"
access_key = "minio"
secret_key = "miniostorage"
client = Minio(url, access_key=access_key, secret_key=secret_key, secure=False)


#############      업로드 코드      #############
# MinIO의 버킷 'raw-data'가 존재하는지 확인하고, 없으면 새로 생성.
bucket_name = "raw-data"
object_name = "iris"
if not client.bucket_exists(bucket_name):
    client.make_bucket(bucket_name)
    # 버킷에 대한 버전 관리를 활성화.
    config = client.set_bucket_versioning(bucket_name, VersioningConfig(ENABLED))

# 'iris.csv' 파일을 'raw-data' 버킷의 'iris' 오브젝트 이름으로 MinIO 서버에 업로드.
client.fput_object(bucket_name, object_name, "iris.csv")

다운로드 코드

object_stat = client.stat_object(bucket_name, object_name)
print(object_stat.version_id)
client.fget_object(bucket_name, object_name, file_path="download_data.csv")

• 클라이언트 선언은 업로드 부분과 같다.

MINIO, MLflow 기본 구조

import uuid

import mlflow
import optuna
import pandas as pd
from minio import Minio
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score


UNIQUE_PREFIX = str(uuid.uuid4())[:8]
BUCKET_NAME = "raw-data"
OBJECT_NAME = "iris"


# 데이터를 MinIO에서 다운로드하는 함수
def download_data():
    # MinIO 클라이언트 설정
    url = "0.0.0.0:9000"
    access_key = "minio"
    secret_key = "miniostorage"
    client = Minio(url, access_key=access_key, secret_key=secret_key, secure=False)

    # MinIO에서 데이터 오브젝트의 메타데이터를 가져오고, 데이터의 버전 ID를 추출
    object_stat = client.stat_object(BUCKET_NAME, OBJECT_NAME)
    data_version_id = object_stat.version_id
    # MinIO에서 데이터 파일을 로컬 시스템으로 다운로드
    client.fget_object(BUCKET_NAME, OBJECT_NAME, file_path="download_data.csv")
    # 데이터 버전 ID 반환
    return data_version_id

# 데이터 로딩과 전처리를 위한 함수
def load_data():
    # 다운로드 함수를 호출하여 데이터와 데이터의 버전 ID를 가져옴
    data_version_id = download_data()
    # CSV 파일을 데이터프레임으로 로딩
    df = pd.read_csv("download_data.csv")
    # 독립변수와 종속변수를 분리하여 X와 y에 할당
    X, y = df.drop(columns=["target"]), df["target"]
    # 데이터와 레이블, 버전 ID를 포함하는 딕셔너리 생성
    data_dict = {"data": X, "target": y, "version_id": data_version_id}
    return data_dict



def objective(trial):
    .
    .
    #
    # load data
    #
    data_dict = load_data()
    mlflow.log_param("bucket_name", BUCKET_NAME)
    mlflow.log_param("object_name", OBJECT_NAME)
    mlflow.log_param("version_id", data_dict["version_id"])
    X, y = data_dict["data"], data_dict["target"]
    .
    .

def train_best_model(params):...


if __name__ == "__main__":
    # MLflow 실험 설정
    experiment_name = "hpo-tutorial"
    mlflow.set_tracking_uri("http://0.0.0.0:5001")
    mlflow.set_experiment(experiment_name)

    # Optuna 실험 설정, 랜덤 샘플러 사용
    sampler = optuna.samplers.RandomSampler(seed=2024)
    # Optuna 스터디 생성
    study = optuna.create_study(sampler=sampler, study_name=experiment_name, direction="maximize")
    # 스터디 최적화 실행
    study.optimize(objective, n_trials=5)

    # 최적의 하이퍼파라미터를 가진 모델을 훈련
    best_params = study.best_params
    best_clf = train_best_model(best_params)