Squashed commit of the following:

iandobbie · iandobbie · commit e790a499a2c4 · 2023-02-09T17:39:05.000-05:00
Added docstrings to DigitalIO class
    Added raspberypi DigitialIO driver.
    First working implmentation of DIO abc and simulator device
    Added an abc for a DigitalIO device class
diff --git a/microscope/abc.py b/microscope/abc.py
@@ -1555,3 +1555,119 @@ def move_to(self, position: typing.Mapping[str, float]) -> None:
 
         """
         raise NotImplementedError()
+
+class DigitalIO(Device, metaclass=abc.ABCMeta):
+    """ABC for digital IO devices.
+
+    Digital IO devices (DIO) have a num,ber of digital lines that can
+    be for output, or optionally input and can switch between a on and
+    off state.
+
+    Args:
+        numLines: total number of digital lines numberes 0 to n-1. 
+
+    """
+
+    def __init__(self, numLines: int, **kwargs) -> None:
+        super().__init__(**kwargs)
+        if numLines < 1:
+            raise ValueError(
+                "NumLines must be a positive number (was %d)" % positions
+            )
+        self._numLines = numLines
+
+        #array to map wether lines are input or output
+        # true is output, start with all lines defined for output. 
+        self._IOMap = [True]*self._numLines
+  
+    @abc.abstractmethod
+    def get_num_lines(self):
+        """Returns the number of Io lines present in this instance"""
+        return self._numLines
+
+    @abc.abstractmethod    
+    def set_IO_state(self, line: int, state: bool):
+        """Sets the state of a single Digital line to either Output or Input
+
+        Args:
+            line: The line to have its mode set.
+        
+            state: True for Output or False for Input."""
+        raise NotImplementedError()
+
+    @abc.abstractmethod    
+    def get_IO_state(self, line):
+        """Returns the state of a single Digital line, either Output or Input
+
+        Args:
+            line: The line to have its mode set.
+        Return value is True for Output and False for Input"""
+        raise NotImplementedError()
+
+    @abc.abstractmethod    
+    def set_all_IO_state(self, stateArray):
+        """Sets the state of all lines to either Input or Output
+        Args:
+            line: The line to have its mode set.
+            stateArray: Boolean array for the lines, True in output False
+                        is Input"""
+        for i, state in enumerate(stateArray):
+                #set each line as defined in stateArray
+                self.set_IO_state(i,state)
+            
+
+    @abc.abstractmethod    
+    def get_all_IO_state(self):
+        """Returns the state of a all Digital line, either Output or Input
+
+        Returns a boolean array one entry for each line, 
+        True for Output and False for Input"""
+
+        stateArray=[None]*self._numLines
+        for i in range(self._numLines):
+              stateArray[i]=self.get_IO_state(i)
+        return stateArray
+
+    @abc.abstractmethod    
+    def write_line(self,line,ouput):
+        """Sets the level of a single output line
+
+        Args:
+            line: the line to be set
+            output: the level True for high and Flase for low."""
+        
+       raise NotImplementedError()
+
+    @abc.abstractmethod    
+    def write_all_lines(self,ouput_array):
+        """Sets the output level of every output line.
+
+        Args:
+            output_array: Boolean array of output states True for high, 
+                          False for low, array entries for lines set 
+                          as inputs are ignored."""
+        
+        if len(output_array) != self._numLines :
+            raise("Output array must be numLines in length")
+        for i in range(self._numLines):
+            #set line i to the IOMap entry, true for output false for input.
+            if(not self._IOMap[i]):
+                self.write_line(i,output_array[i])
+
+    @abc.abstractmethod    
+    def read_line(self,line):
+        """Read a single input line.
+        Args:
+            line: the line to read
+        Return: A boolean of the line state"""
+        raise NotImplementedError()
+
+    @abc.abstractmethod    
+    def read_all_lines(self):
+        """Read all the input lines.
+        Return: Boolean Array with outline enteries set to None."""
+        
+        readarray=[None]*self._numLines
+        for i in range(self._numLines):
+            readarray[i]=self.read_line(i)
+        return(readarray)
diff --git a/microscope/digitalio/raspberrypi.py b/microscope/digitalio/raspberrypi.py
@@ -0,0 +1,120 @@
+#!/usr/bin/env python3
+
+## Copyright (C) 2020 David Miguel Susano Pinto <carandraug@gmail.com>
+## Copyright (C) 2023 Ian Dobbie <ian.dobbie@jhu.edu>
+##
+##
+## This file is part of Microscope.
+##
+## Microscope is free software: you can redistribute it and/or modify
+## it under the terms of the GNU General Public License as published by
+## the Free Software Foundation, either version 3 of the License, or
+## (at your option) any later version.
+##
+## Microscope is distributed in the hope that it will be useful,
+## but WITHOUT ANY WARRANTY; without even the implied warranty of
+## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+## GNU General Public License for more details.
+##
+## You should have received a copy of the GNU General Public License
+## along with Microscope.  If not, see <http://www.gnu.org/licenses/>.
+
+"""Raspberry Pi Digital IO module.
+"""
+
+import contextlib
+import re
+import threading
+import time
+import typing
+import logging
+
+import microscope.abc
+
+import RPi.GPIO as GPIO
+
+
+#Support for async digital IO control on the Raspberryy Pi.
+#Currently supports digital input and output via GPIO lines
+
+
+# Use BCM GPIO references (naming convention for GPIO pins from Broadcom)
+# instead of physical pin numbers on the Raspberry Pi board
+GPIO.setmode(GPIO.BCM)
+_logger = logging.getLogger(__name__)
+
+
+
+class RPiDIO(microscope.abc.DigitalIO):
+    '''Digital IO device implementation for a Raspberry Pi
+
+    gpioMap input array maps line numbers to specific GPIO pins
+    [GPIO pin, GPIO pin]
+    [27,25,29,...]  line 0 in pin 27, 1 is pin 25 etc....
+
+    gpioState input array maps output and input lines.
+    True maps to output
+    False maps to input
+
+    with the gpioMap above [True,False,True,..] would map:
+    27 to out, 
+    25 to in 
+    29 to out'''
+    
+    def __init__(self,gpioMap = [], gpioState = [], **kwargs):
+        super().__init__(numLines=len(gpioMap),**kwargs)
+        #setup io lines 1-n mapped to GPIO lines
+        self._gpioMap=gpioMap
+        self._IOMap=gpioState
+        self._numLines=len(self._gpioMap)
+        self._outputCache = [False]*self._numLines
+        self.set_all_IO_state(self._IOMap)
+
+
+
+    #functions needed
+
+    def set_IO_state(self, line: int, state: bool) -> None:
+        _logger.debug("Line %d set IO state %s"% (line,str(state)))
+        if state:
+            #true maps to output
+            GPIO.setup(self._gpioMap[line],GPIO.OUT)
+            self._IOMap[line] = True
+            #restore state from cache.
+            state=self._outputCache[line]
+            GPIO.output(self._gpioMap[line],state)
+        else:
+            GPIO.setup(self._gpioMap[line],GPIO.IN)
+            self._IOMap[line] = False
+
+    def get_IO_state(self, line: int) -> bool:
+        #returns
+        #  True if the line is Output
+        #  Flase if Input
+        #  None in other cases (i2c, spi etc)
+        pinmode=GPIO.gpio_function(self._gpioMap[line])
+        if pinmode==GPIO.OUT:
+            return True
+        elif pinmode==GPIO.IN:
+            return False
+        return None
+
+    def write_line(self,line: int, state: bool):
+        #Do we need to check if the line can be written?
+        _logger.debug("Line %d set IO state %s"% (line,str(state)))
+        self._outputCache[line]=state
+        GPIO.output(self._gpioMap[line],state)
+        
+    def read_line(self,line: int) -> bool:
+        # Should we check if the line is set to input first?
+        #If input read the real state
+        if (not self._IOMap[line]):
+            state=GPIO.input(self._gpioMap[line])
+            _logger.debug("Line %d returns %s" % (line,str(state)))
+            return state
+        else:
+            #line is an outout so returned cached state
+            return self._outputCache[line]
+
+    def _do_shutdown(self) -> None:
+        pass
diff --git a/microscope/simulators/__init__.py b/microscope/simulators/__init__.py
@@ -2,6 +2,7 @@
 
 ## Copyright (C) 2020 David Miguel Susano Pinto <carandraug@gmail.com>
 ## Copyright (C) 2020 Mick Phillips <mick.phillips@gmail.com>
+## Copyright (C) 2022 Ian Dobbie <ian.dobbie@gmail.com>
 ##
 ## This file is part of Microscope.
 ##
@@ -479,3 +480,38 @@ def move_by(self, delta: typing.Mapping[str, float]) -> None:
     def move_to(self, position: typing.Mapping[str, float]) -> None:
         for name, pos in position.items():
             self.axes[name].move_to(pos)
+
+class SimulatedDigitalIO(microscope.abc.DigitalIO):
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self._cache=[None]*self._numLines
+
+    def set_IO_state(self, line: int, state: bool) -> None:
+        _logger.info("Line %d set IO state %s"% (line,str(state)))
+        self._IOMap[line] = state
+        if not state:
+            #this is an input so needs to have a definite value,
+            #default to False if not already set. If set leave alone
+            if self._cache[line]== None:
+                self._cache[line]=False
+
+    def get_IO_state(self, line: int) -> bool:
+        return(self._IOMap[line])
+
+    def write_line(self,line: int, state: bool):
+        _logger.info("Line %d set IO state %s"% (line,str(state)))
+        self._cache[line]=state
+        
+    def read_line(self,line: int) -> bool:
+        _logger.info("Line %d returns %s" % (line,str(self._cache[line])))
+        return self._cache[line]
+
+    def _do_shutdown(self) -> None:
+        pass
+
+
+#DIO still to do:
+# raise exception if writing to a read line and vis-versa
+# raise exception if line <0 or line>num_lines
+# read all lines to return True,Flase if readable and None if an output
+#
