v0.5 SaltStack proxyminion ESP8266 micropython

README.md

Me asking: https://groups.google.com/forum/#!topic/salt-users/wUCFRcbTexg https://forum.micropython.org/viewtopic.php?f=16&t=1652&start=20#p13968

mpfshell: https://github.com/wendlers/mpfshell/blob/master/mp/mpfexp.py

_modules micropython_esp8266_mpspecific.py

# -*- coding: utf-8 -*-
'''
Provide the module for the MicroPython-specific libraries of a proxy-minion.
'''

from __future__ import absolute_import

# Import python libs
import logging
import salt.utils

log = logging.getLogger(__name__)

__proxyenabled__ = ['micropython_esp8266']
__virtualname__ = 'micropython_esp8266_mpspecific'


def __virtual__():
    '''
    Only work on systems that are a proxy minion
    '''
    log.debug('micropython_esp8266_mpspecific proxy __virtual__() called...')
    try:
        if salt.utils.is_proxy() and __opts__['proxy']['proxytype'] == 'micropython_esp8266':
            return __virtualname__
    except KeyError:
        return (False, 'The micropython_esp8266_mpspecific execution module failed to load.  Check the proxy key in pillar.')

    return (False, 'The micropython_esp8266_mpspecific execution module failed to load: only works on a micropython_esp8266 proxy minion.')
#

#
def check_firmware():
    return __proxy__['micropython_esp8266.check_firmware']()

_proxy micropython_esp8266.py

# -*- coding: utf-8 -*-
'''
This is a proxy minion that connects to a ESP8266 running micropython firmware.
'''
from __future__ import absolute_import

# Import python libs
import logging
import salt.utils.http

# Import 3rd-party libs
try:
    HAS_MPFSHELL = True
    # wraps pyboard with 
    import mp.mpfexp
except ImportError:
    HAS_MPFSHELL = False

__proxyenabled__ = ['micropython_esp8266']
__virtualname__ = 'micropython_esp8266'


GRAINS_CACHE = {}
DETAILS = {}


log = logging.getLogger(__file__)

#
def __virtual__():
#    '''
#    Only return if all the modules are available
#    '''
    if not HAS_MPFSHELL:
        return False, 'Missing dependency mpfshell https://github.com/wendlers/mpfshell'
    return __virtualname__

#def __virtual__():
#    '''
#    Only return if all the modules are available
#    '''
#    log.info('nxos proxy __virtual__() called...')
#
#    if __opts__.get('proxy_merge_grains_in_module', False) is False:
#        salt.utils.warn_until(
#            'Nitrogen',
#            'To use grains with the NXOS proxy minion, '
#            '`proxy_merge_grains_in_module: True` must be set in the '
#            'proxy minion config.'
#        )
#
#    return __virtualname__
#

def init(opts=None):
    '''
    Initialize the library and enter micropythons raw-REPL mode which takes bytes and doesn't echo back.
    '''
    #get option connection string
    DETAILS['conntype'] = opts['proxy'].get('conntype')
    # DETAILS['devicefile'] = opts['proxy'].get('devicefile')
    # DETAILS['port'] = opts['proxy'].get('port') #serial
    # DETAILS['baudrate'] = opts['proxy'].get('baudrate') #serial
    # DETAILS['ip'] = opts['proxy'].get('ip') #telnet, websocket
    # DETAILS['user'] = opts['proxy'].get('user') #telnet
    # DETAILS['password'] = opts['proxy'].get('password') #telnet, websocket
    #try to connect
    DETAILS['pyboard'] = mp.mpfexp.MpFileExplorer('ser:/dev/ttyUSB0') #devicefile
    DETAILS['pyboard'].enter_raw_repl()
    #if successful
    # do whatever
    #else return false
    #
    #DETAILS['initialized'] = True

#def init(opts=None):
#    '''
#    Required.
#    Can be used to initialize the server connection.
#    '''
#    if opts is None:
#        opts = __opts__
#    try:
#        DETAILS[_worker_name()] = SSHConnection(
#            host=opts['proxy']['host'],
#            username=opts['proxy']['username'],
#            password=opts['proxy']['password'],
#            key_accept=opts['proxy'].get('key_accept', False),
#            ssh_args=opts['proxy'].get('ssh_args', ''),
#            prompt='{0}.*#'.format(opts['proxy']['prompt_name']))
#        out, err = DETAILS[_worker_name()].sendline('terminal length 0')
#
#    except TerminalException as e:
#        log.error(e)
#        return False
#    DETAILS['initialized'] = True
#
#
def initialized():
    return True
#def initialized():
#    return DETAILS.get('initialized', False)
#
#
def ping():
    return True
#def ping():
#    '''
#    Required.
#    Ping the device on the other end of the connection
#    .. code-block: bash
#        salt '*' nxos.cmd ping
#    '''
#    if _worker_name() not in DETAILS:
#        init()
#    try:
#        return DETAILS[_worker_name()].conn.isalive()
#    except TerminalException as e:
#        log.error(e)
#        return False
#
#
def shutdown(opts):
    '''
    Required.
    Disconnect from the board.
    '''
    DETAILS['pyboard'].exit_raw_repl()
    return True
#def shutdown(opts):
#    '''
#    Required.
#    Disconnect
#    '''
#    pass
#    #DETAILS[_worker_name()].close_connection()





def grains():
    '''
    Get the grains from the proxy device.
    '''
    if not GRAINS_CACHE:
        return _grains() #maybe use connection settings here?
    return GRAINS_CACHE


def grains_refresh():
    '''
    Refresh the grains from the proxy device.
    '''
    GRAINS_CACHE = {}
    return grains()

#def _grains(host, protocol=None, port=None):
def _grains():
    '''
    Helper function to the grains from the proxied device.
    '''
    ret = {}
    ret['machine.freq'] = '79999'
    ret['machine.unique_id'] = '4711'
    ret['esp.flash_id'] = '0815'
    #todo write a ecex_-eable script to return a dictionary
    #ret['machine.freq'] = DETAILS['pyboard'].machine.freq()
    #ret['machine.unique_id'] = DETAILS['pyboard'].machine.unique_id()
    #ret['esp.flash_id'] = DETAILS['pyboard'].esp.flash_id()
    GRAINS_CACHE.update(ret)
    return GRAINS_CACHE
#

def check_firmware():
    ret = DETAILS['pyboard'].exec_(
        "import esp, ubinascii\r\n"
        "ret['machine_id'] = ubinascii.hexlify(machine.unique_id()).decode('utf-8')\r\n"
        "print(ret)"
    )
    return ret
#

grainstest.py

import esp, os, sys, machine, network, ubinascii, json

ret = {}

ret['fw'] = sys.implementation.name + '_' + sys.platform
ret['fw_family'] = sys.implementation.name
ret['fw_platform'] = sys.platform
ret['fwmajorrelease'] = sys.implementation.version[0]
ret['fwrelease'] = '.'.join(map(str, sys.implementation.version))
ret['fwrelease_info'] = sys.implementation.version

if network.WLAN(network.STA_IF).active() or network.WLAN(network.AP_IF).active():
    ret['ip4_interfaces'] = {}
    ret['ipv4'] = {}

if network.WLAN(network.STA_IF).active() and network.WLAN(network.STA_IF).ifconfig()[0] is not '0.0.0.0':
    ret['ip4_interfaces']['sta'] = network.WLAN(network.STA_IF).ifconfig()[0]
    ret['ipv4']['sta'] = network.WLAN(network.STA_IF).ifconfig()[0]

if network.WLAN(network.AP_IF).active() and network.WLAN(network.AP_IF).ifconfig()[0] is not '0.0.0.0':
    ret['ip4_interfaces']['ap'] = network.WLAN(network.AP_IF).ifconfig()[0]
    ret['ipv4']['ap'] = network.WLAN(network.AP_IF).ifconfig()[0]

ret['machine_id'] = ubinascii.hexlify(machine.unique_id()).decode("utf-8")
ret['mem_total'] = esp.flash_size()
ret['flash_id'] = esp.flash_id()
ret['freq'] = machine.freq()

has_hostname = False
try:
    with open('/etc/hostname', 'r') as f:
        firstline = f.readline().strip()
        has_hostname = True
except:
    pass

try:
    with open('/hostname', 'r') as f:
        firstline = f.readline().strip()
        has_hostname = True
except:
    pass

if has_hostname:
    ret['host'] = firstline
    ret['nodename'] = firstline
    ret['id'] = firstline
    ret['fqdn'] = firstline


ret['pythonversion'] = sys.version_info
ret['serialnumber'] = ubinascii.hexlify(machine.unique_id()).decode("utf-8")
ret['virtual'] = 'physical'
ret['cpu_model'] = 'Xtensa lx106'

ret['hwaddr_interfaces'] = {}
addr = ubinascii.hexlify(network.WLAN(network.STA_IF).config('mac')).decode("utf-8")
ret['hwaddr_interfaces']['sta'] = ':'.join(addr[i:i+2] for i in range(0,len(addr),2))
addr = None

addr = ubinascii.hexlify(network.WLAN(network.AP_IF).config('mac')).decode("utf-8")
ret['hwaddr_interfaces']['ap'] = ':'.join(addr[i:i+2] for i in range(0,len(addr),2))



print(json.dumps(ret))

sample_output.txt

#ampy https://github.com/adafruit/ampy
cpr@T500-2016:~/ESP8266/ampytest$ ampy --port /dev/ttyUSB0 run grainstest.py 
{"fw": "micropython_esp8266", "fwrelease_info": [1, 8, 3], "virtual": "physical", "flash_id": 1458376, "pythonversion": [3, 4, 0], "id": "micropython_proxyminion1", "fw_platform": "esp8266", "cpu_model": "Xtensa lx106", "serialnumber": "052df300", "hwaddr_interfaces": {"ap": "1a:fe:34:f3:2d:05", "sta": "18:fe:34:f3:2d:05"}, "fwmajorrelease": 1, "nodename": "micropython_proxyminion1", "host": "micropython_proxyminion1", "ip4_interfaces": {"ap": "192.168.4.1", "sta": "10.xxx.yyy.zzz"}, "machine_id": "052df300", "fqdn": "micropython_proxyminion1", "fw_family": "micropython", "mem_total": 4194304, "freq": 80000000, "fwrelease": "1.8.3", "ipv4": {"ap": "192.168.4.1", "sta": "10.xxx.yyy.zzz"}}


cpr@T500-2016:~/ESP8266/mpfshell$ sudo salt 'micropython_proxyminion1' test.ping 
micropython_proxyminion1:
    True
cpr@T500-2016:~/ESP8266/mpfshell$ sudo salt 'micropython_proxyminion1' micropython_esp8266_mpspecific.check_firmware
micropython_proxyminion1:
    size: 530440
    md5: 4a5aaa2bcc8f5e2b77090fdef035d93e
    True
cpr@T500-2016:~/ESP8266/mpfshell$