Mutigraph test (#3)

* Update multigraph.py

* cosmetics

* print starting nan if any in multigraph

* removing redefinition of builtin dir

* Update multigraph.py

* Update tests/test.py

Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>

* pylint commit

* oups

* oups

* Update __init__.py

---------

Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>

.pylintrc

@@ -0,0 +1,13 @@
+[MESSAGES CONTROL]
+
+disable=too-many-arguments,
+        too-many-instance-attributes,
+        line-too-long,
+        broad-except,
+
+max-nested-blocks=10
+max-statements=120
+max-branches=35
+max-locals=60
+
+good-names=PyFina,getMeta

PyFina/.pylintrc

@@ -0,0 +1,13 @@
+[MESSAGES CONTROL]
+
+disable=too-many-arguments,
+        too-many-instance-attributes,
+        line-too-long,
+        broad-except,
+
+max-nested-blocks=10
+max-statements=120
+max-branches=35
+max-locals=60
+
+good-names=PyFina,getMeta

PyFina/PyFina.py

@@ -1,20 +1,23 @@
-import numpy as np
+"""pyfina module"""
+
 import struct
 import os
 import math
 
-def trim(id, dir, limit=100):
+import numpy as np
+
+def trim(feed_id, data_dir, limit=100):
     """
     checks and removes anomalies (values above a threshold limit, eg 100)
-    id: feed number
-    dir: feed path (eg /var/opt/emoncms/phpfina)
+    feed_id: feed number
+    data_dir: feed path (eg /var/opt/emoncms/phpfina)
     limit: threshold we don't want to exceed
     """
-    meta = getMeta(id, dir)
+    meta = getMeta(feed_id, data_dir)
     pos = 0
     i = 0
     nbn = 0
-    with open(f"{dir}/{id}.dat", "rb+") as ts:
+    with open(f"{data_dir}/{feed_id}.dat", "rb+") as ts:
         while pos <= meta["npoints"]:
             ts.seek(pos*4, 0)
             hexa = ts.read(4)
@@ -38,17 +41,15 @@ def trim(id, dir, limit=100):
         print(f"{i} anomaly(ies)")
         print(f"{nbn} nan")
 
-def getMeta(id, dir):
+def getMeta(feed_id, data_dir):
     """
     decoding the .meta file
-
-    id (4 bytes, Unsigned integer)
+    feed_id (4 bytes, Unsigned integer)
     npoints (4 bytes, Unsigned integer, Legacy : use instead filesize//4 )
     interval (4 bytes, Unsigned integer)
     start_time (4 bytes, Unsigned integer)
-
     """
-    with open("{}/{}.meta".format(dir,id),"rb") as f:
+    with open(f"{data_dir}/{feed_id}.meta","rb") as f:
         f.seek(8,0)
         hexa = f.read(8)
         aa= bytearray(hexa)
@@ -58,37 +59,35 @@ def getMeta(id, dir):
             print("corrupted meta - aborting")
             return False
     meta = {
-             "interval":decoded[0],
-             "start_time":decoded[1],
-             "npoints":os.path.getsize("{}/{}.dat".format(dir,id))//4
-           }
+        "interval": decoded[0],
+        "start_time": decoded[1],
+        "npoints": os.path.getsize(f"{data_dir}/{feed_id}.dat") // 4
+    }
     return meta
 
 class PyFina(np.ndarray):
-
-    def __new__(cls, id, dir, start, step, npts, remove_nan=True):
-        meta = getMeta(id, dir)
+    """ pyfina class."""
+    def __new__(cls, feed_id, data_dir, start, step, npts, remove_nan=True):
+        meta = getMeta(feed_id, data_dir)
         if not meta:
-            return
-        """
-        decoding and sampling the .dat file
-        values are 32 bit floats, stored on 4 bytes
-        to estimate value(time), position in the dat file is calculated as follow :
-        pos = (time - meta["start_time"]) // meta["interval"]
-        Nota : no NAN value - if a NAN is detected, the algorithm will fetch the first non NAN value in the future
-        """
+            return None
+        # decoding and sampling the .dat file
+        # values are 32 bit floats, stored on 4 bytes
+        # to estimate value(time), position in the dat file is calculated as follow :
+        # pos = (time - meta["start_time"]) // meta["interval"]
+        # Nota : if remove_nan is True and a NAN is detected, the algorithm takes previous value
         obj = np.zeros(npts).view(cls)
         raw_obj = np.empty(npts)
 
         end = start + (npts-1) * step
         time = start
         i = 0
         nb_nan = 0
-        with open(f"{dir}/{id}.dat", "rb") as ts:
+        with open(f"{data_dir}/{feed_id}.dat", "rb") as ts:
             while time < end:
                 time = start + step * i
                 pos = (time - meta["start_time"]) // meta["interval"]
-                if pos >=0 and pos < meta["npoints"]:
+                if 0 <= pos < meta["npoints"]:
                     try:
                         #print(f"trying to find point {i} going to index {pos}")
                         ts.seek(pos*4, 0)
@@ -116,11 +115,10 @@ def __new__(cls, id, dir, start, step, npts, remove_nan=True):
             first_non_nan_value = raw_obj[finiteness_obj][0]
             if starting_by_nan and remove_nan:
                 obj[:first_non_nan_index] = np.ones(first_non_nan_index) * first_non_nan_value
-        """
-        storing the "signature" of the "sampled" feed
-        """
+        # storing the "signature" of the "sampled" feed
         obj.start = start
         obj.step = step
+        obj.nb_nan = nb_nan
         obj.first_non_nan_value = first_non_nan_value
         obj.first_non_nan_index = first_non_nan_index
         obj.starting_by_nan = starting_by_nan
@@ -129,8 +127,8 @@ def __new__(cls, id, dir, start, step, npts, remove_nan=True):
     def __array_finalize__(self, obj):
         if obj is None:
             return
-        self.start = getattr(obj, 'start', None)
-        self.step = getattr(obj, 'step', None)
+        self.start = getattr(obj, 'start', None)   # pylint: disable=W0201
+        self.step = getattr(obj, 'step', None)  # pylint: disable=W0201
 
     def timescale(self):
         """

PyFina/__init__.py

@@ -1 +1,2 @@
-from .PyFina import *
+"""PyFina module"""
+from .PyFina import PyFina, getMeta, trim

tests/multigraph.py

@@ -1,17 +1,28 @@
-from PyFina import getMeta, PyFina
-import matplotlib.pylab as plt
+"""produces some snapshots opening the bloch dataset."""
+
 import datetime
-import time
 import random
+import time
+
+import matplotlib
+import matplotlib.pylab as plt
+from PyFina import getMeta, PyFina
 
 # télécharger le fichier contenant les données de Marc Bloch 2021
-# unzip your files in C:/var/opt/emoncms
-dir = "C:/var/opt/emoncms/phpfina"
-feeds = { "Text" : 5, "TziqNord": 8, "TtecNord": 11, "TdepNord": 21, "TretNord": 22, "pompeNord": 23}
-step = 3600
-verbose = False
-# epL : episode length : 8 days !!
-epL = 8*24*3600
+# puis lancer tar -xvf pour décompresser
+DATA_DIR = "phpfina"
+feeds = {
+    "Text" : 5,
+    "TziqNord": 8,
+    "TtecNord": 11,
+    "TdepNord": 21,
+    "TretNord": 22,
+    "pompeNord": 23
+}
+STEP = 3600
+VERBOSE = False
+# episode length : 8 days !!
+EPISODE_LENGTH = 8 * 24 * 3600
 
 def analyse():
     """
@@ -20,9 +31,9 @@ def analyse():
     """
     starts = []
     ends = []
-    for f in feeds:
-        meta = getMeta(feeds[f], dir)
-        if verbose:
+    for _, feed_nb in feeds.items():
+        meta = getMeta(feed_nb, DATA_DIR)
+        if VERBOSE:
             print(meta)
         start = meta["start_time"]
         length = meta["npoints"] * meta["interval"]
@@ -32,77 +43,67 @@ def analyse():
     start = max(starts)
     end = min(ends)
     length = end - start
-    nbpts = epL // step
-    if epL > length:
-        nbpts = length // step
+    nbpts = EPISODE_LENGTH // STEP
+    if EPISODE_LENGTH > length:
+        nbpts = length // STEP
     return start, end, nbpts
 
-start, end, nbpts = analyse()
-
-def viewEpisode(start_ts):
-    """
-    permet de visualiser un épisode commencant à start_ts
-    """
-    Text = PyFina(feeds["Text"],dir,start_ts,step,nbpts)
-    TziqNord = PyFina(feeds["TziqNord"],dir,start_ts,step,nbpts)
-    TtecNord = PyFina(feeds["TtecNord"],dir,start_ts,step,nbpts)
-    TdepNord = PyFina(feeds["TdepNord"],dir,start_ts,step,nbpts)
-    TretNord = PyFina(feeds["TretNord"],dir,start_ts,step,nbpts)
+def check_starting_nan(feed_name, feed):
+    """check if feed is starting by nan"""
+    print(f"{feed_name} : {feed.nb_nan} nan in the feed")
+    if feed.starting_by_nan:
+        message = f"first non nan value {feed.first_non_nan_value}"
+        message = f"{message} at index {feed.first_non_nan_index}"
+        print(message)
 
+def generate_episode(start_ts, nbpts):
+    """visualise un épisode commencant à start_ts"""
+    temp_ext = PyFina(feeds["Text"], DATA_DIR, start_ts, STEP, nbpts)
+    temp_ziq_nord = PyFina(feeds["TziqNord"], DATA_DIR, start_ts, STEP, nbpts)
+    temp_tec_nord = PyFina(feeds["TtecNord"], DATA_DIR, start_ts, STEP, nbpts)
+    temp_dep_nord = PyFina(feeds["TdepNord"], DATA_DIR, start_ts, STEP, nbpts)
+    temp_ret_nord = PyFina(feeds["TretNord"], DATA_DIR, start_ts, STEP, nbpts)
+    feed_objects = {
+        "Text": temp_ext,
+        "TziqNord": temp_ziq_nord,
+        "TtecNord": temp_tec_nord,
+        "TdepNord": temp_dep_nord,
+        "TretNord": temp_ret_nord
+    }
+    for feed_name, feed_object in feed_objects.items():
+        check_starting_nan(feed_name, feed_object)
     localstart = datetime.datetime.fromtimestamp(start_ts)
     utcstart = datetime.datetime.utcfromtimestamp(start_ts)
-    title = "starting on : UTC {}\n{} {}".format(utcstart,time.tzname[0],localstart)
-
+    title = f"starting on : UTC {utcstart}\n{time.tzname[0]} {localstart}"
+    figure = plt.figure(figsize = (20, 10))
+    matplotlib.rc('font', size=8)
     ax1 = plt.subplot(211)
     plt.title(title)
     plt.ylabel("outdoor Temp °C")
     plt.xlabel("time in hours")
-    plt.plot(Text, label="Text")
+    plt.plot(temp_ext, label="Text")
     plt.legend(loc='upper left')
     ax1 = ax1.twinx()
     plt.ylabel("indoor Temp °C")
-    plt.plot(TziqNord, label = "TziqNord", color="green")
-    plt.plot(TtecNord, label = "TtecNord",  color="orange")
+    plt.plot(temp_ziq_nord, label = "TziqNord", color="green")
+    plt.plot(temp_tec_nord, label = "TtecNord",  color="orange")
     plt.legend(loc='upper right')
-    ax3 = plt.subplot(212, sharex=ax1)
+    plt.subplot(212, sharex=ax1)
     plt.ylabel("hot water Temp °C")
-    plt.plot(TdepNord, label = "TdepNord")
-    plt.plot(TretNord, label = "TretNord")
+    plt.plot(temp_dep_nord, label = "TdepNord")
+    plt.plot(temp_ret_nord, label = "TretNord")
     plt.legend(loc='upper right')
-    plt.show()
-
-import signal
-
-class Loop:
-    """
-    visualisation des épisodes
-    """
-    def __init__(self):
-        self._exit = False
-
-    def run(self):
-        """
-        boucle
-        """
-        signal.signal(signal.SIGINT, self._sigint_handler)
-        signal.signal(signal.SIGTERM, self._sigint_handler)
-        while not self._exit:
-            start_ts = random.randrange(start, end - epL)
-            viewEpisode(start_ts)
+    figure.savefig(f"bloch_{start_ts}.png")
 
-    def _sigint_handler(self, signal, frame):
-        """
-        Réception du signal de fermeture
-        """
-        print("signal de fermeture reçu")
-        self._exit = True
 
-    def close(self):
-        print("fermeture effectuée")
+if __name__ == "__main__":
+    common_start, common_end, common_nbpts = analyse()
 
-if end - epL > start :
-    boucle = Loop()
-    boucle.run()
-    boucle.close()
-else :
-    viewEpisode(start)
+    if (common_end - EPISODE_LENGTH) > common_start :
+        for _ in range(10):
+            start_ts_ep = random.randrange(common_start, common_end - EPISODE_LENGTH)
+            print(start_ts_ep)
+            generate_episode(start_ts_ep, common_nbpts)
+    else :
+        print(common_start)
+    generate_episode(common_start, common_nbpts)

tests/test.py

@@ -1,28 +1,26 @@
 """test : open feed and plot image."""
 
 import datetime
+import time
+
 import matplotlib
 import matplotlib.pylab as plt
+from multigraph import check_starting_nan
 from PyFina import getMeta, PyFina
-import time
-import urllib.request as request
 
-feed_nb = 1
-dir = "./datas"
-meta = getMeta(feed_nb, dir)
+FEED_NB = 1
+DATA_DIR = "./datas"
+meta = getMeta(FEED_NB, DATA_DIR)
 print(meta)
-step = 3600
+STEP = 3600
 start = meta["start_time"]
-window = 8 * 24 * 3600
 length = meta["npoints"] * meta["interval"]
-if window > length:
-    window = length
-nbpts = window // step
-Text = PyFina(feed_nb, dir, start, step, nbpts)
+WINDOW = min(8 * 24 * 3600, length)
+
+nbpts = WINDOW // STEP
+temp_ext = PyFina(FEED_NB, DATA_DIR, start, STEP, nbpts)
 
-if Text.starting_by_nan:
-    print(f"first non nan value {Text.first_non_nan_value}")
-    print(f"at index {Text.first_non_nan_index}")        
+check_starting_nan("température extérieure", temp_ext)
 
 localstart = datetime.datetime.fromtimestamp(start)
 utcstart = datetime.datetime.utcfromtimestamp(start)
@@ -33,5 +31,5 @@
 plt.title(title)
 plt.ylabel("outdoor Temp °C")
 plt.xlabel("time in hours")
-plt.plot(Text)
-figure.savefig(f"feed_{feed_nb}.png")
+plt.plot(temp_ext)
+figure.savefig(f"feed_{FEED_NB}.png")