Raspberry Pi B + python3 + flask + i2c + PWM 完了

Raspberry Pi B + python3 + flask + i2c + PWM ということで、html に slider を１６個表示して、PCA9685 の PWM を制御します。
PCA9685 は「16-channel, 12-bit PWM Fm+ I²C-bus LED controller」です。秋月やアマゾンでも製品が売られています。ここでは、表面実装タイプの素子を秋月の変換基板に乗せて使っています。
PCA9685 はアドレスを持たせることができて、１６個とか３２個とか接続可能です。ひとつで 16ch なので１６個繋げば 256 個の LED を制御できます。

前回のものに修正を加えます。

ディレクトリ構造を示します。
pwm_flask,py が主となるファイルです。

pwm_flask/
├── i2c_lcd.py
├── i2c_pwm.py
├── pwm_flask.py
├── static
│   ├── css
│   │   └── style.css
│   └── js
│       ├── jquery-3.1.1.min.js
│       └── jquery-3.1.1.min.map
└── templates
    ├── base.html
    ├── input_text_for_lcd.html
    └── set_pwm.html

これで clone して、「e55eabb」でチェックアウトすると、今回のバージョンになります。

$ git clone git@bitbucket.org:nagasako/pwm_flask.git
$ cd pwm_flask
$ git checkout e55eabb

i2c_lcd.py

#!/usr/bin/env python
# coding: UTF-8
'''
$ sudo apt-get install python-smbus i2c-tools

comment out this line in /etc/modprobe.d/raspi-blacklist.conf
blacklist i2c-bcm2708

add the following lines to /etc/modules
 i2c-dev
 i2c-bcm2708
and then reboot.

search all addr.
$ sudo i2cdetect -y 1
'''

import time
import datetime
from i2c_pwm import PWM


class MySingleton(object):
    __instance = None

    def __new__(cls):
        if cls.__instance is None:
            cls.__instance = super(MySingleton, cls).__new__(cls)
            cls.__instance.__initialized = False
            cls.__instance.socket = None
        return cls.__instance

    def __init__(self):
        if (self.__initialized):
            return
        self.__initialized = True
        self.st7032i = None
        self.pwm = None

    def init_i2c(self, raspi=True):
        if self.st7032i is None:
            self.st7032i = St7032i(0x3e, 1, 0x20, raspi)
        if self.pwm is None:
            self.pwm = PWM(0x40, False, raspi)


class St7032i(object):
    def __init__(self, addr=0x3e, ch=1, contrast=0x20, raspi=True):
        try:
            self.addr = addr
            self.ch = ch
            self.prev_now = 0
            self.raspi = raspi
            if raspi:
                import smbus
                self.bus = smbus.SMBus(ch)
                self.contrast = contrast
                self.reset()
                self.enable = True
            else:
                self.enable = False
        except:
            self.enable = False

    def reset(self):
        if not self.raspi:
            return
        contrast_h = 0x70 | (self.contrast & 0x0f)
        contrast_l = 0x54 | ((self.contrast >> 4) & 0x03)
        self.bus.write_i2c_block_data(
            self.addr, 0, [0x38, 0x39, 0x14, contrast_h, contrast_l, 0x6c])
#         // ST7032 Initial Program Code Example For 8051 MPU(8 Bit Interface) の初期化サンプルの通り
#         // Function Set : 8bit bus mode, 2-line mode,normal font,normal instruction mode
#         LCD_write(LCD_RS_CMD, 0b00111000, fd);      // 0x38
#         // Function Set : extension instruction mode
#         LCD_write(LCD_RS_CMD, 0b00111001, fd);      // 0x39
#         // Internal OSC frequency(extension instruction mode)
#         LCD_write(LCD_RS_CMD, 0b00010100, fd);      // 0x14
#         // Contrast set(extension instruction mode) コントラスト値下位4bit設定
#         LCD_write(LCD_RS_CMD, 0b01110000 | (LCD_CONTRAST & 0xF), fd); // 0x78 = 0x70 + 0x8
#         // Power/ICON/Contrast set(extension instruction mode) コントラスト値上位2bit設定
#         LCD_write(LCD_RS_CMD, 0b01011100 | ((LCD_CONTRAST >> 4) & 0x3), fd); // 0x5c + 0
#         // Follower control。internal follower on,
#         LCD_write(LCD_RS_CMD, 0b01101100, fd);      // 0x6c

        time.sleep(0.25)
        self.bus.write_i2c_block_data(self.addr, 0, [0x0c, 0x01, 0x06])
#         // Function Set。normal instruction mode。
#         // LCD_write(LCD_RS_CMD, 0b00111000, fd);   // 0x38
#         // Display On
#         LCD_write(LCD_RS_CMD, 0b00001100, fd);      // 0x0c
#         // Clear Display
#         LCD_write(LCD_RS_CMD, 0b00000001, fd);      // 0x01
#         // Entry Mode Set
#         LCD_write(LCD_RS_CMD, 0b00000110, fd);      // 0x06
        time.sleep(0.05)

    def clear(self):
        if self.enable and self.raspi:
            self.bus.write_i2c_block_data(self.addr, 0, [0x01])

    def mov_to(self, row=0, col=0):
        if self.enable and self.raspi:
            self.bus.write_i2c_block_data(self.addr, 0, [0x80 + 0x40 * row + col])

    def put_str(self, the_str):
        if self.enable and self.raspi:
            self.bus.write_i2c_block_data(self.addr, 0x40, list(map(ord, the_str)))


if __name__ == '__main__':
    try:
        my_lcd = St7032i(0x3e, 1)
        if True and my_lcd.enable:
            while True:
                time_str = datetime.datetime.now().strftime("%H:%m:%S")
                date_str = datetime.datetime.now().strftime("%y/%m/%d")
                my_lcd.clear()
                my_lcd.mov_to(0, 0)
                my_lcd.put_str(date_str)
                my_lcd.mov_to(1, 0)
                my_lcd.put_str(time_str)
                print(date_str + ' ' + time_str)
                time.sleep(1)
        # if False and my_lcd.enable:
        #     light_channel = 0
        #     adc = SpiAdc(light_channel)
        #     if adc is not None and my_lcd.enable:
        #         # Define delay between readings
        #         delay = 1
        #         while True:
        #             # Read the light sensor data
        #             light_level = adc.readChannel()
        #             # light_volts = adc.convertVolts(light_level, 2)

        #             # Print out results
        #             # print("Light: {} ({}V)".format(light_level,light_volts))
        #             # print("%4d %1.2f" % (light_level, light_volts))
        #             the_str = "    " + str(light_level)
        #             my_lcd.mov_to(0, 0)
        #             my_lcd.put_str(the_str[-4:])
        #             # Wait before repeating loop
        #             time.sleep(delay)

    except:
        print("Error accessing default I2C bus")

i2c_pwm.py

#!/usr/bin/env python
# coding: UTF-8
'''
i2c_pwm.py
'''

import time


class i2c_lib(object):
    def __init__(self, address, busnum=1, debug=False, raspi=True):
        self.address = address
        self.raspi = raspi
        if raspi:
            import smbus
            self.bus = smbus.SMBus(busnum)
        self.debug = debug

    def errMsg(self, msg):
        print("Error in %s accessing 0x%02X: Check your I2C address" % (msg, self.address))
        return -1

    def write8(self, reg, value):
        """ Writes an 8-bit value to the specified register/address """
        if not self.raspi:
            return 0
        try:
            self.bus.write_byte_data(self.address, reg, value)
            if self.debug:
                print("I2C: Wrote 0x%02X to register 0x%02X" % (value, reg))
            return 0
        except IOError:
            return self.errMsg('write8(reg=0x%02x, value=%d)' % (reg, value))

    def readU8(self, reg):
        """ Read an unsigned byte from the I2C device """
        if not self.raspi:
            return 0
        try:
            result = self.bus.read_byte_data(self.address, reg)
            if self.debug:
                print(
                    "I2C: Device 0x%02X returned 0x%02X from reg 0x%02X" % (
                        self.address, result & 0xFF, reg))
            return result
        except IOError:
            return self.errMsg('readU8(reg=0x%02x)' % (reg))


class PWM(object):
    # Registers/etc.
    __SUBADR1 = 0x02
    __SUBADR2 = 0x03
    __SUBADR3 = 0x04
    __ALLCALLADR = 0x05
    __MODE1 = 0x00
    __MODE2 = 0x01
    __PRESCALE = 0xFE
    __LED0_ON_L = 0x06
    __LED0_ON_H = 0x07
    __LED0_OFF_L = 0x08
    __LED0_OFF_H = 0x09
    __ALLLED_ON_L = 0xFA
    __ALLLED_ON_H = 0xFB
    __ALLLED_OFF_L = 0xFC
    __ALLLED_OFF_H = 0xFD

    def __init__(self, address=0x40, debug=False, raspi=True):
        self.i2c = i2c_lib(address, 1, debug, raspi)
        self.address = address
        self.debug = debug
        if self.debug:
            print("Reseting PCA9685 (%02x)" % (address))
        result = self.i2c.write8(self.__MODE1, 0x00)
        if self.debug:
            print("result = %d" % (result))
        if result < 0:
            self.i2c = None

        if self.debug:
            if self.i2c is not None:
                result = self.i2c.readU8(self.__ALLCALLADR)
                print('__ALLCALLADR = %02x' % (result))
                result = self.i2c.readU8(self.__SUBADR1)
                print('__SUBADR1 = %02x' % (result))
                result = self.i2c.readU8(self.__SUBADR2)
                print('__SUBADR2 = %02x' % (result))
                result = self.i2c.readU8(self.__SUBADR3)
                print('__SUBADR3 = %02x' % (result))

    def setPWMFreq(self, freq):
        """ Sets the PWM frequency """
        prescaleval = 25000000.0  # 25MHz
        prescaleval /= 4096.0  # 12-bit
        prescaleval /= float(freq)
        prescaleval = int(round(prescaleval)) - 1
        if self.debug:
            print("Setting PWM frequency to %d Hz" % (freq))
            print("Estimated pre-scale: %d" % (prescaleval))
        if self.debug:
            print("Final pre-scale: %d" % (prescaleval))

        oldmode = self.i2c.readU8(self.__MODE1)
        newmode = (oldmode & 0x7F) | 0x10  # sleep
        self.i2c.write8(self.__MODE1, newmode)  # go to sleep
        self.i2c.write8(self.__PRESCALE, prescaleval)
        self.i2c.write8(self.__MODE1, oldmode)
        time.sleep(0.005)
        self.i2c.write8(self.__MODE1, oldmode | 0x80)

        # self.i2c.write8(self.__MODE2, 0x10) # INVRT = 1, OUTDRV = 0
        if self.debug:
            result = self.i2c.readU8(self.__MODE1)
            print('__MODE1 = %02x' % (result))
            result = self.i2c.readU8(self.__MODE2)
            print('__MODE2 = %02x' % (result))

    def setPWM(self, channel, on, off):
        """ Sets a single PWM channel """
        self.i2c.write8(self.__LED0_ON_L + 4 * channel, on & 0xFF)
        self.i2c.write8(self.__LED0_ON_H + 4 * channel, on >> 8)
        self.i2c.write8(self.__LED0_OFF_L + 4 * channel, off & 0xFF)
        self.i2c.write8(self.__LED0_OFF_H + 4 * channel, off >> 8)
        # print('ch = %d, on = %d, off = %d' % (channel, on, off))

    def all_led_on(self):
        self.i2c.write8(self.__ALLLED_ON_L, 0x00)
        self.i2c.write8(self.__ALLLED_ON_H, 0x10)

    def all_led_off(self):
        self.i2c.write8(self.__ALLLED_OFF_L, 0x00)
        self.i2c.write8(self.__ALLLED_OFF_H, 0x10)

pwm_flask.py

#!/usr/bin/env python
# coding: UTF-8
'''
$ export FLASK_APP=pwm_flask.py
$ python -m flask run --host=0.0.0.0
'''
from flask import Flask, request, redirect
from flask import url_for, render_template, flash
# from flask import session, g, abort
from i2c_lcd import MySingleton
import datetime
import os
import json


OS_NAME = os.uname()
if 'raspberrypi' in OS_NAME:
    # Raspberry Pi
    RASPI = True
else:
    RASPI = False
print(OS_NAME, RASPI)

DEBUG = True
SECRET_KEY = 'something_secret_pwm_flask'
USERNAME = 'admin'
PASSWORD = 'adminpass'

app = Flask(__name__)
app.config.from_object(__name__)
# app.run(host='0.0.0.0', port=5000)


@app.route('/')
def input_text_for_lcd():
    return render_template('input_text_for_lcd.html')


@app.route('/send_text', methods=['POST'])
def send_text():
    date_str = datetime.datetime.now().strftime("%y/%m/%d")
    time_str = datetime.datetime.now().strftime("%H:%m:%S")
    msg_text = request.form['msg_text']
    my_singleton = MySingleton()
    my_singleton.init_i2c(RASPI)
    if RASPI:
        # my_lcd = St7032i(0x3e, 1)
        my_lcd = my_singleton.st7032i
        if my_lcd:
            my_lcd.clear()
            my_lcd.mov_to(0, 0)
            my_lcd.put_str(time_str)
            my_lcd.mov_to(1, 0)
            if msg_text:
                my_lcd.put_str(msg_text)
    else:
        print('send_text: ', msg_text)
    flash('New entry was successfully posted "' + msg_text + '" at ' + date_str + ' ' + time_str)
    return redirect(url_for('input_text_for_lcd'))


class DevPCA(object):
    def __init__(self, ch, val):
        self.ch = ch
        self.val = val


@app.route('/set_pwm', methods=['GET'])
def set_pwm_get():
    dev_list = []
    my_singleton = MySingleton()
    my_singleton.init_i2c(RASPI)
    pwm = my_singleton.pwm
    val = 0
    for ch in range(16):
        dev_pca = DevPCA(ch, val)
        dev_list.append(dev_pca)
        set_pwm(pwm, ch, val)
    return render_template('set_pwm.html', pca_channels=dev_list)


@app.route('/set_pwm_post', methods=['POST'])
def set_pwm_post():
    ch_str = request.form['ch']
    val_str = request.form['val']
    now_str = request.form['now']
    # print('set_pwm_post ch_str, val_str, now_str=', ch_str, val_str, now_str)
    ch = int(ch_str)
    val = float(val_str)
    now = int(now_str)
    my_singleton = MySingleton()
    my_singleton.init_i2c(RASPI)
    my_lcd = my_singleton.st7032i
    pwm = my_singleton.pwm
    if pwm and my_lcd and (now > my_lcd.prev_now):
        my_lcd.prev_now = now
        set_pwm(pwm, ch, val)
        my_lcd.clear()
        my_lcd.mov_to(0, 0)
        my_lcd.put_str(str(ch + 1))
        my_lcd.mov_to(1, 0)
        my_lcd.put_str(val_str)
    return json.dumps({'status': 'OK', 'ch': ch, 'val': val})


def set_pwm(pwm, ch, val):
    """ val is 0..100% float """
    width = int(4095.0 * val / 100.0)
    pwm.setPWM(ch, 0, width)


if __name__ == '__main__':
    app.run()

style.css

body {
    font-family: sans-serif;
    background: #eee;
}

base.html

<!doctype html>

<html>

<head>

  <title>PWM_Flask</title>

  <meta charset="utf-8">

  <link rel=stylesheet type=text/css href="{{ url_for('static', filename='css/style.css') }}">

  {% block head %}{% endblock %}

</head>

<body>

  {% block body %}{% endblock %}

</body>

  {% block bottom %}{% endblock %}

</html>

input_text_for_lcd.html

{% extends "base.html" %}
{% block body %}
  <form action="{{ url_for('send_text') }}" method=post>
    <dl>
      <dt>Message:
      <dd><input type="text" size="30" name="msg_text">
      <dd><input type="submit" value="send">
    </dl>
  </form>
  <div>
    <a href="{{ url_for('set_pwm_get') }}">set_pwm</a>
  </div>
  <div>
  {% with messages = get_flashed_messages() %}
    {% if messages %}
      <ul class=flashes>
      {% for message in messages %}
        <li>{{ message }}</li>
      {% endfor %}
      </ul>
    {% endif %}
  {% endwith %}
  </div>
{% endblock %}

set_pwm.html

{% extends "base.html" %}
{% block head %}
  <style>
    .range_class {
      width: 400px!important;
    }
  </style>
  <script src="{{ url_for('static', filename='js/jquery-3.1.1.min.js') }}"></script>
{% endblock %}

{% block body %}
  <form action="" method=post>
    <table>
      <thead>
        <tr>
          <th> ch </th>
          <th> PCA9685 value </th>
        </tr>
      </thead>
      <tbody>
      {% for pca_ch in pca_channels %}
        <tr>
          <td>{{ pca_ch.ch + 1 }}</td>
          <td>
            <input type="range" name="{{ pca_ch.ch }}" class="range_class"
                value="{{ pca_ch.val }}" min="0" max="100" step="0.1">
          </td>
        </tr>
      {% endfor %}
      </tbody>
    </table>
  </form>

  <div>
    <span id="pwm_ch"></span>
    <span id="pwm_val"></span>
  </div>

  <div>
    <a href="{{ url_for('input_text_for_lcd') }}">input_text_for_lcd</a>
  </div>

  <div>
  {% with messages = get_flashed_messages() %}
    {% if messages %}
      <ul class=flashes>
      {% for message in messages %}
        <li>{{ message }}</li>
      {% endfor %}
      </ul>
    {% endif %}
  {% endwith %}
  </div>
{% endblock %}

{% block bottom %}
  <script type="text/javascript">
  <!--
    $('.range_class').on('input', function() {
      var the_name = $(this).attr('name');
      var the_ch = +the_name + 1;
      var the_value = $(this).val();
      var now = Date.now();
      var msg = the_ch + ' value=' + the_value;
      var data_str = 'ch=' + the_name;
      data_str += '&val=' + the_value;
      data_str += '&now=' + now;
      $('#pwm_ch').text(the_ch);
      $('#pwm_val').text(the_value);
      // console.log(msg);
      $.ajax({
        type: "POST",
        url: "{{ url_for('set_pwm_post') }}",
        data: data_str,
        success: function(resp) {
          // console.log('success:' + resp);
        },
        error: function(resp) {
          console.log('error:' + resp);
        }
      });
    });
  // -->
  </script>
{% endblock %}

実行コマンドはこれです。

pi~/Documents/env_flask/proj/pwm_flask:develop $ export FLASK_APP=pwm_flask.py
pi~/Documents/env_flask/proj/pwm_flask:develop $ python -m flask run --host=0.0.0.0

python3 でも動作します。

Lock() はやめて Singleton にしました。
i2c_lcd.py の「class MySingleton(object)」がそれです。