orgsyncd/generate_ta.py

from file_object import SFileObject, CordinationObject, DFileAsSingle

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


def prepare_TA_entry_without_altitude(SFiles):
    visited_tower_number = {}  # 记录访问过的塔号
    last_S_mileage = 0
    ta_entry = []
    for S in SFiles:
        SObj = SFileObject(S)
        S_content = SObj.content()
        for S_entry in S_content:
            _entry = {}
            if S_entry[0] == "首端转角号" or S_entry[0] == "塔号":
                continue
            tower_number = S_entry[0]
            _entry["tower_number"] = tower_number
            if tower_number not in visited_tower_number:
                visited_tower_number[tower_number] = 0
            else:
                continue
            mileage = int(float(S_entry[1])) + last_S_mileage
            _entry["mileage"] = mileage
            tower_height = float(S_entry[7])
            _entry["tower_height"] = tower_height
            angel = S_entry[5]
            _entry["angel"] = angel
            string_name = S_entry[8]
            suspend_or_tension = is_tension_tower_by_string(string_name)
            _entry["suspend_or_tension"] = suspend_or_tension
            conductor_type = TA_CONDUCTOR_TYPE
            _entry["conductor_type"] = conductor_type
            # 1是悬垂 2是耐张
            if suspend_or_tension == 1:
                string_length = TA_XC_STRING_LENGTH
                _entry["string_length"] = string_length
                conductor_suspension_point = tower_height - string_length
                _entry["conductor_suspension_point"] = conductor_suspension_point
                pile_number = tower_number
                _entry["pile_number"] = pile_number
                ta_string_name = TA_XC_STRING_NAME
                _entry["ta_string_name"] = ta_string_name
                tower_name = TA_XC_TOWER_NAME
                _entry["tower_name"] = tower_name
            else:
                string_length = TA_NC_STRING_LENGTH
                _entry["string_length"] = string_length
                conductor_suspension_point = tower_height
                _entry["conductor_suspension_point"] = conductor_suspension_point
                pile_number = tower_number
                _entry["pile_number"] = pile_number
                ta_string_name = TA_NC_STRING_NAME
                _entry["ta_string_name"] = ta_string_name
                tower_name = TA_NC_TOWER_NAME
                _entry["tower_name"] = tower_name
            ta_entry.append(_entry)
        last_S_mileage = float(S_content[-1][1]) + last_S_mileage
    return ta_entry


def write_to_TA_file(TA_file, ta_entry):
    with open(TA_file, "w") as file:
        for entry in ta_entry:
            tower_number = entry["tower_number"]
            suspend_or_tension = entry["suspend_or_tension"]
            mileage = entry["mileage"]
            conductor_suspension_point = entry["conductor_suspension_point"]
            altitude = entry["altitude"]
            pile_number = entry["pile_number"]
            tower_name = entry["tower_name"]
            tower_height = entry["tower_height"]
            ta_string_name = entry["ta_string_name"]
            string_length = entry["string_length"]
            angel = entry["angel"]
            conductor_type = entry["conductor_type"]
            file.write(
                TA_ENTRY_TEMPLATE.format(
                    tower_number=tower_number,
                    suspend_or_tension=suspend_or_tension,
                    mileage=mileage,
                    conductor_suspension_poit=conductor_suspension_point,
                    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,
                )
            )


# SFiles中的文件必须是按顺序排列的
def generate_ta_from_D(TA_file, SFiles, D_files_paths):
    # 准备要写入的文件
    ta_entry_without_altitude = prepare_TA_entry_without_altitude(SFiles)
    d_file_object = DFileAsSingle(D_files_paths)
    ta_entry = list(ta_entry_without_altitude)
    for entry in ta_entry:
        mileage = entry["mileage"]
        entry["altitude"] = d_file_object.get_altitude(mileage)
    write_to_TA_file(TA_file, ta_entry)


# SFiles中的文件必须是按顺序排列的
def generate_ta_from_csv(TA_file, SFiles, cordination_file_path):
    ta_entry_without_altitude = prepare_TA_entry_without_altitude(SFiles)
    cord_file_object = CordinationObject(cordination_file_path)
    ta_entry = list(ta_entry_without_altitude)
    for entry in ta_entry:
        pile_number = entry["tower_number"]
        entry["altitude"] = cord_file_object.get_altitude(pile_number)
    write_to_TA_file(TA_file, ta_entry)