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

@@ -32,6 +32,7 @@ E2E_DATABASE = 'scouter_e2e_test'
 E2E_NOTIFICATION_COLLECTION = 'notification_queue'
 DB_SCHEMA_PATH = Path(__file__).resolve().parent / 'db_schema.sql'
 
+
 def _activity_list(activities: Activities) -> list:
     """Return bound activity callables for the E2E worker."""
     return [
@@ -264,7 +265,8 @@ def test_activities(
     Return:
         Live Activities instance (no unittest.mock.patch)
     """
-    pg_parts = postgres_connection_parts(postgres_container.get_connection_url())
+    pg_parts = postgres_connection_parts(
+        postgres_container.get_connection_url())
     redis_host = redis_container.get_container_host_ip()
     redis_port = int(redis_container.get_exposed_port(6379))
 
@@ -309,11 +311,14 @@ async def temporal_worker(temporal_env, test_activities):
         task_queue=SCOUTER_TASK_QUEUE,
         workflows=[Scouter, PIWebAPIScouter, CoreScouter],
         activities=_activity_list(test_activities),
-        activity_executor=ThreadPoolExecutor(max_workers=50, thread_name_prefix='e2e-activity'),
+        activity_executor=ThreadPoolExecutor(
+            max_workers=50, thread_name_prefix='e2e-activity'),
         max_concurrent_workflow_tasks=50,
         max_concurrent_activities=50,
         max_concurrent_local_activities=50,
-        workflow_task_poller_behavior=PollerBehaviorAutoscaling(minimum=1, initial=2, maximum=10),
+        workflow_task_poller_behavior=PollerBehaviorAutoscaling(
-        activity_task_poller_behavior=PollerBehaviorAutoscaling(minimum=1, initial=2, maximum=10),
+            minimum=1, initial=2, maximum=10),
+        activity_task_poller_behavior=PollerBehaviorAutoscaling(
+            minimum=1, initial=2, maximum=10),
     ) as worker:
         yield worker

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': <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)

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)
+# %%

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,

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,