diff --git a/e2e/conftest.py b/e2e/conftest.py index 34042fa..0be0a38 100644 --- a/e2e/conftest.py +++ b/e2e/conftest.py @@ -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) ); """ diff --git a/e2e/test_pi_web_api_scouter_success.py b/e2e/test_pi_web_api_scouter_success.py index 239bd12..db0926b 100644 --- a/e2e/test_pi_web_api_scouter_success.py +++ b/e2e/test_pi_web_api_scouter_success.py @@ -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 + diff --git a/get_data_pims.py b/get_data_pims.py new file mode 100644 index 0000000..ab65b2e --- /dev/null +++ b/get_data_pims.py @@ -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': } 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) \ No newline at end of file diff --git a/pi_web_api_fetch_data.py b/pi_web_api_fetch_data.py new file mode 100644 index 0000000..26761d4 --- /dev/null +++ b/pi_web_api_fetch_data.py @@ -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) +# %% diff --git a/scouter/workflow/sub_workflows/core_scouter.py b/scouter/workflow/sub_workflows/core_scouter.py index 3aac96b..1555a14 100644 --- a/scouter/workflow/sub_workflows/core_scouter.py +++ b/scouter/workflow/sub_workflows/core_scouter.py @@ -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, diff --git a/tests/workflow/sub_workflows/test_core_scouter.py b/tests/workflow/sub_workflows/test_core_scouter.py index 91e0fdd..66a6047 100644 --- a/tests/workflow/sub_workflows/test_core_scouter.py +++ b/tests/workflow/sub_workflows/test_core_scouter.py @@ -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,