さて、本日はArcPyでライン接合を行ってみようと思います。背景などでけっこうラインがぶちぶち切れてしまっていることが多いので、この機能があればラインのフィーチャ数を減らすことができますね。Advancedライセンスがあれば、ArcMapで実行できるのですが（うまくいくかは知らない）、いかんせんAdvancedはお高いですよね（けっこういい車が買えるお値段です）。そんなものを購入できるはずもなく、「使えないなら作ろう！」と思い立った次第であります。ArcMapでのライン接合についてはこのサイトに記載してあります。ラインの接合 (Unsplit Line)—データ管理ツールボックス | ArcGIS Desktop
サンプルソース
ライン接合処理のサンプルです。以下に処理の概要を記載します。
- 接合対象ラインを抽出（つながりのあるラインを抽出）
- 接合対象ラインの中でループしているラインを判定
- 接合対象ラインの中で端のラインとそのラインの端点（他のラインと接していない端点）を抽出（ループしているラインは除く）
- 接合対象ラインの端点から作図順を決める
- 接合対象ラインを作図
- 接合対象外ラインを作図
■unsplitline.py
import arcpy
from collections import Counter
import os
import traceback
import pandas as pd
class UnsplitLine():
column_unsplit_length = "UNSPLIT_LENGTH"
def __init__(self,workspace,in_feature,out_feature):
arcpy.env.workspace = workspace
self.workspace = workspace
self.in_feature = in_feature
self.out_feature = out_feature
self.line_list = []
self.match_point_list = []
self.finish_line_list = []
self.divide_line_list = []
self.list_draw_point_sort = []
self.finish_oid = []
self.count = UnsplitLine.counter(self)
self.check_feature_class()
self.check_columns()
def check_feature_class(self):
"""out_featureに指定したフィーチャクラスが無い場合、作成します"""
exist_out_feature = arcpy.ListFeatureClasses(self.out_feature)
if len(exist_out_feature) == 0:
print u"フィーチャクラスを作成します"
arcpy.CreateFeatureclass_management(self.workspace, self.out_feature, "POLYLINE")
def check_columns(self):
"""out_featureに指定したフィーチャクラスに「UNSPLIT_LENGTH」カラムが存在しない場合、カラムを作成します"""
field_list = []
fields = arcpy.ListFields(self.out_feature)
for field in fields:
if field.type != "Geometry":
field_list.append(field.name)
if not UnsplitLine.column_unsplit_length in field_list:
print u"カラムを作成します"
arcpy.AddField_management(self.out_feature, UnsplitLine.column_unsplit_length, "Double")
def export_upsplit_lines(self):
"""ライン接合後、Shape_lengthやアイテム数をテキスト出力します"""
try:
moto_line_length,saki_line_length = self.count_length()
result_in_feature,result_out_feature = self.count_feature()
file_dir = self.workspace[:self.workspace.rfind("\\")]
with open(os.path.join(file_dir,"export_upsplit_lines.txt"), "w") as f:
message = u"接合元フィーチャの総延長=" + str(moto_line_length)
f.write(message.encode("Shift-JIS"))
f.write("\n")
message = u"接合先フィーチャの総延長=" + str(saki_line_length)
f.write(message.encode("Shift-JIS"))
f.write("\n")
message = u"接合元フィーチャ数=" + str(result_in_feature)
f.write(message.encode("Shift-JIS"))
f.write("\n")
message = u"接合先フィーチャ数=" + str(result_out_feature)
f.write(message.encode("Shift-JIS"))
f.write("\n")
message = u"接合したラインのOID"
f.write(message.encode("Shift-JIS"))
f.write("\n")
for line in self.line_list:
for l in line:
f.write(str(l))
f.write(" ")
f.write("\n")
except IOError:
print u"ファイルの出力に失敗しました。"
except TypeError:
print u"ファイルに出力する内容が不正です。"
def count_feature(self):
"""接合元と接合先のフィーチャクラスのフィーチャ数をカウントします"""
result_in_feature = arcpy.GetCount_management(self.in_feature)
result_out_feature = arcpy.GetCount_management(self.out_feature)
return result_in_feature,result_out_feature
def count_length(self):
"""接合元と接合先のフィーチャクラスのShape_Lengthを集計します"""
moto_line_length = float()
saki_line_length = float()
for line in [row for row in arcpy.da.SearchCursor(self.in_feature, ["OID@","SHAPE@"])]:
moto_line_length += line[1].length
for line in [row for row in arcpy.da.SearchCursor(self.out_feature, ["OID@","SHAPE@"])]:
saki_line_length += line[1].length
return moto_line_length,saki_line_length
def set_join_oid(self,oid_list):
"""oid_listに格納されているOIDを分解します。[1,2,3]の場合、1,2,3に分解します。"""
oid_list = set(oid_list)
oid_list = list(oid_list)
oid = ",".join([str(x) for x in oid_list])
return oid
def copy_feature(self):
"""接合対象ではないラインの作図を行います"""
try:
oid = self.set_join_oid(self.finish_line_list)
with arcpy.da.InsertCursor(self.out_feature, ["OID@","SHAPE@",UnsplitLine.column_unsplit_length]) as cur:
for line in arcpy.da.SearchCursor(self.in_feature, ["OID@","SHAPE@"],"OBJECTID not in (%s)" % oid) :
length = self.get_shape_length(line[0])
for part in line[1]:
array = arcpy.Array()
for pnt in part:
if pnt:
array.add(arcpy.Point(pnt.X, pnt.Y))
cur.insertRow((line[0],arcpy.Polyline(array),length))
except TypeError:
print traceback.format_exc()
def get_lines(self,oid=0):
"""接合元のラインを取得する関数です
returns:
self.moto_lines -- 接合元のラインの全フィーチャ
line -- ラインオブジェクト
"""
if oid == 0:
self.moto_lines = [row for row in arcpy.da.SearchCursor(self.in_feature, ["OID@","SHAPE@"])]
return self.moto_lines
else:
line = [row for row in arcpy.da.SearchCursor(in_feature, ["OID@","SHAPE@"],"OBJECTID = %s" % oid)]
line = [x for x in line for x in x]
return line
def search_line_combination(self,line2):
"""接合するラインの組み合わせを取得する関数です
例：OBJECTID＝1,2と4,5,6と10,11,12,13が接続対象の場合
[[1,2],[4,5,6],[10,11,12,13]]というリストを作成します。
arguments:
line2 -- search_lineでのループから渡されるLineオブジェクト
"""
point_list = []
oid = line2[0]
try:
for line in self.moto_lines:
if line[0] != oid:
if not line[0] in self.finish_line_list:
if (line[1].firstPoint.X == line2[1].lastPoint.X and line[1].firstPoint.Y == line2[1].lastPoint.Y) or \
(line[1].lastPoint.X == line2[1].lastPoint.X and line[1].lastPoint.Y == line2[1].lastPoint.Y) or \
(line[1].lastPoint.X == line2[1].firstPoint.X and line[1].lastPoint.Y == line2[1].firstPoint.Y) or \
(line[1].firstPoint.X == line2[1].firstPoint.X and line[1].firstPoint.Y == line2[1].firstPoint.Y):
point_list.append(line[0])
line_copy = line
if len(point_list) == 0:
if len(self.match_point_list) != 0:
self.match_point_list = set(self.match_point_list)
self.line_list.append(list(self.match_point_list))
self.match_point_list = []
elif len(point_list) == 1:
point_list.append(oid)
self.finish_line_list.extend(point_list)
self.match_point_list.extend(point_list)
self.search_line_combination(line_copy)
elif len(point_list) == 2:
oid_list = []
ret,point_list = self.check_line_combination(oid,point_list)
if ret == -1:
self.divide_line_list.extend(point_list)
else:
oid_list.append(oid)
self.match_point_list.extend(oid_list)
if len(self.match_point_list) > len(point_list):
self.finish_line_list.extend(self.match_point_list)
l = list(x for x in point_list if x not in self.match_point_list)
if len(l) == 1:
line = self.get_lines(l[0])
self.search_line_combination(line)
else:
self.search_line_combination(line_copy)
except Exception:
print traceback.format_exc()
def check_line_combination(self,oid,point_list):
"""接合対象のラインにおいて、端点が三つ以上重なっているか確認する関数です
arguments:
oid -- search_lineでのループから渡されるOBJECTID
point_list -- search_line_combinationで抽出した端点のリスト
returns:
1 or -1 -- 1:端点が三つ重なっていない -1:端点が三つ重なっている
point_list -- point_list
"""
point_list.append(oid)
point_list = set(point_list)
point_list = list(point_list)
oid = ",".join([str(x) for x in point_list])
devide_list = []
for row in arcpy.da.SearchCursor(in_feature, ["OID@", "SHAPE@"],"OBJECTID in (%s)" % oid):
devide_list.append([row[0],row[1].firstPoint.X,row[1].firstPoint.Y,row[1].lastPoint.X,row[1].lastPoint.Y])
df_devide = pd.DataFrame(devide_list)
df_devide.columns = ["OID","FirstX","FirstY","LastX","LastY"]
dic_first = df_devide.groupby("FirstX").groups
dic_last = df_devide.groupby("LastX").groups
list_first = [x for _,x in dic_first.iteritems()]
list_last = [x for _,x in dic_last.iteritems()]
for f,l in zip(list_first,list_last):
if len(f) >= 3 or len(l) >= 3:
return -1,point_list
divide_line_list = []
for key,row in df_devide.iterrows():
for key2,row2 in df_devide.iterrows():
if key != key2:
if row.FirstX == row2.LastX or row.FirstX == row2.FirstX:
divide_line_list.append(row.OID)
if len(divide_line_list) >= 2:
return -1,point_list
divide_line_list = []
return 1,point_list
def search_line(self):
"""search_line_combinationでの結果を返す関数です
returns:
self.line_list -- 接合対象ラインのリスト（多次元）
例：OBJECTID＝1,2と4,5,6と10,11,12,13が接続対象の場合
[[1,2],[4,5,6],[10,11,12,13]]というリストが返されます。
"""
lines = self.get_lines()
for line in lines:
self.search_line_combination(line)
return self.line_list
def pop_list(self,pop_list):
"""リストから要素をポップする関数です
arguments:
pop_list -- ポップ対象のリスト
"""
try:
for p in pop_list:
i = pop_list.index(p)
self.moto_lines.pop(i)
except ValueError:
print traceback.format_exc()
raise Exception
def get_coord(self,oid):
"""接合対象ラインにおいて、各頂点の座標を取得する関数です
arguments:
oid -- OBJECTID → 接合対象のOBJECTIDが1,2,3 の場合、oid = 1,2,3 という形になります
returns:
coord_list -- 接合対象ラインの、各頂点の座標（多次元）
[0] = OBJECTID
[1] = ポイントごとに振ったID
[2] = X
[3] = Y
[4] = 始点X
[5] = 始点Y
[6] = 終点X
[7] = 終点Y
loop_list -- 接合対象ラインがループしているのかどうかの判定用リスト（多次元）
[0] = OBJECTID
[1] = 始点X
[2] = 始点Y
[3] = 終点X
[4] = 終点Y
"""
coord_list = []
loop_list = []
for row in arcpy.da.SearchCursor(self.in_feature, ["OID@", "SHAPE@"],"OBJECTID in (%s)" % oid):
loop_list.append([row[0],row[1].firstPoint.X,row[1].firstPoint.Y,row[1].lastPoint.X,row[1].lastPoint.Y])
for part in row[1]:
for i,pnt in enumerate(part):
if pnt:
pnt.ID = i
coord_list.append([row[0],pnt.ID,pnt.X,pnt.Y,row[1].firstPoint.X,row[1].firstPoint.Y,row[1].lastPoint.X,row[1].lastPoint.Y])
return coord_list,loop_list
def check_geometry(self,df):
"""接合対象ラインにおいて、各頂点が始点なのか終点なのか判定する関数
判定した結果を、データフレームに結合します。
arguments:
df -- Pandasのデータフレーム
columns = ["OID","ID","X","Y","FirstX","FirstY","LastX","LastY"]
returns:
df_merge -- Pandasのデータフレーム
columns = ["OID","ID","X","Y","FirstX","FirstY","LastX","LastY","PointType"]
PointType 1:始点 2:終点 None:それ以外
"""
try:
list_point_type = []
for _ ,row in df.iterrows():
for _ ,row2 in df.iterrows():
if row.OID == row2.OID and row.ID == row2.ID:
if row.X == row2.FirstX and row.Y == row2.FirstY:
list_point_type.append([row2.OID,row2.ID,"1"])
elif row.X == row2.LastX and row.Y == row2.LastY:
list_point_type.append([row2.OID,row2.ID,"2"])
df_point_type = pd.DataFrame(list_point_type)
df_point_type.columns = ["OID","ID","PointType"]
df_point_type = pd.DataFrame(df_point_type.drop_duplicates(["OID","PointType"]))
df_merge = pd.merge(df, df_point_type, on=["OID","ID"],how='left')
return df_merge
except Exception:
print traceback.format_exc()
def check_first_last(self,df):
"""接合対象ラインにおいて、端のラインと端が始点なのか終点なのかを判定する関数です
arguments:
df -- Pandasのデータフレーム。
columns = ["OID","ID","X","Y","FirstX","FirstY","LastX","LastY","PointType"]
PointType 1:始点 2:終点 None:それ以外
returns:
list_first_last -- 端のラインとPointTypeを格納したリスト
[0]:OBJECTID
[1]:頂点ごとに振ったID
[2]:PointType 1:始点 2:終点
"""
list_first_last = []
list_edge = []
if len(df.groupby("OID").groups.keys()) == 2:
for _ , row in df.iterrows():
for _ , row2 in df.iterrows():
if row.OID != row2.OID:
if row.PointType == "1":
if (row.LastX == row2.FirstX and row.LastY == row2.FirstY) or \
(row.LastX == row2.LastX and row.LastY == row2.LastY):
list_first_last.append([row.OID,row.ID,"1"])
elif row.PointType == "2":
if (row.FirstX == row2.LastX and row.FirstY == row2.LastY) or \
(row.FirstX == row2.FirstX and row.FirstY == row2.FirstY):
list_first_last.append([row.OID,row.ID,"2"])
else:
continue
return list_first_last
else:
for _ , row in df.iterrows():
for _ , row2 in df.iterrows():
if row.OID != row2.OID:
if row.PointType == "1":
if (row.LastX == row2.FirstX and row.LastY == row2.FirstY) or \
(row.LastX == row2.LastX and row.LastY == row2.LastY):
list_edge.append([row.OID,row.ID,"1"])
if (row.FirstX == row2.FirstX and row.FirstY == row2.FirstY) or \
(row.FirstX == row2.LastX and row.FirstY == row2.LastY):
list_edge.append([row.OID,row.ID,"2"])
elif row.PointType == "2":
if (row.FirstX == row2.LastX and row.FirstY == row2.LastY) or \
(row.FirstX == row2.FirstX and row.FirstY == row2.FirstY):
list_edge.append([row.OID,row.ID,"2"])
if (row.LastX == row2.FirstX and row.LastY == row2.FirstY) or \
(row.LastX == row2.LastX and row.LastY == row2.LastY):
list_edge.append([row.OID,row.ID,"1"])
c = Counter(tuple(x) for x in list_edge)
list_edge = [list(k) for k,v in c.items()]
if len(list_edge) >= 2:
list_edge = []
elif len(list_edge) == 1:
list_first_last.append([list_edge[0][0],list_edge[0][1],list_edge[0][2]])
return list_first_last
def counter(self):
"""頂点の作図順用のカウンタ"""
n = [0]
def inc():
n[0] += 1
return n[0]
return inc
def search_dataframe(self,df,oid,line_id):
"""データフレームのレコードを抽出する用の関数"""
df = pd.DataFrame(df[df.OID == oid])
df = pd.DataFrame(df[df.ID == line_id])
return df
def draw_point_sort(self,df,oid,line_id,point_type):
"""接合対象ラインにおいて、各頂点の作図順を設定する関数です
self.list_draw_point_sort に作図順を格納して後続処理で使用します
arguments:
df -- Pandasのデータフレーム。
columns = ["OID","ID","X","Y","FirstX","FirstY","LastX","LastY","PointType"]
PointType 1:始点 2:終点 None:それ以外
oid -- OBJECTID（端のライン）
line_id -- 各頂点に割り当てたID（端のライン）
point_type -- 1:始点 2:終点 ←左記以外の値（None）は受け取らない想定（端のラインのどちらかの端点が入っているため）
"""
df_draw_point_sort = self.search_dataframe(df,oid,line_id)
first_x = df_draw_point_sort["FirstX"].values[0]
last_x = df_draw_point_sort["LastX"].values[0]
first_y = df_draw_point_sort["FirstY"].values[0]
last_y = df_draw_point_sort["LastY"].values[0]
if point_type == "":
point_type = df_draw_point_sort["PointType"].values[0]
if point_type == "1":
df_asc = pd.DataFrame(df[df.OID == oid])
df_asc = df_asc.sort(['ID'], ascending=[True])
for _ ,row in df_asc.iterrows():
self.list_draw_point_sort.append([row.OID,row.ID,self.count()])
elif point_type == "2":
df_desc = pd.DataFrame(df[df.OID == oid])
df_desc = df_desc.sort(['ID'], ascending=[False])
for _ ,row in df_desc.iterrows():
self.list_draw_point_sort.append([row.OID,row.ID,self.count()])
for _ ,row in df.iterrows():
if row.OID != oid:
if point_type == "1":
if row.X == last_x and row.Y == last_y:
if not row.OID in self.finish_oid:
self.finish_oid.append(oid)
self.draw_point_sort(df,row.OID,row.ID,"")
elif point_type == "2":
if row.X == first_x and row.Y == first_y:
if not row.OID in self.finish_oid:
self.finish_oid.append(oid)
self.draw_point_sort(df,row.OID,row.ID,"")
def search_loop_line(self,df_loop):
"""接合対象ラインがループしているラインなのか判定する関数
arguments:
df_loop -- Pandasのデータフレーム。
columns = ["OID","FirstX","FirstY","LastX","LastY"]
returns:
1 or -1-- 1:ループしたラインではない -1:ループしたライン
"""
for _,row in df_loop.iterrows():
if not row.OID in self.finish_line_list:
self.finish_line_list.append(row.OID)
if len(df_loop) == 2:
for _,row in df_loop.iterrows():
for _,row2 in df_loop.iterrows():
if row.OID != row2.OID:
if ((row.FirstX == row2.LastX and row.FirstY == row2.LastY) or \
(row.FirstX == row2.FirstX and row.FirstY == row2.FirstY)) and \
((row.LastX == row2.FirstX and row.LastY == row2.FirstY) or \
(row.LastX == row2.LastX and row.LastY == row2.LastY)):
break
else:
return 1
return -1
elif len(df_loop) >= 3:
oid_list = []
for _,row in df_loop.iterrows():
for _,row2 in df_loop.iterrows():
if row.OID != row2.OID:
if (row.FirstX == row2.FirstX and row.FirstY == row2.FirstY) or \
(row.FirstX == row2.LastX and row.FirstY == row2.LastY):
oid_list.append(1)
if (row.LastX == row2.FirstX and row.LastY == row2.FirstY) or \
(row.LastX == row2.LastX and row.LastY == row2.LastY):
oid_list.append(2)
if len(oid_list) != 2:
return 1
oid_list = []
return -1
def get_shape_length(self,oid):
"""接合対象ラインのShape_lengthの合計を取得する関数"""
total_length = float()
for line in arcpy.da.SearchCursor(self.in_feature, ["SHAPE@"],"OBJECTID in (%s)" % oid) :
total_length += line[0].length
return total_length
def upsplit_line(self,cur,df,oid):
"""ラインを接合する関数
arguments:
cur -- カーソル
df -- Pandasのデータフレーム
columns = ["OID","ID","X","Y","FirstX","FirstY","LastX","LastY","PointSort"]
oid -- OBJECTID(1,2,3が接合対象の場合oidには1,2,3 が格納されている)
"""
try:
array = arcpy.Array()
length = self.get_shape_length(oid)
for d in df.index:
array.add(arcpy.Point(df.ix[d][2], df.ix[d][3]))
cur.insertRow((arcpy.Polyline(array),length))
except Exception:
print traceback.format_exc()
def create_line(self,lines):
"""各種関数を呼び出す処理。ここで、ラインの接合や接合対象ではないラインのコピー関数を呼び出す。
arguments:
lines -- search_lineで格納したリスト(多次元)
例：OBJECTID=1,2,3 また、4,5 というラインが接合対象の場合、
[[1,2,3],[4,5]] という要素が格納されている
"""
try:
if len(lines) == 0:
print u"対象データがありません"
raise Exception
with arcpy.da.InsertCursor(self.out_feature, ["SHAPE@",UnsplitLine.column_unsplit_length]) as cur:
for line in lines:
oid = self.set_join_oid(line)
coord_list,loop_list = self.get_coord(oid)
df = pd.DataFrame(loop_list)
df.columns = ["OID","FirstX","FirstY","LastX","LastY"]
result = self.search_loop_line(df)
if result == 1:
df_coord = pd.DataFrame(coord_list)
df_coord.columns = ["OID","ID","X","Y","FirstX","FirstY","LastX","LastY"]
df_checked = self.check_geometry(df_coord)
list_checked = self.check_first_last(df_checked)
if not list_checked:
print u"端点が判断できないラインが存在しました" + " " + "OID=" + oid
raise Exception
oid_one,line_id,point_type = list_checked[0][0],list_checked[0][1],list_checked[0][2]
self.draw_point_sort(df_checked,oid_one,line_id,point_type)
df_point_sort = pd.DataFrame(self.list_draw_point_sort)
df_point_sort.columns = ["OID","ID","PointSort"]
df_checked = pd.merge(df_checked, df_point_sort, on=["OID","ID"],how="left")
df_checked = pd.DataFrame(df_checked.drop_duplicates(["X","Y"]))
df_checked = df_checked.sort(["PointSort"], ascending=[True])
self.upsplit_line(cur,df_checked,oid)
self.list_draw_point_sort = []
else:
df_coord = pd.DataFrame(coord_list)
df_coord.columns = ["OID","ID","X","Y","FirstX","FirstY","LastX","LastY"]
df_checked = self.check_geometry(df_coord)
self.draw_point_sort(df_checked,coord_list[0][0],coord_list[0][1],"")
df_point_sort = pd.DataFrame(self.list_draw_point_sort)
df_point_sort.columns = ["OID","ID","PointSort"]
df_checked = pd.merge(df_checked, df_point_sort, on=["OID","ID"],how="left")
df_checked = df_checked.sort(["PointSort"], ascending=[True])
self.upsplit_line(cur,df_checked,oid)
self.list_draw_point_sort = []
self.copy_feature()
self.export_upsplit_lines()
except Exception:
print traceback.format_exc()
if __name__ == '__main__':
workspace = r"C:\ArcPySample\ArcPyAdvanced.gdb"
in_feature = "LineSplit"
out_feature = "LineUnsplit"
unsplitLine = UnsplitLine(workspace,in_feature,out_feature)
lines = unsplitLine.search_line()
unsplitLine.create_line(lines)
このソースをArcMap上で実行した結果が以下になります。
ちゃんと作られました。このように一本ずつだったラインが
このように接合されました！
プログラムの中で「UNSPLIT_LENGTH」というカラムを作成し、接合前のShape_Lengthを格納するようにしました（接合対象のラインが三本あり、それぞれの長さが10,20,30だった場合、60が格納される）。ただ、ちょっと誤差がでてしまっているので、接合結果を比較する場合は、小数点以下をはずすなどしないとだめそうですね・・・
出力前後の結果確認用にテキストも出力するようにしています（今回はC:\ArcPySampleの直下に作成）。こちらも微妙に誤差が出ていますね・・・
