black

[Adafruit_Blinka-hackapet.git] / src / adafruit_blinka / microcontroller / generic_linux / sysfs_pwmout.py

diff --git a/src/adafruit_blinka/microcontroller/generic_linux/sysfs_pwmout.py b/src/adafruit_blinka/microcontroller/generic_linux/sysfs_pwmout.py
index c966b410a60939f3e3535b349003aa177d2780cc..f751b71a4a017dc3870da1e1445820b6c521e1ce 100644 (file)
--- a/src/adafruit_blinka/microcontroller/generic_linux/sysfs_pwmout.py
+++ b/src/adafruit_blinka/microcontroller/generic_linux/sysfs_pwmout.py
@@ -1,27 +1,44 @@
-# Much code from https://github.com/vsergeev/python-periphery/blob/master/periphery/pwm.py
-# Copyright (c) 2015-2016 vsergeev / Ivan (Vanya) A. Sergeev
-# License: MIT
+"""
+Much code from https://github.com/vsergeev/python-periphery/blob/master/periphery/pwm.py
+Copyright (c) 2015-2016 vsergeev / Ivan (Vanya) A. Sergeev
+License: MIT
+"""

 
 import os
 
 import os

-import digitalio
+from time import sleep
+from errno import EACCES

 
 try:
     from microcontroller.pin import pwmOuts
 except ImportError:
 
 try:
     from microcontroller.pin import pwmOuts
 except ImportError:

-    raise RuntimeError("No PWM outputs defined for this board")
+    raise RuntimeError("No PWM outputs defined for this board") from ImportError

 
 

+
+# pylint: disable=unnecessary-pass

 class PWMError(IOError):
     """Base class for PWM errors."""
 class PWMError(IOError):
     """Base class for PWM errors."""

+

     pass
 
 
     pass
 
 

-class PWMOut(object):
+# pylint: enable=unnecessary-pass
+
+
+class PWMOut:
+    """Pulse Width Modulation Output Class"""
+
+    # Number of retries to check for successful PWM export on open
+    PWM_STAT_RETRIES = 10
+    # Delay between check for scucessful PWM export on open (100ms)
+    PWM_STAT_DELAY = 0.1
+

     # Sysfs paths
     _sysfs_path = "/sys/class/pwm/"
     _channel_path = "pwmchip{}"
 
     # Channel paths
     _export_path = "export"
     # Sysfs paths
     _sysfs_path = "/sys/class/pwm/"
     _channel_path = "pwmchip{}"
 
     # Channel paths
     _export_path = "export"

+    _unexport_path = "unexport"

     _pin_path = "pwm{}"
 
     # Pin attribute paths
     _pin_path = "pwm{}"
 
     # Pin attribute paths


@@ -51,16 +68,18 @@ class PWMOut(object):
         """
 
         self._pwmpin = None
         """
 
         self._pwmpin = None

+        self._channel = None
+        self._period = 0

         self._open(pin, duty_cycle, frequency, variable_frequency)
 
     def __del__(self):
         self._open(pin, duty_cycle, frequency, variable_frequency)
 
     def __del__(self):

-        self.close()
+        self.deinit()

 
     def __enter__(self):
         return self
 
     def __exit__(self, t, value, traceback):
 
     def __enter__(self):
         return self
 
     def __exit__(self, t, value, traceback):

-        self.close()
+        self.deinit()

 
     def _open(self, pin, duty=0, freq=500, variable_frequency=False):
         self._channel = None
 
     def _open(self, pin, duty=0, freq=500, variable_frequency=False):
         self._channel = None


@@ -73,61 +92,129 @@ class PWMOut(object):
         if self._channel is None:
             raise RuntimeError("No PWM channel found for this Pin")
 
         if self._channel is None:
             raise RuntimeError("No PWM channel found for this Pin")
 

-        channel_path = os.path.join(self._sysfs_path, self._channel_path.format(self._channel))
+        if variable_frequency:
+            print("Variable Frequency is not supported, continuing without it...")
+
+        channel_path = os.path.join(
+            self._sysfs_path, self._channel_path.format(self._channel)
+        )

         if not os.path.isdir(channel_path):
         if not os.path.isdir(channel_path):

-            raise ValueError("PWM channel does not exist, check that the required modules are loaded.")
+            raise ValueError(
+                "PWM channel does not exist, check that the required modules are loaded."
+            )

 
 

-        pin_path = os.path.join(channel_path, self._pin_path.format(self._pwmpin))
-        if not os.path.isdir(pin_path):
+        try:
+            with open(
+                os.path.join(channel_path, self._unexport_path), "w"
+            ) as f_unexport:
+                f_unexport.write("%d\n" % self._pwmpin)
+        except IOError as e:
+            pass  # not unusual, it doesnt already exist
+        try:
+            with open(os.path.join(channel_path, self._export_path), "w") as f_export:
+                f_export.write("%d\n" % self._pwmpin)
+        except IOError as e:
+            raise PWMError(e.errno, "Exporting PWM pin: " + e.strerror) from IOError
+
+        # Loop until 'period' is writable, because application of udev rules
+        # after the above pin export is asynchronous.
+        # Without this loop, the following properties may not be writable yet.
+        for i in range(PWMOut.PWM_STAT_RETRIES):

             try:
             try:

-                with open(os.path.join(channel_path, self._export_path), "w") as f_export:
-                    f_export.write("%d\n" % self._pwmpin)
+                with open(
+                    os.path.join(
+                        channel_path, self._pin_path.format(self._pwmpin), "period"
+                    ),
+                    "w",
+                ):
+                    break

             except IOError as e:
             except IOError as e:

-                raise PWMError(e.errno, "Exporting PWM pin: " + e.strerror)
+                if e.errno != EACCES or (
+                    e.errno == EACCES and i == PWMOut.PWM_STAT_RETRIES - 1
+                ):
+                    raise PWMError(e.errno, "Opening PWM period: " + e.strerror) from e
+            sleep(PWMOut.PWM_STAT_DELAY)

 
 

-        self._set_enabled(True)
+        # self._set_enabled(False) # This line causes a write error when trying to enable

 
         # Look up the period, for fast duty cycle updates
         self._period = self._get_period()
 
 
         # Look up the period, for fast duty cycle updates
         self._period = self._get_period()
 

+        # self.duty_cycle = 0  # This line causes a write error when trying to enable
+

         # set frequency
         self.frequency = freq
         # set duty
         self.duty_cycle = duty
 
         # set frequency
         self.frequency = freq
         # set duty
         self.duty_cycle = duty
 

-    def close(self):
-        """Close the sysfs PWM."""
+        self._set_enabled(True)
+
+    def deinit(self):
+        """Deinit the sysfs PWM."""
+        if self._channel is not None:
+            self.duty_cycle = 0
+            try:
+                channel_path = os.path.join(
+                    self._sysfs_path, self._channel_path.format(self._channel)
+                )
+                with open(
+                    os.path.join(channel_path, self._unexport_path), "w"
+                ) as f_unexport:
+                    f_unexport.write("%d\n" % self._pwmpin)
+            except IOError as e:
+                raise PWMError(
+                    e.errno, "Unexporting PWM pin: " + e.strerror
+                ) from IOError
+

         self._channel = None
         self._pwmpin = None
 
         self._channel = None
         self._pwmpin = None
 

+    def _is_deinited(self):
+        if self._pwmpin is None:
+            raise ValueError(
+                "Object has been deinitialize and can no longer "
+                "be used. Create a new object."
+            )
+

     def _write_pin_attr(self, attr, value):
     def _write_pin_attr(self, attr, value):

+        # Make sure the pin is active
+        self._is_deinited()
+

         path = os.path.join(
             self._sysfs_path,
             self._channel_path.format(self._channel),
             self._pin_path.format(self._pwmpin),
         path = os.path.join(
             self._sysfs_path,
             self._channel_path.format(self._channel),
             self._pin_path.format(self._pwmpin),

-            attr)
+            attr,
+        )

 
 

-        with open(path, 'w') as f_attr:
-            #print(value, path)
+        with open(path, "w") as f_attr:
+            # print(value, path)

             f_attr.write(value + "\n")
 
     def _read_pin_attr(self, attr):
             f_attr.write(value + "\n")
 
     def _read_pin_attr(self, attr):

+        # Make sure the pin is active
+        self._is_deinited()
+

         path = os.path.join(
             self._sysfs_path,
             self._channel_path.format(self._channel),
             self._pin_path.format(self._pwmpin),
         path = os.path.join(
             self._sysfs_path,
             self._channel_path.format(self._channel),
             self._pin_path.format(self._pwmpin),

-            attr)
+            attr,
+        )

 
 

-        with open(path, 'r') as f_attr:
+        with open(path, "r") as f_attr:

             return f_attr.read().strip()
 
     # Mutable properties
 
     def _get_period(self):
             return f_attr.read().strip()
 
     # Mutable properties
 
     def _get_period(self):

+        period_ns = self._read_pin_attr(self._pin_period_path)

         try:
         try:

-            period_ns = int(self._read_pin_attr(self._pin_period_path))
+            period_ns = int(period_ns)

         except ValueError:
         except ValueError:

-            raise PWMError(None, "Unknown period value: \"%s\"" % period_ns)
+            raise PWMError(
+                None, 'Unknown period value: "%s"' % period_ns
+            ) from ValueError

 
         # Convert period from nanoseconds to seconds
         period = period_ns / 1e9
 
         # Convert period from nanoseconds to seconds
         period = period_ns / 1e9


@@ -149,6 +236,8 @@ class PWMOut(object):
         # Update our cached period
         self._period = float(period)
 
         # Update our cached period
         self._period = float(period)
 

+    period = property(_get_period, _set_period)
+

     """Get or set the PWM's output period in seconds.
 
     Raises:
     """Get or set the PWM's output period in seconds.
 
     Raises:


@@ -159,10 +248,13 @@ class PWMOut(object):
     """
 
     def _get_duty_cycle(self):
     """
 
     def _get_duty_cycle(self):

+        duty_cycle_ns = self._read_pin_attr(self._pin_duty_cycle_path)

         try:
         try:

-            duty_cycle_ns = int(self._read_pin_attr(self._pin_duty_cycle_path))
+            duty_cycle_ns = int(duty_cycle_ns)

         except ValueError:
         except ValueError:

-            raise PWMError(None, "Unknown duty cycle value: \"%s\"" % duty_cycle_ns)
+            raise PWMError(
+                None, 'Unknown duty cycle value: "%s"' % duty_cycle_ns
+            ) from ValueError

 
         # Convert duty cycle from nanoseconds to seconds
         duty_cycle = duty_cycle_ns / 1e9
 
         # Convert duty cycle from nanoseconds to seconds
         duty_cycle = duty_cycle_ns / 1e9


@@ -172,16 +264,15 @@ class PWMOut(object):
 
         # convert to 16-bit
         duty_cycle = int(duty_cycle * 65535)
 
         # convert to 16-bit
         duty_cycle = int(duty_cycle * 65535)

-

         return duty_cycle
 
     def _set_duty_cycle(self, duty_cycle):
         return duty_cycle
 
     def _set_duty_cycle(self, duty_cycle):

-        # convert from 16-bit
-        duty_cycle /= 65535
-

         if not isinstance(duty_cycle, (int, float)):
             raise TypeError("Invalid duty cycle type, should be int or float.")
         if not isinstance(duty_cycle, (int, float)):
             raise TypeError("Invalid duty cycle type, should be int or float.")

-        elif not 0.0 <= duty_cycle <= 1.0:
+
+        # convert from 16-bit
+        duty_cycle /= 65535.0
+        if not 0.0 <= duty_cycle <= 1.0:

             raise ValueError("Invalid duty cycle value, should be between 0.0 and 1.0.")
 
         # Convert duty cycle from ratio to seconds
             raise ValueError("Invalid duty cycle value, should be between 0.0 and 1.0.")
 
         # Convert duty cycle from ratio to seconds


@@ -227,28 +318,31 @@ class PWMOut(object):
 
         if enabled == "1":
             return True
 
         if enabled == "1":
             return True

-        elif enabled == "0":
+        if enabled == "0":

             return False
 
             return False
 

-        raise PWMError(None, "Unknown enabled value: \"%s\"" % enabled)
+        raise PWMError(None, 'Unknown enabled value: "%s"' % enabled)

 
     def _set_enabled(self, value):
 
     def _set_enabled(self, value):

+        """Get or set the PWM's output enabled state.
+
+        Raises:
+            PWMError: if an I/O or OS error occurs.
+            TypeError: if value type is not bool.
+
+        :type: bool
+        """

         if not isinstance(value, bool):
             raise TypeError("Invalid enabled type, should be string.")
 
         self._write_pin_attr(self._pin_enable_path, "1" if value else "0")
 
         if not isinstance(value, bool):
             raise TypeError("Invalid enabled type, should be string.")
 
         self._write_pin_attr(self._pin_enable_path, "1" if value else "0")
 

-    """Get or set the PWM's output enabled state.
-
-    Raises:
-        PWMError: if an I/O or OS error occurs.
-        TypeError: if value type is not bool.
-
-    :type: bool
-    """
-

     # String representation
 
     def __str__(self):
     # String representation
 
     def __str__(self):

-        return "PWM%d, pin %s (freq=%f Hz, duty_cycle=%f%%)" % \
-            (self._channel, self._pin, self.frequency, self.duty_cycle * 100,)
+        return "PWM%d, pin %s (freq=%f Hz, duty_cycle=%f%%)" % (
+            self._channel,
+            self._pin,
+            self.frequency,
+            self.duty_cycle * 100,
+        )