1.1 데이터 서비스 API 활성화
REST 인터페이스는 Observer Monitor Service에서 제공됩니다. REST 인터페이스를 사용하려면 Monitor 서비스가 설치, 시작, 그리고 구성되어 있어야 합니다.
REST 서비스에 대한 설정은 Observer 프로그램 내에서 다음 경로에서 수행할 수 있습니다: Database > Options > Monitor Service > Phoenix Internal Web Service
설정 방법
Enabled 선택 REST 엔드포인트를 활성화하려면 Enabled 옵션을 선택합니다.
인증 키 생성
Generate Authentication Key 버튼을 클릭하여 영구 인증 키를 생성합니다.
이 키는 인증 과정에서 디지털 서명을 생성하고, 권한 확인 시 디지털 서명을 검증하는 데 사용됩니다.
만약 영구 키를 생성하지 않으면, Monitor가 시작될 때마다 임시 키가 생성됩니다.
리스닝 URL 할당
REST 엔드포인트가 수신 요청을 받을 프로토콜, 주소, 포트를 지정합니다. 여러 URL을 사용할 경우 쉼표(,)로 구분하여 입력할 수 있습니다.
예시:
https://localhost:14050 → loopback(로컬호스트) 인터페이스의 14050 포트에서 HTTPS 프로토콜로 수신
https://10.11.12.13:8080 → 네트워크 인터페이스 10.11.12.13의 8080 포트에서 HTTPS 프로토콜로 수신
https://computer.dns.name:8080 → DNS 이름이 computer.dns.name인 인터페이스의 8080 포트에서 HTTPS 프로토콜로 수신
여러 URL을 조합 가능:
https://computer.dns.name:8080, https://10.11.12.13:8080, https://localhost:14050
URL 구성 요소: <프로토콜>://<IP 주소 또는 호스트명>:<포트 번호>
주요 설명: 프로토콜 권장 프로토콜은 https입니다. 암호화된 통신이므로 보안이 우수합니다. http도 지원되긴 하지만 사용자명, 비밀번호 등 민감 정보가 오가기 때문에 권장하지 않습니다. 참고로, https를 사용하려면 추가적인 윈도우 서버 설정이 필요하며, 이는 1.4 HTTPS 장에서 설명합니다.
IP 주소 또는 호스트명 REST 서비스가 수신 대기할 네트워크 인터페이스를 지정합니다. 일반적으로 물리적 네트워크 인터페이스 카드(NIC)를 사용하지만, loopback, VPN 인터페이스 등 소프트웨어 인터페이스도 가능합니다.
예시: loopback,computername.companyname.com,192.168.1.10
특별히 * (와일드카드 인터페이스)도 있는데, 이는 모든 인터페이스를 의미하므로 주의해서 사용해야 합니다.
포트 번호 REST 인터페이스가 요청을 수신할 포트를 지정합니다. 기본 포트는 14050이며, 사용 중이지 않은 다른 포트도 사용할 수 있습니다. 방화벽에서 해당 포트가 열려 있는지 반드시 확인하십시오.
클라우드 환경 설정 (옵션) 클라우드 환경에서 SKF Rail Track Monitoring과 함께 사용할 경우, Public URL은 외부에서 접근할 수 있는 공용 URL을 의미합니다. 클라우드 제공업체는 공용 IP 또는 DNS 이름을 IaaS 인스턴스에 매핑합니다.
설정 저장
OK 버튼을 클릭하여 설정을 저장합니다.
설정 적용 확인
Observer Monitor는 수 초 내에 변경 사항을 감지하고 REST 인터페이스를 오픈합니다. 적용 여부는 Observer Monitor의 로그 파일을 통해 확인할 수 있습니다.
⚠ 참고:
설정한 포트가 이미 다른 프로세스에 의해 사용 중일 수 있습니다. 이 경우 Observer Monitor의 로그에 관련 내용이 기록되며, 다른 사용 가능한 포트를 선택하거나 해당 프로세스를 종료해야 합니다.
🔑 인증 방식 요약 OAuth2.0 Password Grant 사용
토큰 요청 → access_token + refresh_token 획득
API 호출 시 Authorization: Bearer 헤더 포함
토큰 만료시 refresh_token으로 재발급 가능
[ ]
# 패키지 설치 (한 번만 설치 , 이미 설치되어 있으면 생략 가능)
!pip install requests urllib3
Requirement already satisfied: requests in /usr/local/lib/python3.11/dist-packages (2.32.3)
Requirement already satisfied: urllib3 in /usr/local/lib/python3.11/dist-packages (2.4.0)
Requirement already satisfied: charset-normalizer<4,>=2 in /usr/local/lib/python3.11/dist-packages (from requests) (3.4.2)
Requirement already satisfied: idna<4,>=2.5 in /usr/local/lib/python3.11/dist-packages (from requests) (3.10)
Requirement already satisfied: certifi>=2017.4.17 in /usr/local/lib/python3.11/dist-packages (from requests) (2025.4.26)
[ ]
import requests
import urllib3
import json
import time
import os
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
BASE_URL = "https://localhost:14050"
USERNAME = "admin"
PASSWORD = "admin"
TOKEN_FILE = "./access_token.json"
# ✅ 토큰 저장 함수
def save_token(token_info):
token_info['issued_at'] = time.time() # 저장 시각 기록
save_paths = [
"./access_token.json",
"mnt/access_token.json"
]
for path in save_paths:
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "w") as f:
json.dump(token_info, f, indent=2)
print(f"✅ Access token 저장 완료: {path}")
# ✅ 토큰 발급 함수
def request_token():
url = f"{BASE_URL}/token"
headers = {"Content-Type": "application/x-www-form-urlencoded"}
data = {
"username": USERNAME,
"password": PASSWORD,
"grant_type": "password"
}
try:
response = requests.post(url, headers=headers, data=data, verify=False)
if response.status_code == 200:
token_info = response.json()
save_token(token_info)
return token_info
else:
print(f"❌ 요청 실패 (HTTP {response.status_code}): {response.text}")
return None
except Exception as e:
print("❌ 요청 중 예외 발생:", str(e))
return None
# ✅ 토큰 불러오기 함수
def load_token():
if not os.path.exists(TOKEN_FILE):
print("⚠ 토큰 파일이 존재하지 않습니다. 신규 발급을 시도합니다.")
return request_token()
with open(TOKEN_FILE, "r") as f:
token_info = json.load(f)
return token_info
# ✅ API 호출 함수 예시
def get_asset_list():
token_info = load_token()
if not token_info:
print("❌ 토큰이 없습니다.")
return
access_token = token_info.get("access_token")
url = f"{BASE_URL}/API/DataService/GetAssetList/"
headers = {
"Authorization": f"Bearer {access_token}",
"Accept": "application/json;v=1.0"
}
try:
response = requests.get(url, headers=headers, verify=False)
if response.status_code == 200:
assets = response.json()
print("✅ Asset List 조회 성공!")
print(json.dumps(assets, indent=2))
else:
print(f"❌ API 호출 실패 (HTTP {response.status_code}): {response.text}")
except Exception as e:
print("❌ API 호출 중 예외 발생:", str(e))
# ✅ 전체 테스트 실행
if __name__ == "__main__":
get_asset_list()
3.1 Asset List & Asset 조회
[ ]
# ✅ 3.1 Asset List 조회
def get_asset_list(include_acknowledged: bool = False):
params = {
"IncludeAcknowledged": str(include_acknowledged).lower()
}
url = f"{BASE_URL}/API/DataService/GetAssetList"
try:
response = requests.get(url, headers=headers, params=params, verify=False)
response.raise_for_status()
data = response.json()
except Exception as e:
print(f'에러 발생 : {e} → 임시 데이터 사용')
data = [
{
"ID": 57,
"Name": "Turbine G1",
"Description": "Machine containing MasCon48 measurements",
"Path": "Company\\Example Data\\MasCon48\\Turbine\\Turbine G1",
"Status": [1],
"StatusChanged": "2015-31-16T10:22:55.98",
"Acknowledged": False
},
{
"ID": 85,
"Name": "IMx-8 All Measurement Types",
"Description": "Machine containing all IMx-8 measurement types",
"Path": "Company\\Example Data\\IMx\\IMx-8 All Measurement Types",
"Status": [1, 2, 8, 9, 12, 13, 28],
"StatusChanged": "2019-09-10T08:48:11.557",
"Acknowledged": True
}
]
print("✅ Asset List:")
print(json.dumps(data, indent=2, ensure_ascii=False))
# ✅ 3.2 단일 Asset 조회
def get_asset(asset_id: int):
url = f"{BASE_URL}/API/DataService/GetAsset"
params = {"AssetID": asset_id}
try:
response = requests.get(url, headers=headers, params=params, verify=False)
response.raise_for_status()
data = response.json()
except Exception as e:
print(f"❌ 단일 자산 요청 실패: {e}")
return
print("✅ Asset 정보:")
print(json.dumps(data, indent=2, ensure_ascii=False))
# ✅ 테스트 실행
get_asset_list(include_acknowledged=False)
get_asset(asset_id=85)
에러 발생 : HTTPSConnectionPool(host='localhost', port=14050): Max retries exceeded with url: /API/DataService/GetAssetList?IncludeAcknowledged=false (Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x79bae022f490>: Failed to establish a new connection: [Errno 111] Connection refused')) → 임시 데이터 사용
✅ Asset List:
[
{
"ID": 57,
"Name": "Turbine G1",
"Description": "Machine containing MasCon48 measurements",
"Path": "Company\\Example Data\\MasCon48\\Turbine\\Turbine G1",
"Status": [
1
],
"StatusChanged": "2015-31-16T10:22:55.98",
"Acknowledged": false
},
{
"ID": 85,
"Name": "IMx-8 All Measurement Types",
"Description": "Machine containing all IMx-8 measurement types",
"Path": "Company\\Example Data\\IMx\\IMx-8 All Measurement Types",
"Status": [
1,
2,
8,
9,
12,
13,
28
],
"StatusChanged": "2019-09-10T08:48:11.557",
"Acknowledged": true
}
]
❌ 단일 자산 요청 실패: HTTPSConnectionPool(host='localhost', port=14050): Max retries exceeded with url: /API/DataService/GetAsset?AssetID=85 (Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x79bae022df50>: Failed to establish a new connection: [Errno 111] Connection refused'))
3.3 SubMachine 리스트 조회
[ ]
# ✅ 3.3 SubMachine List 조회
def get_sub_machine_list(asset_id: int):
url = f"{BASE_URL}/API/DataService/GetSubMachineList"
params = {"AssetID": asset_id}
try:
response = requests.get(url, headers=headers, params=params, verify=False)
response.raise_for_status()
data = response.json()
except Exception as e:
print(f"❌ 에러 발생 : {e} → 임시 데이터 사용")
data = [
{
"ID": 330,
"Name": "Order Tracking Group",
"Description": "",
"Path": "Company\\Example Data\\IMx\\IMx-8 All Measurement Types\\Order Tracking Group",
"Status": [1, 2],
"StatusChanged": "2019-09-13T08:15:51.02",
"Acknowledged": True
},
{
"ID": 86,
"Name": "Trend Types",
"Description": "",
"Path": "Company\\Example Data\\IMx\\IMx-8 All Measurement Types\\Trend Types",
"Status": [1, 2, 12],
"StatusChanged": "2019-09-13T08:15:51.13",
"Acknowledged": True
}
]
print(f"✅ AssetID {asset_id}의 Sub-machine 목록:")
print(json.dumps(data, indent=2, ensure_ascii=False))
# ✅ 테스트 실행
get_sub_machine_list(asset_id=85)
❌ 에러 발생 : HTTPSConnectionPool(host='localhost', port=14050): Max retries exceeded with url: /API/DataService/GetSubMachineList?AssetID=85 (Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x79bae022f190>: Failed to establish a new connection: [Errno 111] Connection refused')) → 임시 데이터 사용
✅ AssetID 85의 Sub-machine 목록:
[
{
"ID": 330,
"Name": "Order Tracking Group",
"Description": "",
"Path": "Company\\Example Data\\IMx\\IMx-8 All Measurement Types\\Order Tracking Group",
"Status": [
1,
2
],
"StatusChanged": "2019-09-13T08:15:51.02",
"Acknowledged": true
},
{
"ID": 86,
"Name": "Trend Types",
"Description": "",
"Path": "Company\\Example Data\\IMx\\IMx-8 All Measurement Types\\Trend Types",
"Status": [
1,
2,
12
],
"StatusChanged": "2019-09-13T08:15:51.13",
"Acknowledged": true
}
]
3.4 포인트 리스트 조회
[ ]
# ✅ Points List 조회 (독립 함수 + mock fallback)
def get_points_list(asset_id: int):
url = f"{BASE_URL}/API/DataService/GetPointsList"
params = {
"AssetID": asset_id,
"IncludeLastMeasurement": "false",
"IncludeDiagnoses": "false",
"IncludeAlarmInfo": "false",
"IncludeOverallAlarm": "false",
"IncludeFrequencies": "false",
"IncludeStatus": "false"
}
try:
response = requests.get(url, headers=headers, params=params, verify=False)
response.raise_for_status()
data = response.json()
except Exception as e:
print(f"❌ 요청 실패 : {e} → 임시 데이터 사용")
data = [
{
"ID": 92,
"ParentID": 87,
"Name": "Dynamic TWF",
"Description": "",
"NodeType": 10201,
"NodeTypeName": "Dynamic (IMx)",
"Axes": 1,
"EUType": 1,
"EU": "g",
"Detection": 2,
"DetectionName": "PtP",
"Status": [],
"Frequencies": [],
"AlarmInfo": []
},
{
"ID": 127,
"ParentID": 87,
"Name": "Harmonic S/T/P 2 Channel",
"Description": "With Diagnosis",
"NodeType": 10203,
"NodeTypeName": "Harmonic (IMx)",
"Axes": 2,
"EUType": 1,
"EU": "g",
"Detection": 1,
"DetectionName": "Peak",
"Status": [],
"Frequencies": [],
"AlarmInfo": []
}
]
print(f"✅ AssetID {asset_id}의 측정 포인트:")
print(json.dumps(data, indent=2, ensure_ascii=False))
# ✅ 테스트 실행
get_points_list(asset_id=87)
❌ 요청 실패 : HTTPSConnectionPool(host='localhost', port=14050): Max retries exceeded with url: /API/DataService/GetPointsList?AssetID=87&IncludeLastMeasurement=false&IncludeDiagnoses=false&IncludeAlarmInfo=false&IncludeOverallAlarm=false&IncludeFrequencies=false&IncludeStatus=false (Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x79bae022c850>: Failed to establish a new connection: [Errno 111] Connection refused')) → 임시 데이터 사용
✅ AssetID 87의 측정 포인트:
[
{
"ID": 92,
"ParentID": 87,
"Name": "Dynamic TWF",
"Description": "",
"NodeType": 10201,
"NodeTypeName": "Dynamic (IMx)",
"Axes": 1,
"EUType": 1,
"EU": "g",
"Detection": 2,
"DetectionName": "PtP",
"Status": [],
"Frequencies": [],
"AlarmInfo": []
},
{
"ID": 127,
"ParentID": 87,
"Name": "Harmonic S/T/P 2 Channel",
"Description": "With Diagnosis",
"NodeType": 10203,
"NodeTypeName": "Harmonic (IMx)",
"Axes": 2,
"EUType": 1,
"EU": "g",
"Detection": 1,
"DetectionName": "Peak",
"Status": [],
"Frequencies": [],
"AlarmInfo": []
}
]
3.5 측정 포인트 조회
[ ]
# ✅ Measurement Point 상세 조회 (독립 함수 + mock fallback)
def get_point_detail(point_id: int):
url = f"{BASE_URL}/API/DataService/GetPoint"
params = {
"PointID": point_id,
"IncludeLastMeasurement": "true",
"IncludeDiagnoses": "true",
"IncludeAlarmInfo": "true",
"IncludeOverallAlarm": "true",
"IncludeFrequencies": "true",
"IncludeStatus": "true"
}
try:
response = requests.get(url, headers=headers, params=params, verify=False)
response.raise_for_status()
data = response.json()
except Exception as e:
print(f"❌ 요청 실패 : {e} → 임시 데이터 사용")
data = {
"ID": 211,
"ParentID": 210,
"Name": "Dyn1",
"Description": "",
"NodeType": 10504,
"NodeTypeName": "Vibration (Microlog)",
"Axes": 1,
"EUType": 1,
"EU": "g",
"Detection": 1,
"DetectionName": "Peak",
"Status": [1, 2],
"OverallAlarm": {
"Type": 1,
"TypeName": "Level",
"Summary": "High alarm 1,206919 / High warning 0,6936185",
"ChannelInfo": "X"
},
"Frequencies": [],
"Diagnoses": {
"Hysteresis": 7,
"HysteresisFilter": 4,
"Diagnoses": [
{
"Name": "Bearing Bearing1",
"IsPrivate": True,
"HighAlarm": 0.0,
"HighWarming": 0.0,
"LowWarning": 0.0,
"LowAlarm": 0.0,
"HighAlarmActive": True,
"HighWarmingActive": True,
"LowWarningActive": False,
"LowAlarmActive": False
}
],
"ProteanDiagnoses": []
},
"LastMeasurement": {
"ReadingTimeUTC": "2017-03-16T11:12:24",
"PointID": 211,
"Speed": 0.0,
"SpeedUnits": "RPM",
"Process": 0.0,
"ProcessUnits": "",
"Digital": 0,
"NumberOfChannels": 1,
"Measurements": [
{
"Channel": 1,
"ChannelName": "X",
"Level": 0.2348443,
"Units": "g P",
"BOV": 0.0
}
],
"AlarmInfo": []
},
"AlarmInfo": [
{
"AlarmDate": "2017-03-16T10:19:41",
"Title": "Bearing Bearing1",
"Acknowledged": True,
"AcknowledgedDateTime": "2018-05-24T07:39:47.763",
"AlarmStatusText": "Diagnosis Alarm",
"Source": 5,
"AlarmSourceName": "Diagnose",
"CurrentValue": 0.0,
"AlarmLevel": 0.0,
"EU": ""
}
]
}
print(f"✅ PointID {point_id} 상세 정보:")
print(json.dumps(data, indent=2, ensure_ascii=False))
# ✅ 테스트 실행 (예: PointID = 211)
get_point_detail(211)
❌ 요청 실패 : HTTPSConnectionPool(host='localhost', port=14050): Max retries exceeded with url: /API/DataService/GetPoint?PointID=211&IncludeLastMeasurement=true&IncludeDiagnoses=true&IncludeAlarmInfo=true&IncludeOverallAlarm=true&IncludeFrequencies=true&IncludeStatus=true (Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x79bae0273a50>: Failed to establish a new connection: [Errno 111] Connection refused')) → 임시 데이터 사용
✅ PointID 211 상세 정보:
{
"ID": 211,
"ParentID": 210,
"Name": "Dyn1",
"Description": "",
"NodeType": 10504,
"NodeTypeName": "Vibration (Microlog)",
"Axes": 1,
"EUType": 1,
"EU": "g",
"Detection": 1,
"DetectionName": "Peak",
"Status": [
1,
2
],
"OverallAlarm": {
"Type": 1,
"TypeName": "Level",
"Summary": "High alarm 1,206919 / High warning 0,6936185",
"ChannelInfo": "X"
},
"Frequencies": [],
"Diagnoses": {
"Hysteresis": 7,
"HysteresisFilter": 4,
"Diagnoses": [
{
"Name": "Bearing Bearing1",
"IsPrivate": true,
"HighAlarm": 0.0,
"HighWarming": 0.0,
"LowWarning": 0.0,
"LowAlarm": 0.0,
"HighAlarmActive": true,
"HighWarmingActive": true,
"LowWarningActive": false,
"LowAlarmActive": false
}
],
"ProteanDiagnoses": []
},
"LastMeasurement": {
"ReadingTimeUTC": "2017-03-16T11:12:24",
"PointID": 211,
"Speed": 0.0,
"SpeedUnits": "RPM",
"Process": 0.0,
"ProcessUnits": "",
"Digital": 0,
"NumberOfChannels": 1,
"Measurements": [
{
"Channel": 1,
"ChannelName": "X",
"Level": 0.2348443,
"Units": "g P",
"BOV": 0.0
}
],
"AlarmInfo": []
},
"AlarmInfo": [
{
"AlarmDate": "2017-03-16T10:19:41",
"Title": "Bearing Bearing1",
"Acknowledged": true,
"AcknowledgedDateTime": "2018-05-24T07:39:47.763",
"AlarmStatusText": "Diagnosis Alarm",
"Source": 5,
"AlarmSourceName": "Diagnose",
"CurrentValue": 0.0,
"AlarmLevel": 0.0,
"EU": ""
}
]
}
6.6 설비 파츠 조회
[ ]
# ✅ Machine Parts 조회 (독립 함수 + mock fallback)
def get_machine_parts(asset_id: int):
url = f"{BASE_URL}/machine/parts"
params = {"assetId": asset_id}
try:
response = requests.get(url, headers=headers, params=params, verify=False)
response.raise_for_status()
data = response.json()
except Exception as e:
print(f"❌ 요청 실패 : {e} → 임시 데이터 사용")
data = [
{
"type": "Bearing",
"name": "Bearing1",
"ratio": 1.0,
"id": 20,
"brand": "SKF",
"partType": "02800",
"speedPointId": 89,
"bpfo": 6.37,
"bpfi": 8.62,
"FTF": 0.42,
"BSF": 3.15
},
{
"type": "Bearing",
"name": "Bearing2",
"ratio": 7.142857,
"id": 26,
"brand": "SKF",
"partType": "03062/03162/Q",
"speedPointId": 89,
"bpfo": 36.5,
"bpfi": 56.35714,
"FTF": 2.78571415,
"BSF": 15.5714293
}
]
print(f"✅ Machine Parts (AssetID={asset_id}):")
print(json.dumps(data, indent=2, ensure_ascii=False))
# ✅ 테스트 실행 (예: AssetID = 85)
get_machine_parts(85)
❌ 요청 실패 : HTTPSConnectionPool(host='localhost', port=14050): Max retries exceeded with url: /machine/parts?assetId=85 (Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x79bad0747f10>: Failed to establish a new connection: [Errno 111] Connection refused')) → 임시 데이터 사용
✅ Machine Parts (AssetID=85):
[
{
"type": "Bearing",
"name": "Bearing1",
"ratio": 1.0,
"id": 20,
"brand": "SKF",
"partType": "02800",
"speedPointId": 89,
"bpfo": 6.37,
"bpfi": 8.62,
"FTF": 0.42,
"BSF": 3.15
},
{
"type": "Bearing",
"name": "Bearing2",
"ratio": 7.142857,
"id": 26,
"brand": "SKF",
"partType": "03062/03162/Q",
"speedPointId": 89,
"bpfo": 36.5,
"bpfi": 56.35714,
"FTF": 2.78571415,
"BSF": 15.5714293
}
]
keyboard_arrow_down
> ** 데이터 관련 API **
[ ]
BASE_URL = "https://localhost:14050"
TOKEN_FILE = "access_token.json"
# 🔹 토큰 불러오기
with open(TOKEN_FILE, "r") as f:
token_data = json.load(f)
access_token = token_data.get("access_token")
headers = {
"Authorization": f"Bearer {access_token}",
"Accept": "application/json"
}
Trend 출력
[ ]
# ✅ Trend Measurements 조회 (독립 함수 + mock fallback)
def get_trend_measurements(point_id: int, num_readings: int = 12):
url = f"{BASE_URL}/API/DataService/GetTrendList"
params = {
"PointID": point_id,
"NumReadings": num_readings
}
try:
response = requests.get(url, headers=headers, params=params, verify=False)
response.raise_for_status()
data = response.json()
except Exception as e:
print(f"❌ 요청 실패 : {e} → 임시 데이터 사용")
data = [
{
"ReadingTimeUTC": "2019-09-13T06:55:18.01",
"PointID": point_id,
"Speed": 0.0,
"SpeedUnits": "RPM",
"Process": 0.0,
"ProcessUnits": "",
"Digital": 0,
"NumberOfChannels": 3,
"Measurements": [
{
"Channel": 1,
"ChannelName": "X",
"Level": 33.85868,
"Units": "g P",
"BOV": 2428.45581
},
{
"Channel": 2,
"ChannelName": "Y",
"Level": 33.8709,
"Units": "g P",
"BOV": 2426.80786
},
{
"Channel": 3,
"ChannelName": "Z",
"Level": 33.8628235,
"Units": "g P",
"BOV": 2425.901
}
],
"AlarmInfo": []
},
{
"ReadingTimeUTC": "2017-03-16T11:10:13",
"PointID": point_id,
"Speed": 659.0,
"SpeedUnits": "RPM",
"Process": 0.0,
"ProcessUnits": "",
"Digital": 0,
"NumberOfChannels": 1,
"Measurements": [
{
"Channel": 1,
"ChannelName": "X",
"Level": 658.9861,
"Units": "cpm",
"BOV": 0.0
}
],
"AlarmInfo": []
}
]
print(f"✅ Trend Measurements for PointID={point_id}")
print(json.dumps(data, indent=2, ensure_ascii=False))
# ✅ 테스트 실행 (예: PointID = 105, 최근 5개)
get_trend_measurements(point_id=105, num_readings=5)
❌ 요청 실패 : HTTPSConnectionPool(host='localhost', port=14050): Max retries exceeded with url: /API/DataService/GetTrendList?PointID=105&NumReadings=5 (Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x79bad07e9250>: Failed to establish a new connection: [Errno 111] Connection refused')) → 임시 데이터 사용
✅ Trend Measurements for PointID=105
[
{
"ReadingTimeUTC": "2019-09-13T06:55:18.01",
"PointID": 105,
"Speed": 0.0,
"SpeedUnits": "RPM",
"Process": 0.0,
"ProcessUnits": "",
"Digital": 0,
"NumberOfChannels": 3,
"Measurements": [
{
"Channel": 1,
"ChannelName": "X",
"Level": 33.85868,
"Units": "g P",
"BOV": 2428.45581
},
{
"Channel": 2,
"ChannelName": "Y",
"Level": 33.8709,
"Units": "g P",
"BOV": 2426.80786
},
{
"Channel": 3,
"ChannelName": "Z",
"Level": 33.8628235,
"Units": "g P",
"BOV": 2425.901
}
],
"AlarmInfo": []
},
{
"ReadingTimeUTC": "2017-03-16T11:10:13",
"PointID": 105,
"Speed": 659.0,
"SpeedUnits": "RPM",
"Process": 0.0,
"ProcessUnits": "",
"Digital": 0,
"NumberOfChannels": 1,
"Measurements": [
{
"Channel": 1,
"ChannelName": "X",
"Level": 658.9861,
"Units": "cpm",
"BOV": 0.0
}
],
"AlarmInfo": []
}
]
4.2 동적 측정 치 리스트 출력
[ ]
import requests
import json
import urllib3
# ✅ Measurement List 조회 (독립 함수 + mock fallback)
def get_measurement_list(point_id: int, max_measurements: int = 6):
url = f"{BASE_URL}/API/DataService/GetMeasurementList"
params = {
"PointID": point_id,
"MaxNumberOfMeasurements": max_measurements
}
try:
response = requests.get(url, headers=headers, params=params, verify=False)
response.raise_for_status()
data = response.json()
except Exception as e:
print(f"❌ 요청 실패 : {e} → 임시 데이터 사용")
data = [
{"IDMeasurement": 43164, "ReadingTimeUTC": "2019-09-13T06:54:34.46"},
{"IDMeasurement": 43140, "ReadingTimeUTC": "2019-09-13T06:52:51.11"},
{"IDMeasurement": 43107, "ReadingTimeUTC": "2019-09-13T06:50:16.21"},
{"IDMeasurement": 43081, "ReadingTimeUTC": "2019-09-13T06:48:32.97"},
{"IDMeasurement": 43055, "ReadingTimeUTC": "2019-09-13T06:46:49.64"},
{"IDMeasurement": 43024, "ReadingTimeUTC": "2019-09-13T06:44:14.70"}
]
print(f"✅ Measurement List (PointID={point_id})")
print(json.dumps(data, indent=2, ensure_ascii=False))
# ✅ 테스트 실행 (예: PointID = 92, 최근 6개)
get_measurement_list(point_id=92, max_measurements=6)
❌ 요청 실패 : HTTPSConnectionPool(host='localhost', port=14050): Max retries exceeded with url: /API/DataService/GetMeasurementList?PointID=92&MaxNumberOfMeasurements=6 (Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x79bae022f610>: Failed to establish a new connection: [Errno 111] Connection refused')) → 임시 데이터 사용
✅ Measurement List (PointID=92)
[
{
"IDMeasurement": 43164,
"ReadingTimeUTC": "2019-09-13T06:54:34.46"
},
{
"IDMeasurement": 43140,
"ReadingTimeUTC": "2019-09-13T06:52:51.11"
},
{
"IDMeasurement": 43107,
"ReadingTimeUTC": "2019-09-13T06:50:16.21"
},
{
"IDMeasurement": 43081,
"ReadingTimeUTC": "2019-09-13T06:48:32.97"
},
{
"IDMeasurement": 43055,
"ReadingTimeUTC": "2019-09-13T06:46:49.64"
},
{
"IDMeasurement": 43024,
"ReadingTimeUTC": "2019-09-13T06:44:14.70"
}
]
4.3 TMF/Spectrum 등 출력
[ ]
import requests
import json
import urllib3
import numpy as np
# ✅ GetMeasurement 호출 (의미 있는 랜덤 mock 포함)
def get_measurement(point_id: int, include_measurements: int = 2, format_type: int = 0):
url = f"{BASE_URL}/API/DataService/GetMeasurement"
params = {
"PointID": point_id,
"IncludeMeasurements": include_measurements,
"Format": format_type
}
try:
response = requests.get(url, headers=headers, params=params, verify=False)
response.raise_for_status()
data = response.json()
except Exception as e:
print(f"❌ 요청 실패 : {e} → 랜덤 mock 데이터 생성")
sample_rate = 400.0
samples = 16384
# mock value 생성
if include_measurements == 2: # Spectrum
freqs = np.linspace(0, sample_rate/2, samples)
values = 10*np.exp(-((freqs-50)/20)**2) + np.random.normal(0, 0.2, samples)
measurement_type = 0 # Spectrum
elif include_measurements == 1: # TimeWaveform
t = np.arange(samples) / sample_rate
values = 2.0 * np.sin(2 * np.pi * 5 * t) + 0.5 * np.random.normal(0, 1, samples)
measurement_type = 2 # Time Domain
else:
values = np.zeros(samples)
measurement_type = 2
data = {
"PointID": point_id,
"Speed": 3000.01782,
"SpeedBegin": 3000.0144,
"SpeedEnd": 3000.0144,
"SpeedMin": 3000.014,
"SpeedMax": 3000.01465,
"SpeedUnits": "RPM",
"Process": 0.0,
"ProcessMin": 0.0,
"ProcessMax": 0.0,
"ProcessUnits": "",
"Digital": 0,
"NumberOfChannels": 1,
"SampleRate": sample_rate,
"Samples": samples,
"EUType": 1,
"EU": "g",
"DetectionType": 0,
"DetectionName": "",
"WindowType": 0,
"WindowName": "",
"StartFrequency": 0.0,
"EndFrequency": 0.0,
"Averages": 1,
"AverageType": 0,
"StorageReason": 0,
"MeasurementComment": "Mock Data",
"IncludeMeasurements": include_measurements,
"Content": 0,
"Measurements": [
{
"MeasurementType": measurement_type,
"Direction": 0,
"Values": values.tolist()
}
],
"UFF": [],
"IDMeasurement": 43168,
"ReadingTimeUTC": "2019-09-13T06:55:15.93"
}
# ✅ 테스트 실행
get_measurement(point_id=331, include_measurements=1) # 타임 도메인
get_measurement(point_id=331, include_measurements=2) # 스펙트럼
[ ]
# ✅ 통합형 GetMeasurement + 시뮬레이션 fallback
def get_measurement_with_simulation(point_id: int, include_measurements: int = 2, format_type: int = 0):
url = f"{BASE_URL}/API/DataService/GetMeasurement"
params = {
"PointID": point_id,
"IncludeMeasurements": include_measurements,
"Format": format_type
}
try:
response = requests.get(url, headers=headers, params=params, verify=False)
response.raise_for_status()
data = response.json()
print(f"✅ 실 API 결과 반환 (PointID={point_id})")
except Exception as e:
print(f"❌ API 실패 : {e} → 랜덤 mock 데이터 생성")
data = generate_mock_data(point_id, include_measurements)
# 자동 시각화까지 포함!
plot_measurement(data, include_measurements)
return data
# ✅ 의미 있는 랜덤 mock 생성기
def generate_mock_data(point_id: int, include_measurements: int):
sample_rate = 400.0
samples = 16384
if include_measurements == 2: # Spectrum
freqs = np.linspace(0, sample_rate/2, samples)
values = 10*np.exp(-((freqs-50)/20)**2) + np.random.normal(0, 0.2, samples)
measurement_type = 0
elif include_measurements == 1: # Time Domain
t = np.arange(samples) / sample_rate
values = 2.0 * np.sin(2 * np.pi * 5 * t) + 0.5 * np.random.normal(0, 1, samples)
measurement_type = 2
else:
values = np.zeros(samples)
measurement_type = 2
data = {
"PointID": point_id,
"SampleRate": sample_rate,
"Samples": samples,
"NumberOfChannels": 1,
"EU": "g",
"Measurements": [{
"MeasurementType": measurement_type,
"Direction": 0,
"Values": values.tolist(),
"Channel": 1
}],
"ReadingTimeUTC": datetime.utcnow().isoformat()
}
return data
# ✅ 시각화 통합 함수
# 시뮬레이션 시각화 함수 (다중 채널 지원 + 통합 옵션)
def plot_measurement(data, measurement_type, plot_mode="separate"):
channels = data['NumberOfChannels']
if plot_mode == "overlay":
fig, ax = plt.subplots(figsize=(10, 6))
ax.set_title("Overlay Plot")
ax.grid(True)
for meas in data['Measurements']:
mtype = meas['MeasurementType']
values = meas['Values']
channel = meas['Channel']
if mtype == 2:
x_axis = np.arange(len(values))
x_label = "Sample Index"
y_label = "Amplitude (g)"
elif mtype in [0, 1]:
x_axis = np.linspace(0, data['SampleRate']/2, len(values))
x_label = "Frequency (Hz)"
y_label = "Amplitude" if mtype == 0 else "Phase (radian)"
else:
x_axis = np.arange(len(values))
x_label = "Index"
y_label = "Value"
if plot_mode == "separate":
fig, ax = plt.subplots(figsize=(10, 4))
ax.plot(x_axis, values, label=f"Channel {channel}")
ax.set_title(f"Channel {channel}")
ax.set_xlabel(x_label)
ax.set_ylabel(y_label)
ax.grid(True)
ax.legend()
plt.tight_layout()
plt.show()
elif plot_mode == "overlay":
ax.plot(x_axis, values, label=f"Channel {channel}")
if plot_mode == "overlay":
ax.set_xlabel(x_label)
ax.set_ylabel(y_label)
ax.legend()
plt.tight_layout()
plt.show()
# ✅ 테스트 실행 예제
if __name__ == "__main__":
# 실 API 또는 Mock 테스트 모두 가능
get_measurement_with_simulation(point_id=331, include_measurements=1) # TimeWaveform
get_measurement_with_simulation(point_id=331, include_measurements=2) # Spectrum
[ ]
코딩을 시작하거나 AI로 코드를 생성하세요.
[ ]
import numpy as np
import matplotlib.pyplot as plt
from datetime import datetime
import copy
# 가상 데이터 생성 함수
def generate_mock_data(base_data, measurement_type):
# deepcopy로 복사본 사용 (누적 방지)
data = copy.deepcopy(base_data)
data['Measurements'] = [] # 항상 새로 초기화
for ch in range(data['NumberOfChannels']):
if measurement_type == 1: # Time Domain
values = np.sin(np.linspace(0, 50*np.pi, data['Samples'])) + np.random.normal(0, 0.1, data['Samples'])
meas_type = 2
elif measurement_type == 2: # Spectrum
values = np.abs(np.fft.rfft(np.sin(np.linspace(0, 20*np.pi, data['Samples'])) + np.random.normal(0, 0.05, data['Samples'])))
meas_type = 0
elif measurement_type == 4: # Phase Spectrum
signal = np.sin(np.linspace(0, 20*np.pi, data['Samples'])) + np.random.normal(0, 0.05, data['Samples'])
phase = np.unwrap(np.angle(np.fft.rfft(signal))) # <-- unwrap 적용!
values = phase
meas_type = 1
else:
values = np.zeros(data['Samples'])
meas_type = 2
data['Measurements'].append({
"MeasurementType": meas_type,
"Direction": 0,
"Values": values.tolist(),
"Channel": ch + 1
})
return data
# 시뮬레이션 시각화 함수 (다중 채널 지원 + 통합 옵션)
def plot_measurement(data, measurement_type, plot_mode="separate"):
channels = data['NumberOfChannels']
measurement_type = data['IncludeMeasurements']
if plot_mode == "overlay":
fig, ax = plt.subplots(figsize=(10, 6))
ax.set_title("Overlay Plot")
ax.grid(True)
for meas in data['Measurements']:
mtype = meas['MeasurementType']
values = meas['Values']
channel = meas['Channel']
if mtype == 2:
x_axis = np.arange(len(values))
x_label = "Sample Index"
y_label = "Amplitude (g)"
elif mtype in [0, 1]:
x_axis = np.linspace(0, data['SampleRate']/2, len(values))
x_label = "Frequency (Hz)"
y_label = "Amplitude" if mtype == 0 else "Phase (radian)"
else:
x_axis = np.arange(len(values))
x_label = "Index"
y_label = "Value"
if plot_mode == "separate":
fig, ax = plt.subplots(figsize=(10, 4))
ax.plot(x_axis, values, label=f"Channel {channel}")
ax.set_title(f"Channel {channel}")
ax.set_xlabel(x_label)
ax.set_ylabel(y_label)
ax.grid(True)
ax.legend()
plt.tight_layout()
plt.show()
elif plot_mode == "overlay":
ax.plot(x_axis, values, label=f"Channel {channel}")
if plot_mode == "overlay":
ax.set_xlabel(x_label)
ax.set_ylabel(y_label)
ax.legend()
plt.tight_layout()
plt.show()
# 테스트용 base data
channels = 3
samples = 1024
base_data = {
"PointID": 331,
"Speed": 3000.0,
"SpeedBegin": 3000.0,
"SpeedEnd": 3000.0,
"SpeedMin": 2999.9,
"SpeedMax": 3000.1,
"SpeedUnits": "RPM",
"Process": 0.0,
"ProcessMin": 0.0,
"ProcessMax": 0.0,
"ProcessUnits": "",
"Digital": 0,
"NumberOfChannels": channels,
"SampleRate": 400.0,
"Samples": samples,
"EUType": 1,
"EU": "g",
"DetectionType": 0,
"DetectionName": "",
"WindowType": 0,
"WindowName": "",
"StartFrequency": 0.0,
"EndFrequency": 0.0,
"Averages": 1,
"AverageType": 0,
"StorageReason": 0,
"MeasurementComment": "Mock Test Data",
"IncludeMeasurements": 0,
"Content": 0,
"Measurements": [],
"UFF": [],
"IDMeasurement": 99999,
"ReadingTimeUTC": datetime.utcnow().isoformat()
}
# 각각 테스트
print("===== Time Waveform 테스트 =====")
mock_twf = generate_mock_data(base_data, 1)
plot_measurement(mock_twf, 1, 'overlay')
print("===== Spectrum 테스트 =====")
mock_spectrum = generate_mock_data(base_data, 2)
plot_measurement(mock_spectrum, 2, 'overlay')
print("===== Phase Spectrum 테스트 =====")
mock_phase = generate_mock_data(base_data, 4)
plot_measurement(mock_phase, 4, 'overlay')
[ 옵션 ] 저장 코드
[ ]
✅ 응답 데이터가 파일에 저장되었습니다: ./test/save_path.json
[ ]
코딩을 시작하거나 AI로 코드를 생성하세요.
반응형