snap7.py

Go to the documentation of this file.

# -*- coding: utf-8; -*-
#
# Copyright (c) 2016 Álan Crístoffer
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.

import ahio.abstract_driver

from enum import Enum

try:
    import snap7
except Exception:
    ahioDriverInfo.AVAILABLE = False


class ahioDriverInfo(ahio.abstract_driver.AbstractahioDriverInfo):
    NAME = 'snap7'
    AVAILABLE = True


class Driver(ahio.abstract_driver.AbstractDriver):
    _client = None

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        if self._client:
            self._client.disconnect()
        self._client = None

    
    def setup(self, address, rack=0, slot=1, port=102):
        self._client = snap7.client.Client()
        self._client.connect(address, rack, slot, port)

    def available_pins(self):
        return []

    def _set_pin_direction(self, pin, direction):
        d = self._pin_direction(pin)
        if direction != d and not (type(d) is list and direction in d):
            raise RuntimeError('Port %s does not support this Direction' % pin)

    def _pin_direction(self, pin):
        return {
            'D': [ahio.Direction.Input, ahio.Direction.Output],
            'M': [ahio.Direction.Input, ahio.Direction.Output],
            'Q': ahio.Direction.Output,
            'I': ahio.Direction.Input
        }[pin[0].upper()]

    def _set_pin_type(self, pin, ptype):
        raise RuntimeError('Hardware does not support changing the pin type')

    def _pin_type(self, pin):
        raise RuntimeError('Hardware does not support querying the pin type')

    def _write(self, pin, value, pwm):
        if pwm:
            raise RuntimeError('Pin does not support PWM')
        if self._pin_direction(pin) == ahio.Direction.Input:
            raise RuntimeError('Can not write to Input')
        mem = self._parse_port_name(pin)
        value = {
            ahio.LogicValue.High: 1,
            ahio.LogicValue.Low: 0
        }.get(value, value)
        self._set_memory(mem, value)

    def _read(self, pin):
        mem = self._parse_port_name(pin)
        value = self._get_memory(mem)
        if mem[1] == 'X':
            return {
                0: ahio.LogicValue.Low,
                1: ahio.LogicValue.High
            }.get(value, value)
        else:
            return value

    def analog_references(self):
        return []

    def _set_analog_reference(self, reference, pin):
        raise RuntimeError('Hardware does not support setting analog ref')

    def _analog_reference(self, pin):
        pass

    def _set_pwm_frequency(self, frequency, pin):
        raise RuntimeError(
            'Setting PWM frequency is not supported by hardware')

    def _parse_port_name(self, s):
        s = s.upper()
        area = {
            'D': snap7.snap7types.S7AreaDB,
            'M': snap7.snap7types.S7AreaMK,
            'Q': snap7.snap7types.S7AreaPA,
            'I': snap7.snap7types.S7AreaPE
        }[s[0]]
        length = {'X': 1, 'B': 1, 'W': 2, 'D': 4}[s[1]]
        start = int(s.split('.')[0][2:])
        bit = int(s.split('.')[1]) if s[1] == 'X' else None
        dtype = {
            'X': {
                'get': lambda m: snap7.util.get_bool(m, 0, bit),
                'set': lambda m, v: snap7.util.set_bool(m, 0, bit, v)
            },
            'B': {
                'get': lambda m: snap7.util.get_int(m, 0),
                'set': lambda m, v: snap7.util.set_int(m, 0, v)
            },
            'W': {
                'get': lambda m: snap7.util.get_int(m, 0),
                'set': lambda m, v: snap7.util.set_int(m, 0, v)
            },
            'D': {
                'get': lambda m: snap7.util.get_dword(m, 0),
                'set': lambda m, v: snap7.util.set_dword(m, 0, v)
            }
        }[s[1]]
        return (area, dtype, start, length)

    def _get_memory(self, mem):
        m = self._client.read_area(mem[0], 0, mem[2], mem[3])
        return mem[1]['get'](m)

    def _set_memory(self, mem, value):
        m = self._client.read_area(mem[0], 0, mem[2], mem[3])
        mem[1]['set'](m, value)
        self._client.write_area(mem[0], 0, mem[2], m)