-Aus der Datenbank werden Gasnamen und Kühlerdaten ausgelesen, diese werden an den Browser geschickt.
-Der User wählt ein oder mehrere Gase und Kühler aus, gibt weitere Parameter an und schickt die Daten zurück.
-Die ankommenden Daten werden überprüft, wenn die in Ordnung sind, werden Berechnungen ausgeführt und das Ergebnis im Browser als PDF angezeigt.
-Bei Bedarf kann der User seine eingegeben Daten jetzt in der Datenbank sichern.
Ich dachte, dass das zu unübersichtlich sei, aber ihr seht das vielleicht anders, daher hier mal die vollständige Datei:
Code: Alles auswählen
#!/usr/bin/env python
from json import loads
from math import isclose
from pathlib import Path
from flask_weasyprint import render_pdf
from attrs import define, field
from attrs.validators import (
and_,
ge as is_greater_or_equal,
gt as is_greater,
in_,
le as is_less_or_equal,
min_len,
)
from cattrs import structure, unstructure
from cooler_calculation import calculate_cooler
from flask import Flask, jsonify, Response
from flask_classful import FlaskView, request, route
from flask_cors import CORS
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import text
from icecream import ic
from make_template import make_template
TECHNICAL_DATA_PATH = Path(__file__).parent / "TechnicalData"
COOLER_DATA_FILE = TECHNICAL_DATA_PATH / "CoolerData.json"
VALID_GASES_FILE = TECHNICAL_DATA_PATH / "gases.json"
KELVIN_DIFFERENCE = 273.15
BAR_PA_FACTOR = 1e5
class Base(DeclarativeBase):
pass
database = SQLAlchemy(model_class=Base)
class Cooler(database.Model):
__tablename__ = "Cooler"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(unique=True)
inside_diameter_outer_pipe: Mapped[float] = mapped_column()
outside_diameter_cooling_pipe: Mapped[float] = mapped_column()
inside_diameter_cooling_pipe: Mapped[float] = mapped_column()
number_of_pipes: Mapped[float] = mapped_column()
class Gas(database.Model):
__tablename__ = "Gas"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(unique=True)
class OrderCoolerLink(database.Model):
__tablename__ = "OrderCoolerLink"
id: Mapped[int] = mapped_column(primary_key=True)
cooler_id: Mapped[int] = mapped_column()
order_id: Mapped[int] = mapped_column()
cooling_countercurrent: Mapped[bool] = mapped_column()
gas_humidity_after_cooler: Mapped[float] = mapped_column()
gas_humidity_before_cooler: Mapped[float] = mapped_column()
gas_pressure: Mapped[float] = mapped_column()
gas_temperature_in: Mapped[float] = mapped_column()
gas_temperature_out: Mapped[float] = mapped_column()
temperature_cooling_fluid_in: Mapped[float] = mapped_column()
temperature_cooling_fluid_out: Mapped[float] = mapped_column()
number_of_cooler: Mapped[int] = mapped_column()
class OrderGasLink(database.Model):
__tablename__ = "OrderGasLink"
id: Mapped[int] = mapped_column(primary_key=True)
order_id: Mapped[int] = mapped_column(unique=True)
gas_id: Mapped[int] = mapped_column()
percentage: Mapped[float] = mapped_column()
gas_constant: Mapped[float] = mapped_column()
molar_mass: Mapped[float] = mapped_column()
class Order(database.Model):
__tablename__ = "Order"
id: Mapped[int] = mapped_column(primary_key=True)
order_number: Mapped[int] = mapped_column(unique=True)
glykohl_percent: Mapped[float] = mapped_column()
normvolume_flow: Mapped[float] = mapped_column()
suction_pressure: Mapped[float] = mapped_column()
temperature_suction_filter: Mapped[float] = mapped_column()
special_gas: Mapped[bool] = mapped_column()
def _mm_to_m(value):
return value / 1e3
def _celsius_to_kelvin(value):
return value + KELVIN_DIFFERENCE
def _bar_to_pa(value):
return value * BAR_PA_FACTOR
@define
class General:
overwrite: bool = field()
order_number: int = field()
glykohl_percent: float = field(
validator=and_(is_greater_or_equal(0), is_less_or_equal(100))
)
normvolume_flow: float = field(validator=is_greater(1))
suction_pressure: float = field(
validator=and_(is_greater_or_equal(0.5 * 1e5), is_less_or_equal(1000 * 1e5)),
converter=_bar_to_pa,
)
temperature_suction_filter: float = field(
validator=and_(
is_greater_or_equal(-20 + KELVIN_DIFFERENCE),
is_less_or_equal(300 + KELVIN_DIFFERENCE),
),
converter=_celsius_to_kelvin,
)
special_gas: bool = field()
@define
class GasData:
id: int = field()
name: str = field(
validator=in_(loads(VALID_GASES_FILE.read_bytes())["valid_gases"])
)
percent: float = field(validator=and_(is_greater(0), is_less_or_equal(100)))
gas_constant: float = field(validator=is_greater_or_equal(0))
molar_mass: float = field(validator=is_greater_or_equal(0))
@define
class CoolerData:
id: int = field()
name: str = field()#validator=in_(list(loads(COOLER_DATA_FILE.read_bytes()).keys())))
cooling_countercurrent: bool
gas_humidity_after_cooler: float = field(
validator=and_(is_greater_or_equal(0), is_less_or_equal(100))
)
gas_humidity_before_cooler: float = field(
validator=and_(is_greater_or_equal(0), is_less_or_equal(100))
)
gas_pressure: float = field(
validator=and_(
is_greater_or_equal(0.5 * BAR_PA_FACTOR),
is_less_or_equal(1000 * BAR_PA_FACTOR),
),
converter=_bar_to_pa,
)
gas_temperature_in: float = field(
validator=and_(
is_greater_or_equal(-20 + KELVIN_DIFFERENCE),
is_less_or_equal(300 + KELVIN_DIFFERENCE),
),
converter=_celsius_to_kelvin,
)
gas_temperature_out: float = field(
validator=and_(
is_greater_or_equal(-25 + KELVIN_DIFFERENCE),
is_less_or_equal(300 + KELVIN_DIFFERENCE),
),
converter=_celsius_to_kelvin,
)
inside_diameter_cooling_pipe: float = field(
validator=and_(is_greater_or_equal(0.001), is_less_or_equal(0.015)),
converter=_mm_to_m,
)
inside_diameter_outer_pipe: float = field(converter=_mm_to_m)
number_of_pipes: float
outside_diameter_cooling_pipe: float = field(converter=_mm_to_m)
temperature_cooling_fluid_in: float = field(
validator=and_(
is_greater_or_equal(-20 + KELVIN_DIFFERENCE),
is_less_or_equal(90 + KELVIN_DIFFERENCE),
),
converter=_celsius_to_kelvin,
)
temperature_cooling_fluid_out: float = field(
validator=and_(
is_greater_or_equal(-20 + KELVIN_DIFFERENCE),
is_less_or_equal(90 + KELVIN_DIFFERENCE),
),
converter=_celsius_to_kelvin,
)
number_of_cooler: float = field(validator=is_less_or_equal(1))
@define
class Data:
gases: list[GasData] = field()
general: General
coolers: list[CoolerData] = field(validator=min_len(1))
@gases.validator
def _validate_gases(self, _attribute, value):
total = sum(gas.percent for gas in value)
if not isclose(100, total):
raise ValueError(f"total percent must be 100, not {total}")
@coolers.validator
def _validate_outside_diameter(self, _attribute, coolers):
for cooler in coolers:
if (
cooler.inside_diameter_cooling_pipe
>= cooler.outside_diameter_cooling_pipe
):
raise ValueError(
"Inside diameter of cooling pipe can't be bigger or equal outside diameter"
)
@coolers.validator
def _validate_gas_temperature_out(self, _attribute, coolers):
for cooler in coolers:
if cooler.gas_temperature_out <= cooler.temperature_cooling_fluid_in:
raise ValueError(
"You can't cool down your gas as cooler than your cooling fluid!"
)
@define
class App(FlaskView):
database = field()
user_input = None
results = None
request_id = None
@route("/get_cooler_data", methods=["GET"])
def get_cooler_data(self):
if request.method == "GET":
# TODO Replace `text()`
rows = self.database.session.execute(text("SELECT * FROM Cooler"))
return format_to_dict([row._asdict() for row in rows])
@route("/get_valid_gases", methods=["GET"])
def get_valid_gases(self):
if request.method == "GET":
# TODO Replace `text()`
rows = self.database.session.execute(text("SELECT * FROM Gas"))
return format_to_dict([row._asdict() for row in rows])
@route("/process_calculation", methods=["POST", "GET"])
def index(self):
if request.method == "POST":
input_data = request.get_json()
ic(input_data)
self.process_input(input_data)
return jsonify(request.get_json())
if request.method == "GET":
data = self.merge_result_and_input()
return render_pdf(make_template(data))
@route("/get_order", methods=["GET"])
def get_order(self):
session = self.database.session
with session.begin():
rows = session.execute(database.select(Order)).scalars()
return {row.order_number: row.id for row in rows}
@route("get_order_details", methods=["POST", "GET"])
def get_order_details(self):
if request.method == "POST":
self.request_id = request.get_json()["id"]
return Response(status=200)
elif request.method == "GET":
order = Order.query.filter(Order.id == self.request_id).scalar()
general = {
"glykohl_percent": order.glykohl_percent,
"normvolume_flow": order.normvolume_flow,
"order_number": order.order_number,
"suction_pressure": order.suction_pressure,
"temperature_suction_filter": order.temperature_suction_filter,
"special_gas": order.special_gas
}
ic(general)
gases = []
for row in OrderGasLink.query.filter(
OrderGasLink.order_id == self.request_id
).all():
gas = Gas.query.filter(row.gas_id == Gas.id).scalar()
gases.append(
{"id": row.gas_id, "name": gas.name, "percent": row.percentage}
)
coolers = []
for row in OrderCoolerLink.query.filter(
OrderCoolerLink.order_id == self.request_id
).all():
cooler = Cooler.query.filter(row.cooler_id == Cooler.id).scalar()
coolers.append(
{
"cooling_countercurrent": bool(row.cooling_countercurrent),
"gas_humidity_after_cooler": row.gas_humidity_after_cooler,
"gas_humidity_before_cooler": row.gas_humidity_before_cooler,
"gas_pressure": row.gas_pressure,
"gas_temperature_in": row.gas_temperature_in,
"gas_temperature_out": row.gas_temperature_out,
"id": row.cooler_id,
"inside_diameter_cooling_pipe": cooler.inside_diameter_cooling_pipe,
"inside_diameter_outer_pipe": cooler.inside_diameter_outer_pipe,
"name": cooler.name,
"number_of_cooler": row.number_of_cooler,
"number_of_pipes": cooler.number_of_pipes,
"outside_diameter_cooling_pipe": cooler.outside_diameter_cooling_pipe,
"temperature_cooling_fluid_in": row.temperature_cooling_fluid_in,
"temperature_cooling_fluid_out": row.temperature_cooling_fluid_out,
}
)
return {"coolers": coolers, "gases": gases, "general": general}
@route("/save_input_data", methods=["POST"])
def save_input_data(self):
input_data = structure(request.get_json(), Data)
if input_data.general.overwrite:
self.delete_existing_rows(input_data)
self.save_order(input_data)
return Response(status=200)
def save_order(self, input_data):
session = self.database.session
with session.begin():
order = Order(
order_number=input_data.general.order_number,
glykohl_percent=input_data.general.glykohl_percent,
normvolume_flow=input_data.general.normvolume_flow,
suction_pressure=input_data.general.suction_pressure / BAR_PA_FACTOR,
temperature_suction_filter=input_data.general.temperature_suction_filter
- KELVIN_DIFFERENCE,
special_gas=input_data.general.special_gas
)
session.add(order)
session.flush()
for cooler in input_data.coolers:
order_cooler_link = OrderCoolerLink(
cooler_id=cooler.id,
order_id=order.id,
cooling_countercurrent=cooler.cooling_countercurrent,
gas_humidity_after_cooler=cooler.gas_humidity_after_cooler,
gas_humidity_before_cooler=cooler.gas_humidity_before_cooler,
gas_pressure=cooler.gas_pressure / BAR_PA_FACTOR,
gas_temperature_in=cooler.gas_temperature_in - KELVIN_DIFFERENCE,
gas_temperature_out=cooler.gas_temperature_out - KELVIN_DIFFERENCE,
temperature_cooling_fluid_in=cooler.temperature_cooling_fluid_in
- KELVIN_DIFFERENCE,
temperature_cooling_fluid_out=cooler.temperature_cooling_fluid_out
- KELVIN_DIFFERENCE,
number_of_cooler=cooler.number_of_cooler,
)
session.add(order_cooler_link)
for gas in input_data.gases:
order_gas_link = OrderGasLink(
order_id=order.id,
gas_id=gas.id,
percentage=gas.percent,
gas_constant=gas.gas_constant,
molar_mass=gas.molar_mass
)
session.add(order_gas_link)
def delete_existing_rows(self, input_data):
session = self.database.session
with session.begin():
order_id = (
Order.query.filter(
Order.order_number == input_data.general.order_number
)
.scalar()
.id
)
session.query(Order).filter(Order.id == order_id).delete()
session.query(OrderCoolerLink).filter(
OrderCoolerLink.order_id == order_id
).delete()
session.query(OrderGasLink).filter(
OrderGasLink.order_id == order_id
).delete()
session.flush()
def merge_result_and_input(self):
user_input = unstructure(self.user_input)
results = {"results": unstructure(self.results)}
return user_input | results
def process_input(self, input_data):
try:
self.user_input = structure(input_data, Data)
except ValueError as error:
ic(error)
else:
gas = {gas.name: gas.percent for gas in self.user_input.gases}
if self.user_input.general.special_gas:
self.results = calculate_cooler(
self.user_input.coolers,
gas,
self.user_input.general.glykohl_percent,
self.user_input.general.normvolume_flow,
self.user_input.general.temperature_suction_filter,
self.user_input.general.suction_pressure,
self.user_input.general.special_gas,
# Index 0 because it's the same for all gases
self.user_input.gases[0].gas_constant,
self.user_input.gases[0].molar_mass,
)
else:
self.results = calculate_cooler(
self.user_input.coolers,
gas,
self.user_input.general.glykohl_percent,
self.user_input.general.normvolume_flow,
self.user_input.general.temperature_suction_filter,
self.user_input.general.suction_pressure,
self.user_input.general.special_gas
)
def format_to_dict(rows):
return {
row["name"]: {key: value for key, value in row.items() if key != "name"}
for row in rows
}
def main():
app = Flask(__name__)
app.config.from_object(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = (
"mysql+pymysql://dennis:Dennis89@127.0.0.1:3306/CoolerCalculation"
)
CORS(app, resources={r"/*": {"origins": "*"}})
database.init_app(app)
App.register(
app,
route_base="/",
init_argument=database,
)
app.run()
if __name__ == "__main__":
main()
