Merge pull request #10 from FROM-THE-EARTH/develop

Develop

docs/flightcam.md

@@ -69,10 +69,11 @@ Log files are useful to analysis state of the application, especially during deb
 Below is an example of the log output.
 
 ```log
-[2022-08-01 17:43:46,239] [DEBUG] Start the Flight camera mode. Setting; glm_name: main, timeout: 210.0, pin_flight: 22, pin_led: 12, file_mov: mov-220801-174346.h264, file_log: mov-220801-174346.log, parent_dir: None, resolution: (1920, 1080), framerate: 30, interval: 0.1, led_blink_freq: 2.0
-[2022-08-01 17:43:46,415] [DEBUG] Waiting a flight pin to be connected...
-[2022-08-01 17:43:50,004] [DEBUG] A flight pin was connected.
-[2022-08-01 17:43:50,005] [DEBUG] Wating the flight pin to be disconnected...
-[2022-08-01 17:43:58,528] [DEBUG] The flight pin was disconnected.
-[2022-08-01 17:43:58,528] [DEBUG] Start recording.
+[2022-08-02 03:59:53,126] [INFO] Start the Flight camera mode. Setting; glm_name: main, timeout: 210.0, pin_flight: 22, pin_led: 12, file_mov: mov-220802-035953.h264, file_log: mov-220802-035953.log, parent_dir: None, resolution: (1920, 1080), framerate: 30, interval: 0.1, led_blink_freq: 2.0, log_level: 20, check_waiting_time: False
+[2022-08-02 03:59:53,218] [INFO] Waiting a flight pin to be connected...
+[2022-08-02 04:00:01,074] [INFO] Detected that the flight pin was connected.
+[2022-08-02 04:00:01,075] [INFO] Wating the flight pin to be disconnected...
+[2022-08-02 04:00:03,034] [INFO] Detected that the flight pin was disconnected.
+[2022-08-02 04:00:03,034] [INFO] Start recording.
+[2022-08-02 04:03:33,151] [INFO] Stop recording.
 ```

docs/setting.md

@@ -49,12 +49,20 @@ Wating time for recording, `0.1`.
 
 Frequency of LED blinking, defaults `2.`.
 
+### `log_level`
+
+Log level used in the application, defaults `logging.INFO` (`20`).
+
+### `check_waiting_time`
+
+If making log outputs during waiting time for disconnection of the flight pin or not, defaults `False`.
+
 ## Examples
 
 ```python
+import logging
 from obcam import OBCamGileum
 
-
 glm = OBCamGileum(
     # Make the timeout shorter for tests.
     timeout=10,
@@ -71,6 +79,12 @@ glm = OBCamGileum(
     # Make the frequency smaller.
     led_blink_freq=1.,
 
+    # Make verbose log outputs for tests.
+    log_level=logging.DEBUG,
+
+    # Observe waiting time until the flight pin is disconnected.
+    check_waiting_time=True,
+
     # Use the default settings.
     # To use defualt values, you don't have to give corresponding arguments.
     pin_flight=22,

glm.py

@@ -1,3 +1,5 @@
+import logging
+
 from obcam import OBCamGileum
 
 
@@ -31,4 +33,11 @@
 
     # Frequency of LED blinking.
     led_blink_freq=2.,
+
+    # Log level used in the application.
+    log_level=logging.INFO,
+
+    # If making log outputs during waiting time for disconnection
+    # of the flight pin or not
+    check_waiting_time=False,
 )

obcam/__init__.py

@@ -1 +1,2 @@
+from .cli import run_flight_camera
 from .glm import OBCamGileum

obcam/cli.py

@@ -16,15 +16,7 @@
 LOG_PLACEHOLDER = "[%(asctime)s] [%(levelname)s] %(message)s"
 
 
-@click.command()
-@click.argument(
-    "glm_file",
-    type=click.Path(exists=True),
-)
-def main(glm_file: str) -> None:
-    gileum.load_glms_at(glm_file)
-    glm = gileum.get_glm(OBCamGileum)
-
+def run_flight_camera(glm: OBCamGileum) -> None:
     # Setup file paths.
     if glm.file_mov is None:
         glm.file_mov = get_timestamp("mov", "h264")
@@ -40,31 +32,46 @@ def main(glm_file: str) -> None:
 
     # Setup the logger.
     handler = logging.FileHandler(glm.file_log)
-    handler.setLevel(logging.DEBUG)
+    handler.setLevel(glm.log_level)
     handler.setFormatter(logging.Formatter(LOG_PLACEHOLDER))
     logger = logging.getLogger()
-    logger.setLevel(logging.DEBUG)
+    logger.setLevel(glm.log_level)
     logger.addHandler(handler)
 
-    logger.debug(
+    logger.info(
         "Start the Flight camera mode. "
         f"Setting; {', '.join([f'{k}: {v}' for k, v in glm.dict().items()])}"
     )
 
+    recorder = IORecorder(
+        glm.pin_flight,
+        glm.pin_led,
+        logger,
+        file_mov=glm.file_mov,
+        resolution=glm.resolution,
+        framerate=glm.framerate,
+        led_blink_freq=glm.led_blink_freq,
+    )
     try:
-        with IORecorder(
-            glm.pin_flight,
-            glm.pin_led,
-            logger,
-            file_mov=glm.file_mov,
-            resolution=glm.resolution,
-            framerate=glm.framerate,
-            led_blink_freq=glm.led_blink_freq,
-        ) as rec:
-            rec.start_record(
-                glm.timeout,
-                interval=glm.interval,
-            )
+        recorder.record(
+            glm.timeout,
+            interval=glm.interval,
+            check_waiting_time=glm.check_waiting_time,
+        )
     except Exception as e:
         logger.exception("Finish with an exception.", exc_info=e)
         raise e
+    finally:
+        recorder.cleanup()
+
+
+@click.command()
+@click.argument(
+    "glm_file",
+    type=click.Path(exists=True),
+)
+def main(glm_file: str) -> None:
+    gileum.load_glms_at(glm_file)
+    glm = gileum.get_glm(OBCamGileum)
+
+    run_flight_camera(glm)

obcam/glm.py

@@ -1,3 +1,4 @@
+import logging
 import typing as t
 
 import gileum
@@ -47,3 +48,10 @@ class OBCamGileum(gileum.BaseGileum):
 
     led_blink_freq: float = 2.
     """Frequency of LED blinking."""
+
+    log_level: int = logging.INFO
+    """Log level used in the application."""
+
+    check_waiting_time: bool = False
+    """If making log outputs during waiting time for disconnection
+    of the flight pin or not."""

obcam/recorder.py

@@ -29,7 +29,7 @@ def verify_video_format(path: str) -> bool:
     return ext[1:] in _VIDEO_FORMATS
 
 
-RISING_TIME_THRESHOLD = 0.5     # seconds
+FALLING_TIME_THRESHOLD = 0.5     # seconds
 
 
 class IORecorder:
@@ -70,6 +70,9 @@ def __init__(
         self._format = os.path.splitext(file_mov)[1][1:]
         self._thread: t.Optional[threading.Thread] = None
 
+        self._lock_in_flight = threading.RLock()
+        self._in_flight = False
+
         gpio.setmode(gpio.BCM)
         gpio.setup(self._pin_flight, gpio.IN, pull_up_down=gpio.PUD_DOWN)
         gpio.setup(self._pin_led, gpio.OUT)
@@ -84,23 +87,43 @@ def in_flight(self) -> bool:
 
     def blink_led(self) -> None:
         self._pwm_led.ChangeDutyCycle(50)
+        self._logger.debug("LED started to be blinking.")
 
     def turn_on_led(self) -> None:
         self._pwm_led.ChangeDutyCycle(100)
+        self._logger.debug("Turned the LED on.")
 
     def turn_off_led(self) -> None:
         self._pwm_led.ChangeDutyCycle(0)
+        self._logger.debug("Turned the LED off.")
+
+    def _log_waiting_time(self, interval: float = 1.) -> None:
+        time_init = time.time()
+        while True:
+            self._lock_in_flight.acquire()
+            if self._in_flight:
+                break
+            self._lock_in_flight.release()
+
+            self._logger.debug(
+                "Waiting the flight pin to be disconnected, "
+                f"{round(time.time() - time_init, 3)} elapsing."
+            )
+            time.sleep(interval)
 
     def record(
         self,
         timeout: float,
         interval: float = 0.1,
+        check_waiting_time: bool = False,
     ) -> None:
         """Record a video while obseving state of the body.
 
         Args:
             timeout: Length of recording time.
             interval: Wating time for recording.
+            check_waiting_time: If making log outputs during waiting time for
+                disconnection of the flight pin or not.
 
         Notes:
             `timeout` must be positive.
@@ -114,60 +137,95 @@ def record(
             raise ValueError("timeout must be positive.")
 
         # Wait until the flight pin is to be connected.
-        self._logger.debug("Waiting a flight pin to be connected...")
+        self._logger.info("Waiting a flight pin to be connected...")
         self.blink_led()
 
             # Noise measures
         is_flightpin_connected = False
         while True:
             gpio.wait_for_edge(self._pin_flight, gpio.FALLING)
-            time_init = time.time()
+            self._logger.debug(
+                "Falling edge of the flight pin level was detected. "
+                f"Waiting {FALLING_TIME_THRESHOLD} seconds to verify "
+                "the flight pin was pulled out exactly."
+            )
 
+            time_init = time.time()
             while True:
                 if self.in_flight:
+                    self._logger.debug(
+                        "Level of the flight pin is high. It means that "
+                        "the flight pin is disconnected before time of "
+                        "waiting threshold elapsed. Thus, detecting "
+                        "the flight pin is to be continued."
+                    )
                     break
                 else:
-                    if time.time() - time_init > RISING_TIME_THRESHOLD:
+                    if time.time() - time_init > FALLING_TIME_THRESHOLD:
                         is_flightpin_connected = True
+                        self._logger.debug(
+                            "Time of waiting threshold elapsed while level of "
+                            "the flight pin is stable."
+                        )
                         break
 
             if is_flightpin_connected:
                 break
 
         self.turn_off_led()
-        self._logger.debug("A flight pin was connected.")
+        self._logger.info("Detected that the flight pin was connected.")
 
         # Wait until the level of the flight pin becomes low.
-        self._logger.debug("Wating the flight pin to be disconnected...")
+        self._logger.info("Wating the flight pin to be disconnected...")
+        if check_waiting_time:
+            th = threading.Thread(target=self._log_waiting_time)
+            th.start()
+            # This thread automatically exits after the flight pin
+            # is disconnected.
+
         gpio.wait_for_edge(self._pin_flight, gpio.RISING)
+        self._lock_in_flight.acquire()
+        self._in_flight = True
+        self._lock_in_flight.release()
+
+        self._logger.debug(
+            "Rising edge of the flight pin level was detected."
+        )
+        self._logger.info("Detected that the flight pin was disconnected.")
         self.turn_on_led()
-        self._logger.debug("The flight pin was disconnected.")
 
-        # Start recording
-        self._logger.debug("Start recording.")
+        # Recording
+        self._logger.info("Start recording.")
         self._camera.start_recording(file, format=self._format)
         time_init = time.time()
         try:
             while True:
-                if 0 < timeout <= (time.time() - time_init):
+                elapsed_time = time.time() - time_init
+                if 0 < timeout <= elapsed_time:
                     break
                 self._camera.wait_recording(timeout=interval)
                 file.flush()
+                self._logger.debug(
+                    "Recording in progress while writing into the file, "
+                    f"remaining {round(timeout - elapsed_time, 3)} seconds..."
+                )
+            self._logger.debug("Timeout is over.")
         finally:
             file.flush()
             self._camera.stop_recording()
-            self._logger.debug("Stop recording.")
+            self._logger.info("Stop recording.")
 
             self.turn_off_led()
 
     def start_record(
         self,
         timeout: float,
         interval: float = 1.,
+        check_waiting_time: bool = False,
     ) -> IORecorder:
         self._thread = threading.Thread(
             target=self.record,
-            args=(timeout, interval),
+            args=(timeout, interval, check_waiting_time),
         )
         self._thread.start()
         return self

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "obcam"
-version = "0.1.2"
+version = "0.1.3"
 description = "Flight camera module and program for the OB team rocket in NSE 2022."
 authors = ["Yunhyeon Jeong <[email protected]>"]