d7cd9043b2
Add unique constraint on (model_id, timestamp, variable) in schema and implement idempotent export test to ensure no duplicates are created. Update CoreScouter and related tests to handle conflict resolution by ignoring duplicates.
307 lines
9.8 KiB
Python
307 lines
9.8 KiB
Python
"""
|
|
Script to fetch WebIds from PI Web API and then retrieve historical values
|
|
for a list of tags over a given period in 30-day chunks, storing results
|
|
in a DataFrame indexed by timestamp.
|
|
|
|
Based on pi_web_api_client.py and tests.ipynb. Run with project venv active.
|
|
Use # %% cell separators: run each cell in order (Run Cell / Shift+Enter).
|
|
"""
|
|
|
|
# %% 1. Imports and configuration
|
|
import asyncio
|
|
import concurrent.futures
|
|
import json
|
|
import os
|
|
import time
|
|
from typing import Any
|
|
|
|
import pandas as pd
|
|
import requests
|
|
from unittest.mock import MagicMock, AsyncMock
|
|
|
|
from sientia_do.repository.pi_web_api_client import PIWebAPIClient
|
|
|
|
BASE_URL = 'https://pivision.votorantimcimentos.com/piwebapi'
|
|
AUTH_TOKEN = 'dmlkX3ZjbmV0XHN2Yy5waW9zaS5wcmQud2ViYXBpOlN2Y1ByRFdlQkBQaQ=='
|
|
WEBID_LOOKUP_PATH = 'dataservers/F1DS-7fYgsRTtUOa7V9NIwSujAUElIQVZD/points'
|
|
PERIOD_DAYS = (365 * 3) + 50
|
|
CHUNK_DAYS = 5
|
|
ENDPOINT = '/streamsets/recorded'
|
|
API_TIMEOUT = 60
|
|
WEB_IDS_SAVE_PATH = 'web_ids.json'
|
|
TAG_NAMES = [
|
|
'CI-J3J01S1', 'CI-J3P01T1A', 'CI-J3P03S1', 'CI-W3A05F1',
|
|
'CI-W3A50A1', 'CI-W3A50A2', 'CI-W3A50A3', 'CI-W3A50P1', 'CI-W3A50T1',
|
|
'CI-W3A55P1', 'CI-W3A55T1', 'CI-W3A65_Cl', 'CI-W3A65_SO3', 'CI-W3A71P1',
|
|
'CI-W3A71P2', 'CI-W3A71P3', 'CI-W3E01F1', 'CI-W3K01S1', 'CI-W3K01T1',
|
|
'CI-W3K01T2', 'CI-W3K01T3', 'CI-W3K01T4', 'CI-W3K14P1', 'CI-W3P17S1', 'CI-W3V04P1',
|
|
'CI-W3V04P3', 'CI-W3V21F1', 'CI-W3V21P1', 'CI-W3V30F1', 'CI-W3V33P1',
|
|
'CI-W3W01A1', 'CI-W3W01A2', 'CI-W3W01G1', 'CI-W3W01P1',
|
|
'CI-W3W01P2', 'CI-W3W03I1', 'CI-W3W03S1', 'CI-W3X21IN', 'CI-W3_C3S',
|
|
'CI-W3_CAO', 'CI-W3_MA', 'CI-W3_MS', 'CI-W3_PL'
|
|
]
|
|
|
|
print(f'Config: BASE_URL={BASE_URL}, PERIOD_DAYS={PERIOD_DAYS}, CHUNK_DAYS={CHUNK_DAYS}, tags={len(TAG_NAMES)}, WEB_IDS_SAVE_PATH={WEB_IDS_SAVE_PATH}')
|
|
|
|
|
|
def run_async(coro):
|
|
"""
|
|
Run a coroutine from sync code. Works in scripts and in Jupyter (where an event loop is already running).
|
|
|
|
Args:
|
|
coro: Coroutine to run.
|
|
|
|
Return:
|
|
Result of the coroutine.
|
|
"""
|
|
try:
|
|
asyncio.get_running_loop()
|
|
except RuntimeError:
|
|
return asyncio.run(coro)
|
|
with concurrent.futures.ThreadPoolExecutor() as pool:
|
|
future = pool.submit(asyncio.run, coro)
|
|
return future.result()
|
|
|
|
|
|
# %% 2. Helper: fetch WebIds from PI Web API
|
|
def fetch_webids(
|
|
tag_names: list[str],
|
|
base_url: str,
|
|
auth_token: str,
|
|
webid_lookup_path: str,
|
|
delay_seconds: float = 0.5,
|
|
) -> dict[str, dict[str, Any]]:
|
|
"""
|
|
Resolve WebIds for the given tag names via PI Web API points endpoint.
|
|
|
|
Args:
|
|
tag_names: List of tag names to resolve.
|
|
base_url: PI Web API base URL (no trailing slash).
|
|
auth_token: Basic auth token (base64-encoded user:password).
|
|
webid_lookup_path: Path relative to base_url, with {tag} placeholder for namefilter.
|
|
delay_seconds: Delay between requests to avoid rate limiting.
|
|
|
|
Returns:
|
|
dict mapping tag name to {'webid': str, 'aggr_func': str, 'data_range': list}.
|
|
"""
|
|
base_url = base_url.rstrip('/')
|
|
url_template = f'{base_url}/{webid_lookup_path}'
|
|
if '?' in url_template:
|
|
url_template = f'{url_template}&namefilter={{tag}}'
|
|
else:
|
|
url_template = f'{url_template}?namefilter={{tag}}'
|
|
|
|
headers = {
|
|
'Content-Type': 'application/json',
|
|
'Accept': 'application/json',
|
|
'X-Requested-With': 'piwebapistreams',
|
|
'Authorization': f'Basic {auth_token}',
|
|
}
|
|
|
|
web_ids: dict[str, dict[str, Any]] = {}
|
|
for idx, tag in enumerate(tag_names, start=1):
|
|
url = url_template.format(tag=tag)
|
|
resp = requests.get(url, headers=headers, timeout=API_TIMEOUT)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
items = data.get('Items', [])
|
|
if not items:
|
|
raise ValueError(f'No point found for tag: {tag}')
|
|
web_ids[tag] = {
|
|
'webid': items[0]['WebId'],
|
|
'aggr_func': 'lts',
|
|
'data_range': [-100000, 100000],
|
|
}
|
|
print(f' Resolved tag {idx}/{len(tag_names)}: {tag}')
|
|
time.sleep(delay_seconds)
|
|
|
|
return web_ids
|
|
|
|
|
|
# %% 3. Helper: load WebIds from JSON
|
|
def load_web_ids(json_path: str | None) -> dict[str, dict[str, Any]] | None:
|
|
"""
|
|
Load web_ids from a JSON file if path is provided.
|
|
|
|
Args:
|
|
json_path: Path to JSON file with tag -> {webid, ...} structure.
|
|
|
|
Return:
|
|
Loaded dict or None if json_path is None or file missing.
|
|
"""
|
|
if not json_path or not os.path.isfile(json_path):
|
|
return None
|
|
with open(json_path, encoding='utf-8') as f:
|
|
out = json.load(f)
|
|
print(f' Loaded {len(out)} web_ids from {json_path}')
|
|
return out
|
|
|
|
|
|
# %% 4. Helper: fetch values in chunks (async)
|
|
async def fetch_values_chunked(
|
|
web_ids: dict[str, dict[str, Any]],
|
|
period_days: int,
|
|
chunk_days: int,
|
|
base_url: str,
|
|
auth_token: str,
|
|
endpoint: str,
|
|
request_timeout: int,
|
|
) -> pd.DataFrame:
|
|
"""
|
|
Fetch historical values for web_ids over period_days in chunks of chunk_days.
|
|
|
|
Args:
|
|
web_ids: Dict mapping tag name to at least {'webid': str}.
|
|
period_days: Total period to fetch (e.g. 180 for last 180 days).
|
|
chunk_days: Size of each time chunk in days (e.g. 30).
|
|
base_url: PI Web API base URL.
|
|
auth_token: Basic auth token.
|
|
endpoint: PI Web API endpoint (e.g. /streamsets/recorded).
|
|
request_timeout: Request timeout in seconds.
|
|
|
|
Returns:
|
|
DataFrame with timestamp index and one column per tag (values).
|
|
"""
|
|
logger = MagicMock()
|
|
notification_handler = AsyncMock()
|
|
metrics_controller = AsyncMock()
|
|
|
|
client = PIWebAPIClient(
|
|
base_url=base_url,
|
|
auth_config={'type': 'basic', 'token': auth_token},
|
|
logger=logger,
|
|
notification_handler=notification_handler,
|
|
metrics_controller=metrics_controller,
|
|
headers_config={
|
|
'Content-Type': 'application/json',
|
|
'Accept': 'application/json',
|
|
'x-requested-with': 'piwebapistreams',
|
|
'User-Agent': 'PiWebApiFetchData/1.0',
|
|
},
|
|
)
|
|
|
|
metadata: dict[str, Any] = {}
|
|
chunks: list[pd.DataFrame] = []
|
|
|
|
try:
|
|
for i in range(period_days, 0, -chunk_days):
|
|
j = i - chunk_days
|
|
start_time_pi = f'*-{i}d'
|
|
end_time_pi = f'*-{j}d' if j > 0 else '*'
|
|
print(f' Fetching chunk: {start_time_pi} to {end_time_pi}')
|
|
df = await client.get_latest_values_df(
|
|
web_ids=web_ids,
|
|
endpoint=endpoint,
|
|
start_time=start_time_pi,
|
|
end_time=end_time_pi,
|
|
max_count=None,
|
|
request_timeout=request_timeout,
|
|
metadata=metadata,
|
|
)
|
|
if not df.empty:
|
|
chunks.append(df)
|
|
print(f' Chunk size: {df.shape}')
|
|
print(f' Chunk sample: {df.head(3)}')
|
|
else:
|
|
print(f' Chunk size: 0 ')
|
|
await asyncio.sleep(1)
|
|
finally:
|
|
client.close()
|
|
|
|
if not chunks:
|
|
return pd.DataFrame()
|
|
|
|
data = pd.concat(chunks, ignore_index=True)
|
|
|
|
print(f"Amount of names: {len(data['name'].unique())}")
|
|
|
|
raw_count = len(data)
|
|
# data['timestamp'] = pd.to_datetime(data['timestamp'], utc=True).dt.floor('s')
|
|
data_timestamp_na = data[data['timestamp'].isna()]
|
|
|
|
print(f"NA timestamp: {data_timestamp_na}")
|
|
|
|
print(f"Amount of names: {len(data['name'].unique())}")
|
|
|
|
data = data.sort_values('timestamp')
|
|
data = data.drop_duplicates(subset=['timestamp', 'name'], keep='last')
|
|
|
|
print(f"Amount of names: {len(data['name'].unique())}")
|
|
|
|
dedup_count = len(data)
|
|
print(f' Total raw rows: {raw_count}, after dedup: {dedup_count}')
|
|
|
|
pivot = data.pivot(index='timestamp', columns='name', values='value')
|
|
pivot.sort_index(inplace=True)
|
|
print(f' Pivot shape: {pivot.shape} (index=timestamp, columns=tags)')
|
|
return pivot
|
|
|
|
|
|
# %% 5. Step: load or fetch WebIds (saved to WEB_IDS_SAVE_PATH after fetch for continuity)
|
|
print('Step 5: Load or fetch WebIds')
|
|
print(f' Trying WEB_IDS_SAVE_PATH={WEB_IDS_SAVE_PATH}')
|
|
web_ids = load_web_ids(WEB_IDS_SAVE_PATH)
|
|
if web_ids is None:
|
|
if not AUTH_TOKEN:
|
|
raise ValueError('Set AUTH_TOKEN at top to fetch WebIds.')
|
|
print(f'Fetching WebIds for {len(TAG_NAMES)} tags...')
|
|
web_ids = fetch_webids(
|
|
tag_names=TAG_NAMES,
|
|
base_url=BASE_URL,
|
|
auth_token=AUTH_TOKEN,
|
|
webid_lookup_path=WEBID_LOOKUP_PATH,
|
|
)
|
|
print(f'Resolved {len(web_ids)} WebIds.')
|
|
with open(WEB_IDS_SAVE_PATH, 'w', encoding='utf-8') as f:
|
|
json.dump(web_ids, f, indent=4)
|
|
print(f'Saved web_ids to {WEB_IDS_SAVE_PATH} for continuity.')
|
|
else:
|
|
print(f'Loaded {len(web_ids)} WebIds from {WEB_IDS_SAVE_PATH}.')
|
|
web_ids
|
|
|
|
|
|
# %% 6. Step: fetch values in chunks
|
|
print('Step 6: Fetch values in chunks')
|
|
print(f' Period: {PERIOD_DAYS} days, chunk size: {CHUNK_DAYS} days, tags: {list(web_ids.keys())}')
|
|
df = run_async(
|
|
fetch_values_chunked(
|
|
web_ids=web_ids,
|
|
period_days=PERIOD_DAYS,
|
|
chunk_days=CHUNK_DAYS,
|
|
base_url=BASE_URL,
|
|
auth_token=AUTH_TOKEN,
|
|
endpoint=ENDPOINT,
|
|
request_timeout=API_TIMEOUT,
|
|
)
|
|
)
|
|
if df.empty:
|
|
print(' Done. No data returned.')
|
|
else:
|
|
print(f' Done. Shape: {df.shape}, index range: {df.index.min()} to {df.index.max()}')
|
|
df
|
|
|
|
|
|
# %% 7. Step: inspect and optionally save
|
|
print('Step 7: Inspect and optionally save')
|
|
if df.empty:
|
|
print(' DataFrame is empty.')
|
|
else:
|
|
print(f' Shape: {df.shape}, columns: {list(df.columns)}')
|
|
print(f' Index (timestamp) range: {df.index.min()} to {df.index.max()}')
|
|
df.head()
|
|
df.to_csv('pi_web_api_data.csv')
|
|
# df.to_parquet('pi_web_api_data.parquet')
|
|
|
|
# %%
|
|
|
|
from pandas import read_csv, to_datetime
|
|
|
|
data = read_csv('pi_web_api_data.csv')
|
|
data['timestamp'] = to_datetime(data['timestamp'])
|
|
print(data['timestamp'].min())
|
|
print(data['timestamp'].max())
|
|
|
|
# %%
|
|
print(data.shape)
|
|
# %%
|