{ "cells": [ { "cell_type": "code", "execution_count": 1, "metadata": {}, "outputs": [], "source": [ "import pandas as pd\n", "import numpy as np\n", "import os, sys, copy, argparse, json, pickle\n", "import sumolib, traci\n", "from tqdm import tqdm\n", "from datetime import datetime\n", "path_root = os.path.dirname(os.path.dirname(os.path.abspath('.')))\n", "path_scr = os.path.join(path_root, 'scripts')\n", "sys.path.append(path_scr)\n", "from preprocess_daily import DailyPreprocessor\n", "# from generate_signals import SignalGenerator" ] }, { "cell_type": "code", "execution_count": 2, "metadata": {}, "outputs": [], "source": [ "self = DailyPreprocessor(config_name = 'test_0731',\n", " file_net = 'new_sungnam_network_internal_target_0721.net.xml')" ] }, { "cell_type": "code", "execution_count": 3, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "1-1. 네트워크가 로드되었습니다.\n" ] } ], "source": [ "# 1-1. 네트워크 불러오기\n", "self.load_networks()" ] }, { "cell_type": "code", "execution_count": 4, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "1-2. 테이블들이 로드되었습니다.\n" ] } ], "source": [ "# 1-2. 테이블 불러오기\n", "self.load_tables()" ] }, { "cell_type": "code", "execution_count": 5, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "1-5. 테이블을 표준화했습니다.\n" ] } ], "source": [ "# 1-5. 테이블 표준화\n", "self.standardize()" ] }, { "cell_type": "code", "execution_count": 6, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "1-5. 테이블을 표준화했습니다.\n" ] } ], "source": [ "# 1-5. 테이블 표준화\n", "self.standardize()" ] }, { "cell_type": "code", "execution_count": 7, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "1-6. 주요 객체 (리스트, 딕셔너리)들을 저장했습니다.\n" ] } ], "source": [ "# 1-6. 주요 객체 (리스트, 딕셔너리) 저장\n", "self.store_objects()" ] }, { "cell_type": "code", "execution_count": 8, "metadata": {}, "outputs": [], "source": [ "self.make_match1()\n", "self.make_match2()\n", "self.make_match3()\n", "self.make_match4()\n", "self.make_match5()" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "self.uturn = pd.merge(self.uturn, self.u_condition, on='child_id')\n", "\n", "# p2inc_edge2angle : node_id to inc_edge2angle\n", "p2inc_edge2angle = dict()\n", "# p2out_edge2angle : node_id to out_edge2angle\n", "p2out_edge2angle = dict()\n", "# p2inc_angle2edge : node_id to inc_angle2edge\n", "p2inc_angle2edge = dict()\n", "# p2out_angle2edge : node_id to out_angle2edge\n", "p2out_angle2edge = dict()\n", "for node_id in self.parent_ids:\n", " m5 = self.match5[self.match5.node_id==node_id]\n", " m5 = m5.dropna(subset=['inc_edge_id', 'out_edge_id'])\n", " # inc_edge2angle : inc_edge_id to inc_angle\n", " inc_edge2angle = dict(zip(m5.inc_edge_id, m5.inc_angle.astype(int)))\n", " p2inc_edge2angle[node_id] = inc_edge2angle\n", " # out_edge2angle : out_edge_id to out_angle\n", " out_edge2angle = dict(zip(m5.out_edge_id, m5.out_angle.astype(int)))\n", " p2out_edge2angle[node_id] = out_edge2angle\n", " # inc_angle2edge : inc_angle to inc_edge_id\n", " inc_angle2edge = dict(zip(m5.inc_angle.astype(int), m5.inc_edge_id))\n", " p2inc_angle2edge[node_id] = inc_angle2edge\n", " # out_angle2edge : out_angle to out_edge_id\n", " out_angle2edge = dict(zip(m5.out_angle.astype(int), m5.out_edge_id))\n", " p2out_angle2edge[node_id] = out_angle2edge\n", "# 각 uturn node에 대하여 (inc_edge_id, out_edge_id) 부여\n", "cmatches = []\n", "row = self.uturn.iloc[0]\n", "parent_id = row.parent_id\n", "child_id = row.child_id\n", "condition = row.condition\n", "inc_edge_id = row.inc_edge_id\n", "out_edge_id = row.out_edge_id\n", "print(type(inc_edge_id))\n", "adj_inc_edge_id = row.adj_inc_edge_id\n", "adj_out_edge_id = row.adj_out_edge_id\n", "\n", "# match5에서 부모노드id에 해당하는 행들을 가져옴 (cmatch)\n", "cmatch = self.match5.copy()[self.match5.node_id==parent_id] # match dataframe for a child node\n", "cmatch = cmatch.sort_values(by=['phase_no', 'ring_type']).reset_index(drop=True)\n", "cmatch['node_id'] = child_id\n", "cmatch['node_type'] = 'u_turn'\n", "\n", "# 진입엣지 각도\n", "inc_angle = p2inc_edge2angle[parent_id][adj_inc_edge_id]\n", "\n", "# 이격각도\n", "self.angle_separation = 10\n", "\n", "# 진입로 각도 목록\n", "inc_angles = cmatch.dropna(subset=['inc_angle', 'out_angle']).inc_angle.astype(int).unique()\n", "inc_angles = np.sort(inc_angles)\n", "inc_angles = list(inc_angles - 360) + list(inc_angles) + list(inc_angles + 360)\n", "inc_angles = np.array(inc_angles)\n", "\n", "# 보행신호시의 진입로 각도\n", "inc_angles_left = inc_angles[inc_angles >= inc_angle + self.angle_separation]\n", "inc_angle_pedes = np.sort(inc_angles_left)[0] % 360\n", "\n", "# 보행신호시의 진입로 엣지id\n", "inc_angle2edge = p2inc_angle2edge[parent_id]\n", "inc_edge_id_pedes = inc_angle2edge[inc_angle_pedes]\n", "\n", "# 진출로 각도 목록\n", "out_angles = cmatch.dropna(subset=['inc_angle', 'out_angle']).out_angle.astype(int).unique()\n", "out_angles = np.sort(out_angles)\n", "out_angles = list(out_angles - 360) + list(out_angles) + list(out_angles + 360)\n", "out_angles = np.array(out_angles)\n", "\n", "# 보행신호시의 진출로 각도\n", "out_angles_right = out_angles[out_angles <= inc_angle - self.angle_separation]\n", "out_angle_pedes = np.sort(out_angles_right)[-1] % 360\n", "\n", "# 보행신호시의 진출로 엣지id\n", "out_angle2edge = p2out_angle2edge[parent_id]\n", "out_edge_id_pedes = out_angle2edge[out_angle_pedes]\n", "\n", "# 진입엣지/진출엣지 포함 조건\n", "inc_true = (cmatch.inc_edge_id==adj_inc_edge_id)\n", "out_true = (cmatch.out_edge_id==adj_out_edge_id)\n", "\n", "# 보행신호시 조건\n", "pedes_flag = (cmatch.inc_edge_id==inc_edge_id_pedes) & (cmatch.out_edge_id==out_edge_id_pedes)\n", "\n", "# 좌회전시 조건\n", "right_flag = inc_true & (cmatch.turn_type=='left')\n", "\n", "# 보행신호이동류(17) 조건\n", "crosswalk_on = (cmatch.move_no==17) & ~ out_true\n", "\n", "# 신호없음이동류(18) 조건\n", "all_redsigns = (cmatch.move_no==18) & ~ out_true" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "print(type(inc_edge_id))\n", "print(type(out_edge_id))\n", "print(cmatch[['inc_edge_id', 'out_edge_id']].info())" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "print(inc_edge_id, out_edge_id)\n", "# 보행신호시/좌회전시 진입/진출 엣지id 배정\n", "cmatch[['inc_edge_id', 'out_edge_id']] = None\n", "if condition == \"보행신호시\":\n", " cmatch.loc[pedes_flag, ['inc_edge_id', 'out_edge_id']] = [inc_edge_id, out_edge_id]\n", "elif condition == \"좌회전시\":\n", " cmatch.loc[right_flag, ['inc_edge_id', 'out_edge_id']] = [inc_edge_id, out_edge_id]\n", "\n", "uturn_not_assigned = cmatch[['inc_edge_id','out_edge_id']].isna().any(axis=1).all()\n" ] }, { "cell_type": "code", "execution_count": 9, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "2-1. 매칭 테이블들을 생성했습니다.\n", "2-2. 초기화 신호가 지정되었습니다. (우회전 : g)\n", "2-3. 유턴 인덱스 / 비보호좌회전 인덱스를 지정했습니다.\n", "2-4. 직진 및 좌회전(G)을 배정했습니다.\n", "2-5. node2num_cycles.json를 저장했습니다.\n" ] } ], "source": [ "# 2. 중간산출물 만들기\n", "# 2-1 매칭테이블 생성\n", "self.get_matches()\n", "# 2-2 신호 초기화\n", "self.initialize_state()\n", "# 2-3 유턴 인덱스 / 비보호좌회전 인덱스 지정\n", "self.assign_indices()\n", "# 2-4 신호배정\n", "self.assign_signals()\n", "# 2-5 기반파일 저장\n", "self.save_intermediates()" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "self.write_issues()" ] } ], "metadata": { "kernelspec": { "display_name": "siggen_env", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.12.4" } }, "nbformat": 4, "nbformat_minor": 2 }