SIENTIAPDE-1712
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.
This commit is contained in:
vitor-aignosi
@@ -89,6 +89,7 @@ def _create_schema_and_table(engine):
|
||||
value numeric NULL,
|
||||
"timestamp" timestamptz NOT NULL,
|
||||
created_at timestamptz DEFAULT CURRENT_TIMESTAMP NOT NULL,
|
||||
CONSTRAINT unique_timestamp_variable UNIQUE (model_id, "timestamp", variable),
|
||||
PRIMARY KEY (id, created_at)
|
||||
);
|
||||
"""
|
||||
|
||||
@@ -296,3 +296,101 @@ async def test_scenario_1_1_3_success_with_debug_data_package(
|
||||
# The package should be a dict with 'data' and 'held_data' keys
|
||||
assert isinstance(package_data, dict), "Expected data package to be a dict"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.integration
|
||||
async def test_scenario_1_1_4_idempotent_export_ignores_duplicates(
|
||||
temporal_test_env: WorkflowEnvironment,
|
||||
temporal_worker: Worker,
|
||||
mock_pi_web_api_client,
|
||||
postgres_engine,
|
||||
):
|
||||
"""
|
||||
Scenario 1.1.4: Idempotent export ignores duplicates.
|
||||
|
||||
Running the same batch twice must not increase row count for the same
|
||||
(model_id, timestamp, variable) keys.
|
||||
"""
|
||||
client = temporal_test_env.client
|
||||
unique_id = int(datetime.now().timestamp() * 1000) % 1000000
|
||||
|
||||
input_data = {
|
||||
'model_name': 'PI Web API Scouter Test Model',
|
||||
'model_id': str(unique_id),
|
||||
'schedule_name': f'pi-web-api-scouter-idempotent-{unique_id}',
|
||||
'model_tags': {
|
||||
'tag1': {
|
||||
'webid': 'webid1',
|
||||
'aggr_function': 'avg',
|
||||
'data_range': [0, 100],
|
||||
'frequency': 60000,
|
||||
},
|
||||
'tag2': {
|
||||
'webid': 'webid2',
|
||||
'aggr_function': 'avg',
|
||||
'data_range': [0, 100],
|
||||
'frequency': 60000,
|
||||
},
|
||||
},
|
||||
'trigger_laborious': False,
|
||||
'filters': {},
|
||||
'schema': 'sientia_data',
|
||||
'table_name': 'laborious_data',
|
||||
'retention_time': 3600,
|
||||
'fill_missing_tags': False,
|
||||
'pi_web_api_query': {
|
||||
'endpoint': '/streamsets/recorded',
|
||||
'period': '*-1d',
|
||||
'max_count': 10,
|
||||
'api_timeout': 30,
|
||||
},
|
||||
}
|
||||
|
||||
first_handle = await client.start_workflow(
|
||||
PIWebAPIScouter.run,
|
||||
input_data,
|
||||
id=f'test-workflow-idempotent-1-{unique_id}',
|
||||
task_queue='test-queue',
|
||||
)
|
||||
await first_handle.result()
|
||||
|
||||
second_handle = await client.start_workflow(
|
||||
PIWebAPIScouter.run,
|
||||
input_data,
|
||||
id=f'test-workflow-idempotent-2-{unique_id}',
|
||||
task_queue='test-queue',
|
||||
)
|
||||
await second_handle.result()
|
||||
|
||||
schema_name = 'sientia_data'
|
||||
table_name = 'laborious_data'
|
||||
full_table_name = f"{schema_name}.{table_name}"
|
||||
|
||||
with postgres_engine.connect() as conn:
|
||||
total_count_query = text(
|
||||
f"""
|
||||
SELECT COUNT(*)
|
||||
FROM {full_table_name}
|
||||
WHERE model_id = :model_id
|
||||
"""
|
||||
)
|
||||
total_count = conn.execute(total_count_query, {'model_id': unique_id}).scalar()
|
||||
|
||||
unique_count_query = text(
|
||||
f"""
|
||||
SELECT COUNT(*)
|
||||
FROM (
|
||||
SELECT DISTINCT model_id, "timestamp", variable
|
||||
FROM {full_table_name}
|
||||
WHERE model_id = :model_id
|
||||
) unique_rows
|
||||
"""
|
||||
)
|
||||
unique_count = conn.execute(unique_count_query, {'model_id': unique_id}).scalar()
|
||||
|
||||
assert total_count > 0, "Expected exported rows for idempotency scenario"
|
||||
assert total_count == unique_count, (
|
||||
"Expected no duplicate rows for same (model_id, timestamp, variable)"
|
||||
)
|
||||
assert mock_pi_web_api_client.get_latest_values_df.call_count == 2
|
||||
|
||||
|
||||
604
get_data_pims.py
Normal file
604
get_data_pims.py
Normal file
@@ -0,0 +1,604 @@
|
||||
import requests # type: ignore
|
||||
import pandas as pd # type: ignore
|
||||
from typing import Optional, Dict, List
|
||||
|
||||
|
||||
class PIMSClient:
|
||||
"""
|
||||
Cliente para interagir com a API do PIMS (PI System) da Votorantim.
|
||||
|
||||
Esta classe fornece métodos para autenticação e busca de dados de streams/tags
|
||||
do sistema PIMS através da API REST.
|
||||
"""
|
||||
|
||||
def __init__(self, base_url: str, api_key: Optional[str] = None, api_key_header: Optional[str] = "apikey", additional_headers: Optional[Dict[str, str]] = None):
|
||||
"""
|
||||
Inicializa o cliente PIMS.
|
||||
|
||||
Args:
|
||||
base_url (str): URL base da API (ex: https://votorantim.apimanagement.br10.hana.ondemand.com/v2/webapi/piwebapi)
|
||||
api_key (Optional[str]): API Key para autenticação
|
||||
api_key_header (Optional[str]): Nome do header onde a chave deve ser enviada (ex: "X-API-Key", "Ocp-Apim-Subscription-Key", "apikey")
|
||||
additional_headers (Optional[Dict[str, str]]): Cabeçalhos adicionais para incluir em todas as requisições
|
||||
"""
|
||||
self.base_url = base_url.rstrip('/')
|
||||
self.api_key = api_key
|
||||
self.api_key_header = api_key_header
|
||||
self.session = requests.Session()
|
||||
self.additional_headers = additional_headers or {}
|
||||
self._authenticated = False
|
||||
|
||||
def authenticate(self) -> bool:
|
||||
"""
|
||||
Configura a autenticação via cabeçalhos.
|
||||
|
||||
Returns:
|
||||
bool: True se a configuração foi bem-sucedida, False caso contrário
|
||||
"""
|
||||
try:
|
||||
default_headers: Dict[str, str] = {
|
||||
'Content-Type': 'application/json',
|
||||
'Accept': 'application/json'
|
||||
}
|
||||
if self.api_key and self.api_key_header:
|
||||
default_headers[self.api_key_header] = self.api_key
|
||||
|
||||
# Mescla cabeçalhos adicionais (sobrescrevem os padrões se necessário)
|
||||
default_headers.update(self.additional_headers)
|
||||
|
||||
self.session.headers.update(default_headers)
|
||||
|
||||
# Não faz chamada de teste aqui para evitar 401 em endpoints protegidos; assume headers configurados
|
||||
self._authenticated = True
|
||||
return True
|
||||
|
||||
except requests.exceptions.RequestException as e:
|
||||
print(f"Erro na configuração da autenticação: {e}")
|
||||
return False
|
||||
|
||||
def get_stream_data(self, web_ids: Dict[str, str], start_time: str = "*-3d", end_time: str = "*") -> Optional[pd.DataFrame]:
|
||||
"""
|
||||
Busca dados de múltiplos streams/tags.
|
||||
|
||||
Args:
|
||||
web_ids (Dict[str, str]): Dicionário no formato {tag_name: web_id}
|
||||
start_time (str): Data/hora de início (formato: "*-3d" ou "yyyy-mm-dd")
|
||||
end_time (str): Data/hora de fim (formato: "*" ou "yyyy-mm-dd")
|
||||
|
||||
Returns:
|
||||
pd.DataFrame: DataFrame onde as colunas são o nome da tag, o índice é o Timestamp, e os valores são os valores das tags
|
||||
"""
|
||||
if not self._authenticated:
|
||||
if not self.authenticate():
|
||||
print("Falha na autenticação")
|
||||
return None
|
||||
|
||||
all_series = []
|
||||
|
||||
for tag_name, web_id in web_ids.items():
|
||||
try:
|
||||
url = f"{self.base_url}/streams/{web_id}/recorded"
|
||||
params = {
|
||||
"startTime": start_time,
|
||||
"endTime": end_time
|
||||
}
|
||||
response = self.session.get(url, params=params)
|
||||
response.raise_for_status()
|
||||
|
||||
data = response.json()
|
||||
if 'Items' in data and data['Items']:
|
||||
df = pd.DataFrame(data['Items'])
|
||||
if 'Timestamp' in df.columns and 'Value' in df.columns:
|
||||
# Converte timestamp
|
||||
try:
|
||||
df['Timestamp'] = pd.to_datetime(
|
||||
df['Timestamp'],
|
||||
format='ISO8601',
|
||||
utc=True,
|
||||
errors='coerce'
|
||||
)
|
||||
except TypeError:
|
||||
df['Timestamp'] = pd.to_datetime(
|
||||
df['Timestamp'],
|
||||
utc=True,
|
||||
errors='coerce'
|
||||
)
|
||||
# Arredonda timestamps para precisão de segundos
|
||||
df['Timestamp'] = df['Timestamp'].dt.floor('s')
|
||||
# Normaliza valores: quando a API retorna um dict, tenta extrair um número
|
||||
def _extract_numeric(v):
|
||||
if isinstance(v, dict):
|
||||
# Casos comuns: {'Value': <num>} ou aninhados
|
||||
inner = v.get('Value')
|
||||
if isinstance(inner, (int, float)):
|
||||
return inner
|
||||
# Tenta outros campos conhecidos
|
||||
for k in ('NumericValue',):
|
||||
inner2 = v.get(k)
|
||||
if isinstance(inner2, (int, float)):
|
||||
return inner2
|
||||
return None
|
||||
if isinstance(v, (int, float)):
|
||||
return v
|
||||
# Converte strings numéricas, demais viram NaN
|
||||
return pd.to_numeric(v, errors='coerce')
|
||||
|
||||
df['Value'] = df['Value'].apply(_extract_numeric)
|
||||
df['Value'] = pd.to_numeric(df['Value'], errors='coerce')
|
||||
|
||||
df.set_index('Timestamp', inplace=True)
|
||||
# Agrega valores por segundo para remover índices duplicados
|
||||
series = (
|
||||
df['Value']
|
||||
.groupby(level=0)
|
||||
.mean()
|
||||
.sort_index()
|
||||
.rename(tag_name)
|
||||
)
|
||||
all_series.append(series)
|
||||
else:
|
||||
print(f"Nenhum dado encontrado para a tag '{tag_name}' no período especificado")
|
||||
|
||||
except requests.exceptions.RequestException as e:
|
||||
print(f"Erro ao buscar dados do stream {tag_name}: {e}")
|
||||
|
||||
if all_series:
|
||||
result_df = pd.concat(all_series, axis=1)
|
||||
return result_df
|
||||
else:
|
||||
print("Nenhum dado encontrado para as tags informadas.")
|
||||
return pd.DataFrame()
|
||||
|
||||
def search_streams(self, name_filter: Optional[str] = None, tag_filter: Optional[str] = None) -> Optional[List[Dict]]:
|
||||
"""
|
||||
Busca streams disponíveis com filtros opcionais.
|
||||
|
||||
Args:
|
||||
name_filter (str): Filtro por nome do stream
|
||||
tag_filter (str): Filtro por tag
|
||||
|
||||
Returns:
|
||||
List[Dict]: Lista de streams encontrados ou None se houver erro
|
||||
"""
|
||||
if not self._authenticated:
|
||||
if not self.authenticate():
|
||||
print("Falha na autenticação")
|
||||
return None
|
||||
|
||||
try:
|
||||
search_url = f"{self.base_url}/streams"
|
||||
params = {}
|
||||
|
||||
if name_filter:
|
||||
params['nameFilter'] = name_filter
|
||||
if tag_filter:
|
||||
params['tag'] = tag_filter
|
||||
|
||||
response = self.session.get(search_url, params=params)
|
||||
response.raise_for_status()
|
||||
|
||||
return response.json().get('Items', [])
|
||||
|
||||
except requests.exceptions.RequestException as e:
|
||||
print(f"Erro ao buscar streams: {e}")
|
||||
return None
|
||||
|
||||
def get_stream_info(self, web_id: str) -> Optional[Dict]:
|
||||
"""
|
||||
Obtém informações detalhadas de um stream específico.
|
||||
|
||||
Args:
|
||||
web_id (str): WebID do stream
|
||||
|
||||
Returns:
|
||||
Dict: Informações do stream ou None se houver erro
|
||||
"""
|
||||
if not self._authenticated:
|
||||
if not self.authenticate():
|
||||
print("Falha na autenticação")
|
||||
return None
|
||||
|
||||
try:
|
||||
info_url = f"{self.base_url}/streams/{web_id}"
|
||||
response = self.session.get(info_url)
|
||||
response.raise_for_status()
|
||||
|
||||
return response.json()
|
||||
|
||||
except requests.exceptions.RequestException as e:
|
||||
print(f"Erro ao buscar informações do stream: {e}")
|
||||
return None
|
||||
|
||||
def get_web_ids_by_tags(self, data_server_id: str, tag_names: List[str]) -> Dict[str, Optional[str]]:
|
||||
"""
|
||||
Retorna os WebIds para uma lista de tags (pontos) em um Data Server específico.
|
||||
|
||||
Args:
|
||||
data_server_id (str): ID/WebId do Data Server (ex: "F1DS-...")
|
||||
tag_names (List[str]): Lista com os nomes exatos das tags
|
||||
|
||||
Returns:
|
||||
Dict[str, Optional[str]]: Dicionário mapeando tag -> WebId (ou None se não encontrada)
|
||||
"""
|
||||
|
||||
# Garante autenticação, similar ao script de teste
|
||||
if not self._authenticated:
|
||||
if not self.authenticate():
|
||||
print("Falha na autenticação")
|
||||
return {tag: None for tag in tag_names}
|
||||
|
||||
results: Dict[str, Optional[str]] = {}
|
||||
|
||||
for tag in tag_names:
|
||||
try:
|
||||
# Monta a URL seguindo a lógica do script de teste fornecido no contexto
|
||||
url = f"{self.base_url}/dataservers/{data_server_id}/points"
|
||||
params = {"namefilter": tag}
|
||||
print(url, params)
|
||||
response = self.session.get(url, params=params)
|
||||
response.raise_for_status()
|
||||
|
||||
data = response.json()
|
||||
|
||||
# Corrige: procurar a lista 'Items' como no script de teste
|
||||
items = data.get("Items", []) if isinstance(data, dict) else []
|
||||
|
||||
web_id_value: Optional[str] = None
|
||||
if items:
|
||||
# Emula exatamente o resultado do script: pega primeiro item se disponível
|
||||
first_item = items[0]
|
||||
if isinstance(first_item, dict):
|
||||
web_id_value = first_item.get("WebId")
|
||||
|
||||
results[tag] = web_id_value
|
||||
except requests.exceptions.RequestException as e:
|
||||
print(f"Erro ao buscar WebId para tag '{tag}': {e}")
|
||||
results[tag] = None
|
||||
|
||||
return results
|
||||
|
||||
def multi_tags_agregadas(
|
||||
self,
|
||||
web_ids: List[str],
|
||||
start_time: str,
|
||||
end_time: str,
|
||||
summary_duration: str = "15m",
|
||||
summary_type: str = "average",
|
||||
selected_fields: str = "Items.Name;Items.Items.Type;Items.Items.Value.Timestamp;Items.Items.Value.Value;Items.Items.Value.Good",
|
||||
batch_size: int = 50,
|
||||
) -> Optional[Dict]:
|
||||
"""
|
||||
Chama o endpoint /streamsets/summary com múltiplos webids via GET e retorna o JSON bruto.
|
||||
Para evitar URLs muito longas, realiza chamadas em lotes e agrega os resultados.
|
||||
|
||||
Args:
|
||||
web_ids (List[str]): Lista de WebIds a consultar
|
||||
start_time (str): Início do período (ex.: "2024-09-05" ou "*-1d")
|
||||
end_time (str): Fim do período (ex.: "2024-09-06" ou "*")
|
||||
summary_duration (str): Duração do resumo (ex.: "15m")
|
||||
summary_type (str): Tipo de resumo (ex.: "average", "minimum", "maximum", etc.)
|
||||
selected_fields (str): Campos a retornar
|
||||
batch_size (int): Tamanho do lote de WebIds por requisição
|
||||
|
||||
Returns:
|
||||
Optional[Dict]: JSON com "Items" unificados ou None em caso de erro
|
||||
"""
|
||||
if not self._authenticated:
|
||||
if not self.authenticate():
|
||||
print("Falha na autenticação")
|
||||
return None
|
||||
|
||||
try:
|
||||
url = f"{self.base_url}/streamsets/summary"
|
||||
all_items: List[Dict] = []
|
||||
|
||||
for i in range(0, len(web_ids), batch_size):
|
||||
chunk = web_ids[i:i + batch_size]
|
||||
# Constrói lista de tuplas para repetir 'webid' como múltiplos params
|
||||
params: List[tuple] = [("webid", wid) for wid in chunk]
|
||||
params.extend([
|
||||
("startTime", start_time),
|
||||
("endtime", end_time), # conforme imagem
|
||||
("summaryDuration", summary_duration),
|
||||
("summaryType", summary_type),
|
||||
("selectedFields", selected_fields),
|
||||
])
|
||||
|
||||
response = self.session.get(url, params=params, timeout=600000)
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
items = data.get("Items", []) if isinstance(data, dict) else []
|
||||
if isinstance(items, list):
|
||||
all_items.extend(items)
|
||||
|
||||
return {"Items": all_items}
|
||||
|
||||
except requests.exceptions.RequestException as e:
|
||||
print(f"Erro ao buscar dados brutos de múltiplas tags: {e}")
|
||||
return None
|
||||
|
||||
def multi_tags_agregadas_df(
|
||||
self,
|
||||
web_ids: List[str],
|
||||
start_time: str,
|
||||
end_time: str,
|
||||
summary_duration: str = "1m",
|
||||
summary_type: str = "average",
|
||||
selected_fields: str = "Items.Name;Items.Items.Type;Items.Items.Value.Timestamp;Items.Items.Value.Value;Items.Items.Value.Good"
|
||||
) -> pd.DataFrame:
|
||||
"""
|
||||
Chama /streamsets/summary para múltiplos webids e retorna DataFrame:
|
||||
- índice: Timestamp (precisão de segundos)
|
||||
- colunas: nome da tag
|
||||
- células: Value (numérico)
|
||||
"""
|
||||
raw = self.multi_tags_agregadas(
|
||||
web_ids=web_ids,
|
||||
start_time=start_time,
|
||||
end_time=end_time,
|
||||
summary_duration=summary_duration,
|
||||
summary_type=summary_type,
|
||||
selected_fields=selected_fields,
|
||||
)
|
||||
|
||||
if not raw or 'Items' not in raw or not isinstance(raw['Items'], list):
|
||||
return pd.DataFrame()
|
||||
|
||||
records = []
|
||||
|
||||
def _extract_numeric(v):
|
||||
if isinstance(v, dict):
|
||||
inner = v.get('Value')
|
||||
if isinstance(inner, (int, float)):
|
||||
return inner
|
||||
for k in ('NumericValue',):
|
||||
inner2 = v.get(k)
|
||||
if isinstance(inner2, (int, float)):
|
||||
return inner2
|
||||
return None
|
||||
if isinstance(v, (int, float)):
|
||||
return v
|
||||
return pd.to_numeric(v, errors='coerce')
|
||||
|
||||
for entry in raw['Items']:
|
||||
tag_name = entry.get('Name')
|
||||
series_items = entry.get('Items') or []
|
||||
for it in series_items:
|
||||
v = (it.get('Value') or {}) if isinstance(it, dict) else {}
|
||||
ts = v.get('Timestamp') if isinstance(v, dict) else None
|
||||
val = v.get('Value') if isinstance(v, dict) else None
|
||||
val = _extract_numeric(val)
|
||||
if ts is not None:
|
||||
records.append({
|
||||
'Timestamp': ts,
|
||||
'Tag': tag_name,
|
||||
'Value': val,
|
||||
})
|
||||
|
||||
if not records:
|
||||
return pd.DataFrame()
|
||||
|
||||
df = pd.DataFrame.from_records(records)
|
||||
# Converte e arredonda timestamps
|
||||
try:
|
||||
df['Timestamp'] = pd.to_datetime(df['Timestamp'], format='ISO8601', utc=True, errors='coerce')
|
||||
except TypeError:
|
||||
df['Timestamp'] = pd.to_datetime(df['Timestamp'], utc=True, errors='coerce')
|
||||
df['Timestamp'] = df['Timestamp'].dt.floor('s')
|
||||
|
||||
# Pivot: índice timestamp, colunas nome da tag, valores numéricos
|
||||
df_pivot = df.pivot_table(index='Timestamp', columns='Tag', values='Value', aggfunc='mean')
|
||||
df_pivot.sort_index(inplace=True)
|
||||
return df_pivot
|
||||
|
||||
def multi_tags_brutas(
|
||||
self,
|
||||
web_ids: List[str],
|
||||
start_time: str,
|
||||
end_time: str,
|
||||
max_count: int = 10000,
|
||||
batch_size: int = 50,
|
||||
) -> Optional[Dict]:
|
||||
"""
|
||||
Chama o endpoint /streamsets/recorded com múltiplos webids via GET e retorna o JSON bruto.
|
||||
Para evitar URLs muito longas, realiza chamadas em lotes e agrega os resultados.
|
||||
|
||||
Args:
|
||||
web_ids (List[str]): Lista de WebIds a consultar
|
||||
start_time (str): Início do período (ex.: "2024-09-01" ou "*-1d")
|
||||
end_time (str): Fim do período (ex.: "2024-09-09" ou "*")
|
||||
max_count (int): Número máximo de registros a retornar por requisição
|
||||
batch_size (int): Tamanho do lote de WebIds por requisição
|
||||
|
||||
Returns:
|
||||
Optional[Dict]: JSON com "Items" unificados ou None em caso de erro
|
||||
"""
|
||||
if not self._authenticated:
|
||||
if not self.authenticate():
|
||||
print("Falha na autenticação")
|
||||
return None
|
||||
|
||||
try:
|
||||
url = f"{self.base_url}/streamsets/recorded"
|
||||
all_items: List[Dict] = []
|
||||
|
||||
for i in range(0, len(web_ids), batch_size):
|
||||
chunk = web_ids[i:i + batch_size]
|
||||
# Constrói lista de tuplas para repetir 'webid' como múltiplos params
|
||||
params: List[tuple] = [("webid", wid) for wid in chunk]
|
||||
params.extend([
|
||||
("startTime", start_time),
|
||||
("endTime", end_time),
|
||||
("maxCount", str(max_count)),
|
||||
])
|
||||
|
||||
response = self.session.get(url, params=params, timeout=600000)
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
items = data.get("Items", []) if isinstance(data, dict) else []
|
||||
if isinstance(items, list):
|
||||
all_items.extend(items)
|
||||
|
||||
return {"Items": all_items}
|
||||
|
||||
except requests.exceptions.RequestException as e:
|
||||
print(f"Erro ao buscar dados brutos de múltiplas tags: {e}")
|
||||
return None
|
||||
|
||||
def multi_tags_brutas_df(
|
||||
self,
|
||||
web_ids: List[str],
|
||||
start_time: str,
|
||||
end_time: str,
|
||||
max_count: int = 10000,
|
||||
) -> pd.DataFrame:
|
||||
"""
|
||||
Chama /streamsets/recorded para múltiplos webids e retorna DataFrame:
|
||||
- índice: Timestamp (precisão de segundos)
|
||||
- colunas: nome da tag
|
||||
- células: Value (numérico)
|
||||
|
||||
Args:
|
||||
web_ids (List[str]): Lista de WebIds a consultar
|
||||
start_time (str): Início do período (ex.: "2024-09-01" ou "*-1d")
|
||||
end_time (str): Fim do período (ex.: "2024-09-09" ou "*")
|
||||
max_count (int): Número máximo de registros a retornar por requisição
|
||||
|
||||
Returns:
|
||||
pd.DataFrame: DataFrame com timestamp como índice e tags como colunas
|
||||
"""
|
||||
raw = self.multi_tags_brutas(
|
||||
web_ids=web_ids,
|
||||
start_time=start_time,
|
||||
end_time=end_time,
|
||||
max_count=max_count,
|
||||
)
|
||||
|
||||
if not raw or 'Items' not in raw or not isinstance(raw['Items'], list):
|
||||
return pd.DataFrame()
|
||||
|
||||
records = []
|
||||
|
||||
def _extract_numeric(v):
|
||||
if isinstance(v, dict):
|
||||
inner = v.get('Value')
|
||||
if isinstance(inner, (int, float)):
|
||||
return inner
|
||||
for k in ('NumericValue',):
|
||||
inner2 = v.get(k)
|
||||
if isinstance(inner2, (int, float)):
|
||||
return inner2
|
||||
return None
|
||||
if isinstance(v, (int, float)):
|
||||
return v
|
||||
return pd.to_numeric(v, errors='coerce')
|
||||
|
||||
for entry in raw['Items']:
|
||||
tag_name = entry.get('Name')
|
||||
# Para /streamsets/recorded, cada entry tem uma lista 'Items' com objetos contendo Timestamp e Value diretamente
|
||||
series_items = entry.get('Items') or []
|
||||
|
||||
# Processa items aninhados
|
||||
for it in series_items:
|
||||
if isinstance(it, dict):
|
||||
# Estrutura: {'Timestamp': '2024-09-01T23:00:00Z', 'Value': 4431.94141, ...}
|
||||
if 'Timestamp' in it and 'Value' in it:
|
||||
ts = it.get('Timestamp')
|
||||
val = it.get('Value')
|
||||
val = _extract_numeric(val)
|
||||
if ts is not None:
|
||||
records.append({
|
||||
'Timestamp': ts,
|
||||
'Tag': tag_name,
|
||||
'Value': val,
|
||||
})
|
||||
|
||||
if not records:
|
||||
return pd.DataFrame()
|
||||
|
||||
df = pd.DataFrame.from_records(records)
|
||||
# Converte e arredonda timestamps
|
||||
try:
|
||||
df['Timestamp'] = pd.to_datetime(df['Timestamp'], format='ISO8601', utc=True, errors='coerce')
|
||||
except TypeError:
|
||||
df['Timestamp'] = pd.to_datetime(df['Timestamp'], utc=True, errors='coerce')
|
||||
df['Timestamp'] = df['Timestamp'].dt.floor('s')
|
||||
|
||||
# Pivot: índice timestamp, colunas nome da tag, valores numéricos
|
||||
# Agrega valores duplicados no mesmo timestamp usando média
|
||||
df_pivot = df.pivot_table(index='Timestamp', columns='Tag', values='Value', aggfunc='mean')
|
||||
df_pivot.sort_index(inplace=True)
|
||||
return df_pivot
|
||||
|
||||
def close(self):
|
||||
"""Fecha a sessão HTTP."""
|
||||
self.session.close()
|
||||
|
||||
|
||||
|
||||
|
||||
# Exemplo de uso
|
||||
if __name__ == "__main__":
|
||||
# Configuração do cliente
|
||||
base_url = "https://votorantim.apimanagement.br10.hana.ondemand.com/v2/webapi/piwebapi"
|
||||
api_key = "zK4WbZAZGBwSaQ5GJzhPpp06P1PGueqP"
|
||||
|
||||
# Cria instância do cliente
|
||||
pims_client = PIMSClient(base_url, api_key)
|
||||
|
||||
# Exemplo: buscar dados de um stream específico
|
||||
tag_forms = [
|
||||
'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-W3W01A3', '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'
|
||||
]
|
||||
web_ids = pims_client.get_web_ids_by_tags("F1DS-7fYgsRTtUOa7V9NIwSujAUElIQVZD", tag_forms)
|
||||
|
||||
web_ids_list: List[str] = [wid for wid in web_ids.values() if isinstance(wid, str)]
|
||||
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
# Parâmetros iniciais apenas até o dia (granulometria diária)
|
||||
inicio = datetime(2023, 1, 1) # Somente a data, sem horas/minutos/segundos
|
||||
fim = datetime.today().replace(hour=0, minute=0, second=0, microsecond=0) # Até hoje à 00:00 (começo do dia atual)
|
||||
delta = timedelta(days=1)
|
||||
|
||||
|
||||
dfs = [] # lista para armazenar os dataframes parciais
|
||||
|
||||
while inicio < fim:
|
||||
proximo = min(inicio + delta, fim) # garante que não passa da data atual
|
||||
|
||||
print(f"Buscando de {inicio:%Y-%m-%d} até {proximo:%Y-%m-%d}...")
|
||||
|
||||
df_parcial = pims_client.multi_tags_brutas_df(
|
||||
web_ids_list,
|
||||
inicio.strftime("%Y-%m-%d"),
|
||||
proximo.strftime("%Y-%m-%d"),
|
||||
max_count=1000
|
||||
)
|
||||
|
||||
dfs.append(df_parcial)
|
||||
inicio = proximo # avança o cursor
|
||||
# break
|
||||
|
||||
# concatena todos em um único dataframe
|
||||
df_final = pd.concat(dfs, ignore_index=False)
|
||||
df_final.reset_index(inplace=True)
|
||||
df_final.rename(columns={'index': 'timestamp'}, inplace=True)
|
||||
# df_final = df_final.ffill()
|
||||
# df_final = df_final.bfill()
|
||||
|
||||
# save to csv
|
||||
if not df_final.empty:
|
||||
df_final.to_parquet("data_brutos_pims_no_fill.parquet", index=False)
|
||||
|
||||
print(df_final.head())
|
||||
print(df_final.shape)
|
||||
print(df_final.columns)
|
||||
306
pi_web_api_fetch_data.py
Normal file
306
pi_web_api_fetch_data.py
Normal file
@@ -0,0 +1,306 @@
|
||||
"""
|
||||
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)
|
||||
# %%
|
||||
@@ -118,6 +118,7 @@ class CoreScouter:
|
||||
'schema': input_data['schema'],
|
||||
'table_name': input_data['table_name'],
|
||||
'data': held_data,
|
||||
'on_conflict': 'ignore',
|
||||
'timestamp_conversion': {'column': 'timestamp', 'format': DATETIME_FORMAT_WITH_TZ},
|
||||
},
|
||||
retry_policy=retry_policy,
|
||||
|
||||
@@ -114,6 +114,7 @@ async def test_core_scouter_workflow_success(mock_workflow, core_scouter):
|
||||
'schema': 'test_schema',
|
||||
'table_name': 'test_table',
|
||||
'data': 'held_data',
|
||||
'on_conflict': 'ignore',
|
||||
'timestamp_conversion': {
|
||||
'column': 'timestamp',
|
||||
'format': DATETIME_FORMAT_WITH_TZ,
|
||||
@@ -299,6 +300,7 @@ async def test_core_scouter_workflow_with_zero_affected_rows(mock_workflow, core
|
||||
'schema': 'test_schema',
|
||||
'table_name': 'test_table',
|
||||
'data': 'held_data',
|
||||
'on_conflict': 'ignore',
|
||||
'timestamp_conversion': {
|
||||
'column': 'timestamp',
|
||||
'format': DATETIME_FORMAT_WITH_TZ,
|
||||
@@ -369,6 +371,7 @@ async def test_core_scouter_workflow_without_debug_data_package(mock_workflow, c
|
||||
'schema': 'test_schema',
|
||||
'table_name': 'test_table',
|
||||
'data': 'held_data',
|
||||
'on_conflict': 'ignore',
|
||||
'timestamp_conversion': {
|
||||
'column': 'timestamp',
|
||||
'format': DATETIME_FORMAT_WITH_TZ,
|
||||
|
||||
Reference in New Issue
Block a user