fixing #138 making velocity processor more serializable

pyorc/api/frames.py

@@ -74,8 +74,10 @@ def get_piv(self, **kwargs):
             search_area_size=kwargs["search_area_size"],
             overlap=kwargs["overlap"]
         )
+        cols_vector = cols[0].astype(np.int64)
+        rows_vector = rows[:, 0].astype(np.int64)
         # retrieve the x and y-axis belonging to the results
-        x, y = helpers.get_axes(cols, rows, self.camera_config.resolution)
+        x, y = helpers.get_axes(cols_vector, rows_vector, frames1.x.values, frames1.y.values)
         # convert in projected and latlon coordinates
         xs, ys = helpers.get_xs_ys(
             cols,

pyorc/api/video.py

@@ -43,6 +43,7 @@ def __init__(
             end_frame: Optional[int] = None,
             freq: Optional[int] = 1,
             stabilize: Optional[List[List]] = None,
+            lazy: bool = True,
             rotation: Optional[int] = None,
     ):
         """
@@ -69,6 +70,10 @@ def __init__(
             set of coordinates, that together encapsulate the polygon that defines the mask, separating land from water.
             The mask is used to select region (on land) for rigid point search for stabilization. If not set, then no
             stabilization will be performed
+        lazy : bool, optional
+            If set, frames are read lazily. This slows down the processing, but makes interaction with large videos
+            easier and consuming less memory. For operational processing with short videos, it is recommended to set
+            this explicitly to False.
         rotation : int, optional
             can be 0, 90, 180, 270. If provided, images will be forced to rotate along the provided angle.
         """
@@ -79,6 +84,7 @@ def __init__(
         self.feats_errs = None
         self.ms = None
         self.mask = None
+        self.lazy = lazy
         self.stabilize = stabilize
         if camera_config is not None:
             self.camera_config = camera_config
@@ -117,8 +123,18 @@ def __init__(
             end_frame = np.minimum(end_frame, self.frame_count)
         else:
             end_frame = self.frame_count
-        # extract times and frame numbers as far as available
-        time, frame_number = cv.get_time_frames(cap, start_frame, end_frame)
+        self.fps = cap.get(cv2.CAP_PROP_FPS)
+        self.rotation = cap.get(cv2.CAP_PROP_ORIENTATION_META)
+        # extract times, frame numbers and frames as far as available
+        time, frame_number, frames = cv.get_time_frames(
+            cap,
+            start_frame,
+            end_frame,
+            lazy=lazy,
+            rotation=self.rotation,
+            method="grayscale",
+        )
+        self.frames = frames
         # check if end_frame changed
         if frame_number[-1] != end_frame:
             warnings.warn(f"End frame {end_frame} cannot be read from file. End frame is adapted to {frame_number[-1]}")
@@ -131,7 +147,6 @@ def __init__(
         self.start_frame = start_frame
         if self.stabilize is not None:
             # select the right recipe dependent on the movie being fixed or moving
-            # recipe = const.CLASSIFY_CAM[self.stabilize] if self.stabilize in const.CLASSIFY_CAM else []
             self.get_ms(cap)
 
         self.fps = cap.get(cv2.CAP_PROP_FPS)
@@ -150,6 +165,22 @@ def __init__(
         cap.release()
         del cap
 
+    @property
+    def lazy(self):
+        """
+
+        Returns
+        -------
+        np.ndarray
+            Mask of region of interest
+        """
+        return self._lazy
+
+    @lazy.setter
+    def lazy(self, lazy):
+        self._lazy = lazy
+
+
     @property
     def mask(self):
         """
@@ -174,7 +205,10 @@ def camera_config(self):
 
         :return: CameraConfig object
         """
-        return self._camera_config
+        if hasattr(self, "_camera_config"):
+            return self._camera_config
+        else:
+            return None
 
     @camera_config.setter
     def camera_config(self, camera_config_input):
@@ -284,6 +318,18 @@ def start_frame(
         else:
             self._start_frame = start_frame
 
+
+    @property
+    def frames(self):
+        return self._frames
+
+    @frames.setter
+    def frames(
+            self,
+            frames: Optional[List] = None
+    ):
+        self._frames = frames
+
     @property
     def fps(self):
         """
@@ -341,19 +387,16 @@ def get_frame(
             self,
             n: int,
             method: Optional[str] = "grayscale",
-            lens_corr: Optional[bool] = False
     ) -> np.ndarray:
         """
-        Retrieve one frame. Frame will be corrected for lens distortion if lens parameters are given.
+        Retrieve one frame.
 
         Parameters:
         -----------
         n : int
             frame number to retrieve
         method : str
             can be "rgb", "grayscale", or "hsv", default: "grayscale"
-        lens_corr: bool, optional
-            if set to True, lens parameters will be used to undistort image
 
         Returns
         -------
@@ -367,27 +410,12 @@ def get_frame(
                            "hsv"]), f'method must be "grayscale", "rgb" or "hsv", method is "{method}"'
         cap = cv2.VideoCapture(self.fn)
         cap.set(cv2.CAP_PROP_POS_FRAMES, n + self.start_frame)
-        try:
-            ret, img = cap.read()
-            if self.rotation is not None:
-                img = cv2.rotate(img, self.rotation)
-        except:
-            raise IOError(f"Cannot read")
-        if ret:
-            if self.ms is not None:
-                img = cv.transform(img, self.ms[n])
-            # apply lens distortion correction
-            # if hasattr(self, "camera_config"):
-            #     img = cv.undistort_img(img, self.camera_config.camera_matrix, self.camera_config.dist_coeffs)
-            if method == "grayscale":
-                # apply gray scaling, contrast- and gamma correction
-                # img = _corr_color(img, alpha=None, beta=None, gamma=0.4)
-                img = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)  # mean(axis=2)
-            elif method == "rgb":
-                # turn bgr to rgb for plotting purposes
-                img = cv2.cvtColor(img, cv2.COLOR_BGR2RGB)
-            elif method == "hsv":
-                img = cv2.cvtColor(img, cv2.COLOR_BGR2HSV)
+        ret, img = cv.get_frame(
+            cap,
+            rotation=self.rotation,
+            ms=self.ms[n] if self.ms else None,
+            method=method
+        )
         self.frame_count = n + 1
         cap.set(cv2.CAP_PROP_POS_FRAMES, 0)
         cap.release()
@@ -416,15 +444,23 @@ def get_frames(
                         "_camera_config")), "No camera configuration is set, add it to the video using the .camera_config method"
         # camera_config may be altered for the frames object, so copy below
         camera_config = copy.deepcopy(self.camera_config)
-        get_frame = dask.delayed(self.get_frame, pure=True)  # Lazy version of get_frame
-        # get all listed frames
-        frames = [get_frame(n=n, **kwargs) for n, f_number in enumerate(self.frame_number)]
-        sample = frames[0].compute()
-        data_array = [da.from_delayed(
-            frame,
-            dtype=sample.dtype,
-            shape=sample.shape
-        ) for frame in frames]
+
+        if self.frames is None or len(kwargs) > 0:
+            # a specific method for collecting frames is requested or lazy access is requested.
+            get_frame = dask.delayed(self.get_frame, pure=True)  # Lazy version of get_frame
+            # get all listed frames
+            frames = [get_frame(n=n, **kwargs) for n, f_number in enumerate(self.frame_number)]
+            sample = frames[0].compute()
+            data_array = [da.from_delayed(
+                frame,
+                dtype=sample.dtype,
+                shape=sample.shape
+            ) for frame in frames]
+            da_stack = da.stack(data_array, axis=0)
+        else:
+            sample = self.frames[0]
+            da_stack = self.frames
+
         # undistort source control points
         # if hasattr(camera_config, "gcps"):
         #     camera_config.gcps["src"] = cv.undistort_points(
@@ -435,8 +471,10 @@ def get_frames(
         time = np.array(
             self.time) * 0.001  # measure in seconds to comply with CF conventions # np.arange(len(data_array))*1/self.fps
         # y needs to be flipped up down to match the order of rows followed by coordinate systems (bottom to top)
-        y = np.flipud(np.arange(data_array[0].shape[0]))
-        x = np.arange(data_array[0].shape[1])
+        # y = np.flipud(np.arange(data_array[0].shape[0]))
+        # x = np.arange(data_array[0].shape[1])
+        y = np.flipud(np.arange(sample.shape[0]))
+        x = np.arange(sample.shape[1])
         # perspective column and row coordinate grids
         xp, yp = np.meshgrid(x, y)
         coords = {
@@ -454,11 +492,12 @@ def get_frames(
             "h_a": json.dumps(self.h_a)
         }
         frames = xr.DataArray(
-            da.stack(data_array, axis=0),
+            da_stack,
             dims=dims,
             coords=coords,
-            attrs=attrs
+            attrs=attrs,
         )[::self.freq]
+        frames = frames.chunk({"time": 1})  # set chunks over time dimension
         del coords["time"]
         if len(sample.shape) == 3:
             del coords["rgb"]

pyorc/cli/cli_utils.py

@@ -156,6 +156,7 @@ def validate_file(ctx, param, value):
             raise click.FileError(f"{value}")
         return value
 
+
 def validate_dir(ctx, param, value):
     if not(os.path.isdir(value)):
         os.makedirs(value)
@@ -167,7 +168,6 @@ def validate_rotation(ctx, param, value):
             raise click.UsageError(f"Rotation value must be either 90, 180 or 270")
         return value
 
-
 def parse_camconfig(ctx, param, camconfig_file):
     """
     Read and validate cam config file
@@ -189,6 +189,7 @@ def parse_camconfig(ctx, param, camconfig_file):
     # return dict formatted
     return camconfig.to_dict_str()
 
+
 def parse_recipe(ctx, param, recipe_file):
     """
     Read and validate entire recipe from top to bottom, add compulsory classes where needed
@@ -238,8 +239,17 @@ def parse_str_num(ctx, param, value):
             return float(value)
 
 
-def read_shape(fn):
-    gdf = gpd.read_file(fn)
+def read_shape(fn=None, geojson=None):
+    if fn is None and geojson is None:
+        raise click.UsageError(f"Either fn or geojson must be provided")
+    if geojson:
+        if "crs" in geojson:
+            crs = geojson["crs"]["properties"]["name"]
+        else:
+            crs = None
+        gdf = gpd.GeoDataFrame().from_features(geojson, crs=crs)
+    else:
+        gdf = gpd.read_file(fn)
     # check if all geometries are points
     assert(all([isinstance(geom, Point) for geom in gdf.geometry])), f'shapefile may only contain geometries of type ' \
                                                                      f'"Point"'
@@ -249,10 +259,14 @@ def read_shape(fn):
     else:
         coords = [[p.x, p.y] for p in gdf.geometry]
     if not(hasattr(gdf, "crs")):
-        raise click.FileError(f"{fn} does not contain CRS, use a GIS program to add a valid CRS.")
-    if gdf.crs is None:
-        raise click.FileError(f"{fn} does not contain CRS, use a GIS program to add a valid CRS.")
-    return coords, gdf.crs.to_wkt()
+        click.echo(f"shapefile or geojson does not contain CRS, assuming CRS is the same as camera config CRS")
+        crs = None
+    elif gdf.crs is None:
+        click.echo(f"shapefile or geojson does not contain CRS, assuming CRS is the same as camera config CRS")
+        crs = None
+    else:
+        crs = gdf.crs.to_wkt()
+    return coords, crs
 
 def validate_dst(value):
     if value is not None:

pyorc/cli/main.py

@@ -360,12 +360,11 @@ def velocimetry(
     )
     logger.info(f"Preparing your velocimetry result in {output}")
     # load in recipe and camera config
-    if h_a is not None:
-        recipe["video"]["h_a"] = h_a
     pyorc.service.velocity_flow(
         recipe=recipe,
         videofile=videofile,
         cameraconfig=cameraconfig,
+        h_a=h_a,
         prefix=prefix,
         output=output,
         update=update,