Merge pull request #26 from imubit/add-history-harvester

add historical data harvester

Author: cloud-rocket

.github/workflows/ci.yml

@@ -85,7 +85,7 @@ jobs:
       - name: Generate coverage report
         run: pipx run coverage lcov -o coverage.lcov
       - name: Upload partial coverage report
-        uses: coverallsapp/github-action@master
+        uses: coverallsapp/github-action@v2
         with:
           path-to-lcov: coverage.lcov
           github-token: ${{ secrets.GITHUB_TOKEN }}
@@ -97,7 +97,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Finalize coverage report
-        uses: coverallsapp/github-action@master
+        uses: coverallsapp/github-action@v2
         with:
           github-token: ${{ secrets.GITHUB_TOKEN }}
           parallel-finished: true

.gitignore

@@ -8,7 +8,7 @@
 *.orig
 *.log
 *.pot
-__pycache__/*
+__pycache__/
 .cache/*
 .*.swp
 */.ipynb_checkpoints/*

AUTHORS.md

@@ -1,3 +1,3 @@
 # Contributors
 
-* Meir Tseitlin [meir@imubit.com](mailto:meir@imubit.com)
+* Meir Tseitlin [opensource@imubit.com](mailto:opensource@imubit.com)

README.md

@@ -6,9 +6,10 @@
 # Data Agent
 
 Python package for accessing real-time and historical data on industrial historians and control systems.
-Different historian protocols and APIs are implemented through standalone plugins.
+Different historian protocols and APIs are implemented through standalone plugins:
 
-*THIS PACKAGE IS USELESS WITHOUT EXTERNAL PLUGINS IMPLEMENTING TARGET SYSTEM CUSTOM DATA ACCESS PROTOCOLS*
+* https://github.com/imubit/data-agent-osisoft-pi
+* https://github.com/imubit/data-agent-aspen-ip21
 
 ## Description
 

setup.cfg

@@ -32,6 +32,8 @@ package_dir =
 install_requires =
     dynaconf
     pandas
+    msgpack
+    zstandard
     aiomisc
     aiodebug
     apscheduler

src/data_agent/abstract_connector.py

@@ -1,4 +1,5 @@
 from abc import ABC, abstractmethod
+from enum import IntEnum
 from functools import wraps
 from typing import Union
 
@@ -7,7 +8,7 @@
 from .exceptions import ConnectionNotActive, TagsGroupNotFound
 
 
-class SupportedOperation:
+class SupportedOperation(IntEnum):
     READ_TAG_VALUE = 1
     WRITE_TAG_VALUE = 2
     READ_TAG_PERIOD = 3
@@ -20,6 +21,12 @@ class SupportedOperation:
     DELETE_TAG = 10
 
 
+class HistDataFormat(IntEnum):
+    DATAFRAME = (1,)
+    SERIES_LIST = (2,)
+    DICTIONARY = 3
+
+
 STANDARD_ATTRIBUTES = {
     "Name": {"Type": "str", "Name": "Tag Name"},
     "Type": {"Type": "str", "Name": "Data Type"},

src/data_agent/daq_scheduler.py

@@ -1,3 +1,4 @@
+import json
 import logging
 import time
 from datetime import timedelta
@@ -80,12 +81,13 @@ async def _job_func(self, job_id, conn, broker, tags, from_cache, refresh_rate_m
 
             read_time = time.time() - start_time
 
-            msg = {
-                "job_id": job_id,
+            payload = {
                 "sample_id": self._job_state[job_id]["iter_counter"],
                 "data": tag_values,
             }
 
+            msg = json.dumps(payload, sort_keys=True, default=str).encode()
+
             # Publish
             to_publish = [f'{t}={tag_values[t]["Value"]}' for t in tag_values]
             self._total_iterations_counter += 1

src/data_agent/exceptions.py

@@ -76,3 +76,7 @@ class SafetyErrorManipulateOutsideOfRateBound(Exception):
 
 class DaqJobAlreadyExists(Exception):
     pass
+
+
+class HistoryHarvesterJobAlreadyExists(Exception):
+    pass

src/data_agent/history_harvester.py

@@ -0,0 +1,135 @@
+import datetime as dt
+import logging
+import time as tm
+
+from apscheduler.executors.pool import ThreadPoolExecutor
+from apscheduler.schedulers.asyncio import AsyncIOScheduler
+
+from data_agent.exceptions import HistoryHarvesterJobAlreadyExists
+from data_agent.msg_packer import encode_dataframe
+
+log = logging.getLogger(__name__)
+
+
+class HistoryHarvester:
+    def __init__(self, connection_manager, broker):
+        self.connection_manager = connection_manager
+        self.broker = broker
+
+        self._thread_pool_executor = ThreadPoolExecutor(max_workers=20)
+        self._scheduler = AsyncIOScheduler(thread_pool=self._thread_pool_executor)
+        self._scheduler.start()
+
+    async def _delivery_job_func(
+        self,
+        job_id,
+        conn,
+        tags,
+        first_timestamp,
+        last_timestamp,
+        time_frequency,
+        batch_size,
+        iteration,
+    ):
+        try:
+            start_time = tm.time()
+
+            next_period_end = min(last_timestamp, first_timestamp + batch_size)
+
+            df = conn.read_tag_values_period(
+                tags=tags,
+                first_timestamp=first_timestamp,
+                last_timestamp=next_period_end,
+                time_frequency=time_frequency,
+            )
+
+            read_time = tm.time() - start_time
+
+            if df.empty:
+                log.warning(
+                    f"No data read for job '{job_id}' for period {first_timestamp} - {next_period_end}"
+                )
+
+            else:  # Publish data
+                headers = {
+                    "data_category": "historical",
+                    "connection": conn.name,
+                    "job_id": job_id,
+                    "batch_num": iteration,
+                }
+
+                payload = encode_dataframe(df)
+
+                log.debug(
+                    f"(#{iteration}): Job {job_id}: "
+                    f"Data publish: read time={read_time:.2f}s), {len(df)} samples, "
+                    f"period: {first_timestamp} - {next_period_end}"
+                )
+                self.broker.publish_data(payload, headers=headers)
+
+            if next_period_end < last_timestamp:
+                # Reschedule next run
+                self._scheduler.add_job(
+                    func=self._delivery_job_func,
+                    trigger="date",
+                    # next_run_time=dt.datetime.now(),
+                    coalesce=True,  # Always run once
+                    id=job_id,
+                    max_instances=2,
+                    replace_existing=True,
+                    args=[
+                        job_id,
+                        conn,
+                        tags,
+                        next_period_end,
+                        last_timestamp,
+                        time_frequency,
+                        batch_size,
+                        iteration + 1,
+                    ],
+                )
+
+        except Exception as e:
+            log.exception(f'Exception in history harvester job "{job_id}" - {e}')
+
+    def create_delivery_job(
+        self,
+        job_id: str,
+        conn_name: str,
+        tags: list,
+        first_timestamp: dt.datetime,
+        last_timestamp: dt.datetime,
+        time_frequency: dt.timedelta,
+        batch_size: dt.timedelta = None,
+        progress_callback=None,
+    ):
+        # order tags alphabetically
+        tags.sort()
+
+        existing_job = self._scheduler.get_job(job_id)
+        if existing_job:
+            raise HistoryHarvesterJobAlreadyExists(
+                f"History loader Job {job_id} already exists."
+            )
+
+        conn = self.connection_manager.connection(conn_name, check_enabled=False)
+
+        self._scheduler.add_job(
+            func=self._delivery_job_func,
+            trigger="date",
+            # next_run_time=dt.datetime.now(),
+            coalesce=True,  # Always run once
+            id=job_id,
+            max_instances=2,
+            replace_existing=True,
+            args=[
+                job_id,
+                conn,
+                tags,
+                first_timestamp,
+                last_timestamp,
+                time_frequency,
+                batch_size,
+                0,
+            ],
+        )

src/data_agent/linux/config_default.yaml

@@ -160,6 +160,11 @@ log: # standard logging dictConfig
       level: 'DEBUG'
       propagate: False
 
+    data_agent.history_harvester:
+      handlers: ['console', 'amqp']
+      level: 'DEBUG'
+      propagate: False
+
     data_agent.connection_manager:
       handlers: ['console', 'amqp']
       level: 'DEBUG'