From d7cd9043b2732954d08d11b15aae151495f7e203 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 24 Mar 2026 11:34:57 -0300 Subject: [PATCH 1/5] 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. --- e2e/conftest.py | 1 + e2e/test_pi_web_api_scouter_success.py | 98 +++ get_data_pims.py | 604 ++++++++++++++++++ pi_web_api_fetch_data.py | 306 +++++++++ .../workflow/sub_workflows/core_scouter.py | 1 + .../sub_workflows/test_core_scouter.py | 3 + 6 files changed, 1013 insertions(+) create mode 100644 get_data_pims.py create mode 100644 pi_web_api_fetch_data.py 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, From 09658bee5281f726e30b8c8eff2003621d97a899 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 24 Mar 2026 11:47:04 -0300 Subject: [PATCH 2/5] SIENTIAPDE-1712 Update values.yaml for GITHUB_BRANCH and POSTGRES credentials; add unique_columns to CoreScouter and related tests for conflict resolution. --- scouter/workflow/sub_workflows/core_scouter.py | 1 + tests/workflow/sub_workflows/test_core_scouter.py | 3 +++ values.yaml | 6 +++--- 3 files changed, 7 insertions(+), 3 deletions(-) diff --git a/scouter/workflow/sub_workflows/core_scouter.py b/scouter/workflow/sub_workflows/core_scouter.py index 1555a14..1daa235 100644 --- a/scouter/workflow/sub_workflows/core_scouter.py +++ b/scouter/workflow/sub_workflows/core_scouter.py @@ -119,6 +119,7 @@ class CoreScouter: 'table_name': input_data['table_name'], 'data': held_data, 'on_conflict': 'ignore', + 'unique_columns': ['model_id', 'timestamp', 'variable'], '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 66a6047..387118d 100644 --- a/tests/workflow/sub_workflows/test_core_scouter.py +++ b/tests/workflow/sub_workflows/test_core_scouter.py @@ -115,6 +115,7 @@ async def test_core_scouter_workflow_success(mock_workflow, core_scouter): 'table_name': 'test_table', 'data': 'held_data', 'on_conflict': 'ignore', + 'unique_columns': ['model_id', 'timestamp', 'variable'], 'timestamp_conversion': { 'column': 'timestamp', 'format': DATETIME_FORMAT_WITH_TZ, @@ -301,6 +302,7 @@ async def test_core_scouter_workflow_with_zero_affected_rows(mock_workflow, core 'table_name': 'test_table', 'data': 'held_data', 'on_conflict': 'ignore', + 'unique_columns': ['model_id', 'timestamp', 'variable'], 'timestamp_conversion': { 'column': 'timestamp', 'format': DATETIME_FORMAT_WITH_TZ, @@ -372,6 +374,7 @@ async def test_core_scouter_workflow_without_debug_data_package(mock_workflow, c 'table_name': 'test_table', 'data': 'held_data', 'on_conflict': 'ignore', + 'unique_columns': ['model_id', 'timestamp', 'variable'], 'timestamp_conversion': { 'column': 'timestamp', 'format': DATETIME_FORMAT_WITH_TZ, diff --git a/values.yaml b/values.yaml index c07158e..9197507 100644 --- a/values.yaml +++ b/values.yaml @@ -163,7 +163,7 @@ env: - name: GITHUB_REPO_URL value: "git@github.com:Aignosi/sientia-dataops-scouter_temporal.git" - name: GITHUB_BRANCH - value: "fix/SIENTIAPDE-1478" + value: "fix/SIENTIAPDE-1712" - name: PYTHON_APP value: "scouter.worker.worker" @@ -173,9 +173,9 @@ env: - name: POSTGRES_PORT value: "5432" - name: POSTGRES_USER - value: "sientia" + value: "posgres" - name: POSTGRES_PASSWORD - value: "sientia" + value: "nFqc81y6kwmr2zuAIx43DhiOosFCVPpeEfTtTWZflkNjB2j1KtEeIANkhFR9mAX3" - name: POSTGRES_DBNAME value: "sientia" - name: POSTGRES_MIN_CONNECTIONS From aa91f69c4b2fb93c3193b1c696ed4af38dd32542 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 24 Mar 2026 11:48:58 -0300 Subject: [PATCH 3/5] SIENTIAPDE-1712 Remove validate.sh script and fix typo in POSTGRES_USER value in values.yaml --- validate.sh | 124 ---------------------------------------------------- values.yaml | 2 +- 2 files changed, 1 insertion(+), 125 deletions(-) delete mode 100755 validate.sh diff --git a/validate.sh b/validate.sh deleted file mode 100755 index 72f2e22..0000000 --- a/validate.sh +++ /dev/null @@ -1,124 +0,0 @@ -#!/bin/bash -# Model Manager Code Validation Script -# This script runs all code quality checks before committing or deploying - -set -e # Exit on any error - -# Colors for output -RED='\033[0;31m' -GREEN='\033[0;32m' -YELLOW='\033[1;33m' -BLUE='\033[0;34m' -NC='\033[0m' # No Color - -# Args -FIX_MODE=false -while [[ $# -gt 0 ]]; do - case "$1" in - --fix) - FIX_MODE=true - shift - ;; - -h|--help) - echo "Usage: $0 [--fix]" - echo " --fix Apply Ruff auto-fixes (format and lint fixes)." - exit 0 - ;; - *) - echo -e "${RED}Unknown option: $1${NC}" - echo "Usage: $0 [--fix]" - exit 2 - ;; - esac -done - -echo -e "${BLUE}╔════════════════════════════════════════════════════════╗${NC}" -echo -e "${BLUE}║ Model Manager - Code Validation Suite ║${NC}" -echo -e "${BLUE}╚════════════════════════════════════════════════════════╝${NC}" -echo "" - -# Check if virtual environment is activated -if [[ -z "${VIRTUAL_ENV}" ]] && [[ -z "${CONDA_DEFAULT_ENV}" ]]; then - echo -e "${YELLOW}⚠️ Warning: No virtual environment detected${NC}" - echo -e "${YELLOW} Consider activating your venv/conda environment${NC}" - echo "" -fi - -# Function to run a validation step -run_step() { - local step_name=$1 - local step_command=$2 - - echo -e "${BLUE}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━${NC}" - echo -e "${BLUE}▶ ${step_name}${NC}" - echo -e "${BLUE}━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━${NC}" - - if eval "$step_command"; then - echo -e "${GREEN}✅ ${step_name} - PASSED${NC}" - echo "" - return 0 - else - echo -e "${RED}❌ ${step_name} - FAILED${NC}" - echo "" - return 1 - fi -} - -# Track failures -FAILED_STEPS=() - -# Step 1: Code Formatting Check (Ruff) -# - default: check only -# - --fix: write changes -if ! run_step "1. Code Formatting (Ruff)" "if \$FIX_MODE; then ruff format scouter/ tests/; else ruff format --check scouter/ tests/; fi"; then - FAILED_STEPS+=("Code Formatting") -fi - -# Step 2: Linting (Ruff) -# - default: check only -# - --fix: apply autofixes -if ! run_step "2. Code Linting (Ruff)" "if \$FIX_MODE; then ruff check --fix scouter/ tests/; else ruff check scouter/ tests/; fi"; then - FAILED_STEPS+=("Linting") -fi - -# Step 3: Type Checking (mypy) -if ! run_step "3. Type Checking (mypy)" "mypy scouter/"; then - FAILED_STEPS+=("Type Checking") -fi - -# Step 4: Security Analysis (Bandit) -if ! run_step "4. Security Analysis (Bandit)" "bandit -r scouter/ -ll -q"; then - FAILED_STEPS+=("Security Analysis") -fi - -# Step 5: Unit Tests (pytest) -if ! run_step "5. Unit Tests (pytest)" "pytest tests/ --cov=scouter --cov-report=term-missing --cov-report=xml --cov-report=html --cov-fail-under=80 -q"; then - FAILED_STEPS+=("Unit Tests") -fi - -# Summary -echo -e "${BLUE}╔════════════════════════════════════════════════════════╗${NC}" -echo -e "${BLUE}║ Validation Summary ║${NC}" -echo -e "${BLUE}╚════════════════════════════════════════════════════════╝${NC}" -echo "" - -if [ ${#FAILED_STEPS[@]} -eq 0 ]; then - echo -e "${GREEN}✅ All validation checks passed!${NC}" - echo -e "${GREEN} Your code is ready for commit/deployment.${NC}" - echo "" - exit 0 -else - echo -e "${RED}❌ Validation failed for the following steps:${NC}" - for step in "${FAILED_STEPS[@]}"; do - echo -e "${RED} • ${step}${NC}" - done - echo "" - echo -e "${YELLOW}💡 Tips:${NC}" - echo -e "${YELLOW} • Run 'ruff format scouter/ tests/' to auto-fix formatting${NC}" - echo -e "${YELLOW} • Run 'ruff check --fix scouter/ tests/' to auto-fix linting issues${NC}" - echo -e "${YELLOW} • Review mypy errors and add type hints where needed${NC}" - echo -e "${YELLOW} • Check bandit warnings for security issues${NC}" - echo -e "${YELLOW} • Fix failing tests or improve test coverage${NC}" - echo "" - exit 1 -fi \ No newline at end of file diff --git a/values.yaml b/values.yaml index 9197507..828e059 100644 --- a/values.yaml +++ b/values.yaml @@ -173,7 +173,7 @@ env: - name: POSTGRES_PORT value: "5432" - name: POSTGRES_USER - value: "posgres" + value: "postgres" - name: POSTGRES_PASSWORD value: "nFqc81y6kwmr2zuAIx43DhiOosFCVPpeEfTtTWZflkNjB2j1KtEeIANkhFR9mAX3" - name: POSTGRES_DBNAME From 724c2291a05fc292ab8d5541d82a91cf303cbb50 Mon Sep 17 00:00:00 2001 From: Bruno Domingues Date: Wed, 1 Jul 2026 21:18:44 -0300 Subject: [PATCH 4/5] chore(sonar): Update SonarQube project key --- sonar-project.properties | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sonar-project.properties b/sonar-project.properties index f98e3ab..5adaaf8 100644 --- a/sonar-project.properties +++ b/sonar-project.properties @@ -1,4 +1,4 @@ -sonar.projectKey=Aignosi_sientia-dataops-scouter_temporal_f79e575e-be40-4cc6-acce-b7c5ebb9a5b0 +sonar.projectKey=Aignosi_sientia-dataops-scouter_temporal_82f6f501-b2d4-45ec-b2bc-d8fdc1a8a06d sonar.projectName=sientia-dataops-scouter_temporal sonar.sources=scouter sonar.tests=tests From f560467aa151727755e2ebb1c8d855cf168a4723 Mon Sep 17 00:00:00 2001 From: Bruno Domingues Date: Wed, 8 Jul 2026 22:13:16 -0300 Subject: [PATCH 5/5] SIENTIAPDE-1945: Delete values.yaml file. --- values.yaml | 277 ---------------------------------------------------- 1 file changed, 277 deletions(-) delete mode 100644 values.yaml diff --git a/values.yaml b/values.yaml deleted file mode 100644 index c07158e..0000000 --- a/values.yaml +++ /dev/null @@ -1,277 +0,0 @@ -# Default values for sientia-module. -# This is a YAML-formatted file. -# Declare variables to be passed into your templates. - -# This will set the replicaset count more information can be found here: https://kubernetes.io/docs/concepts/workloads/controllers/replicaset/ -replicaCount: 1 - -# This sets the container image more information can be found here: https://kubernetes.io/docs/concepts/containers/images/ -image: - repository: aignosi.azurecr.io/sientia-module - # This sets the pull policy for images. - pullPolicy: Always - # Overrides the image tag whose default is the chart appVersion. - tag: "1.0.0" - -# This is for the secrets for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/ -imagePullSecrets: -- name: docker-hub-secret -# This is to override the chart name. -nameOverride: "sientia-scouter-worker" -fullnameOverride: "sientia-scouter-worker" -namespace: sientia - -# This section builds out the service account more information can be found here: https://kubernetes.io/docs/concepts/security/service-accounts/ -serviceAccount: - # Specifies whether a service account should be created - create: true - # Automatically mount a ServiceAccount's API credentials? - automount: true - # Annotations to add to the service account - annotations: {} - # The name of the service account to use. - # If not set and create is true, a name is generated using the fullname template - name: "sientia-scouter-worker" - -# This is for setting Kubernetes Annotations to a Pod. -# For more information checkout: https://kubernetes.io/docs/concepts/overview/working-with-objects/annotations/ -podAnnotations: {} -# This is for setting Kubernetes Labels to a Pod. -# For more information checkout: https://kubernetes.io/docs/concepts/overview/working-with-objects/labels/ -podLabels: {} - -podSecurityContext: {} - # fsGroup: 2000 - -securityContext: {} - # capabilities: - # drop: - # - ALL - # readOnlyRootFilesystem: true - # runAsNonRoot: true - # runAsUser: 1000 - - -resources: - # We usually recommend not to specify default resources and to leave this as a conscious - # choice for the user. This also increases chances charts run on environments with little - # resources, such as Minikube. If you do want to specify resources, uncomment the following - # lines, adjust them as necessary, and remove the curly braces after 'resources:'. - # limits: - # cpu: 100m - # memory: 128Mi - # requests: - # cpu: 100m - # memory: 128Mi - - limits: - cpu: 1000m - memory: 2048Mi - requests: - cpu: 300m - memory: 256Mi - -# This is to setup the liveness and readiness probes more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/configure-liveness-readiness-startup-probes/ -livenessProbe: - exec: - command: - - sh - - -c - - | - curl -sf http://localhost:9090/metrics | grep -q '^app_up{.*} 1' - initialDelaySeconds: 30 - periodSeconds: 15 - timeoutSeconds: 5 - failureThreshold: 3 - -readinessProbe: - exec: - command: - - sh - - -c - - | - curl -sf http://localhost:9090/metrics | grep -q '^app_up{.*} 1' - initialDelaySeconds: 20 - periodSeconds: 10 - timeoutSeconds: 3 - failureThreshold: 2 - - -# This section is for setting up autoscaling more information can be found here: https://kubernetes.io/docs/concepts/workloads/autoscaling/ -autoscaling: - enabled: false - minReplicas: 1 - maxReplicas: 100 - targetCPUUtilizationPercentage: 80 - # targetMemoryUtilizationPercentage: 80 - -# Additional volumes on the output Deployment definition. -volumes: [] -# - name: foo -# secret: -# secretName: mysecret -# optional: false - -# Additional volumeMounts on the output Deployment definition. -volumeMounts: [] -# - name: foo -# mountPath: "/etc/foo" -# readOnly: true - -nodeSelector: {} - -tolerations: [] - -affinity: {} - -services: - sdk-metrics: - enabled: true - type: ClusterIP - port: 9091 - targetPort: 9091 - name: sdk-metrics - - metrics: - enabled: true - type: ClusterIP - port: 9090 - targetPort: 9090 - name: metrics - -# Configuração do ServiceMonitor para o Prometheus Operator -# ref: https://github.com/prometheus-operator/prometheus-operator -serviceMonitor: - # Se true, um recurso ServiceMonitor será criado. - enabled: true - # O intervalo no qual as métricas devem ser coletadas (ex: 30s, 1m). - endpoints: - - port: metrics - path: /metrics - interval: 30s - relabelings: [] - - port: sdk-metrics - path: /metrics - interval: 30s - relabelings: [] - - additionalLabels: - release: kube-prometheus-stack - -env: - # Entrypoint variables - - name: GITHUB_REPO_URL - value: "git@github.com:Aignosi/sientia-dataops-scouter_temporal.git" - - name: GITHUB_BRANCH - value: "fix/SIENTIAPDE-1478" - - name: PYTHON_APP - value: "scouter.worker.worker" - - # Application variables - - name: POSTGRES_HOST - value: "paradedb-rw.paradedb.svc.cluster.local" - - name: POSTGRES_PORT - value: "5432" - - name: POSTGRES_USER - value: "sientia" - - name: POSTGRES_PASSWORD - value: "sientia" - - name: POSTGRES_DBNAME - value: "sientia" - - name: POSTGRES_MIN_CONNECTIONS - value: "10" - - name: POSTGRES_MAX_CONNECTIONS - value: "40" - - - name: REDIS_HOST - value: "redis-master.redis.svc.cluster.local" - - name: REDIS_PORT - value: "6379" - - name: REDIS_USERNAME - valueFrom: - secretKeyRef: - name: redis - key: redis-username - - name: REDIS_PASSWORD - valueFrom: - secretKeyRef: - name: redis - key: redis-password - - - name: LOG_LEVEL - value: "DEBUG" - - name: HTTP_METRICS_PORT - value: "9090" - - name: PROJECT_NAME - value: "sientia-scouter" - - - name: TEMPORAL_HOST - value: "temporal-frontend.temporal.svc.cluster.local:7233" - - name: TEMPORAL_NAMESPACE - value: "scouter" - - - name: MONGODB_USERNAME - value: "root" - - name: MONGODB_PASSWORD - value: "wKZDbMNU1c" - - name: MONGODB_URL - value: "my-release-mongodb.mongodb.svc.cluster.local:27017" - - name: MONGODB_DATABASE - value: "sientia" - - - name: PI_WEB_API_BASE_URL - value: "https://pivision.votorantimcimentos.com/piwebapi" - - name: PI_WEB_API_AUTH_TYPE - value: "basic" - - name: PI_WEB_API_AUTH_TOKEN - valueFrom: - secretKeyRef: - name: pi-web-api-auth-token - key: token - - - name: PYPI_SERVER - value: "http://library-distribution-server.library.svc.cluster.local:5000" - - # Temporal worker tuning - - name: SCOUTER_MAX_CONCURRENT_WORKFLOW_TASKS - value: "200" - - name: SCOUTER_MAX_CONCURRENT_ACTIVITIES - value: "200" - - name: SCOUTER_MAX_CONCURRENT_LOCAL_ACTIVITIES - value: "200" - - name: SCOUTER_MAX_CACHED_WORKFLOWS - value: "200" - - - name: SCOUTER_WORKFLOW_POLLER_BEHAVIOUR_MINIMUM - value: "10" - - name: SCOUTER_WORKFLOW_POLLER_BEHAVIOUR_INITIAL - value: "100" - - name: SCOUTER_WORKFLOW_POLLER_BEHAVIOUR_MAXIMUM - value: "200" - - - name: SCOUTER_ACTIVITY_POLLER_BEHAVIOUR_MINIMUM - value: "10" - - name: SCOUTER_ACTIVITY_POLLER_BEHAVIOUR_INITIAL - value: "100" - - name: SCOUTER_ACTIVITY_POLLER_BEHAVIOUR_MAXIMUM - value: "200" - - -ssh: - enabled: true - secretName: git-ssh-key-sientia-scouter-worker - sshPath: /mnt/.ssh - knownHostsPath: /mnt/known_hosts - -# kubectl create secret docker-registry docker-hub-secret --namespace sientia --docker-server=http://aignosi.azurecr.io --docker-username=aignosi --docker-password=5I5zpQ6sRaHqX1hD3dr+2mo647yO3FRc359/wu6gsP+ACRDRz5mp - -# helm upgrade --install sientia-scouter-worker sientia/sientia-module -n sientia --create-namespace -f ./values.yaml --version 0.6.0 - -# kubectl create secret generic git-ssh-key-sientia-scouter-worker \ -# --namespace sientia \ -# --from-file=ssh-privatekey=git_key \ -# --type=kubernetes.io/ssh-auth - -# kubectl create secret generic pi-web-api-auth-token \ -# --namespace sientia \ -# --from-literal=token=your-token-here \ No newline at end of file