1.完成了同步TA文件和S文件塔高的功能
2.完成了在ORG文件后面追加测量桩号的功能 3.完成了将S文件第一次同步到TA文件的功能
This commit is contained in:
commit
3a19a81224
|
|
@ -0,0 +1,170 @@
|
|||
import re
|
||||
import datetime
|
||||
import shutil
|
||||
import os.path
|
||||
import re
|
||||
|
||||
def file_backup_time():
|
||||
now = datetime.datetime.now()
|
||||
return now.strftime("%Y%I%d%H%M%S")
|
||||
|
||||
|
||||
def get_directory(file_path):
|
||||
return os.path.dirname(file_path)
|
||||
|
||||
|
||||
def get_file_name_with_extention(file_path):
|
||||
dir_part = get_directory(file_path)
|
||||
file_name = file_path.replace(dir_part, "").replace("\\", "").replace(r"/", "")
|
||||
(name, ext) = file_name.split(".")
|
||||
return (name, ext)
|
||||
|
||||
|
||||
class SFileObject:
|
||||
def __init__(self, file_path):
|
||||
self._content = []
|
||||
self._file_path = file_path
|
||||
self._read(file_path)
|
||||
pass
|
||||
|
||||
def _read(self, file_path):
|
||||
with open(file_path) as f:
|
||||
for line in f:
|
||||
trimeed_line = line.strip()
|
||||
if trimeed_line == "":
|
||||
continue
|
||||
norm_trimed_line = re.sub(r"(\s+)", ",", trimeed_line)
|
||||
sep = norm_trimed_line.split(",")
|
||||
self._content.append(sep)
|
||||
|
||||
def _save_backup(self, back_file_path):
|
||||
shutil.copy(self._file_path, back_file_path)
|
||||
|
||||
def write(self, target_file_path):
|
||||
(file_name, ext) = get_file_name_with_extention(self._file_path)
|
||||
src_file_dir = get_directory(self._file_path)
|
||||
self._save_backup(
|
||||
"{dir}/{file_name}-{time}.{ext}".format(
|
||||
dir=src_file_dir, file_name=file_name, time=file_backup_time(), ext=ext
|
||||
)
|
||||
)
|
||||
with open(target_file_path, "wt") as file:
|
||||
for content in self._content:
|
||||
file.write("{content}\n".format(content=" ".join(content)))
|
||||
|
||||
def has(self, tower_number): # 塔位号是否存在S文件中。
|
||||
for index, content in enumerate(self._content):
|
||||
if content[0] == tower_number:
|
||||
return index
|
||||
return 0
|
||||
|
||||
def content(self):
|
||||
return self._content
|
||||
|
||||
def update_height(self, index, new_height): # 更新内存中S文件中的塔高
|
||||
content = self._content
|
||||
content[index][7] = new_height
|
||||
|
||||
|
||||
class TaFileObject:
|
||||
def __init__(self, file_path, SFile):
|
||||
# self._content = []
|
||||
self._file_path = file_path
|
||||
self._read(file_path)
|
||||
self._old_tower_height_record = None
|
||||
self._SFile = SFile
|
||||
pass
|
||||
|
||||
def _read(self, file_path):
|
||||
contents = []
|
||||
with open(file_path) as f:
|
||||
for line in f:
|
||||
trimeed_line = line.strip()
|
||||
if trimeed_line == "":
|
||||
continue
|
||||
norm_trimed_line = re.sub(r"(\s+)", "", trimeed_line)
|
||||
sep = norm_trimed_line.split(",")
|
||||
contents.append(sep)
|
||||
return contents
|
||||
|
||||
def _make_tower_height_record(self, contents): # 将TA文件读入内存字典中
|
||||
tower_height_record = {}
|
||||
for content in contents:
|
||||
sep = content
|
||||
if len(sep) > 9:
|
||||
tower_number = sep[0]
|
||||
tower_height = sep[9]
|
||||
tower_height_record[tower_number] = tower_height
|
||||
return tower_height_record
|
||||
|
||||
def sync_tower_height(self, new_contents, old_tower_height_record):
|
||||
updated = False
|
||||
for content in new_contents:
|
||||
sep = content
|
||||
if len(sep) > 9:
|
||||
tower_number = sep[0]
|
||||
old_tower_height = float(old_tower_height_record[tower_number])
|
||||
new_tower_height = float(sep[9])
|
||||
if abs(old_tower_height - new_tower_height) > 1e-5:
|
||||
print(
|
||||
"{tower_number} height changes to {new_tower_height}".format(
|
||||
tower_number=tower_number, new_tower_height=new_tower_height
|
||||
)
|
||||
)
|
||||
SFile = self._SFile
|
||||
norm_tower_number = tower_number.replace("G", "")
|
||||
for Ss in SFile:
|
||||
d_file_obj = SFileObject(Ss)
|
||||
index = d_file_obj.has(norm_tower_number)
|
||||
if index > 0:
|
||||
d_file_obj.update_height(index, str(new_tower_height))
|
||||
d_file_obj.write(Ss)
|
||||
print("update S file {Ss}".format(Ss=Ss))
|
||||
break
|
||||
updated = True
|
||||
return updated
|
||||
|
||||
def start(self):
|
||||
file_path = self._file_path
|
||||
while True:
|
||||
new_contents = self._read(file_path) # 每一次循环都会重新读一遍TA文件
|
||||
if self._old_tower_height_record is None:
|
||||
self._old_tower_height_record = self._make_tower_height_record(
|
||||
new_contents
|
||||
)
|
||||
if self.sync_tower_height(new_contents, self._old_tower_height_record):
|
||||
self._old_tower_height_record = self._make_tower_height_record(
|
||||
new_contents
|
||||
)
|
||||
input("press enter to continue\n")
|
||||
|
||||
|
||||
class CordinationObject:
|
||||
def __init__(self, cord_file_path):
|
||||
content = []
|
||||
cord_dic = {}
|
||||
with open(cord_file_path) as cord_file:
|
||||
for line in cord_file:
|
||||
if line.strip() == "":
|
||||
continue
|
||||
sep = line.split(",")
|
||||
J_or_Z=sep[0][0]
|
||||
pile=re.findall(r'[JZ]0*([0-9]+)',sep[0])[0]
|
||||
JZ_pile=J_or_Z+pile
|
||||
# pile = sep[0][1:len(sep[0])]
|
||||
altitude = sep[1]
|
||||
mileage = sep[2]
|
||||
content.append((JZ_pile, altitude, mileage))
|
||||
index = len(content) - 1
|
||||
cord_dic[pile] = index#字典是不带J或者Z的
|
||||
self._content = content
|
||||
self._cord_dic = cord_dic
|
||||
|
||||
def get_altitude(self,pile):
|
||||
cord_dic=self._cord_dic
|
||||
if pile in cord_dic:
|
||||
index=cord_dic[pile]
|
||||
return self._content[index][1]
|
||||
return None
|
||||
def content(self):
|
||||
return self._content
|
||||
|
|
@ -0,0 +1,75 @@
|
|||
from file_object import SFileObject,CordinationObject
|
||||
|
||||
TA_ENTRY_TEMPLATE = "{tower_number},{suspend_or_tension},{mileage}, {conductor_suspension_poit},0.3212,0.000,{altitude},{pile_number},{tower_name}_{tower_height},{tower_height},{string_name},{string_length}, 0.000,0.1142,0.000000,0.000000,,15,15.500000,,{angel},2.5,15.5,2.500,3.000,1,610.4,2.000,0,2,0,2,{conductor_type},GJ-50,0,0,0,0.000,,0,0.000,9,750,1.000,,0,0.000,0,750,1.000,ZHCH_1,28,1.000,9.5,750,1.000,9.5,750,1.000,{string_name},1,{string_length},{string_name},1,{string_length},,0,0.000,,0,0.000,0,0,0.000, 0.000,0,0,0.000,0,0,4.000,0,1,1.000,,0,0.000,,0,0.000,,0,0.000,1,1,0.000,0,0,0.000, 0.000,0,0, 0.000,1.000,1.000,0.000,0.000,0.000,,0,\n"
|
||||
TA_NC_STRING_NAME = "750-NC"
|
||||
TA_NC_STRING_LENGTH = 14
|
||||
TA_NC_TOWER_NAME = 'WNSJ'
|
||||
TA_XC_STRING_NAME = "750-FXBW"
|
||||
TA_XC_STRING_LENGTH = 9.5
|
||||
TA_XC_TOWER_NAME = 'WNSZ'
|
||||
TA_CONDUCTOR_TYPE = "LGJ-400/50"
|
||||
S_NC_STRING_NAME = "2*42-50"
|
||||
# 通过串型来判别是不是耐张塔
|
||||
def is_tension_tower_by_string(string_name):
|
||||
if string_name == S_NC_STRING_NAME:
|
||||
return 2
|
||||
return 1
|
||||
|
||||
|
||||
# SFiles中的文件应该是按顺序排列的`
|
||||
def generate_ta(TA_file, SFiles,cordination_file_path):
|
||||
visited_tower_number={}#记录访问过的塔号
|
||||
cord_file_object=CordinationObject(cordination_file_path)
|
||||
with open(TA_file, "w") as file:
|
||||
# 生成所有需要用到的数据
|
||||
last_S_mileage = 0
|
||||
for S in SFiles:
|
||||
SObj = SFileObject(S)
|
||||
S_content = SObj.content()
|
||||
for S_entry in S_content:
|
||||
if S_entry[0] == "首端转角号" or S_entry[0] == "塔号":
|
||||
continue
|
||||
tower_number = S_entry[0]
|
||||
if tower_number not in visited_tower_number:
|
||||
visited_tower_number[tower_number]=0
|
||||
else:
|
||||
continue
|
||||
mileage = int(S_entry[1]) + last_S_mileage
|
||||
tower_height = float(S_entry[7])
|
||||
angel = S_entry[5]
|
||||
string_name = S_entry[8]
|
||||
suspend_or_tension = is_tension_tower_by_string(string_name)
|
||||
conductor_type = TA_CONDUCTOR_TYPE
|
||||
#1是悬垂 2是耐张
|
||||
if suspend_or_tension == 1:
|
||||
string_length = TA_XC_STRING_LENGTH
|
||||
conductor_suspension_poit = tower_height - string_length
|
||||
pile_number = tower_number
|
||||
ta_string_name=TA_XC_STRING_NAME
|
||||
tower_name=TA_XC_TOWER_NAME
|
||||
else:
|
||||
string_length = TA_NC_STRING_LENGTH
|
||||
conductor_suspension_poit = tower_height
|
||||
pile_number = tower_number
|
||||
ta_string_name = TA_NC_STRING_NAME
|
||||
tower_name = TA_NC_TOWER_NAME
|
||||
altitude = cord_file_object.get_altitude(pile_number)
|
||||
file.write(
|
||||
TA_ENTRY_TEMPLATE.format(
|
||||
tower_number=tower_number,
|
||||
suspend_or_tension=suspend_or_tension,
|
||||
mileage=mileage,
|
||||
conductor_suspension_poit=conductor_suspension_poit,
|
||||
altitude=altitude,
|
||||
pile_number=pile_number,
|
||||
tower_name=tower_name,
|
||||
tower_height=tower_height,
|
||||
string_name=ta_string_name,
|
||||
string_length=string_length,
|
||||
angel=angel,
|
||||
conductor_type=conductor_type,
|
||||
)
|
||||
)
|
||||
last_S_mileage = float(S_content[-1][1])+last_S_mileage
|
||||
|
||||
|
||||
|
|
@ -0,0 +1,15 @@
|
|||
import generate_ta_from_csv
|
||||
from file_object import TaFileObject
|
||||
|
||||
if __name__ == "__main__":
|
||||
SFile = [
|
||||
r"d:\工程\灵州-青山\排位\0-21\S000.DAT",
|
||||
r"d:\工程\灵州-青山\排位\21-52\S021.DAT",
|
||||
r"d:\工程\灵州-青山\排位\52-68\S052.DAT",
|
||||
r"d:\工程\灵州-青山\排位\68-72\S068.DAT",
|
||||
r"d:\工程\灵州-青山\排位\72-169\S072.DAT",
|
||||
]
|
||||
ta_object = TaFileObject(r"d:\工程\灵州-青山\排位\道亨\现场排位\现场排位.TA", SFile)
|
||||
ta_object.start()
|
||||
# generate_ta_from_csv.generate_ta(r"d:\工程\灵州-青山\排位\道亨\最终排位\最终排位.TA",SFile,r'd:\工程\灵州-青山\排位\道亨\成果表.csv')
|
||||
print("Finished.")
|
||||
|
|
@ -0,0 +1,19 @@
|
|||
# 用于处理勘测的文件
|
||||
from file_object import CordinationObject
|
||||
|
||||
# 在道亨的ORG文件里面添加测量的桩位
|
||||
def output_pile(cordination_file, output_file):
|
||||
cord_file_object = CordinationObject(cordination_file)
|
||||
out_content = cord_file_object.content()
|
||||
with open(output_file, "w") as out_file:
|
||||
for c in out_content:
|
||||
out_file.write(
|
||||
"0,{pile},0.0000,{mileage},{altitude}, 96=00+ 96, 0.0000, 0.0000\n".format(
|
||||
pile=c[0], altitude=c[1], mileage=c[2]
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
output_pile(r"d:\工程\灵州-青山\排位\道亨\成果表.csv", r"d:\工程\灵州-青山\排位\道亨\最终排位\org_append.TA")
|
||||
print("survey. Finished.")
|
||||
Loading…
Reference in New Issue