diff --git a/docs/local_conf.md b/docs/local_conf.md index bf2c1d3..f6e6344 100644 --- a/docs/local_conf.md +++ b/docs/local_conf.md @@ -30,6 +30,8 @@ With no other instruction, EthoPy loads: The file is read once, at `import ethopy`. If it does not exist, EthoPy runs entirely on its [built-in defaults](#built-in-defaults). +**Editing the file has no effect on a running EthoPy.** The configuration is read once, at import, and kept in memory for the lifetime of the process. Restart EthoPy for a change to take effect. + ### 2. A different config file on the command line Every EthoPy run can point at another file with `-c` / `--config`: diff --git a/src/ethopy/experiments/calibrate.py b/src/ethopy/experiments/calibrate.py index 94dac09..def7f60 100644 --- a/src/ethopy/experiments/calibrate.py +++ b/src/ethopy/experiments/calibrate.py @@ -1,12 +1,12 @@ import logging import time +import os from importlib import import_module import pygame try: import pygame_menu - IMPORT_PYGAME_MENU = True except ImportError: IMPORT_PYGAME_MENU = False @@ -15,9 +15,10 @@ class Experiment: - """_summary_ - I created a main menu where every time i want to move to new one a clean it - and i render the new components + """Calibration experiment with Pi 5 compatibility + + Main menu where every time we want to move to new one, clean it + and render the new components Menu order: 1. pressure menu: define the air pressure in PSI @@ -28,30 +29,38 @@ class Experiment: """ def __init__(self): - # self.interface = None self.session_params = None self.logger = None self.sync = False self.cal_idx = 0 self.msg = "" self.pulse = 0 + + # Screen dimensions - will be auto-detected self.screen_width = 800 self.screen_height = 480 + self.ports = None self.port = None + self.interface = None + self.screen = None + self.menu = None + self.theme = None + + # Pi 5 compatibility flags + self.is_fullscreen = False + self.display_scale = 1.0 + if not globals()["IMPORT_PYGAME_MENU"]: raise ImportError( "You need to install the pygame-menu: pip install pygame-menu" ) def setup(self, logger, params): - """setup _summary_ - - _extended_summary_ - """ self.session_params = params self.logger = logger + # Get interface configuration interface_module = self.logger.get( schema="interface", table="SetupConfiguration", @@ -63,25 +72,90 @@ def setup(self, logger, params): ) self.setup_conf_idx = self.session_params["setup_conf_idx"] - self.interface = interface(exp=self, callbacks=False) + # Initialize interface (this will use our Pi 5 compatible RPPorts) + try: + self.interface = interface(exp=self, callbacks=False) + log.info("Interface initialized successfully") + except Exception as e: + log.error(f"Failed to initialize interface: {e}") + raise + + # Initialize pygame with Pi 5 compatibility + self._init_pygame() + + # Setup pygame menu theme + self._setup_theme() + + # Create main menu + self._create_main_menu() + + # Initialize calibration variables + self.pressure = None + self.curr = "" + self.stop = False + + # Start with the pressure menu + self.create_pressure_menu() + + # Run the experiment + self.run() - pygame.init() - self.screen = pygame.display.set_mode((800, 480)) - if self.logger.is_pi: - self.screen = pygame.display.set_mode( - (self.screen_width, self.screen_height), pygame.FULLSCREEN - ) + def _init_pygame(self): + """Initialize pygame with Pi 5 compatibility""" + if not pygame.get_init(): + pygame.init() + + # Set display mode with Pi 5 compatibility + try: + if self.logger.is_pi: + # Ask SDL for the desktop resolution. Requesting a size that + # differs from it returns a surface that segfaults on draw, and + # /sys/class/graphics/fb0/virtual_size reports its dimensions + # in the opposite order on some Pis. + self.screen = pygame.display.set_mode( + (0, 0), + pygame.FULLSCREEN | pygame.DOUBLEBUF | pygame.HWSURFACE + ) + self.is_fullscreen = True + log.info("Fullscreen mode activated") + + # Hide mouse cursor for kiosk mode + pygame.mouse.set_visible(False) + + else: + # Windowed mode for development + self.screen = pygame.display.set_mode((self.screen_width, self.screen_height)) + log.info("Windowed mode activated") + + except pygame.error as e: + log.error(f"Failed to set display mode: {e}") + # Fallback to windowed mode + self.screen = pygame.display.set_mode((800, 480)) + log.info("Fallback to 800x480 windowed mode") + + # Every widget is sized from the surface pygame actually gave us. + self.screen_width, self.screen_height = self.screen.get_size() + self.display_scale = min(self.screen_width / 800, self.screen_height / 480) + log.info(f"Display surface size: {self.screen_width}x{self.screen_height}") + + pygame.display.set_caption("EthoPy Calibration") - # Configure self.theme + def _setup_theme(self): + """Setup pygame menu theme with scaling""" self.theme = pygame_menu.themes.THEME_DARK.copy() self.theme.background_color = (0, 0, 0) self.theme.title_background_color = (43, 43, 43) - self.theme.title_font_size = 35 + + # Scale font sizes based on display + self.theme.title_font_size = int(35 * self.display_scale) + self.theme.widget_font_size = int(30 * self.display_scale) + self.theme.widget_alignment = pygame_menu.locals.ALIGN_CENTER self.theme.widget_font_color = (255, 255, 255) - self.theme.widget_font_size = 30 self.theme.widget_padding = 0 + def _create_main_menu(self): + """Create the main menu with proper dimensions""" self.menu = pygame_menu.Menu( "", self.screen_width, @@ -93,62 +167,94 @@ def setup(self, logger, params): theme=self.theme, ) - self.pressure = None - self.curr = "" - self.stop = False - # Start with the pressure menu - - self.create_pressure_menu() - self.run() - def run(self) -> None: - """ - Calibration mainloop. - """ + """Calibration mainloop with improved error handling""" + clock = pygame.time.Clock() # Add FPS control + try: - while not self.stop: # Changed from self.stop == False for better style + while not self.stop: events = pygame.event.get() for event in events: if event.type == pygame.QUIT: self.stop = True break - - if self.menu.is_enabled() and not self.stop: # Added stop check - self.menu.update(events) + # Add escape key to exit fullscreen/application + elif event.type == pygame.KEYDOWN: + if event.key == pygame.K_ESCAPE: + self.stop = True + break + elif event.key == pygame.K_F11 and not self.logger.is_pi: + # Toggle fullscreen in development mode + self._toggle_fullscreen() + + if self.menu and self.menu.is_enabled() and not self.stop: try: + self.menu.update(events) + # Clear screen before drawing + self.screen.fill((0, 0, 0)) self.menu.draw(self.screen) - pygame.display.flip() - except pygame.error: - # Display was probably quit, exit gracefully + pygame.display.flip() # Use flip() for better performance + except pygame.error as e: + log.error(f"Display error: {e}") break + + # Limit FPS to reduce CPU usage + clock.tick(60) + + except KeyboardInterrupt: + log.info("Keyboard interrupt received") + except Exception as e: + log.error(f"Unexpected error in main loop: {e}") finally: self.cleanup() + def _toggle_fullscreen(self): + """Toggle fullscreen mode (development only)""" + try: + if self.is_fullscreen: + self.screen = pygame.display.set_mode((800, 480)) + self.is_fullscreen = False + pygame.mouse.set_visible(True) + else: + self.screen = pygame.display.set_mode((0, 0), pygame.FULLSCREEN) + self.is_fullscreen = True + pygame.mouse.set_visible(False) + except Exception as e: + log.error(f"Failed to toggle fullscreen: {e}") + def cleanup(self): - """Cleanup pygame and interface resources.""" + """Cleanup pygame and interface resources with better error handling""" + log.info("Starting cleanup...") + + # Cleanup pygame if pygame.get_init(): try: - # Clear any remaining events pygame.event.clear() - - if hasattr(self, "menu"): + + if hasattr(self, "menu") and self.menu: self.menu.disable() + pygame.mouse.set_visible(True) # Show cursor before exit # Not pygame.quit(): it frees fonts still cached by # pygame_menu, segfaulting the next menu built. pygame.display.quit() + log.info("Pygame cleaned up successfully") except Exception as e: log.warning(f"Error during pygame cleanup: {e}") - if hasattr(self, "interface"): + # Cleanup interface + if hasattr(self, "interface") and self.interface: try: self.interface.cleanup() + log.info("Interface cleaned up successfully") except Exception as e: log.warning(f"Error during interface cleanup: {e}") - if hasattr(self, "logger"): + # Update logger status + if hasattr(self, "logger") and self.logger: try: self.logger.update_setup_info({"status": "ready"}) + log.info("Logger status updated") except Exception as e: log.warning(f"Error updating logger status: {e}") @@ -167,8 +273,8 @@ def _clear_menu(self): float=True, padding=(5, 10, 5, 10), background_color=(153, 0, 0), - font_size=25, - ).translate(650, 350) + font_size=int(25 * self.display_scale), + ).translate(int(650 * self.display_scale), int(350 * self.display_scale)) def abort(self): """Stop the calibration immediately. @@ -180,8 +286,10 @@ def abort(self): try: self.menu.clear() self.menu.add.label( - "Calibration aborted!", float=True, font_size=30 - ).translate(20, 80) + "Calibration aborted!", + float=True, + font_size=int(30 * self.display_scale), + ).translate(int(20 * self.display_scale), int(80 * self.display_scale)) try: self.menu.draw(self.screen) pygame.display.flip() @@ -194,30 +302,33 @@ def abort(self): self.stop = True def exit(self): - """exit _summary_ - - exit function after the Experiment has finished - """ + """Exit function after the Experiment has finished""" try: - self.menu.clear() - self.menu.add.label("Done calibrating!!", float=True, font_size=30).translate(20, 80) - try: - self.menu.draw(self.screen) - pygame.display.flip() - time.sleep(2) - except pygame.error: - pass # Display might already be quit + if self.menu: + self.menu.clear() + # Scale exit message font + exit_font_size = int(30 * self.display_scale) + self.menu.add.label( + "Done calibrating!!", + float=True, + font_size=exit_font_size + ).translate(int(20 * self.display_scale), int(80 * self.display_scale)) + + try: + self.screen.fill((0, 0, 0)) + self.menu.draw(self.screen) + pygame.display.flip() + time.sleep(2) + except pygame.error: + pass # Display might already be quit self.stop = True except Exception as e: log.warning(f"Error during exit: {e}") self.stop = True - self.interface.cleanup() - self.logger.update_setup_info({"status": "ready"}) - time.sleep(1) def create_pressure_menu(self): - """The First menu in Calibration where is definde the air pressure in PSI""" + """The First menu in Calibration where air pressure in PSI is defined""" self._clear_menu() self.button_input("Enter air pressure (PSI)", self.create_pulsenum_menu) @@ -227,9 +338,17 @@ def create_pulsenum_menu(self): self.curr = "" if self.cal_idx < len(self.session_params["pulsenum"]): self._clear_menu() + + # Scale UI elements + label_font_size = int(30 * self.display_scale) + button_font_size = int(30 * self.display_scale) + self.menu.add.label( - "Place zero-weighted pad under the port", float=True, font_size=30 - ).translate(20, 80) + "Place zero-weighted pad under the port", + float=True, + font_size=label_font_size + ).translate(int(20 * self.display_scale), int(80 * self.display_scale)) + self.menu.add.button( "OK", self.create_pulse_num, @@ -237,30 +356,33 @@ def create_pulsenum_menu(self): float=True, padding=(10, 10, 10, 10), background_color=(0, 128, 0), - font_size=30, - ).translate(400, 140) + font_size=button_font_size, + ).translate(int(400 * self.display_scale), int(140 * self.display_scale)) else: self.exit() def create_pulse_num(self): - """ - Display the pulses - """ + """Display the pulses""" self.pulse = 0 msg = f"Pulse {self.pulse + 1}/{self.session_params['pulsenum'][self.cal_idx]}" self._clear_menu() + + # Scale pulse label + pulse_font_size = int(40 * self.display_scale) + pulses_label = self.menu.add.label( msg, float=True, label_id="pulses_label", - font_size=40, + font_size=pulse_font_size, background_color=(0, 15, 15), - ).translate(0, 50) - # Adds a function to the Widget to be executed each time the label is drawn. + ).translate(0, int(50 * self.display_scale)) + + # Adds a function to the Widget to be executed each time the label is drawn pulses_label.add_draw_callback(self.run_pulses) def run_pulses(self, widget, menu): - """This function is executed each time the label is drawm + """This function is executed each time the label is drawn Args: widget (_type_): The widget that uses the function @@ -281,28 +403,27 @@ def run_pulses(self, widget, menu): self.interface.give_liquid( port, self.session_params["duration"][self.cal_idx] ) - pass except Exception as error: - # ToDo update notes in control table - log.info(f"Calibration Error {error}") + log.error(f"Calibration Error: {error}") self.exit() + return time.sleep( self.session_params["duration"][self.cal_idx] / 1000 + self.session_params["pulse_interval"][self.cal_idx] / 1000 ) - self.pulse += 1 # update trial + self.pulse += 1 else: self.cal_idx += 1 self.ports = self.session_params["ports"].copy() self.create_port_weight() def create_port_weight(self): - """A menu with numpad for defining the water in every port""" + """A menu with numpad for defining the water weight in every port""" self._clear_menu() cal_idx = self.cal_idx - 1 + if self.session_params["save"]: - self._clear_menu() if len(self.ports) != 0: if len(self.ports) != len(self.session_params["ports"]): self.log_pulse_weight( @@ -315,7 +436,7 @@ def create_port_weight(self): self.port = self.ports.pop(0) self.button_input( - f"Enter weight for port {self.port }", self.create_port_weight + f"Enter weight for port {self.port}", self.create_port_weight ) else: self.log_pulse_weight( @@ -325,34 +446,18 @@ def create_port_weight(self): self.curr, self.pressure, ) - self.create_pulsenum_menu() else: self.create_pulsenum_menu() def button_input(self, message: str, _func): - """button_input _summary_ - - Create a label with a numpad - - Args: - message (str): a string to display in as label - _func (method): a method to run after the OK is pressed in the numpad - """ - self.menu.add.label( - message, - font_size=25, - ) + """Create a label with a numpad""" + label_font_size = int(25 * self.display_scale) + self.menu.add.label(message, font_size=label_font_size) self.num_pad(_func) def num_pad(self, _func): - """num_pad _summary_ - - _extended_summary_ - - Args: - log_function (_type_): _description_ - """ + """Create numpad with scaling for different screen sizes""" self.num_pad_disp = self.menu.add.label( "", background_color=None, @@ -360,13 +465,18 @@ def num_pad(self, _func): selectable=True, selection_effect=None, ) - self.menu.add.vertical_margin(10) + self.menu.add.vertical_margin(int(10 * self.display_scale)) cursor = pygame_menu.locals.CURSOR_HAND - self.curr = "" - # Add horizontal frames - f1 = self.menu.add.frame_h(299, 54, margin=(0, 0)) + # Scale button dimensions + frame_width = int(299 * self.display_scale) + frame_height = int(54 * self.display_scale) + button_width = int(74 * self.display_scale) + button_height = int(54 * self.display_scale) + + # Add horizontal frames with scaled dimensions + f1 = self.menu.add.frame_h(frame_width, frame_height, margin=(0, 0)) b1 = f1.pack(self.menu.add.button("1", lambda: self._press(1), cursor=cursor)) b2 = f1.pack( self.menu.add.button("2", lambda: self._press(2), cursor=cursor), @@ -376,9 +486,9 @@ def num_pad(self, _func): self.menu.add.button("3", lambda: self._press(3), cursor=cursor), align=pygame_menu.locals.ALIGN_RIGHT, ) - self.menu.add.vertical_margin(5) + self.menu.add.vertical_margin(int(5 * self.display_scale)) - f2 = self.menu.add.frame_h(299, 54, margin=(0, 0)) + f2 = self.menu.add.frame_h(frame_width, frame_height, margin=(0, 0)) b4 = f2.pack(self.menu.add.button("4", lambda: self._press(4), cursor=cursor)) b5 = f2.pack( self.menu.add.button("5", lambda: self._press(5), cursor=cursor), @@ -388,9 +498,9 @@ def num_pad(self, _func): self.menu.add.button("6", lambda: self._press(6), cursor=cursor), align=pygame_menu.locals.ALIGN_RIGHT, ) - self.menu.add.vertical_margin(5) + self.menu.add.vertical_margin(int(5 * self.display_scale)) - f3 = self.menu.add.frame_h(299, 54, margin=(0, 0)) + f3 = self.menu.add.frame_h(frame_width, frame_height, margin=(0, 0)) b7 = f3.pack(self.menu.add.button("7", lambda: self._press(7), cursor=cursor)) b8 = f3.pack( self.menu.add.button("8", lambda: self._press(8), cursor=cursor), @@ -400,9 +510,9 @@ def num_pad(self, _func): self.menu.add.button("9", lambda: self._press(9), cursor=cursor), align=pygame_menu.locals.ALIGN_RIGHT, ) - self.menu.add.vertical_margin(5) + self.menu.add.vertical_margin(int(5 * self.display_scale)) - f4 = self.menu.add.frame_h(299, 54, margin=(0, 0)) + f4 = self.menu.add.frame_h(frame_width, frame_height, margin=(0, 0)) delete = f4.pack( self.menu.add.button("<", lambda: self._press("<"), cursor=cursor), align=pygame_menu.locals.ALIGN_RIGHT, @@ -415,30 +525,33 @@ def num_pad(self, _func): self.menu.add.button(" .", lambda: self._press("."), cursor=cursor), align=pygame_menu.locals.ALIGN_LEFT, ) - self.menu.add.vertical_margin(5) + self.menu.add.vertical_margin(int(5 * self.display_scale)) - f5 = self.menu.add.frame_h(299, 54, margin=(0, 0)) + f5 = self.menu.add.frame_h(frame_width, frame_height, margin=(0, 0)) ok = f5.pack( self.menu.add.button("OK", lambda: self._press("ok", _func), cursor=cursor), align=pygame_menu.locals.ALIGN_CENTER, ) - # Add decorator for each object + # Add decorator for each object with scaled dimensions + rect_offset_x = int(-37 * self.display_scale) + rect_offset_y = int(-27 * self.display_scale) + rect_width = int(button_width) + rect_height = int(button_height) + for widget in (b1, b2, b3, b4, b5, b6, b7, b8, b9, b0, ok, delete, dot): w_deco = widget.get_decorator() if widget != ok: - w_deco.add_rectangle(-37, -27, 74, 54, (15, 15, 15)) - on_layer = w_deco.add_rectangle(-37, -27, 74, 54, (84, 84, 84)) + w_deco.add_rectangle(rect_offset_x, rect_offset_y, rect_width, rect_height, (15, 15, 15)) + on_layer = w_deco.add_rectangle(rect_offset_x, rect_offset_y, rect_width, rect_height, (84, 84, 84)) else: - w_deco.add_rectangle(-37, -27, 74, 54, (0, 128, 0)) - on_layer = w_deco.add_rectangle(-37, -27, 74, 54, (40, 171, 187)) + w_deco.add_rectangle(rect_offset_x, rect_offset_y, rect_width, rect_height, (0, 128, 0)) + on_layer = w_deco.add_rectangle(rect_offset_x, rect_offset_y, rect_width, rect_height, (40, 171, 187)) w_deco.disable(on_layer) widget.set_attribute("on_layer", on_layer) def widget_select(sel: bool, wid: "pygame_menu.widgets.Widget", _): - """ - Function triggered if widget is selected - """ + """Function triggered if widget is selected""" lay = wid.get_attribute("on_layer") if sel: wid.get_decorator().enable(lay) @@ -446,47 +559,62 @@ def widget_select(sel: bool, wid: "pygame_menu.widgets.Widget", _): wid.get_decorator().disable(lay) widget.set_onselect(widget_select) - widget.set_padding((2, 19, 0, 23)) + widget.set_padding( + ( + int(2 * self.display_scale), + int(19 * self.display_scale), + 0, + int(23 * self.display_scale), + ) + ) def _press(self, digit, _func=None) -> None: - """ - Press numpad digit. - - :param digit: Number or symbol - """ + """Press numpad digit""" if digit == "ok": - if not self.curr == "": + if self.curr != "": _func() elif digit == "<": self.curr = "" - self.num_pad_disp.set_title(str("")) + self.num_pad_disp.set_title("") else: if len(self.curr) <= 9: self.curr += str(digit) self.num_pad_disp.set_title(self.curr) def log_pulse_weight(self, pulse_dur, port, pulse_num, weight=0, pressure=0): - key = dict(setup=self.logger.setup, port=port, date=time.strftime("%Y-%m-%d")) - self.logger.put( - table="PortCalibration", - tuple=key, - schema="interface", - priority=5, - ignore_extra_fields=True, - validate=True, - block=True, - replace=False, - ) - self.logger.put( - table="PortCalibration.Liquid", - schema="interface", - replace=True, - ignore_extra_fields=True, - tuple=dict( - key, - pulse_dur=pulse_dur, - pulse_num=pulse_num, - weight=weight, - pressure=pressure, - ), - ) + """Log calibration data to database""" + try: + key = dict( + setup=self.logger.setup, + port=port, + date=time.strftime("%Y-%m-%d") + ) + + self.logger.put( + table="PortCalibration", + tuple=key, + schema="interface", + priority=5, + ignore_extra_fields=True, + validate=True, + block=True, + replace=False, + ) + + self.logger.put( + table="PortCalibration.Liquid", + schema="interface", + replace=True, + ignore_extra_fields=True, + tuple=dict( + key, + pulse_dur=pulse_dur, + pulse_num=pulse_num, + weight=weight, + pressure=pressure, + ), + ) + log.info(f"Logged calibration data for port {port}") + + except Exception as e: + log.error(f"Failed to log calibration data: {e}") \ No newline at end of file diff --git a/src/ethopy/interfaces/RPPorts.py b/src/ethopy/interfaces/RPPorts.py index 1e167d8..7da357e 100644 --- a/src/ethopy/interfaces/RPPorts.py +++ b/src/ethopy/interfaces/RPPorts.py @@ -248,11 +248,10 @@ def _position_change(self, channel=0): port = self._channel2port(channel, 'Proximity') # Check if the animal is in position in_position = self._get_position(port.port) - # Start the timer if the animal is in position - if in_position: - self.timer_ready.start() # Log the in_position event and update the position if there is a change in position if in_position and not self.position.port: + # Start the ready timer on a real entry, not on every edge + self.timer_ready.start() self.position_tmst = self.beh.log_activity({**port.__dict__, 'in_position': 1}) self.position = port elif not in_position and self.position.port: diff --git a/src/ethopy/interfaces/RPPorts5.py b/src/ethopy/interfaces/RPPorts5.py new file mode 100644 index 0000000..a7f735a --- /dev/null +++ b/src/ethopy/interfaces/RPPorts5.py @@ -0,0 +1,374 @@ +import logging +from concurrent.futures import ThreadPoolExecutor +from time import sleep +import threading + +import numpy as np + +from ethopy import local_conf +from ethopy.core.interface import Interface, Port +from ethopy.utils.helper_functions import convert_numeric_keys + +try: + from RPi import GPIO + import lgpio + + IMPORT_RP = True +except ImportError: + IMPORT_RP = False + +log = logging.getLogger(__name__) + + +class RPPorts5(Interface): + def __init__(self, **kwargs): + if not globals()["IMPORT_RP"]: + raise ImportError( + "Could not import RPi packages (rpi-lgpio)! " + "Please install: pip install rpi-lgpio" + ) + super(RPPorts5, self).__init__(**kwargs) + self.GPIO = GPIO + self.GPIO.setmode(self.GPIO.BCM) + + # Store the lgpio chip handle for tx_pulse + self.chip = GPIO._chip # Access the internal chip handle + + self.frequency = 15 + self.pulses = dict() + self.sound_pulses = [] + self.channels = convert_numeric_keys(local_conf.get("Channels")) + self.thread = ThreadPoolExecutor(max_workers=4) + + # For sound generation + self._sound_pwm = None + self._sound_stop_event = threading.Event() + + matched_ports = set(self.rew_ports) & set(self.channels["Liquid"].keys()) + assert matched_ports == set(self.rew_ports), ( + "All reward ports must have assigned a liquid delivery port!" + ) + + if "Lick" in self.channels: + self.GPIO.setup( + list(self.channels["Lick"].values()), + self.GPIO.IN, + pull_up_down=GPIO.PUD_DOWN, + ) + if self.callbacks: + for channel in self.channels["Lick"]: + self.GPIO.add_event_detect( + self.channels["Lick"][channel], + self.GPIO.RISING, + callback=self._lick_port_activated, + bouncetime=100, + ) + + if "Proximity" in self.channels: + self.GPIO.setup( + list(self.channels["Proximity"].values()), + self.GPIO.IN, + pull_up_down=GPIO.PUD_DOWN, + ) + if self.callbacks: + for channel in self.channels["Proximity"]: + self.GPIO.add_event_detect( + self.channels["Proximity"][channel], + self.GPIO.BOTH, + callback=self._position_change, + bouncetime=50, + ) + + if "Odor" in self.channels: + self.GPIO.setup( + list(self.channels["Odor"].values()), + self.GPIO.OUT, + initial=self.GPIO.LOW, + ) + + if "Opto" in self.channels: + self.GPIO.setup(self.channels["Opto"], self.GPIO.OUT, initial=self.GPIO.LOW) + + if "Liquid" in self.channels: + for channel in self.channels["Liquid"]: + # Set up liquid delivery pins using RPi.GPIO + self.GPIO.setup( + self.channels["Liquid"][channel], + self.GPIO.OUT, + initial=self.GPIO.LOW, + ) + + if "Sound" in self.channels: + for channel in self.channels["Sound"]: + # Set up sound pins using RPi.GPIO + self.GPIO.setup( + self.channels["Sound"][channel], + self.GPIO.OUT, + initial=self.GPIO.LOW, + ) + + if "Status" in self.channels: + self.GPIO.setup( + self.channels["Status"], self.GPIO.OUT, initial=self.GPIO.LOW + ) + + if "Sync" in self.channels and "out" in self.channels["Sync"]: + self.GPIO.setup( + self.channels["Sync"]["out"], self.GPIO.OUT, initial=self.GPIO.LOW + ) + + if self.exp.sync: + self.GPIO.setup( + self.channels["Sync"]["rec"], self.GPIO.IN, pull_up_down=GPIO.PUD_DOWN + ) + self.GPIO.setup( + self.channels["Sync"]["in"], self.GPIO.IN, pull_up_down=GPIO.PUD_DOWN + ) + self.GPIO.add_event_detect( + self.channels["Sync"]["in"], + self.GPIO.BOTH, + callback=self._sync_in, + bouncetime=20, + ) + self.dataset = self.logger.createDataset( + dataset_name="sync", dataset_type=np.dtype([("sync_times", np.double)]) + ) + + def give_liquid(self, port, duration=False): + if not duration: + duration = self.duration[port] + self.thread.submit(self._give_pulse, port, duration) + + def give_odor(self, delivery_port, odor_id, odor_duration, dutycycle): + for i, _ in enumerate(odor_id): + self.thread.submit( + self._pwd_out, + self.channels["Odor"][delivery_port[i]], + odor_duration, + dutycycle[i], + ) + + def opto_stim(self, duration, dutycycle): + self.thread.submit(self._pwd_out, self.channels["Opto"], duration, dutycycle) + + def sync_out(self, state=True): + self.GPIO.output(self.channels["Sync"]["out"], state) + + def give_sound(self, sound_freq=40500, volume=100, pulse_freq=0): + self.thread.submit( + self.__pulse_out, self.channels["Sound"][1], sound_freq, volume, pulse_freq + ) + + def stop_sound(self): + # Signal the sound generation thread to stop + self._sound_stop_event.set() + # Stop PWM if it exists + if self._sound_pwm: + self._sound_pwm.stop() + self._sound_pwm = None + # Ensure the sound pin is off + if "Sound" in self.channels: + self.GPIO.output(self.channels["Sound"][1], self.GPIO.LOW) + + def setup_touch_exit(self): + pass + + def set_operation_status(self, operation_status): + if self.exp.sync: + while not self.is_recording(): + log.info("Waiting for recording to start...") + sleep(1) + self.GPIO.output(self.channels["Status"], operation_status) + + def is_recording(self): + if self.exp.sync: + return self.GPIO.input(self.channels["Sync"]["rec"]) + else: + return False + + def cleanup(self): + self.set_operation_status(False) + self._sound_stop_event.set() + + # Clean up sound PWM + if self._sound_pwm: + self._sound_pwm.stop() + self._sound_pwm = None + + if self.callbacks: + if "Lick" in self.channels: + for channel in self.channels["Lick"]: + self.GPIO.remove_event_detect(self.channels["Lick"][channel]) + if "Proximity" in self.channels: + for channel in self.channels["Proximity"]: + self.GPIO.remove_event_detect(self.channels["Proximity"][channel]) + + self.GPIO.cleanup() + + if self.exp.sync: + if "Sync" in self.channels: + for channel in self.channels["Sync"]: + self.GPIO.remove_event_detect(self.channels["Sync"][channel]) + self.closeDatasets() + + def in_position(self, port=0): + """Determine if the specified port is in position and return the position data. + + Args: + port (int, optional): The port to check the position of. Defaults to 0. + + Returns: + tuple: A tuple containing the position data for the specified port in the following format: + - position (Port): A Port object representing the position of the specified port. + - position_dur (float): The duration in ms that the specified port has been in its current position. + - position_tmst (float): The timestamp in ms that the specified port activated. + + If the specified port is not in position, the tuple will be (0, 0, 0). + + """ + # self.position is maintained by the debounced proximity callbacks + if not self.position.port or (port and self.position.port != port): + return 0, 0, 0 + return self.position, self.timer_ready.elapsed_time(), self.position_tmst + + def off_proximity(self): + """Check that no proximity port is activated. + + Used to make sure that none of the ports is activated before moving on to + the next trial. The answer comes from self.position, which the debounced + GPIO callbacks maintain, so the trial logic and the logged activity always + agree on where the animal is. + + Returns: + bool: True if all proximity ports are not activated + """ + return not self.position.port + + def _get_position(self, ports=0): + """get the position of the proximity ports + + _extended_summary_ + + Args: + ports (int, optional): The port to check the position of. Defaults is 0 which means check all the ports. + + Returns: + int: the id of the activated port else 0 + """ + # if port is not specified check all proximity ports + if not ports: + ports = self.proximity_ports + elif not type(ports) is list: + ports = [ports] + for port in ports: + # find the position of the port + in_position = self.GPIO.input(self.channels["Proximity"][port]) + # if port invert take the opposite + if self.ports[Port(type="Proximity", port=port) == self.ports][0].invert: + in_position = not in_position + # return the port id if any port is in position + if in_position: + return port + return 0 + + def _position_change(self, channel=0): + """Update the position of the animal and log the in_position event. + + Position_change is called in as callback at event_detect of GPIO.BOTH of the proximity channels. + Also called from function in_position in the case where the callback has not run but the position has changed. + We want to log the port change and update the self.position with the activated port or reset it. + Also we calculate + - position_dur (float): The duration in ms that the specified port has been in its current position. + - position_tmst (float): The timestamp in ms that the specified port activated. + Before we log the position we check that it has been changed, because due to the small bouncetime + most proximity sensors(switches) will flicker back and forth between the two values before settling down. + + Args: + channel (int, optional): The channel number of the proximity sensor. Defaults to 0. + """ + # Get the port number corresponding to the proximity sensor channel + port = self._channel2port(channel, "Proximity") + # Check if the animal is in position + in_position = self._get_position(port.port) + # Log the in_position event and update the position if there is a change in position + if in_position and not self.position.port: + # The ready timer starts on a real entry, not on every edge + self.timer_ready.start() + self.position_tmst = self.beh.log_activity({**port.__dict__, 'in_position': 1}) + self.position = port + elif not in_position and self.position.port: + tmst = self.beh.log_activity({**port.__dict__, 'in_position': 0}) + self.position_dur = tmst - self.position_tmst + self.position = Port() + + def _give_pulse(self, port, duration): + """Generate liquid reward pulse using lgpio.tx_pulse for precise timing.""" + pin = self.channels["Liquid"][port] + + # Convert milliseconds to microseconds + pulse_us = int(duration * 1000) + + # Generate a single hardware-timed pulse + # tx_pulse(chip, gpio, pulse_on_us, pulse_off_us, offset_us, cycles) + lgpio.tx_pulse(self.chip, pin, pulse_us, 0, 0, 1) + + def _lick_port_activated(self, channel): + self.resp_tmst = self.logger.logger_timer.elapsed_time() + self.response = self._channel2port(channel, "Lick") + self.beh.log_activity({**self.response.__dict__, "time": self.resp_tmst}) + return self.response, self.resp_tmst + + def _sync_in(self, channel): + self.dataset.append("sync_data", [self.logger.logger_timer.elapsed_time()]) + + def _pwd_out(self, channel, duration, dutycycle): + """PWM output using RPi.GPIO (now rpi-lgpio).""" + pwm = self.GPIO.PWM(channel, self.frequency) + pwm.ChangeFrequency(self.frequency) + pwm.start(dutycycle) + sleep(duration / 1000) # to add a delay in seconds + pwm.stop() + + def __pulse_out(self, channel, freq, dutycycle=100, pulse_freq=0): + """Generate sound using RPi.GPIO.PWM instead of pigpio wave generation.""" + self._sound_stop_event.clear() + + if dutycycle == 0: + return + + # Speaker has non-monotonic response with ~50% duty cycle is maximum response + # Normalize percentage by /2 for speaker response + normalized_duty = dutycycle / 2 + + try: + # Create PWM instance + self._sound_pwm = self.GPIO.PWM(channel, freq) + + if pulse_freq == 0: + # Continuous tone + self._sound_pwm.start(normalized_duty) + # Keep running until stop is requested + while not self._sound_stop_event.is_set(): + sleep(0.01) # Small sleep to prevent CPU spinning + else: + # Pulsed tone - alternate between sound and silence + pulse_period = 1.0 / (pulse_freq * 2) # Time for one pulse (on + off) + + while not self._sound_stop_event.is_set(): + # Turn on sound + self._sound_pwm.start(normalized_duty) + if self._sound_stop_event.wait(timeout=pulse_period): + break + + # Turn off sound (silence period) + self._sound_pwm.stop() + if self._sound_stop_event.wait(timeout=pulse_period): + break + + except Exception as e: + log.error(f"Error in sound generation: {e}") + finally: + # Clean up PWM + if self._sound_pwm: + self._sound_pwm.stop() + self._sound_pwm = None diff --git a/src/ethopy/utils/start.py b/src/ethopy/utils/start.py index 06ce312..1909e8a 100644 --- a/src/ethopy/utils/start.py +++ b/src/ethopy/utils/start.py @@ -2,37 +2,72 @@ import os from typing import Union +import logging import pygame import pygame_menu +log = logging.getLogger(__name__) + class PyWelcome: def __init__(self, logger) -> None: self.logger = logger - self.SCREEN_WIDTH = 800 - self.SCREEN_HEIGHT = 480 + # Initialize pygame first if not pygame.get_init(): pygame.init() - if self.logger.is_pi: - self.screen = pygame.display.set_mode( - (self.SCREEN_WIDTH, self.SCREEN_HEIGHT), pygame.FULLSCREEN - ) - else: + self.SCREEN_WIDTH = 800 + self.SCREEN_HEIGHT = 480 + + # Set display mode - always try fullscreen first on Pi + try: + if self.logger.is_pi: + # Ask SDL for the desktop resolution. Requesting a size that + # differs from it returns a surface that segfaults on draw, and + # /sys/class/graphics/fb0/virtual_size reports its dimensions + # in the opposite order on some Pis. + self.screen = pygame.display.set_mode( + (0, 0), + pygame.FULLSCREEN | pygame.DOUBLEBUF | pygame.HWSURFACE, + ) + log.debug("Fullscreen mode activated") + else: + # Windowed mode for testing + self.screen = pygame.display.set_mode( + (self.SCREEN_WIDTH, self.SCREEN_HEIGHT) + ) + print("Windowed mode activated") + except pygame.error as e: + print(f"Failed to set fullscreen mode: {e}") + # Fallback to windowed mode self.screen = pygame.display.set_mode( (self.SCREEN_WIDTH, self.SCREEN_HEIGHT) ) + print("Fallback to windowed mode") + + # Every menu is sized from the surface pygame actually gave us. + self.SCREEN_WIDTH, self.SCREEN_HEIGHT = self.screen.get_size() + log.debug(f"Display surface size: {self.SCREEN_WIDTH}x{self.SCREEN_HEIGHT}") - # Configure self.theme + # pygame.display.set_caption("EthoPy") + + # Hide mouse cursor using multiple methods + # self.hide_cursor_comprehensive() + + # Configure self.theme - adjust font sizes based on screen size self.theme = pygame_menu.themes.THEME_DARK.copy() self.theme.background_color = (0, 0, 0) self.theme.title_background_color = (43, 43, 43) - self.theme.title_font_size = 35 + + # Scale font sizes based on screen resolution + font_scale = min(self.SCREEN_WIDTH / 800, self.SCREEN_HEIGHT / 480) + self.theme.title_font_size = int(35 * font_scale) + self.theme.widget_font_size = int(30 * font_scale) + self.theme.widget_alignment = pygame_menu.locals.ALIGN_CENTER self.theme.widget_font_color = (255, 255, 255) - self.theme.widget_font_size = 30 self.theme.widget_padding = 0 # variables @@ -52,6 +87,8 @@ def setup_menus(self) -> None: def mainloop(self) -> None: """App mainloop.""" + clock = pygame.time.Clock() # Add clock for better performance + while ( self.logger.setup_status != "running" and self.logger.setup_status != "exit" ): @@ -59,18 +96,26 @@ def mainloop(self) -> None: for event in events: if event.type == pygame.QUIT: break + # Add escape key to exit fullscreen + elif event.type == pygame.KEYDOWN: + if event.key == pygame.K_ESCAPE: + self.logger.update_setup_info({"status": "exit"}) + break if self.main_menu.is_enabled(): self.main_menu.update(events) + + # Clear screen before drawing + self.screen.fill((0, 0, 0)) self.main_menu.draw(self.screen) - pygame.display.update() - pygame.time.wait(2) + pygame.display.flip() # Use flip() instead of update() for better performance + clock.tick(60) # Limit to 60 FPS pygame_menu.events.CLOSE if pygame.get_init(): self.main_menu.disable() - pygame.mouse.set_visible(1) + pygame.mouse.set_visible(True) pygame.display.quit() def create_main(self) -> "pygame_menu.Menu": @@ -85,11 +130,15 @@ def create_main(self) -> "pygame_menu.Menu": theme=self.theme, ) + # Scale positions based on screen size + scale_x = self.SCREEN_WIDTH / 800 + scale_y = self.SCREEN_HEIGHT / 480 + menu.add.label( f"ip: {self.logger.get_ip()}, setup: {self.logger.setup}", - font_size=15, + font_size=int(15 * min(scale_x, scale_y)), align=pygame_menu.locals.ALIGN_LEFT, - ).translate(5, 10) + ).translate(int(5 * scale_x), int(10 * scale_y)) menu.add.button( f"Animal id: {self.animal_id}", @@ -98,7 +147,7 @@ def create_main(self) -> "pygame_menu.Menu": float=True, padding=(5, 10, 5, 10), background_color=(0, 15, 15), - ).translate(300, 40) + ).translate(int(300 * scale_x), int(40 * scale_y)) menu.add.button( f"Task id: {self.task_id}", @@ -107,7 +156,7 @@ def create_main(self) -> "pygame_menu.Menu": float=True, padding=(5, 10, 5, 10), background_color=(0, 15, 15), - ).translate(300, 130) + ).translate(int(300 * scale_x), int(130 * scale_y)) menu.add.button( "Start experiment", @@ -116,8 +165,8 @@ def create_main(self) -> "pygame_menu.Menu": float=True, padding=(10, 15, 10, 15), background_color=(0, 153, 0), - font_size=35, - ).translate(250, 260) + font_size=int(35 * min(scale_x, scale_y)), + ).translate(int(250 * scale_x), int(260 * scale_y)) menu.add.button( "Restart", @@ -126,7 +175,7 @@ def create_main(self) -> "pygame_menu.Menu": float=True, padding=(5, 23, 5, 30), background_color=(128, 128, 128), - ).translate(630, 290) + ).translate(int(630 * scale_x), int(290 * scale_y)) menu.add.button( "Weight", @@ -144,7 +193,7 @@ def create_main(self) -> "pygame_menu.Menu": float=True, background_color=(255, 0, 0), padding=(5, 10, 5, 10), - ).translate(630, 350) + ).translate(int(630 * scale_x), int(350 * scale_y)) return menu @@ -159,7 +208,7 @@ def create_animal(self): ) menu_animal.add.label( "Select Animal id: ", - font_size=20, + font_size=int(20 * min(self.SCREEN_WIDTH / 800, self.SCREEN_HEIGHT / 480)), ) self.curr_animal = "" menu_animal.add.vertical_margin(5) @@ -216,7 +265,7 @@ def create_task(self): ) menu_task.add.label( "Select Task id: ", - font_size=20, + font_size=int(20 * min(self.SCREEN_WIDTH / 800, self.SCREEN_HEIGHT / 480)), ) self.curr_task = "" menu_task.add.vertical_margin(5) @@ -242,7 +291,7 @@ def start_experiment(self): self.logger.update_setup_info({"status": "running"}) def close(self): - pygame.mouse.set_visible(1) + pygame.mouse.set_visible(True) pygame.quit() def add_num_pad(self, menu, log_function, screen) -> "pygame_menu.Menu": @@ -252,8 +301,13 @@ def add_num_pad(self, menu, log_function, screen) -> "pygame_menu.Menu": cursor = pygame_menu.locals.CURSOR_HAND + # Scale button sizes based on screen resolution + button_width = int(74 * min(self.SCREEN_WIDTH / 800, self.SCREEN_HEIGHT / 480)) + button_height = int(54 * min(self.SCREEN_WIDTH / 800, self.SCREEN_HEIGHT / 480)) + frame_width = int(299 * (self.SCREEN_WIDTH / 800)) + # Add horizontal frames - f1 = menu.add.frame_h(299, 54, margin=(5, 0)) + f1 = menu.add.frame_h(frame_width, button_height, margin=(5, 0)) b1 = f1.pack( menu.add.button("1", lambda: self._press(1, screen), cursor=cursor) ) @@ -267,7 +321,7 @@ def add_num_pad(self, menu, log_function, screen) -> "pygame_menu.Menu": ) menu.add.vertical_margin(5) - f2 = menu.add.frame_h(299, 54, margin=(5, 0)) + f2 = menu.add.frame_h(frame_width, button_height, margin=(5, 0)) b4 = f2.pack( menu.add.button("4", lambda: self._press(4, screen), cursor=cursor) ) @@ -281,7 +335,7 @@ def add_num_pad(self, menu, log_function, screen) -> "pygame_menu.Menu": ) menu.add.vertical_margin(5) - f3 = menu.add.frame_h(299, 54, margin=(5, 0)) + f3 = menu.add.frame_h(frame_width, button_height, margin=(5, 0)) b7 = f3.pack( menu.add.button("7", lambda: self._press(7, screen), cursor=cursor) ) @@ -295,7 +349,7 @@ def add_num_pad(self, menu, log_function, screen) -> "pygame_menu.Menu": ) menu.add.vertical_margin(5) - f4 = menu.add.frame_h(299, 54, margin=(5, 0)) + f4 = menu.add.frame_h(frame_width, button_height, margin=(5, 0)) delete = f4.pack( menu.add.button("<", lambda: self._press("<", screen), cursor=cursor), align=pygame_menu.locals.ALIGN_RIGHT, @@ -310,7 +364,7 @@ def add_num_pad(self, menu, log_function, screen) -> "pygame_menu.Menu": ) menu.add.vertical_margin(5) - f5 = menu.add.frame_h(299, 54, margin=(5, 0)) + f5 = menu.add.frame_h(frame_width, button_height, margin=(5, 0)) ok = f5.pack( menu.add.button( "Ok", lambda: self._press("ok", screen, log_function), cursor=cursor @@ -318,15 +372,42 @@ def add_num_pad(self, menu, log_function, screen) -> "pygame_menu.Menu": align=pygame_menu.locals.ALIGN_CENTER, ) - # Add decorator for each object + # Add decorator for each object - scale rectangle sizes + rect_width = int(button_width) + rect_height = int(button_height) + for widget in (b1, b2, b3, b4, b5, b6, b7, b8, b9, b0, ok, delete, dot): w_deco = widget.get_decorator() if widget != ok: - w_deco.add_rectangle(-37, -27, 74, 54, (15, 15, 15)) - on_layer = w_deco.add_rectangle(-37, -27, 74, 54, (84, 84, 84)) + w_deco.add_rectangle( + -rect_width // 2, + -rect_height // 2, + rect_width, + rect_height, + (15, 15, 15), + ) + on_layer = w_deco.add_rectangle( + -rect_width // 2, + -rect_height // 2, + rect_width, + rect_height, + (84, 84, 84), + ) else: - w_deco.add_rectangle(-37, -27, 74, 54, (38, 96, 103)) - on_layer = w_deco.add_rectangle(-37, -27, 74, 54, (40, 171, 187)) + w_deco.add_rectangle( + -rect_width // 2, + -rect_height // 2, + rect_width, + rect_height, + (38, 96, 103), + ) + on_layer = w_deco.add_rectangle( + -rect_width // 2, + -rect_height // 2, + rect_width, + rect_height, + (40, 171, 187), + ) w_deco.disable(on_layer) widget.set_attribute("on_layer", on_layer)