implement

cpdctrl/led_script.py

@@ -5,10 +5,38 @@ Created on Fri Jan 24 16:46:06 2025
 @author: CPD
 """
 import time
+import re
+import numpy as np
 
 class LedScript:
-    def __init__(self):
-        self.t_start = 0
+    def __init__(self, script:np.ndarray|str|int=0):
+        """
+        
+
+        Parameters
+        ----------
+        script : np.ndarray|str|int
+            If np.ndarray: numpy array in this form:
+                [(duration, cumulative time, state), ...]
+                Where <duration> is the duration of <state>, and <cumulative duration>
+                is the sum of all previous durations, including the current one.
+            If str: path to a led script file
+            If int: constant led state value
+        constantValue : TYPE, optional
+            DESCRIPTION. The default is None.
+
+        Returns
+        -------
+        None.
+
+        """
+        self.t_start = 0   
+        if type(script) == int:  
+            self.script = np.array([(0., 0., script)])
+        elif type(script) == np.ndarray:
+            self.script = script
+        elif type(script) == str:
+            self.script = LedScript.parse_script(script, ignore_errors=False)
         
     def start(self) -> int:
         """
@@ -57,7 +85,111 @@ class LedScript:
         int
             LED Intensity [0,100]
         """
-        # TODO remove hard coded script
-        if (dt // 3600) % 2 == 0:
-            return 100
-        return 0
+        if self.script.shape[0] == 1:
+            return self.script[0, 2]
+        distance = self.script[:,1] - dt
+        idx = np.where(distance >= 0, distance, np.inf).argmin()
+        return int(self.script[idx, 2])
+
+    
+    @staticmethod
+    def parse_script(filepath, ignore_errors=False):
+        with open(filepath, "r") as file:
+            lines = file.readlines()
+        
+        # for just checking scripts, with ignore_errors=True this function returns all errors in an array
+        errors = []
+        if ignore_errors:
+           def raiseError(*args, **kwargs):
+               e = InvalidScript(*args, **kwargs)
+               errors.append(e)
+               print(e)
+        else:   
+            def raiseError(*args, **kwargs):
+                raise InvalidScript(*args, **kwargs)
+                
+        # parse into <cumulative t in seconds, t state in seconds, state>
+        states = []
+        timesuffixes = {"h": 3600, "m": 60, "s": 1}
+        for i, l in enumerate(lines):
+            # 0) Check if empty or comment
+            l = l.strip("\n").strip(" ")
+            if l.startswith("#") or len(l) == 0: continue
+            # 1) Separate statements
+            matches = [ match for match in re.finditer(r"[^ \t\n\r]+", l) ]
+            if len(matches) < 2:
+                raiseError(i+1, f"Line has only one statement: '{l}'.", fix="The line needs to have the duration and led state separated by a space.")
+                continue
+            # 2) parse the duration
+            s_duration = matches[0].group()
+            duration_match = re.fullmatch(r"(\d+[hms]?)+", s_duration)
+            if not duration_match:
+                raiseError(i+1, f"Invalid duration: '{s_duration}'.", fix="Duration needs to consist of one ore more <digits><h/m/s> pairs, without spaces.")
+                continue
+            duration = 0.0
+            for m in re.findall(r"\d+[hms]?", s_duration): 
+                if m[-1] in timesuffixes:
+                    try:
+                        t = int(m[:-1])
+                    except:
+                        raiseError(i+1, f"Invalid number in duration '{m[-1]}': '{m[:-1]}'")
+                        continue
+                    t *= timesuffixes[m[-1]]
+                else:
+                    try:
+                        t = int(m)
+                    except:
+                        raiseError(i+1, f"Invalid number in duration: '{m}'")
+                        continue
+                duration += t          
+            # 3) parse the state
+            s_state = matches[1].group()
+            state_match = re.fullmatch(r"(?:(\d+%?)|(on)|(off))(#.*)?", s_state, re.IGNORECASE)
+            if not state_match: 
+                raiseError(i+1, f"Not a valid LED state: '{s_state}'", fix="LED state needs to be either 'on', 'off', an integer between [0,100], optionally ending with '%'")
+                continue
+            state_percent, state_on, state_off, ends_with_comment = state_match.groups()
+            if state_percent:
+                state_percent = state_percent.strip("%")
+                try:
+                    state = int(state_percent)
+                except:
+                    raiseError(i+1, f"Invalid number in led state: '{m}'")
+                    continue
+            elif state_on:
+                state = 100
+            elif state_off:
+                state = 0
+            else:
+                raiseError(i+1, f"Not a valid LED state: '{s_state}'")
+                continue
+            if state > 100:
+                raiseError(i+1, f"State is larger than 100%: '{state}'")
+                continue
+            elif state < 0:
+                raiseError(i+1, f"State is smaller than 0%: '{state}'")
+            # 4) Check for missing comment symbol # AFTER parsing duration and state, since the source of an error might be an invalid duration or state
+            if not ends_with_comment and len(matches) >= 3:
+                rest = matches[2].group()
+                if not rest.startswith("#"):
+                    raiseError(i+1, f"Garbage after statement: '{rest}'.", fix="If this was meant to be a comment, use '#'.")
+                    continue
+            # 5) Calculate cumulative duration
+            if len(states) == 0:
+                cum_duration = duration
+            else:
+                cum_duration = states[-1][1] + duration
+            # 6) append
+            states.append((duration, cum_duration, state))
+        states = np.array(states)
+        if ignore_errors:
+            return states, errors
+        return states
+        
+class InvalidScript(Exception):
+    def __init__(self, lineNr, message, fix=""):
+        self.lineNr = lineNr
+        self.message = message
+        self.fix = fix
+        self.full_message = f"Line {lineNr}: {message} {fix}"
+        super().__init__(self.full_message)