PP Converter

mail@pastecode.io avatarunknown
python
a month ago
9.5 kB
6
Indexable
Never
class pandapipesConverter:

    def __init__(self, nodes, edges, buildings, house_connections, heat_generation, heat_generation_node):

        self.__grid = pp.create_empty_network(fluid="water")
        self.__nodes = nodes
        self.__edges = edges
        self.__buildings = buildings
        self.__house_connections = house_connections
        self.__heat_generation = heat_generation
        self.__heat_generation_node = heat_generation_node
        self.__settings()

    def __settings(self):

        self.__nominal_diameters_in_mm = [20, 25, 32, 40, 50, 65, 80, 100, 125, 150, 200, 250, 300, 350, 400]
        self.__c_p_water_in_kj_per_kgk = 4.817
        self.__rho_water_in_kg_per_m3 = 1000
        self.__nominal_pressure_in_bar = 16
        self.__vl_temperature_in_k = 273.15 + 110
        self.__delta_temperature_in_k = 40

    def create_grid(self):

        self.__add_junctions()
        self.__add_pipes()
        self.__add_heat_exchangers()
        self.__add_house_connections()
        if self.__heat_generation == "pump":
            self.__add_pump()

    def __add_junctions(self):

        self.__junctions_vl, self.__junctions_rl = {}, {}
        for index, row in self.__nodes.iterrows():
            for vr, j_vr in zip(["vl", "rl"], [self.__junctions_vl, self.__junctions_rl]):
                j_vr.update({row["osmid"]: pp.create_junction(net=self.__grid,
                                                              pn_bar=self.__nominal_pressure_in_bar,
                                                              tfluid_k=TFLUID_K,
                                                              name=f"{row['osmid']}_{row['highway']}_{vr}",
                                                              geodata=(row["x"], row["y"]))})

    def __calculate_pipe_nominal_diameter(self, p_in_kw):

        def nominal_diamter(d_in_mm):
            return [d for d in self.__nominal_diameters_in_mm if d >= d_in_mm][0]

        v_in_m_per_s = 1.1
        d_in_m = np.sqrt(4 * p_in_kw / (np.pi * v_in_m_per_s * self.__delta_temperature_in_k * self.__c_p_water_in_kj_per_kgk * self.__rho_water_in_kg_per_m3))

        return nominal_diamter(d_in_m * 1000)

    def __add_pipes(self):

        self.__pipes_vl, self.__pipes_rl = {}, {}
        for index, row in self.__edges.iterrows():
            for vr, p_vr in zip(["vl", "rl"], [self.__pipes_vl, self.__pipes_rl]):
                p_vr.update({row["osmid"]: pp.create_pipe_from_parameters(net=self.__grid,
                                                                          length_km=row["length"] / 1000,
                                                                          geodata=list(row["geometry"].coords),
                                                                          diameter_m=self.__calculate_pipe_nominal_diameter(row["p_in_kw"]) / 1000,
                                                                          k_mm=0.01,
                                                                          alpha_w_per_m2k=10,
                                                                          sections=1,
                                                                          text_k=283,
                                                                          name=f"({row['u']},{row['v']})_street_{vr}",
                                                                          from_junction=self.__junctions_vl[row["u"]] if vr== "vl" else self.__junctions_rl[row["v"]],
                                                                          to_junction=self.__junctions_vl[row["v"]] if vr == "vl" else self.__junctions_rl[row["u"]])
                             })

            self.__edges.at[index, "DN"] = self.__calculate_pipe_nominal_diameter(row["p_in_kw"])

    def __add_heat_exchangers(self):

        self.__heat_exchangers = {}
        self.__junctions_building_vl, self.__junctions_building_rl = {}, {}
        for index, row in self.__buildings.iterrows():
            x, y = self.__house_connections.at[index, "x_b"], self.__house_connections.at[index, "y_b"]
            self.__junctions_building_vl.update({row["osmid"]: pp.create_junction(net=self.__grid,
                                                                                  pn_bar=self.__nominal_pressure_in_bar,
                                                                                  tfluid_k=self.__vl_temperature_in_k,
                                                                                  name=f"hc_{row['osmid']}_vl",
                                                                                  geodata=(x, y))})
            self.__junctions_building_rl.update({row["osmid"]: pp.create_junction(net=self.__grid,
                                                                                  pn_bar=self.__nominal_pressure_in_bar,
                                                                                  tfluid_k=self.__vl_temperature_in_k,
                                                                                  name=f"hc_{row['osmid']}_rl",
                                                                                  geodata=(x, y))})
            self.__heat_exchangers.update({row["osmid"]: pp.create_heat_exchanger(net=self.__grid,
                                                                                  diameter_m=self.__calculate_pipe_nominal_diameter(row["P_peak"]+row["P_base"])/1000,
                                                                                  qext_w=(row["P_peak"] + row["P_base"]) * 1000,
                                                                                  name=row["osmid"],
                                                                                  from_junction=self.__junctions_building_vl[row["osmid"]],
                                                                                  to_junction=self.__junctions_building_rl[row["osmid"]])
                                           })

    def __add_house_connections(self):

        self.__house_connections_vl, self.__house_connections_rl = {}, {}
        for index, row in self.__house_connections.iterrows():
            street_node_osmid = self.__nodes[self.__nodes["connected_buildings"].str.contains(str(row["building_osmid"]), na=False)]
                                    .reset_index(drop=True).at[0, "osmid"]
            for vr, hc_vr in zip(["vl", "rl"], [self.__house_connections_vl, self.__house_connections_rl]):
                hc_vr.update({row["building_osmid"]: pp.create_pipe_from_parameters(net=self.__grid,
                                                                                    length_km=row["length"] / 1000,
                                                                                    geodata=list(row["geometry"].coords),
                                                                                    diameter_m=self.__calculate_pipe_nominal_diameter(self.__buildings.at[index,                       "P_peak"] + self.__buildings.at[index, "P_base"]) / 1000,
                                                                                    k_mm=0.01,
                                                                                    alpha_w_per_m2k=10,
                                                                                    sections=1,
                                                                                    text_k=283,
                                                                                    name=f"hc_{row['building_osmid']}_{vr}",
                                                                                    from_junction=self.__junctions_vl[street_node_osmid] if vr == "vl" 
                                                                                                          else self.__junctions_building_rl[row["building_osmid"]],
                                                                                    to_junction=self.__junctions_building_vl[row["building_osmid"]] if vr == "vl" 
                                                                                                          else self.__junctions_rl[street_node_osmid])
                              })
            self.__house_connections.at[index, "DN"] = self.__calculate_pipe_nominal_diameter(self.__buildings.at[index, "P_peak"] 
                                                                                                    + self.__buildings.at[index, "P_base"])

    def __add_pump(self):

        p_in_kw = self.__buildings["P_peak"].sum() + self.__buildings["P_base"].sum()
        self.__mass_flow_in_kg_per_s = p_in_kw / (self.__c_p_water_in_kj_per_kgk * self.__delta_temperature_in_k)

        pp.create_circ_pump_const_mass_flow(net=self.__grid,
                                            p_flow_bar=self.__nominal_pressure_in_bar,
                                            mdot_flow_kg_per_s=self.__mass_flow_in_kg_per_s,
                                            t_flow_k=self.__vl_temperature_in_k,
                                            flow_junction=self.__junctions_vl[self.__heat_generation_node],
                                            return_junction=self.__junctions_rl[self.__heat_generation_node])


    def start_pipeflow(self):

        pp.pipeflow(self.__grid, mode="all")