|
|
@ -52,6 +52,7 @@ class SignalGenerator(): |
|
|
self.uturn_ids = ids['uturn_ids'] |
|
|
self.uturn_ids = ids['uturn_ids'] |
|
|
self.coord_ids = ids['coord_ids'] |
|
|
self.coord_ids = ids['coord_ids'] |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# 1. 데이터 준비 |
|
|
# 1. 데이터 준비 |
|
|
def prepare_data(self): |
|
|
def prepare_data(self): |
|
|
print("1. 데이터를 준비합니다.") |
|
|
print("1. 데이터를 준비합니다.") |
|
|
@ -66,11 +67,13 @@ class SignalGenerator(): |
|
|
self.prepare_auxiliaries() |
|
|
self.prepare_auxiliaries() |
|
|
self.time15 = datetime.now() |
|
|
self.time15 = datetime.now() |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# 1-1. 네트워크 불러오기 |
|
|
# 1-1. 네트워크 불러오기 |
|
|
def load_networks(self): |
|
|
def load_networks(self): |
|
|
self.net = sumolib.net.readNet(os.path.join(self.path_networks, 'sn.net.xml')) |
|
|
self.net = sumolib.net.readNet(os.path.join(self.path_networks, 'sn.net.xml')) |
|
|
print("1-1. 네트워크가 로드되었습니다.") |
|
|
print("1-1. 네트워크가 로드되었습니다.") |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# 1-2. 테이블 불러오기 |
|
|
# 1-2. 테이블 불러오기 |
|
|
def load_tables(self): |
|
|
def load_tables(self): |
|
|
if self.config_name == 'draft': |
|
|
if self.config_name == 'draft': |
|
|
@ -197,6 +200,7 @@ class SignalGenerator(): |
|
|
self.plan.at[i, f'dura_A{j}'] = row[f'dura_B{j}'] |
|
|
self.plan.at[i, f'dura_A{j}'] = row[f'dura_B{j}'] |
|
|
print("1-2. 테이블들이 로드되었습니다.") |
|
|
print("1-2. 테이블들이 로드되었습니다.") |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# 1-3. 네트워크 무결성 검사 |
|
|
# 1-3. 네트워크 무결성 검사 |
|
|
def check_networks(self): |
|
|
def check_networks(self): |
|
|
# https://sumo.dlr.de/docs/Netedit/neteditUsageExamples.html#simplify_tls_program_state_after_changing_connections |
|
|
# https://sumo.dlr.de/docs/Netedit/neteditUsageExamples.html#simplify_tls_program_state_after_changing_connections |
|
|
@ -218,6 +222,7 @@ class SignalGenerator(): |
|
|
traci.close() |
|
|
traci.close() |
|
|
print("1-3. 네트워크의 모든 clean state requirement들을 체크했습니다.") |
|
|
print("1-3. 네트워크의 모든 clean state requirement들을 체크했습니다.") |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# 1-4. 테이블 무결성 검사 |
|
|
# 1-4. 테이블 무결성 검사 |
|
|
def check_tables(self): |
|
|
def check_tables(self): |
|
|
self.check_history() |
|
|
self.check_history() |
|
|
@ -226,6 +231,7 @@ class SignalGenerator(): |
|
|
# self.check_moves() # 이동류번호에 대한 무결성검사 필요하나 아직 작성하지 않음. (24. 2. 5 화) |
|
|
# self.check_moves() # 이동류번호에 대한 무결성검사 필요하나 아직 작성하지 않음. (24. 2. 5 화) |
|
|
print("1-4. 테이블들의 무결성 검사를 완료했습니다.") |
|
|
print("1-4. 테이블들의 무결성 검사를 완료했습니다.") |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# 1-4-1. 신호이력(history) 검사 |
|
|
# 1-4-1. 신호이력(history) 검사 |
|
|
def check_history(self): |
|
|
def check_history(self): |
|
|
# 1-4-1-1. inter_no 검사 |
|
|
# 1-4-1-1. inter_no 검사 |
|
|
@ -252,6 +258,7 @@ class SignalGenerator(): |
|
|
if invalid_inter_nos: |
|
|
if invalid_inter_nos: |
|
|
msg = f"1-4-1-3. 음수이거나 200보다 큰 현시시간이 존재합니다. : {invalid_inter_nos}" |
|
|
msg = f"1-4-1-3. 음수이거나 200보다 큰 현시시간이 존재합니다. : {invalid_inter_nos}" |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# 1-5. 보조 딕셔너리, 데이터프레임, 리스트 등 만들기 |
|
|
# 1-5. 보조 딕셔너리, 데이터프레임, 리스트 등 만들기 |
|
|
def prepare_auxiliaries(self): |
|
|
def prepare_auxiliaries(self): |
|
|
# inter2node : a dictionary that maps inter_no to the node_id |
|
|
# inter2node : a dictionary that maps inter_no to the node_id |
|
|
@ -346,7 +353,8 @@ class SignalGenerator(): |
|
|
self.time22 = datetime.now() |
|
|
self.time22 = datetime.now() |
|
|
self.make_hrhists() |
|
|
self.make_hrhists() |
|
|
self.time23 = datetime.now() |
|
|
self.time23 = datetime.now() |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# 2-1. rhistory |
|
|
# 2-1. rhistory |
|
|
def make_rhistory(self): |
|
|
def make_rhistory(self): |
|
|
# 1. 조회시점의 유닉스 타임 이전의 신호이력 수집 |
|
|
# 1. 조회시점의 유닉스 타임 이전의 신호이력 수집 |
|
|
@ -398,6 +406,7 @@ class SignalGenerator(): |
|
|
self.rhistory['start_unix'] = self.rhistory['start_unix'].astype(int) |
|
|
self.rhistory['start_unix'] = self.rhistory['start_unix'].astype(int) |
|
|
self.rhistory = self.rhistory[['inter_no', 'start_unix'] + [f'dura_{alph}{j}' for alph in ['A', 'B'] for j in range(1,9)] + ['cycle']] |
|
|
self.rhistory = self.rhistory[['inter_no', 'start_unix'] + [f'dura_{alph}{j}' for alph in ['A', 'B'] for j in range(1,9)] + ['cycle']] |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def load_prow(self, inter_no, time): |
|
|
def load_prow(self, inter_no, time): |
|
|
''' |
|
|
''' |
|
|
load the planned row |
|
|
load the planned row |
|
|
@ -418,6 +427,7 @@ class SignalGenerator(): |
|
|
|
|
|
|
|
|
return program_start, prow |
|
|
return program_start, prow |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# 2-2. rhists |
|
|
# 2-2. rhists |
|
|
def make_rhists(self): |
|
|
def make_rhists(self): |
|
|
self.rhists = [] |
|
|
self.rhists = [] |
|
|
@ -516,6 +526,7 @@ class SignalGenerator(): |
|
|
self.rhists = pd.concat(self.rhists)#.sort_values(by=['start_unix','inter_no']) |
|
|
self.rhists = pd.concat(self.rhists)#.sort_values(by=['start_unix','inter_no']) |
|
|
self.rhists = self.rhists[self.rhists.start_unix >= self.present_time - self.subtractor // 2] |
|
|
self.rhists = self.rhists[self.rhists.start_unix >= self.present_time - self.subtractor // 2] |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def calculate_DS(self, rhist, curr_unix): |
|
|
def calculate_DS(self, rhist, curr_unix): |
|
|
# program_starts = np.array(self.timetable.start_seconds) |
|
|
# program_starts = np.array(self.timetable.start_seconds) |
|
|
# idx = (program_starts <= self.present_time).sum() - 1 |
|
|
# idx = (program_starts <= self.present_time).sum() - 1 |
|
|
@ -543,6 +554,7 @@ class SignalGenerator(): |
|
|
S_n = S_n_durs.values.sum() // 2 |
|
|
S_n = S_n_durs.values.sum() // 2 |
|
|
return D_n, S_n |
|
|
return D_n, S_n |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# 2-3. hrhists |
|
|
# 2-3. hrhists |
|
|
def make_hrhists(self): |
|
|
def make_hrhists(self): |
|
|
# 계층화된 형태로 변환 |
|
|
# 계층화된 형태로 변환 |
|
|
@ -578,6 +590,7 @@ class SignalGenerator(): |
|
|
self.hrhists['start_unix'] = self.hrhists['start_unix'].astype(int) |
|
|
self.hrhists['start_unix'] = self.hrhists['start_unix'].astype(int) |
|
|
# self.hrhists = self.hrhists.sort_values(by = ['start_unix', 'inter_no', 'phas_A', 'phas_B']).reset_index(drop=True) |
|
|
# self.hrhists = self.hrhists.sort_values(by = ['start_unix', 'inter_no', 'phas_A', 'phas_B']).reset_index(drop=True) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# 3. 이동류정보 전처리 |
|
|
# 3. 이동류정보 전처리 |
|
|
def process_movement(self): |
|
|
def process_movement(self): |
|
|
print("3. 이동류정보 테이블을 변환합니다.") |
|
|
print("3. 이동류정보 테이블을 변환합니다.") |
|
|
@ -586,6 +599,7 @@ class SignalGenerator(): |
|
|
self.update_movement() |
|
|
self.update_movement() |
|
|
self.time32 = datetime.now() |
|
|
self.time32 = datetime.now() |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# 3-1. movement |
|
|
# 3-1. movement |
|
|
def make_movement(self): |
|
|
def make_movement(self): |
|
|
if self.config_name == 'draft' or self.config_name == 'test_0721': |
|
|
if self.config_name == 'draft' or self.config_name == 'test_0721': |
|
|
@ -628,21 +642,56 @@ class SignalGenerator(): |
|
|
self.movement = movement |
|
|
self.movement = movement |
|
|
|
|
|
|
|
|
elif self.config_name == 'test_0731': |
|
|
elif self.config_name == 'test_0731': |
|
|
self.movement = pd.read_csv(os.path.join(self.path_tables, 'TL_IF_SIGL.csv')) |
|
|
|
|
|
self.movement = self.movement.drop(columns=['FRST_REG_DT', 'RINGA_FLOW', 'RINGB_FLOW']) |
|
|
|
|
|
self.movement = self.movement.rename(columns={ |
|
|
|
|
|
|
|
|
phases = pd.read_csv(os.path.join(self.path_tables, 'TL_IF_SIGL.csv')) |
|
|
|
|
|
phases = phases.drop(columns=['FRST_REG_DT', 'RINGA_FLOW', 'RINGB_FLOW']) |
|
|
|
|
|
phases = phases.rename(columns={ |
|
|
'PHASE_DT':'start_unix','CRSRD_ID':'inter_no', |
|
|
'PHASE_DT':'start_unix','CRSRD_ID':'inter_no', |
|
|
'RINGA_PHASE':'phas_A', 'RINGB_PHASE':'phas_B', |
|
|
'RINGA_PHASE':'phas_A', 'RINGB_PHASE':'phas_B', |
|
|
'MAP_MODE':'STOS_NO' |
|
|
'MAP_MODE':'STOS_NO' |
|
|
}) |
|
|
}) |
|
|
isp2move_A = self.isp2move['A'] |
|
|
isp2move_A = self.isp2move['A'] |
|
|
isp2move_B = self.isp2move['B'] |
|
|
isp2move_B = self.isp2move['B'] |
|
|
self.movement['move_A'] = self.movement.apply( |
|
|
|
|
|
|
|
|
phases['move_A'] = phases.apply( |
|
|
lambda row: int(isp2move_A.get((row.inter_no, row.STOS_NO, row.phas_A), -1)), axis=1) |
|
|
lambda row: int(isp2move_A.get((row.inter_no, row.STOS_NO, row.phas_A), -1)), axis=1) |
|
|
self.movement['move_B'] = self.movement.apply( |
|
|
|
|
|
|
|
|
phases['move_B'] = phases.apply( |
|
|
lambda row: int(isp2move_B.get((row.inter_no, row.STOS_NO, row.phas_B), -1)), axis=1) |
|
|
lambda row: int(isp2move_B.get((row.inter_no, row.STOS_NO, row.phas_B), -1)), axis=1) |
|
|
self.movement['start_unix'] = pd.to_datetime(self.movement['start_unix']) |
|
|
|
|
|
self.movement['start_unix'] = self.movement['start_unix'].apply(lambda x: int(x.timestamp())) |
|
|
|
|
|
|
|
|
phases['start_unix'] = phases['start_unix'].apply(lambda x:int(datetime.strptime(x, '%Y-%m-%d %H:%M:%S').timestamp())) |
|
|
|
|
|
|
|
|
|
|
|
for fsec in range(self.present_time - self.sim_timespan, self.present_time + 1, 5): |
|
|
|
|
|
# 1. 상태 테이블 조회해서 전체 데이터중 필요데이터(교차로번호, A링 현시번호, A링 이동류번호, B링 현시번호, B링 이동류번호)만 수집 : A |
|
|
|
|
|
moves = [group.iloc[-1:] for _, group in phases[phases.start_unix < fsec].groupby('inter_no')] |
|
|
|
|
|
if not moves: |
|
|
|
|
|
move = pd.DataFrame({ |
|
|
|
|
|
'start_unix':[], 'inter_no':[], 'phas_A':[], 'phas_B':[], |
|
|
|
|
|
'STOS_NO':[], 'move_A':[], 'move_B':[]}) |
|
|
|
|
|
else: |
|
|
|
|
|
move = pd.concat(moves) |
|
|
|
|
|
move = move.drop(columns='start_unix') |
|
|
|
|
|
# 2. 이력 테이블 조회해서 교차로별로 유닉스시간 최대인 데이터(교차로번호, 종료유닉스타임)만 수집 : B |
|
|
|
|
|
recent_histories = [group.iloc[-1:] for _, group in self.history[self.history['end_unix'] < fsec].groupby('inter_no')] # 교차로별로 유닉스시간이 최대인 행들 |
|
|
|
|
|
|
|
|
|
|
|
if not recent_histories: |
|
|
|
|
|
rhistory = pd.DataFrame({'inter_no':[], 'end_unix':[]}) # recent history |
|
|
|
|
|
else: |
|
|
|
|
|
rhistory = pd.concat(recent_histories) |
|
|
|
|
|
recent_unix = rhistory[['inter_no', 'end_unix']] |
|
|
|
|
|
# 3. 상태 테이블 조회정보(A)와 이력 테이블 조회정보(B) 조인(키값 : 교차로번호) : C |
|
|
|
|
|
move = pd.merge(move, recent_unix, how='left', on='inter_no') |
|
|
|
|
|
move['end_unix'] = move['end_unix'].fillna(0).astype(int) |
|
|
|
|
|
# # 4. C데이터 프레임에 신규 컬럼(시작 유닉스타임) 생성 후 종료유닉스 타임 값 입력, 종료 유닉스 타임 컬럼 제거 |
|
|
|
|
|
move = move.rename(columns = {'end_unix':'start_unix'}) |
|
|
|
|
|
# 5. 이동류 이력정보 READ |
|
|
|
|
|
# - CSV 파일로 서버에 저장된 이동류정보를 읽어옴(파일이 없는 경우에는 데이터가 없는 프레임 D 생성) |
|
|
|
|
|
try: # movement가 존재할 경우 그걸 그대로 씀. |
|
|
|
|
|
self.movement |
|
|
|
|
|
except AttributeError: # movement가 존재하지 않는 경우 생성 |
|
|
|
|
|
self.movement = pd.DataFrame() |
|
|
|
|
|
# 6. 이동류 이력정보 데이터테이블(D)에 C데이터 add |
|
|
|
|
|
self.movement = pd.concat([self.movement, move]) |
|
|
|
|
|
# 7. D데이터 프레임에서 중복데이터 제거(교차로번호, 시작 유닉스타임, A링 현시번호, B링 현시번호 같은 행은 제거) |
|
|
|
|
|
self.movement = self.movement.drop_duplicates(['inter_no','phas_A','phas_B','start_unix']) |
|
|
|
|
|
# 8. D데이터 보관 시간 기준시간을 시작 유닉스 타임의 최대값 - self.subtractor // 2을 값으로 산출하고, 보관 시간 기준시간보다 작은 시작 유닉스 타임을 가진 행은 모두 제거(1시간 데이터만 보관) |
|
|
|
|
|
self.movement = self.movement[self.movement.start_unix > fsec - self.subtractor // 2] |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# 3-2. movement_updated |
|
|
# 3-2. movement_updated |
|
|
@ -670,6 +719,7 @@ class SignalGenerator(): |
|
|
else: |
|
|
else: |
|
|
self.movement_updated = self.movement |
|
|
self.movement_updated = self.movement |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# 4. 통합테이블 생성 |
|
|
# 4. 통합테이블 생성 |
|
|
def make_histids(self): |
|
|
def make_histids(self): |
|
|
print("4. 통합 테이블을 생성합니다.") |
|
|
print("4. 통합 테이블을 생성합니다.") |
|
|
@ -680,11 +730,13 @@ class SignalGenerator(): |
|
|
self.attach_children() |
|
|
self.attach_children() |
|
|
self.time43 = datetime.now() |
|
|
self.time43 = datetime.now() |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# 4-1. movedur : movements and durations |
|
|
# 4-1. movedur : movements and durations |
|
|
def merge_dfs(self): |
|
|
def merge_dfs(self): |
|
|
self.movedur = pd.merge(self.hrhists, self.movement_updated, how='inner', on=['inter_no', 'start_unix', 'phas_A', 'phas_B']) |
|
|
self.movedur = pd.merge(self.hrhists, self.movement_updated, how='inner', on=['inter_no', 'start_unix', 'phas_A', 'phas_B']) |
|
|
self.movedur = self.movedur[['inter_no', 'start_unix', 'phas_A', 'phas_B', 'move_A', 'move_B', 'duration']] |
|
|
self.movedur = self.movedur[['inter_no', 'start_unix', 'phas_A', 'phas_B', 'move_A', 'move_B', 'duration']] |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# 4-2. histid |
|
|
# 4-2. histid |
|
|
def assign_signals(self): |
|
|
def assign_signals(self): |
|
|
self.histid = self.movedur.copy() |
|
|
self.histid = self.movedur.copy() |
|
|
@ -695,7 +747,6 @@ class SignalGenerator(): |
|
|
mapping_dict = self.matching.set_index(['node_id', 'move_no'])['state'].to_dict() |
|
|
mapping_dict = self.matching.set_index(['node_id', 'move_no'])['state'].to_dict() |
|
|
# matching : 가능한 모든 (노드id, 이동류번호)에 대한 신호 * 시차제와 연관 有 |
|
|
# matching : 가능한 모든 (노드id, 이동류번호)에 대한 신호 * 시차제와 연관 有 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
for i, row in self.histid.iterrows(): |
|
|
for i, row in self.histid.iterrows(): |
|
|
node_id = row.node_id |
|
|
node_id = row.node_id |
|
|
move_A = row.move_A |
|
|
move_A = row.move_A |
|
|
@ -715,6 +766,7 @@ class SignalGenerator(): |
|
|
state_B = ''.join(self.node2init[node_id]) |
|
|
state_B = ''.join(self.node2init[node_id]) |
|
|
self.histid.at[i, 'state_B'] = state_B |
|
|
self.histid.at[i, 'state_B'] = state_B |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# 4-3. histids |
|
|
# 4-3. histids |
|
|
def attach_children(self): |
|
|
def attach_children(self): |
|
|
new_histids = [] |
|
|
new_histids = [] |
|
|
@ -745,6 +797,7 @@ class SignalGenerator(): |
|
|
self.histids = self.histids.sort_values(by=['start_unix', 'node_id', 'phas_A', 'phas_B']).reset_index(drop=True) |
|
|
self.histids = self.histids.sort_values(by=['start_unix', 'node_id', 'phas_A', 'phas_B']).reset_index(drop=True) |
|
|
self.histids = self.histids[['inter_no', 'node_id', 'start_unix', 'phas_A', 'phas_B', 'move_A', 'move_B', 'duration', 'state_A', 'state_B']] |
|
|
self.histids = self.histids[['inter_no', 'node_id', 'start_unix', 'phas_A', 'phas_B', 'move_A', 'move_B', 'duration', 'state_A', 'state_B']] |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# 5. 신호 생성 |
|
|
# 5. 신호 생성 |
|
|
def get_signals(self): |
|
|
def get_signals(self): |
|
|
print("5. 신호를 생성합니다.") |
|
|
print("5. 신호를 생성합니다.") |
|
|
@ -755,6 +808,7 @@ class SignalGenerator(): |
|
|
self.make_tl_file() |
|
|
self.make_tl_file() |
|
|
self.time53 = datetime.now() |
|
|
self.time53 = datetime.now() |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# 5-1. 신호 파일의 시작 및 종료시각 설정 |
|
|
# 5-1. 신호 파일의 시작 및 종료시각 설정 |
|
|
def set_timepoints(self): |
|
|
def set_timepoints(self): |
|
|
self.offsets = {} |
|
|
self.offsets = {} |
|
|
@ -783,6 +837,7 @@ class SignalGenerator(): |
|
|
self.sigtable = pd.concat(self.sigtable).reset_index(drop=True) |
|
|
self.sigtable = pd.concat(self.sigtable).reset_index(drop=True) |
|
|
self.sigtable['phase_sumo'] = self.sigtable.groupby(['node_id', 'start_unix']).cumcount() |
|
|
self.sigtable['phase_sumo'] = self.sigtable.groupby(['node_id', 'start_unix']).cumcount() |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# 5-2. 적색 및 황색신호 부여 |
|
|
# 5-2. 적색 및 황색신호 부여 |
|
|
def assign_red_yellow(self): |
|
|
def assign_red_yellow(self): |
|
|
''' |
|
|
''' |
|
|
@ -847,6 +902,7 @@ class SignalGenerator(): |
|
|
self.SIGTABLE.append(SIG) |
|
|
self.SIGTABLE.append(SIG) |
|
|
self.SIGTABLE = pd.concat(self.SIGTABLE) |
|
|
self.SIGTABLE = pd.concat(self.SIGTABLE) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# 5-2-1 helper function of 5-2 |
|
|
# 5-2-1 helper function of 5-2 |
|
|
def get_red(self, pre_state:str, cur_state:str): |
|
|
def get_red(self, pre_state:str, cur_state:str): |
|
|
assert len(pre_state) == len(cur_state), "cur_state, nex_state의 길이가 서로 다릅니다." |
|
|
assert len(pre_state) == len(cur_state), "cur_state, nex_state의 길이가 서로 다릅니다." |
|
|
@ -862,6 +918,7 @@ class SignalGenerator(): |
|
|
raise ValueError(f"예상치 못한 신호조합: previous={p}, current={c}") |
|
|
raise ValueError(f"예상치 못한 신호조합: previous={p}, current={c}") |
|
|
return state_r |
|
|
return state_r |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# 5-2-2 helper function of 5-2 |
|
|
# 5-2-2 helper function of 5-2 |
|
|
def get_yellow(self, cur_state:str, nex_state:str): |
|
|
def get_yellow(self, cur_state:str, nex_state:str): |
|
|
assert len(cur_state) == len(nex_state), "cur_state, nex_state의 길이가 서로 다릅니다." |
|
|
assert len(cur_state) == len(nex_state), "cur_state, nex_state의 길이가 서로 다릅니다." |
|
|
@ -879,6 +936,7 @@ class SignalGenerator(): |
|
|
raise ValueError(f"예상치 못한 신호조합: current={c}, next={n}") |
|
|
raise ValueError(f"예상치 못한 신호조합: current={c}, next={n}") |
|
|
return state_y |
|
|
return state_y |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# 5-2-3 helper function of 5-2 |
|
|
# 5-2-3 helper function of 5-2 |
|
|
def cumulate(self, sig, alph): |
|
|
def cumulate(self, sig, alph): |
|
|
|
|
|
|
|
|
@ -950,6 +1008,7 @@ class SignalGenerator(): |
|
|
csig = pd.concat(csig).reset_index(drop=True) |
|
|
csig = pd.concat(csig).reset_index(drop=True) |
|
|
return csig |
|
|
return csig |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# 5-3. 신호파일 생성 |
|
|
# 5-3. 신호파일 생성 |
|
|
def make_tl_file(self): |
|
|
def make_tl_file(self): |
|
|
strings = ['<additional>\n'] |
|
|
strings = ['<additional>\n'] |
|
|
@ -967,6 +1026,7 @@ class SignalGenerator(): |
|
|
with open(self.path_output, 'w') as f: |
|
|
with open(self.path_output, 'w') as f: |
|
|
f.write(strings) |
|
|
f.write(strings) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# 6. 이슈사항 저장 |
|
|
# 6. 이슈사항 저장 |
|
|
def write_issues(self): |
|
|
def write_issues(self): |
|
|
print('6. 이슈사항을 저장합니다.') |
|
|
print('6. 이슈사항을 저장합니다.') |
|
|
|
