Azure Function에서 CoolProp을 사용하여 Copilot Studio와 연동할 수 있는 API를 만들기 위한 코드를 제공해드리겠습니다. 이 코드는 HTTP 트리거를 사용하여 다양한 유체 물성치를 계산하고, 단위 변환을 지원합니다.
필수 파일 구조
1. requirements.txt
azure-functions
CoolProp
pint
fastjsonschema
2. function_app.py
import azure.functions as func
import logging
import json
from CoolProp.CoolProp import PropsSI, get_global_param_string, get_fluid_param_string
import pint
import fastjsonschema
from typing import Dict, Any, List, Optional
# 앱 설정
app = func.FunctionApp()
# 단위 시스템 초기화
ureg = pint.UnitRegistry()
# 요청 스키마 정의
request_schema = {
"type": "object",
"properties": {
"operation": {"type": "string", "enum": ["property", "units", "fluids", "info"]},
"fluid": {"type": "string"},
"property": {"type": "string"},
"conditions": {
"type": "array",
"items": {
"type": "object",
"properties": {
"parameter": {"type": "string"},
"value": {"type": "number"},
"unit": {"type": "string"}
},
"required": ["parameter", "value", "unit"]
}
},
"output_unit": {"type": "string"}
},
"required": ["operation"]
}
validate_request = fastjsonschema.compile(request_schema)
# 지원되는 물성치 및 파라미터 매핑
PROPERTY_MAP = {
"pressure": "P",
"temperature": "T",
"density": "D",
"enthalpy": "H",
"entropy": "S",
"internal_energy": "U",
"specific_volume": "V",
"quality": "Q",
"cp": "C",
"cv": "O",
"sound_speed": "A",
"viscosity": "V",
"thermal_conductivity": "L"
}
PARAMETER_MAP = {
"pressure": "P",
"temperature": "T",
"density": "D",
"enthalpy": "H",
"entropy": "S",
"quality": "Q"
}
@app.route(route="fluid_properties")
async def fluid_properties(req: func.HttpRequest) -> func.HttpResponse:
logging.info('Python HTTP trigger function processed a request.')
try:
# 요청 본문 파싱
req_body = req.get_json()
validate_request(req_body)
operation = req_body.get("operation")
# 작업 유형에 따른 처리
if operation == "property":
result = calculate_property(req_body)
elif operation == "units":
result = get_supported_units()
elif operation == "fluids":
result = get_supported_fluids()
elif operation == "info":
result = get_fluid_info(req_body.get("fluid"))
else:
return func.HttpResponse(
json.dumps({"error": "Unknown operation"}),
mimetype="application/json",
status_code=400
)
return func.HttpResponse(
json.dumps(result),
mimetype="application/json"
)
except ValueError as e:
return func.HttpResponse(
json.dumps({"error": "Invalid JSON in request body", "details": str(e)}),
mimetype="application/json",
status_code=400
)
except fastjsonschema.exceptions.JsonSchemaException as e:
return func.HttpResponse(
json.dumps({"error": "Invalid request format", "details": str(e)}),
mimetype="application/json",
status_code=400
)
except Exception as e:
logging.error(f"Error processing request: {str(e)}")
return func.HttpResponse(
json.dumps({"error": "Error processing request", "details": str(e)}),
mimetype="application/json",
status_code=500
)
def calculate_property(data: Dict[str, Any]) -> Dict[str, Any]:
"""유체 물성치 계산 함수"""
fluid = data.get("fluid")
property_name = data.get("property")
conditions = data.get("conditions", [])
output_unit = data.get("output_unit")
if not fluid or not property_name or not conditions:
raise ValueError("Missing required parameters: fluid, property, and conditions")
# 물성치 코드와 입력 파라미터 변환
property_code = PROPERTY_MAP.get(property_name.lower())
if not property_code:
raise ValueError(f"Unsupported property: {property_name}")
# 조건 파라미터 변환
inputs = []
for condition in conditions:
param = condition.get("parameter")
value = condition.get("value")
unit = condition.get("unit")
param_code = PARAMETER_MAP.get(param.lower())
if not param_code:
raise ValueError(f"Unsupported parameter: {param}")
# 단위 변환
if unit:
try:
value_with_unit = value * ureg(unit)
# CoolProp 기본 SI 단위로 변환
si_unit = get_si_unit(param.lower())
value = value_with_unit.to(si_unit).magnitude
except:
raise ValueError(f"Invalid unit: {unit} for parameter: {param}")
inputs.append((param_code, value))
# CoolProp 호출하여 물성치 계산
try:
# 조건 파라미터 구성
props = []
vals = []
for param, val in inputs:
props.append(param)
vals.append(val)
# PropsSI 호출
result_value = PropsSI(property_code, props, vals, props, vals, fluid)
# 출력 단위 변환
if output_unit:
try:
result_with_unit = result_value * ureg(get_si_unit(property_name.lower()))
result_value = result_with_unit.to(output_unit).magnitude
except:
raise ValueError(f"Invalid output unit: {output_unit} for property: {property_name}")
return {
"fluid": fluid,
"property": property_name,
"value": result_value,
"unit": output_unit if output_unit else get_si_unit(property_name.lower())
}
except Exception as e:
raise Exception(f"Error calculating property: {str(e)}")
def get_si_unit(property_name: str) -> str:
"""물성치에 대한 SI 단위 반환"""
unit_map = {
"pressure": "pascal",
"temperature": "kelvin",
"density": "kg/m^3",
"enthalpy": "J/kg",
"entropy": "J/kg/K",
"internal_energy": "J/kg",
"specific_volume": "m^3/kg",
"quality": "dimensionless",
"cp": "J/kg/K",
"cv": "J/kg/K",
"sound_speed": "m/s",
"viscosity": "Pa*s",
"thermal_conductivity": "W/m/K"
}
return unit_map.get(property_name, "dimensionless")
def get_supported_units() -> Dict[str, Any]:
"""지원되는 단위 목록 반환"""
return {
"pressure": ["Pa", "kPa", "MPa", "bar", "atm", "psi"],
"temperature": ["K", "C", "F", "R"],
"density": ["kg/m^3", "g/cm^3", "kg/L", "lb/ft^3"],
"enthalpy": ["J/kg", "kJ/kg", "BTU/lb"],
"entropy": ["J/kg/K", "kJ/kg/K", "BTU/lb/R"],
"specific_volume": ["m^3/kg", "L/kg", "ft^3/lb"],
"viscosity": ["Pa*s", "cP", "lb/ft/s"],
"thermal_conductivity": ["W/m/K", "BTU/hr/ft/F"]
}
def get_supported_fluids() -> Dict[str, Any]:
"""CoolProp에서 지원하는 유체 목록 반환"""
fluids_str = get_global_param_string("fluids_list")
fluids = fluids_str.split(',')
# 유체 분류
categories = {
"refrigerants": [],
"hydrocarbons": [],
"inorganic": [],
"other": []
}
for fluid in fluids:
if fluid.startswith('R') and any(c.isdigit() for c in fluid):
categories["refrigerants"].append(fluid)
elif fluid in ['Methane', 'Ethane', 'Propane', 'Butane', 'Pentane', 'Hexane', 'Heptane', 'Octane']:
categories["hydrocarbons"].append(fluid)
elif fluid in ['Water', 'Ammonia', 'CarbonDioxide', 'Nitrogen', 'Oxygen', 'Hydrogen']:
categories["inorganic"].append(fluid)
else:
categories["other"].append(fluid)
return categories
def get_fluid_info(fluid: str) -> Dict[str, Any]:
"""특정 유체에 대한 정보 반환"""
if not fluid:
raise ValueError("Fluid name is required")
try:
# 기본 정보 수집
critical_temperature = PropsSI("Tcrit", fluid)
critical_pressure = PropsSI("pcrit", fluid)
molar_mass = PropsSI("molar_mass", fluid)
# 삼중점 정보 (있는 경우)
try:
triple_temperature = PropsSI("T_triple", fluid)
triple_pressure = PropsSI("p_triple", fluid)
except:
triple_temperature = None
triple_pressure = None
return {
"name": fluid,
"critical_point": {
"temperature": {
"value": critical_temperature,
"unit": "K"
},
"pressure": {
"value": critical_pressure,
"unit": "Pa"
}
},
"triple_point": {
"temperature": {
"value": triple_temperature,
"unit": "K"
} if triple_temperature else None,
"pressure": {
"value": triple_pressure,
"unit": "Pa"
} if triple_pressure else None
},
"molar_mass": {
"value": molar_mass,
"unit": "kg/mol"
}
}
except Exception as e:
raise Exception(f"Error getting fluid information: {str(e)}")
3. host.json
{
"version": "2.0",
"logging": {
"applicationInsights": {
"samplingSettings": {
"isEnabled": true,
"excludedTypes": "Request"
}
}
},
"extensionBundle": {
"id": "Microsoft.Azure.Functions.ExtensionBundle",
"version": "[3.*, 4.0.0)"
},
"concurrency": {
"dynamicConcurrencyEnabled": true,
"snapshotPersistenceEnabled": true
}
}
사용 방법 설명
이 Azure Function은 다음과 같은 기능을 제공합니다:
물성치 계산 (operation: "property")
- 특정 유체의 물성치를 주어진 조건에서 계산합니다
- 다양한 공학 단위 지원 및 단위 변환
지원되는 단위 조회 (operation: "units")
- 각 물성치에 대해 지원되는 단위 목록 반환
지원되는 유체 목록 (operation: "fluids")
- CoolProp에서 지원하는 유체 목록 카테고리별 제공
유체 정보 조회 (operation: "info")
- 특정 유체에 대한 상세 정보 제공 (임계점, 삼중점, 몰질량 등)
API 사용 예시
// 물성치 계산 예시 요청
{
"operation": "property",
"fluid": "Water",
"property": "density",
"conditions": [
{
"parameter": "temperature",
"value": 25,
"unit": "C"
},
{
"parameter": "pressure",
"value": 101325,
"unit": "Pa"
}
],
"output_unit": "kg/m^3"
}
// 유체 목록 조회
{
"operation": "fluids"
}
// 단위 목록 조회
{
"operation": "units"
}
// 유체 정보 조회
{
"operation": "info",
"fluid": "Water"
}
Microsoft Copilot Studio 연동 방법
- Azure 포털에서 Function을 배포하고 URL 엔드포인트를 얻습니다
- Copilot Studio에서:
- Custom connector 생성
- 인증 방식 설정 (API 키 또는 OAuth)
- 액션 정의 (위 예시와 같은 요청/응답 형식)
- 대화 흐름에 액션 통합
이 코드는 다양한 유체 물성치 계산과 단위 변환을 지원하며, Copilot에서 사용자가 자연어로 유체 물성치에 관한 질문을 할 때 적절한 응답을 제공할 수 있습니다.
댓글
댓글 쓰기