revpi/revpimodio2
1
0
Fork 0
mirror of https://github.com/naruxde/revpimodio2.git synced 2025-11-08 13:53:53 +01:00
Code Issues Packages Releases Wiki Activity

new project started

Browse Source
This commit is contained in:
Sven Sager
2017-08-13 12:05:23 +02:00
commit 8105323d18
11 changed files with 3528 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
.hgignore Normal file
View File

@@ -0,0 +1,19 @@
glob:.eric6project
glob:_eric6project
glob:.eric5project
glob:_eric5project
glob:.eric4project
glob:_eric4project
glob:.ropeproject
glob:_ropeproject
glob:.directory
glob:**.pyc
glob:**.pyo
glob:**.orig
glob:**.bak
glob:**.rej
glob:**~
glob:cur
glob:tmp
glob:__pycache__
glob:**.DS_Store

2
MANIFEST.in Normal file
View File

@@ -0,0 +1,2 @@
global-exclude test/*
global-exclude *.pyc

336
revpimodio2.e4p Normal file
View File

@@ -0,0 +1,336 @@
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE Project SYSTEM "Project-5.1.dtd">
<!-- eric project file for project revpimodio2 -->
<!-- Saved: 2017-08-13, 12:05:22 -->
<!-- Copyright (C) 2017 Sven Sager, akira@narux.de -->
<Project version="5.1">
<Language>en_US</Language>
<Hash>7ea159534ad3516e9069331120048abf9b00151e</Hash>
<ProgLanguage mixed="0">Python3</ProgLanguage>
<ProjectType>Console</ProjectType>
<Description></Description>
<Version>2.0.0</Version>
<Author>Sven Sager</Author>
<Email>akira@narux.de</Email>
<Eol index="-1"/>
<Sources>
<Source>setup.py</Source>
<Source>revpimodio2/modio.py</Source>
<Source>revpimodio2/summary.py</Source>
<Source>revpimodio2/app.py</Source>
<Source>revpimodio2/io.py</Source>
<Source>revpimodio2/__init__.py</Source>
<Source>revpimodio2/device.py</Source>
<Source>revpimodio2/helper.py</Source>
</Sources>
<Forms/>
<Translations/>
<Resources/>
<Interfaces/>
<Others>
<Other>doc</Other>
<Other>.hgignore</Other>
</Others>
<Vcs>
<VcsType>Mercurial</VcsType>
<VcsOptions>
<dict>
<key>
<string>add</string>
</key>
<value>
<list>
<string></string>
</list>
</value>
<key>
<string>checkout</string>
</key>
<value>
<list>
<string></string>
</list>
</value>
<key>
<string>commit</string>
</key>
<value>
<list>
<string></string>
</list>
</value>
<key>
<string>diff</string>
</key>
<value>
<list>
<string></string>
</list>
</value>
<key>
<string>export</string>
</key>
<value>
<list>
<string></string>
</list>
</value>
<key>
<string>global</string>
</key>
<value>
<list>
<string></string>
</list>
</value>
<key>
<string>history</string>
</key>
<value>
<list>
<string></string>
</list>
</value>
<key>
<string>log</string>
</key>
<value>
<list>
<string></string>
</list>
</value>
<key>
<string>remove</string>
</key>
<value>
<list>
<string></string>
</list>
</value>
<key>
<string>status</string>
</key>
<value>
<list>
<string></string>
</list>
</value>
<key>
<string>tag</string>
</key>
<value>
<list>
<string></string>
</list>
</value>
<key>
<string>update</string>
</key>
<value>
<list>
<string></string>
</list>
</value>
</dict>
</VcsOptions>
<VcsOtherData>
<dict/>
</VcsOtherData>
</Vcs>
<FiletypeAssociations>
<FiletypeAssociation pattern="*.idl" type="INTERFACES"/>
<FiletypeAssociation pattern="*.py" type="SOURCES"/>
<FiletypeAssociation pattern="*.py3" type="SOURCES"/>
<FiletypeAssociation pattern="*.pyw" type="SOURCES"/>
<FiletypeAssociation pattern="*.pyw3" type="SOURCES"/>
</FiletypeAssociations>
<Documentation>
<DocumentationParams>
<dict>
<key>
<string>ERIC4API</string>
</key>
<value>
<dict>
<key>
<string>ignoreDirectories</string>
</key>
<value>
<list>
<string>deb</string>
<string>dist</string>
<string>doc</string>
<string>test</string>
</list>
</value>
<key>
<string>ignoreFilePatterns</string>
</key>
<value>
<list>
<string>setup.py</string>
</list>
</value>
<key>
<string>languages</string>
</key>
<value>
<list>
<string>Python3</string>
</list>
</value>
<key>
<string>outputFile</string>
</key>
<value>
<string>eric-revpimodio.api</string>
</value>
<key>
<string>useRecursion</string>
</key>
<value>
<bool>True</bool>
</value>
</dict>
</value>
<key>
<string>ERIC4DOC</string>
</key>
<value>
<dict>
<key>
<string>ignoreDirectories</string>
</key>
<value>
<list>
<string>deb</string>
<string>dist</string>
<string>doc</string>
<string>test</string>
</list>
</value>
<key>
<string>ignoreFilePatterns</string>
</key>
<value>
<list>
<string>setup.py</string>
</list>
</value>
<key>
<string>noindex</string>
</key>
<value>
<bool>True</bool>
</value>
<key>
<string>outputDirectory</string>
</key>
<value>
<string>doc</string>
</value>
<key>
<string>qtHelpEnabled</string>
</key>
<value>
<bool>False</bool>
</value>
<key>
<string>sourceExtensions</string>
</key>
<value>
<list>
<string></string>
</list>
</value>
<key>
<string>useRecursion</string>
</key>
<value>
<bool>True</bool>
</value>
</dict>
</value>
</dict>
</DocumentationParams>
</Documentation>
<Checkers>
<CheckersParams>
<dict>
<key>
<string>Pep8Checker</string>
</key>
<value>
<dict>
<key>
<string>DocstringType</string>
</key>
<value>
<string>pep257</string>
</value>
<key>
<string>ExcludeFiles</string>
</key>
<value>
<string></string>
</value>
<key>
<string>ExcludeMessages</string>
</key>
<value>
<string>E123,E226,E24</string>
</value>
<key>
<string>FixCodes</string>
</key>
<value>
<string></string>
</value>
<key>
<string>FixIssues</string>
</key>
<value>
<bool>False</bool>
</value>
<key>
<string>HangClosing</string>
</key>
<value>
<bool>False</bool>
</value>
<key>
<string>IncludeMessages</string>
</key>
<value>
<string></string>
</value>
<key>
<string>MaxLineLength</string>
</key>
<value>
<int>79</int>
</value>
<key>
<string>NoFixCodes</string>
</key>
<value>
<string>E501</string>
</value>
<key>
<string>RepeatMessages</string>
</key>
<value>
<bool>True</bool>
</value>
<key>
<string>ShowIgnored</string>
</key>
<value>
<bool>False</bool>
</value>
</dict>
</value>
</dict>
</CheckersParams>
</Checkers>
</Project>

34
revpimodio2/__init__.py Normal file
View File

@@ -0,0 +1,34 @@
#
# python3-RevPiModIO
#
# Webpage: https://revpimodio.org/
# (c) Sven Sager, License: LGPLv3
#
# -*- coding: utf-8 -*-
"""Stellt alle Klassen fuer den RevolutionPi zur Verfuegung.
Stellt Klassen fuer die einfache Verwendung des Revolution Pis der
Kunbus GmbH (https://revolution.kunbus.de/) zur Verfuegung. Alle I/Os werden
aus der piCtory Konfiguration eingelesen und mit deren Namen direkt zugreifbar
gemacht. Fuer Gateways sind eigene IOs ueber mehrere Bytes konfigurierbar
Mit den definierten Namen greift man direkt auf die gewuenschten Daten zu.
Auf alle IOs kann der Benutzer Funktionen als Events registrieren. Diese
fuehrt das Modul bei Datenaenderung aus.
"""
import warnings
from .modio import *
__all__ = ["RevPiModIO", "RevPiModIOSelected", "RevPiModIODriver"]
__author__ = "Sven Sager <akira@revpimodio.org>"
__version__ = "2.0.0"
# Global package values
OFF = 0
GREEN = 1
RED = 2
RISING = 31
FALLING = 32
BOTH = 33
warnings.simplefilter(action="always")

22
revpimodio2/app.py Normal file
View File

@@ -0,0 +1,22 @@
#
# python3-RevPiModIO
#
# Webpage: https://revpimodio.org/
# (c) Sven Sager, License: LGPLv3
#
# -*- coding: utf-8 -*-
"""Bildet die App Sektion von piCtory ab."""
class App(object):
"""Bildet die App Sektion der config.rsc ab."""
def __init__(self, app):
"""Instantiiert die App-Klasse.
@param app piCtory Appinformationen"""
self.name = app["name"]
self.version = app["version"]
self.language = app["language"]
# TODO: Layout untersuchen und anders abbilden
self.layout = app["layout"]

1170
revpimodio2/device.py Normal file
View File

File diff suppressed because it is too large Load Diff

355
revpimodio2/helper.py Normal file
View File

@@ -0,0 +1,355 @@
#
# python3-RevPiModIO
#
# Webpage: https://revpimodio.org/
# (c) Sven Sager, License: LGPLv3
#
# -*- coding: utf-8 -*-
import warnings
from threading import Event, Lock, Thread
from timeit import default_timer
class EventCallback(Thread):
"""Thread fuer das interne Aufrufen von Event-Funktionen.
Der Eventfunktion, welche dieser Thread aufruft, wird der Thread selber
als Parameter uebergeben. Darauf muss bei der definition der Funktion
geachtet werden z.B. "def event(th):". Bei umfangreichen Funktionen kann
dieser ausgewertet werden um z.B. doppeltes Starten zu verhindern.
Ueber EventCallback.ioname kann der Name des IO-Objekts abgerufen werden,
welches das Event ausgeloest hast. EventCallback.iovalue gibt den Wert des
IO-Objekts zum Ausloesezeitpunkt zurueck.
Der Thread stellt das EventCallback.exit Event als Abbruchbedingung fuer
die aufgerufene Funktion zur Verfuegung.
Durch Aufruf der Funktion EventCallback.stop() wird das exit-Event gesetzt
und kann bei Schleifen zum Abbrechen verwendet werden.
Mit dem .exit() Event auch eine Wartefunktion realisiert
werden: "th.exit.wait(0.5)" - Wartet 500ms oder bricht sofort ab, wenn
fuer den Thread .stop() aufgerufen wird.
while not th.exit.is_set():
# IO-Arbeiten
th.exit.wait(0.5)
"""
def __init__(self, func, name, value):
"""Init EventCallback class.
@param func Funktion die beim Start aufgerufen werden soll
@param name IO-Name
@param value IO-Value zum Zeitpunkt des Events
"""
super().__init__()
self.daemon = True
self.exit = Event()
self.func = func
self.ioname = name
self.iovalue = value
def run(self):
"""Ruft die registrierte Funktion auf."""
self.func(self)
def stop(self):
"""Setzt das exit-Event mit dem die Funktion beendet werden kann."""
self.exit.set()
class Cycletools():
"""Werkzeugkasten fuer Cycleloop-Funktion.
Diese Klasse enthaelt Werkzeuge fuer Zyklusfunktionen, wie Taktmerker
und Flankenmerker.
Zu beachten ist, dass die Flankenmerker beim ersten Zyklus alle den Wert
True haben! Ueber den Merker Cycletools.first kann ermittelt werden,
ob es sich um den ersten Zyklus handelt.
Taktmerker flag1c, flag5c, flag10c, usw. haben den als Zahl angegebenen
Wert an Zyklen jeweils False und True.
Beispiel: flag5c hat 5 Zyklen den Wert False und in den naechsten 5 Zyklen
den Wert True.
Flankenmerker flank5c, flank10c, usw. haben immer im, als Zahl angebenen
Zyklus fuer einen Zyklusdurchlauf den Wert True, sonst False.
Beispiel: flank5c hat immer alle 5 Zyklen den Wert True.
Diese Merker koennen z.B. verwendet werden um, an Outputs angeschlossene,
Lampen synchron blinken zu lassen.
"""
def __init__(self):
"""Init Cycletools class."""
self.__cycle = 0
self.__ucycle = 0
self.__dict_ton = {}
self.__dict_tof = {}
self.__dict_tp = {}
# Taktmerker
self.first = True
self.flag1c = False
self.flag5c = False
self.flag10c = False
self.flag15c = False
self.flag20c = False
# Flankenmerker
self.flank5c = True
self.flank10c = True
self.flank15c = True
self.flank20c = True
def _docycle(self):
"""Zyklusarbeiten."""
# Einschaltverzoegerung
for tof in self.__dict_tof:
if self.__dict_tof[tof] > 0:
self.__dict_tof[tof] -= 1
# Ausschaltverzoegerung
for ton in self.__dict_ton:
if self.__dict_ton[ton][1]:
if self.__dict_ton[ton][0] > 0:
self.__dict_ton[ton][0] -= 1
self.__dict_ton[ton][1] = False
else:
self.__dict_ton[ton][0] = -1
# Impuls
for tp in self.__dict_tp:
if self.__dict_tp[tp][1]:
if self.__dict_tp[tp][0] > 0:
self.__dict_tp[tp][0] -= 1
else:
self.__dict_tp[tp][1] = False
else:
self.__dict_tp[tp][0] = -1
# Flankenmerker
self.flank5c = False
self.flank10c = False
self.flank15c = False
self.flank20c = False
# Logische Flags
self.first = False
self.flag1c = not self.flag1c
# Berechnete Flags
self.__cycle += 1
if self.__cycle == 5:
self.__ucycle += 1
if self.__ucycle == 3:
self.flank15c = True
self.flag15c = not self.flag15c
self.__ucycle = 0
if self.flag5c:
if self.flag10c:
self.flank20c = True
self.flag20c = not self.flag20c
self.flank10c = True
self.flag10c = not self.flag10c
self.flank5c = True
self.flag5c = not self.flag5c
self.__cycle = 0
def get_tofc(self, name):
"""Wert der Ausschaltverzoegerung.
@param name Eindeutiger Name des Timers
@return Wert der Ausschaltverzoegerung"""
return self.__dict_tof.get(name, 0) > 0
def set_tofc(self, name, cycles):
"""Startet bei Aufruf einen ausschaltverzoegerten Timer.
@param name Eindeutiger Name fuer Zugriff auf Timer
@param cycles Zyklusanzahl, der Verzoegerung wenn nicht neu gestartet
"""
self.__dict_tof[name] = cycles
def get_tonc(self, name):
"""Einschaltverzoegerung.
@param name Eindeutiger Name des Timers
@return Wert der Einschaltverzoegerung"""
return self.__dict_ton.get(name, [-1])[0] == 0
def set_tonc(self, name, cycles):
"""Startet einen einschaltverzoegerten Timer.
@param name Eindeutiger Name fuer Zugriff auf Timer
@param cycles Zyklusanzahl, der Verzoegerung wenn neu gestartet
"""
if self.__dict_ton.get(name, [-1])[0] == -1:
self.__dict_ton[name] = [cycles, True]
else:
self.__dict_ton[name][1] = True
def get_tpc(self, name):
"""Impulstimer.
@param name Eindeutiger Name des Timers
@return Wert der des Impulses"""
return self.__dict_tp.get(name, [-1])[0] > 0
def set_tpc(self, name, cycles):
"""Startet einen Impuls Timer.
@param name Eindeutiger Name fuer Zugriff auf Timer
@param cycles Zyklusanzahl, die der Impuls anstehen soll
"""
if self.__dict_tp.get(name, [-1])[0] == -1:
self.__dict_tp[name] = [cycles, True]
else:
self.__dict_tp[name][1] = True
class ProcimgWriter(Thread):
"""Klasse fuer Synchroniseriungs-Thread.
Diese Klasse wird als Thread gestartet, wenn das Prozessabbild zyklisch
synchronisiert werden soll. Diese Funktion wird hauptsaechlich fuer das
Event-Handling verwendet.
"""
def __init__(self, parentmodio):
"""Init ProcimgWriter class.
@param parentmodio Parent Object"""
super().__init__()
self._adjwait = 0
self._ioerror = 0
self._modio = parentmodio
self._refresh = 0.05
self._work = Event()
self.daemon = True
self.lck_refresh = Lock()
self.maxioerrors = 0
self.newdata = Event()
def _gotioerror(self):
"""IOError Verwaltung fuer auto_refresh."""
self._ioerror += 1
if self.maxioerrors != 0 and self._ioerror >= self.maxioerrors:
raise RuntimeError(
"reach max io error count {} on process image".format(
self.maxioerrors
)
)
warnings.warn(
"count {} io errors on process image".format(self._ioerror),
RuntimeWarning
)
def get_refresh(self):
"""Gibt Zykluszeit zurueck.
@return int() Zykluszeit in Millisekunden"""
return int(self._refresh * 1000)
def run(self):
"""Startet die automatische Prozessabbildsynchronisierung."""
fh = self._modio._create_myfh()
self._adjwait = self._refresh
while not self._work.is_set():
ot = default_timer()
# Lockobjekt holen und Fehler werfen, wenn nicht schnell genug
if not self.lck_refresh.acquire(timeout=self._adjwait):
warnings.warn(
"cycle time of {} ms exceeded on lock".format(
int(self._refresh * 1000)
),
RuntimeWarning
)
continue
try:
fh.seek(0)
bytesbuff = bytearray(fh.read(self._modio._length))
except IOError:
self._gotioerror()
self.lck_refresh.release()
self._work.wait(self._adjwait)
continue
if self._modio._monitoring:
# Inputs und Outputs in Puffer
for dev in self._modio._lst_refresh:
dev._filelock.acquire()
dev._ba_devdata[:] = bytesbuff[dev.slc_devoff]
dev._filelock.release()
else:
# Inputs in Puffer, Outputs in Prozessabbild
ioerr = False
for dev in self._modio._lst_refresh:
dev._filelock.acquire()
dev._ba_devdata[dev.slc_inp] = bytesbuff[dev.slc_inpoff]
try:
fh.seek(dev.slc_outoff.start)
fh.write(dev._ba_devdata[dev.slc_out])
except IOError:
ioerr = True
finally:
dev._filelock.release()
if self._modio._buffedwrite:
try:
fh.flush()
except IOError:
ioerr = True
if ioerr:
self._gotioerror()
self.lck_refresh.release()
self._work.wait(self._adjwait)
continue
self.lck_refresh.release()
# Alle aufwecken
self.newdata.set()
self._work.wait(self._adjwait)
# Wartezeit anpassen um echte self._refresh zu erreichen
if default_timer() - ot >= self._refresh:
self._adjwait -= 0.001
if self._adjwait < 0:
warnings.warn(
"cycle time of {} ms exceeded".format(
int(self._refresh * 1000)
),
RuntimeWarning
)
self._adjwait = 0
else:
self._adjwait += 0.001
# Alle am Ende erneut aufwecken
self.newdata.set()
fh.close()
def stop(self):
"""Beendet die automatische Prozessabbildsynchronisierung."""
self._work.set()
def set_refresh(self, value):
"""Setzt die Zykluszeit in Millisekunden.
@param value int() Millisekunden"""
if value >= 10 and value < 2000:
self._refresh = value / 1000
self._adjwait = self._refresh
else:
raise ValueError(
"refresh time must be 10 to 2000 milliseconds"
)
refresh = property(get_refresh, set_refresh)

631
revpimodio2/io.py Normal file
View File

@@ -0,0 +1,631 @@
#
# python3-RevPiModIO
#
# Webpage: https://revpimodio.org/
# (c) Sven Sager, License: LGPLv3
#
# -*- coding: utf-8 -*-
import struct
from .__init__ import RISING, FALLING, BOTH
from .device import Gateway
from threading import Event
class IOType(object):
"""IO Typen."""
INP = 300
OUT = 301
MEM = 302
class IOList(object):
"""Basisklasse fuer direkten Zugriff auf IO Objekte."""
def __init__(self):
"""Init IOList clacc."""
self.__dict_iobyte = {k: [] for k in range(4096)}
def __contains__(self, key):
"""Prueft ob IO existiert.
@param key IO-Name str()
@return True, wenn IO vorhanden"""
if type(key) == int:
return key in self.__dict_iobyte \
and len(self.__dict_iobyte[key]) > 0
else:
return hasattr(self, key)
def __delattr__(self, key):
"""Entfernt angegebenen IO.
@param key IO zum entfernen"""
# TODO: Prüfen ob auch Bit sein kann
dev = getattr(self, key)
self.__dict_iobyte[dev.address].remove(dev)
object.__delattr__(self, key)
def __getitem__(self, key):
"""Ruft angegebenen IO ab.
@param key IO Name oder Byte
@return IO Object"""
if type(key) == int:
if key in self.__dict_iobyte:
return self.__dict_iobyte[key]
else:
raise KeyError("byte '{}' does not exist".format(key))
else:
return getattr(self, key)
def __setitem__(self, key, value):
"""Setzt IO Wert.
@param key IO Name oder Byte
@param value Wert, auf den der IO gesetzt wird"""
if type(key) == int:
if key in self.__dict_iobyte:
if len(self.__dict_iobyte[key]) == 1:
self.__dict_iobyte[key][0].value = value
elif len(self.__dict_iobyte[key]) == 0:
raise KeyError("byte '{}' contains no input".format(key))
else:
raise KeyError(
"byte '{}' contains more than one bit-input"
"".format(key)
)
else:
raise KeyError("byte '{}' does not exist".format(key))
else:
getattr(self, key).value = value
def __setattr__(self, key, value):
"""Setzt IO Wert.
@param key IO Name oder Byte
@param value Wert, auf den der IO gesetzt wird"""
if issubclass(type(value), IOBase):
if hasattr(self, key):
raise AttributeError(
"attribute {} already exists - can not set io".format(key)
)
object.__setattr__(self, key, value)
# Bytedict erstellen für Adresszugriff
if value._bitaddress < 0:
self.__dict_iobyte[value.address].append(value)
else:
if len(self.__dict_iobyte[value.address]) != 8:
# "schnell" 8 Einträge erstellen da es BIT IOs sind
self.__dict_iobyte[value.address] += [
None, None, None, None, None, None, None, None
]
self.__dict_iobyte[value.address][value._bitaddress] = value
elif key == "_IOList__dict_iobyte":
object.__setattr__(self, key, value)
else:
getattr(self, key).value = value
def _testme(self):
# NOTE: Nur Debugging
for x in self.__dict_iobyte:
if len(self.__dict_iobyte[x]) > 0:
print(x, self.__dict_iobyte[x])
def reg_inp(self, name, frm, **kwargs):
"""Registriert einen neuen Input an Adresse von Diesem.
@param name Name des neuen Inputs
@param frm struct() formatierung (1 Zeichen)
@param kwargs Weitere Parameter:
- bmk: Bezeichnung fuer Input
- bit: Registriert Input als bool() am angegebenen Bit im Byte
- byteorder: Byteorder fuer den Input, Standardwert=little
- defaultvalue: Standardwert fuer Input, Standard ist 0
- event: Funktion fuer Eventhandling registrieren
- as_thread: Fuehrt die event-Funktion als RevPiCallback-Thread aus
- edge: event-Ausfuehren bei RISING, FALLING or BOTH Wertaenderung
@see <a target="_blank"
href="https://docs.python.org/3/library/struct.html#format-characters"
>Python3 struct()</a>
"""
if not issubclass(self._parentdevice, Gateway):
raise RuntimeError(
"this function can be used on gatway or virtual devices only"
)
self._create_io(name, startinp, frm, IOType.INP, **kwargs)
# Optional Event eintragen
reg_event = kwargs.get("event", None)
if reg_event is not None:
as_thread = kwargs.get("as_thread", False)
edge = kwargs.get("edge", None)
self.reg_event(name, reg_event, as_thread=as_thread, edge=edge)
def reg_out(self, name, startout, frm, **kwargs):
"""Registriert einen neuen Output.
@param name Name des neuen Outputs
@param startout Outputname ab dem eingefuegt wird
@param frm struct() formatierung (1 Zeichen)
@param kwargs Weitere Parameter:
- bmk: Bezeichnung fuer Output
- bit: Registriert Outputs als bool() am angegebenen Bit im Byte
- byteorder: Byteorder fuer den Output, Standardwert=little
- defaultvalue: Standardwert fuer Output, Standard ist 0
- event: Funktion fuer Eventhandling registrieren
- as_thread: Fuehrt die event-Funktion als RevPiCallback-Thread aus
- edge: event-Ausfuehren bei RISING, FALLING or BOTH Wertaenderung
@see <a target="_blank"
href="https://docs.python.org/3/library/struct.html#format-characters"
>Python3 struct()</a>
"""
if not issubclass(self._parentdevice, Gateway):
raise RuntimeError(
"this function can be used on gatway or virtual devices only"
)
self._create_io(name, startout, frm, IOType.OUT, **kwargs)
# Optional Event eintragen
reg_event = kwargs.get("event", None)
if reg_event is not None:
as_thread = kwargs.get("as_thread", False)
edge = kwargs.get("edge", None)
self.reg_event(name, reg_event, as_thread=as_thread, edge=edge)
class IOBase(object):
"""Basisklasse fuer alle IO-Objekte.
Die Basisfunktionalitaet ermoeglicht das Lesen und Schreiben der Werte
als bytes() oder bool(). Dies entscheidet sich bei der Instantiierung.
Wenn eine Bittadresse angegeben wird, werden bool()-Werte erwartet
und zurueckgegeben, ansonsten bytes().
Diese Klasse dient als Basis fuer andere IO-Klassen mit denen die Werte
auch als int() verwendet werden koennen.
"""
def __init__(self, parentdevice, valuelist, iotype, byteorder):
"""Instantiierung der IOBase()-Klasse.
@param parentdevice Parentdevice auf dem der IO liegt
@param valuelist Datenliste fuer Instantiierung
@param iotype IOType() Wert
@param byteorder Byteorder 'little' / 'big' fuer int() Berechnung
"""
self._parentdevice = parentdevice
# Bitadressen auf Bytes aufbrechen und umrechnen
self._bitaddress = -1 if valuelist[7] == "" else int(valuelist[7]) % 8
# Längenberechnung
self._bitlength = int(valuelist[2])
self._length = 1 if self._bitaddress == 0 else int(self._bitlength / 8)
self._byteorder = byteorder
self._iotype = iotype
self._name = valuelist[0]
self._signed = False
self.bmk = valuelist[6]
int_startaddress = int(valuelist[3])
if self._bitaddress == -1:
self.slc_address = slice(
int_startaddress, int_startaddress + self._length
)
# Defaultvalue aus Zahl in Bytes umrechnen
if str(valuelist[1]).isnumeric():
self.defaultvalue = int(valuelist[1]).to_bytes(
self._length, byteorder=self._byteorder
)
else:
# Defaultvalue direkt von bytes übernehmen
if type(valuelist[1]) == bytes:
if len(valuelist[1]) != self._length:
raise ValueError(
"given bytes for default value must have a length "
"of {}".format(self._length)
)
else:
self.defaultvalue = valuelist[1]
else:
self.defaultvalue = bytes(self._length)
else:
# Höhere Bits als 7 auf nächste Bytes umbrechen
int_startaddress += int((int(valuelist[7]) % 16) / 8)
self.slc_address = slice(
int_startaddress, int_startaddress + 1
)
self.defaultvalue = bool(int(valuelist[1]))
def __bool__(self):
"""bool()-wert der Klasse.
@return IO-Wert als bool(). Nur False wenn False oder 0 sonst True"""
return bool(self.get_value())
def __bytes__(self):
"""bytes()-wert der Klasse.
@return IO-Wert als bytes()"""
if self._bitaddress >= 0:
int_byte = int.from_bytes(
self._parentdevice._ba_devdata[self.slc_address],
byteorder=self._byteorder
)
return b'\x01' if bool(int_byte & 1 << self._bitaddress) \
else b'\x00'
else:
return bytes(self._parentdevice._ba_devdata[self.slc_address])
def __str__(self):
"""str()-wert der Klasse.
@return Namen des IOs"""
return self._name
def _get_byteorder(self):
"""Gibt konfigurierte Byteorder zurueck.
@return str() Byteorder"""
return self._byteorder
def get_address(self):
"""Gibt die absolute Byteadresse im Prozessabbild zurueck.
@return Absolute Byteadresse"""
return self._parentdevice.offset + self.slc_address.start
def get_length(self):
"""Gibt die Bytelaenge des IO zurueck.
@return Bytelaenge des IO"""
return self._length
def get_name(self):
"""Gibt den Namen des IOs zurueck.
@return IO Name"""
return self._name
def get_value(self):
"""Gibt den Wert des IOs als bytes() oder bool() zurueck.
@return IO-Wert"""
if self._bitaddress >= 0:
int_byte = int.from_bytes(
self._parentdevice._ba_devdata[self.slc_address],
byteorder=self._byteorder
)
return bool(int_byte & 1 << self._bitaddress)
else:
return bytes(self._parentdevice._ba_devdata[self.slc_address])
def reg_event(self, func, edge=BOTH, as_thread=False):
"""Registriert ein Event bei der Eventueberwachung.
@param func Funktion die bei Aenderung aufgerufen werden soll
@param edge Ausfuehren bei RISING, FALLING or BOTH Wertaenderung
@param as_thread Bei True, Funktion als EventCallback-Thread ausfuehren
"""
# Prüfen ob Funktion callable ist
if not callable(func):
raise RuntimeError(
"registered function '{}' ist not callable".format(func)
)
if edge != BOTH and self._bitaddress < 0:
raise AttributeError(
"parameter 'edge' can be used with bit io objects only"
)
if self not in self._parentdevice._dict_events:
self._parentdevice._dict_events[self] = [(func, edge, as_thread)]
else:
# Prüfen ob Funktion schon registriert ist
for regfunc in self._parentdevice._dict_events[self]:
if regfunc[0] == func and edge == BOTH:
if self._bitaddress < 0:
raise AttributeError(
"io '{}' with function '{}' already in list."
"".format(self._name, func)
)
else:
raise AttributeError(
"io '{}' with function '{}' already in list. "
"edge 'BOTH' not allowed anymore".format(
self._name, func
)
)
elif regfunc[0] == func and regfunc[1] == edge:
raise AttributeError(
"io '{}' with function '{}' for given edge "
"already in list".format(self._name, func)
)
else:
self._parentdevice._dict_events[self].append(
(func, edge, as_thread)
)
break
def set_value(self, value):
"""Setzt den Wert des IOs mit bytes() oder bool().
@param value IO-Wert als bytes() oder bool()"""
if self._iotype == IOType.OUT:
if self._bitaddress >= 0:
# Versuchen egal welchen Typ in Bool zu konvertieren
value = bool(value)
# ganzes Byte laden
byte_buff = self._parentdevice._ba_devdata[self.slc_address]
# Bytes in integer umwandeln
int_len = len(byte_buff)
int_byte = int.from_bytes(byte_buff, byteorder=self._byteorder)
int_bit = 1 << self._bitaddress
# Aktuellen Wert vergleichen und ggf. setzen
if not bool(int_byte & int_bit) == value:
if value:
int_byte += int_bit
else:
int_byte -= int_bit
# Zurückschreiben wenn verändert
self._parentdevice._ba_devdata[self.slc_address] = \
int_byte.to_bytes(int_len, byteorder=self._byteorder)
else:
if type(value) == bytes:
if self._length == len(value):
self._parentdevice._ba_devdata[self.slc_address] = \
value
else:
raise ValueError(
"requires a bytes() object of length {}, but"
" {} was given".format(self._length, len(value))
)
else:
raise ValueError(
"requires a bytes() object, not {}".format(type(value))
)
elif self._iotype == IOType.INP:
raise AttributeError("can not write to input")
elif self._iotype == IOType.MEM:
raise AttributeError("can not write to memory")
def unreg_event(self, func=None, edge=None):
"""Entfernt ein Event aus der Eventueberwachung.
@param func Nur Events mit angegebener Funktion
@param edge Nur Events mit angegebener Funktion und angegebener Edge
"""
if self in self._parentdevice._dict_events:
if func is None:
del self._parentdevice._dict_events[self]
else:
newlist = []
for regfunc in self._parentdevice._dict_events[self]:
if regfunc[0] != func or edge is not None \
and regfunc[1] != edge:
newlist.append(regfunc)
# Wenn Funktionen übrig bleiben, diese übernehmen
if len(newlist) > 0:
self._parentdevice._dict_events[self] = newlist
else:
del self._parentdevice._dict_events[self]
def wait(self, edge=BOTH, exitevent=None, okvalue=None, timeout=0):
"""Wartet auf Wertaenderung eines IOs.
Die Wertaenderung wird immer uerberprueft, wenn fuer Devices
in Devicelist.auto_refresh() neue Daten gelesen wurden.
Bei Wertaenderung, wird das Warten mit 0 als Rueckgabewert beendet.
HINWEIS: Wenn ProcimgWriter() keine neuen Daten liefert, wird
bis in die Ewigkeit gewartet (nicht bei Angabe von "timeout").
Wenn edge mit RISING oder FALLING angegeben wird muss diese Flanke
ausgeloest werden. Sollte der Wert 1 sein beim Eintritt mit Flanke
RISING, wird das Warten erst bei Aenderung von 0 auf 1 beendet.
Als exitevent kann ein threading.Event()-Objekt uebergeben werden,
welches das Warten bei is_set() sofort mit 1 als Rueckgabewert
beendet.
Wenn der Wert okvalue an dem IO fuer das Warten anliegt, wird
das Warten sofort mit -1 als Rueckgabewert beendet.
Der Timeoutwert bricht beim Erreichen das Warten sofort mit
Wert 2 Rueckgabewert ab. (Das Timeout wird ueber die Zykluszeit
der auto_refresh Funktion berechnet, entspricht also nicht exact den
angegeben Millisekunden! Es wird immer nach oben gerundet!)
@param edge Flanke RISING, FALLING, BOTH bei der mit True beendet wird
@param exitevent thrading.Event() fuer vorzeitiges Beenden mit False
@param okvalue IO-Wert, bei dem das Warten sofort mit True beendet wird
@param timeout Zeit in ms nach der mit False abgebrochen wird
@return int() erfolgreich Werte <= 0
- Erfolgreich gewartet
Wert 0: IO hat den Wert gewechselt
Wert -1: okvalue stimmte mit IO ueberein
- Fehlerhaft gewartet
Wert 1: exitevent wurde gesetzt
Wert 2: timeout abgelaufen
Wert 100: Devicelist.exit() wurde aufgerufen
"""
# Prüfen ob Device in auto_refresh ist
if not self._parentdevice._selfupdate:
raise RuntimeError(
"auto_refresh is not activated for device '{}|{}' - there "
"will never be new data".format(
self._parentdevice.position, self._parentdevice.name
)
)
if edge != BOTH and self._bitaddress < 0:
raise AttributeError(
"parameter 'edge' can be used with bit Inputs only"
)
# Abbruchwert prüfen
if okvalue == self.value:
return -1
# WaitExit Event säubern
self._parentdevice._parent._waitexit.clear()
val_start = self.value
timeout = timeout / 1000
bool_timecount = timeout > 0
if exitevent is None:
exitevent = Event()
flt_timecount = 0 if bool_timecount else -1
while not self._parentdevice._parent._waitexit.is_set() \
and not exitevent.is_set() \
and flt_timecount < timeout:
if self._parentdevice._parent.imgwriter.newdata.wait(2.5):
self._parentdevice._parent.imgwriter.newdata.clear()
if val_start != self.value:
if edge == BOTH \
or edge == RISING and not val_start \
or edge == FALLING and val_start:
return 0
else:
val_start = not val_start
if bool_timecount:
flt_timecount += \
self._parentdevice._parent.imgwriter._refresh
elif bool_timecount:
# TODO: Prüfen
flt_timecount += 1
# Abbruchevent wurde gesetzt
if exitevent.is_set():
return 1
# RevPiModIO mainloop wurde verlassen
if self._parentdevice._parent._waitexit.is_set():
return 100
# Timeout abgelaufen
return 2
address = property(get_address)
length = property(get_length)
name = property(get_name)
value = property(get_value, set_value)
class IntIO(IOBase):
"""Klasse fuer den Zugriff auf die Daten mit Konvertierung in int().
Diese Klasse erweitert die Funktion von IOBase() um Funktionen,
ueber die mit int() Werten gearbeitet werden kann. Fuer die Umwandlung
koennen 'Byteorder' (Default 'little') und 'signed' (Default False) als
Parameter gesetzt werden.
@see #IOBase IOBase
"""
def __int__(self):
"""Gibt IO als int() Wert zurueck mit Beachtung byteorder/signed.
@return int() ohne Vorzeichen"""
return self.get_int()
def _get_signed(self):
"""Ruft ab, ob der Wert Vorzeichenbehaftet behandelt werden soll.
@return True, wenn Vorzeichenbehaftet"""
return self._signed
def _set_byteorder(self, value):
"""Setzt Byteorder fuer int() Umwandlung.
@param value str() 'little' or 'big'"""
if not (value == "little" or value == "big"):
raise ValueError("byteorder must be 'little' or 'big'")
self._byteorder = value
def _set_signed(self, value):
"""Left fest, ob der Wert Vorzeichenbehaftet behandelt werden soll.
@param value True, wenn mit Vorzeichen behandel"""
if type(value) != bool:
raise ValueError("signed must be bool() True or False")
self._signed = value
def get_int(self):
"""Gibt IO als int() Wert zurueck mit Beachtung byteorder/signed.
@return int() Wert"""
return int.from_bytes(
self._parentdevice._ba_devdata[self.slc_address],
byteorder=self._byteorder,
signed=self._signed
)
def set_int(self, value):
"""Setzt IO mit Beachtung byteorder/signed.
@param value int()"""
if type(value) == int:
self.set_value(value.to_bytes(
self._length,
byteorder=self._byteorder,
signed=self._signed
))
else:
raise ValueError(
"need an int() value, but {} was given".format(type(value))
)
byteorder = property(IOBase._get_byteorder, _set_byteorder)
signed = property(_get_signed, _set_signed)
value = property(get_int, set_int)
class StructIO(IOBase):
"""Klasse fuer den Zugriff auf Daten ueber ein definierten struct().
Diese Klasse ueberschreibt get_value() und set_value() der IOBase()
Klasse. Sie stellt ueber struct die Werte in der gewuenschten Formatierung
bereit. Der struct-Formatwert wird bei der Instantiierung festgelegt.
@see #IOBase IOBase
"""
def __init__(self, parentdevice, valuelist, iotype, byteorder, frm):
"""Erweitert IOBase um struct-Formatierung.
@see #IOBase.__init__ IOBase.__init__(...)"""
super().__init__(parentdevice, valuelist, iotype, byteorder)
self.frm = frm
def get_structvalue(self):
"""Gibt den Wert mit struct Formatierung zurueck.
@return Wert vom Typ der struct-Formatierung"""
if self._bitaddress >= 0:
return self.get_value()
else:
return struct.unpack(self.frm, self.get_value())[0]
def set_structvalue(self, value):
"""Setzt den Wert mit struct Formatierung.
@param value Wert vom Typ der struct-Formatierung"""
if self._bitaddress >= 0:
self.set_value(value)
else:
self.set_value(struct.pack(self.frm, value))
byteorder = property(IOBase._get_byteorder)
value = property(get_structvalue, set_structvalue)

899
revpimodio2/modio.py Normal file
View File

@@ -0,0 +1,899 @@
#
# python3-RevPiModIO
#
# Webpage: https://revpimodio.org/
# (c) Sven Sager, License: LGPLv3
#
# -*- coding: utf-8 -*-
import warnings
from . import app as appmodule
from . import device as devicemodule
from . import helper as helpermodule
from . import io as iomodule
from . import summary as summarymodule
from .__init__ import RISING, FALLING, BOTH
from json import load as jload
from os import access, F_OK, R_OK
from signal import signal, SIG_DFL, SIGINT, SIGTERM
from threading import Thread, Event
class RevPiModIO(object):
"""Klasse fuer die Verwaltung aller piCtory Informationen.
Diese Klasse uebernimmt die gesamte Konfiguration aus piCtory und bilded
die Devices und IOs ab. Sie uebernimmt die exklusive Verwaltung des
Prozessabbilds und stellt sicher, dass die Daten synchron sind.
Sollten nur einzelne Devices gesteuert werden, verwendet man
RevPiModIOSelected() und uebergibt bei Instantiierung eine Liste mit
Device Positionen oder Device Namen.
"""
def __init__(self, **kwargs):
"""Instantiiert die Grundfunktionen.
@param kwargs Weitere Parameter:
- auto_refresh: Wenn True, alle Devices zu auto_refresh hinzufuegen
- configrsc: Pfad zur piCtory Konfigurationsdatei
- procimg: Pfad zum Prozessabbild
- monitoring: In- und Outputs werden gelesen, niemals geschrieben
- simulator: Laed das Modul als Simulator und vertauscht IOs
- syncoutputs: Aktuell gesetzte Outputs vom Prozessabbild einlesen
"""
self._auto_refresh = kwargs.get("auto_refresh", False)
self._configrsc = kwargs.get("configrsc", None)
self._monitoring = kwargs.get("monitoring", False)
self._procimg = kwargs.get("procimg", "/dev/piControl0")
self._simulator = kwargs.get("simulator", False)
self._syncoutputs = kwargs.get("syncoutputs", True)
# TODO: bei simulator und procimg prüfen ob datei existiert / anlegen?
# Private Variablen
self.__cleanupfunc = None
self._buffedwrite = False
self._device = []
self._exit = Event()
self._imgwriter = None
self._length = 0
self._looprunning = False
self._lst_devselect = []
self._lst_refresh = []
self._myfh = self._create_myfh()
self._th_mainloop = None
self._waitexit = Event()
# Modulvariablen
self.core = None
# piCtory Klassen
self.app = None
self.device = None
self.devices = None
self.io = None
self.summary = None
# Nur Konfigurieren, wenn nicht vererbt
if type(self) == RevPiModIO:
self._configure()
def __del__(self):
"""Zerstoert alle Klassen um aufzuraeumen."""
self.exit(full=True)
self._myfh.close()
def __evt_exit(self, signum, sigframe):
"""Eventhandler fuer Programmende.
@param signum Signalnummer
@param sigframe Signalframe"""
signal(SIGINT, SIG_DFL)
signal(SIGTERM, SIG_DFL)
self.exit(full=True)
if self.__cleanupfunc is not None:
self.readprocimg()
self.__cleanupfunc()
self.writeprocimg()
def _configure(self):
"""Verarbeitet die piCtory Konfigurationsdatei."""
jconfigrsc = self.get_jconfigrsc()
# App Klasse instantiieren
self.app = appmodule.App(jconfigrsc["App"])
# Devicefilter anwenden
if len(self._lst_devselect) > 0:
lst_found = []
if type(self) == RevPiModIODriver:
_searchtype = "VIRTUAL"
else:
_searchtype = None
# Angegebene Devices suchen
for dev in jconfigrsc["Devices"]:
if _searchtype is None or dev["type"] == _searchtype:
if dev["name"] in self._lst_devselect:
lst_found.append(dev)
elif dev["position"].isnumeric() \
and int(dev["position"]) in self._lst_devselect:
lst_found.append(dev)
# Devices aus JSON oder Filter übernehmen
lst_devices = jconfigrsc["Devices"] if len(self._lst_devselect) == 0 \
else lst_found
# Device und IO Klassen anlegen
self.device = devicemodule.DeviceList()
self.io = iomodule.IOList()
# Devices initialisieren
err_names = []
for device in sorted(lst_devices, key=lambda x: x["position"]):
# Bei VDev in alter piCtory Version, Position eindeutig machen
if device["position"] == "adap.":
device["position"] = -1
# NOTE: Testen mit alter piCtory Version
while device["position"] in self.device:
device["position"] -= 1
if device["type"] == "BASE":
# Core
dev_new = devicemodule.Core(
self, device, simulator=self._simulator
)
self.core = dev_new
# Für RS485 errors defaults laden und schreiben
# NOTE: Soll das wirklich gemacht werden?
for io in dev_new.get_outs():
io.set_value(io.defaultvalue)
if not self._monitoring:
self.writeprocimg(True, dev_new)
elif device["type"] == "LEFT_RIGHT":
# IOs
dev_new = devicemodule.Device(
self, device, simulator=self._simulator
)
elif device["type"] == "VIRTUAL":
# Virtuals
dev_new = devicemodule.Virtual(
self, device, simulator=self._simulator
)
elif device["type"] == "EDGE":
# Gateways
dev_new = devicemodule.Gateway(
self, device, simulator=self._simulator
)
else:
# Device-Type nicht gefunden
warnings.warn(
"device type {} unknown",
Warning
)
dev_new = None
if dev_new is not None:
self._device.append(dev_new)
# Offset prüfen, muss mit Länge übereinstimmen
if self._length < dev_new.offset:
self._length = dev_new.offset
self._length += dev_new._length
# Auf doppelte Namen prüfen, da piCtory dies zulässt
if hasattr(self.device, dev_new.name):
err_names.append(dev_new.name)
# DeviceList für direkten Zugriff aufbauen
setattr(self.device, dev_new.name, dev_new)
# dict_devname zerstören, wenn doppelte Namen vorhanden sind
for errdev in err_names:
delattr(self.device, errdev)
warnings.warn(
"equal device name in pictory configuration. can not "
"build device to acces by name. you can access all devices "
"by position number pos_XX only!",
Warning
)
# ImgWriter erstellen
self._imgwriter = helpermodule.ProcimgWriter(self)
# Aktuellen Outputstatus von procimg einlesen
if self._syncoutputs:
self.syncoutputs(force=True)
# NOTE: Nur noch bis Final für kompatibilität
# Devices Klasse instantiieren
self.devices = devicemodule.Devicelist(self)
# Optional ins auto_refresh aufnehmen
if self._auto_refresh:
for dev in self._device:
dev.auto_refresh()
# Summary Klasse instantiieren
self.summary = summarymodule.Summary(jconfigrsc["Summary"])
def _create_myfh(self):
"""Erstellt FileObject mit Pfad zum procimg.
return FileObject"""
self._buffedwrite = False
return open(self._procimg, "r+b", 0)
def _get_configrsc(self):
"""Getter function.
@return Pfad der verwendeten piCtory Konfiguration"""
return self._configrsc
def _get_cycletime(self):
"""Gibt Aktualisierungsrate in ms der Prozessabbildsynchronisierung aus.
@return Millisekunden"""
return self._imgwriter.refresh
def _get_length(self):
"""Getter function.
@return Laenge in Bytes der Devices"""
return self._length
def _get_monitoring(self):
"""Getter function.
@return True, wenn als Monitoring gestartet"""
return self._monitoring
def _get_procimg(self):
"""Getter function.
@return Pfad des verwendeten Prozessabbilds"""
return self._procimg
def _get_simulator(self):
"""Getter function.
@return True, wenn als Simulator gestartet"""
return self._simulator
def _set_cycletime(self, milliseconds):
"""Setzt Aktualisierungsrate der Prozessabbild-Synchronisierung.
@param milliseconds int() in Millisekunden"""
self._imgwriter.refresh = milliseconds
def auto_refresh_maxioerrors(self, value=None):
"""Maximale IO Fehler fuer auto_refresh.
@param value Setzt maximale Anzahl bis exception ausgeloest wird
@return Maximale Anzahl bis exception ausgeloest wird"""
if value is None:
return self._imgwriter.maxioerrors
elif type(value) == int and value >= 0:
self._imgwriter.maxioerrors = value
def auto_refresh_resetioerrors(self):
"""Setzt aktuellen IOError-Zaehler auf 0 zurueck."""
self._imgwriter.maxioerrors = 0
def cleanup(self):
"""Beendet auto_refresh und alle Threads."""
# TODO: wirklich alles löschen
self.exit(full=True)
self._myfh.close()
self.app = None
self.device = None
self.devices = None
self.io = None
self.summary = None
def cycleloop(self, func, cycletime=50):
"""Startet den Cycleloop.
Der aktuelle Programmthread wird hier bis Aufruf von
RevPiDevicelist.exit() "gefangen". Er fuehrt nach jeder Aktualisierung
des Prozessabbilds die uebergebene Funktion "func" aus und arbeitet sie
ab. Waehrend der Ausfuehrung der Funktion wird das Prozessabbild nicht
weiter aktualisiert. Die Inputs behalten bis zum Ende den aktuellen
Wert. Gesetzte Outputs werden nach Ende des Funktionsdurchlaufs in das
Prozessabbild geschrieben.
Verlassen wird der Cycleloop, wenn die aufgerufene Funktion einen
Rueckgabewert nicht gleich None liefert, oder durch Aufruf von
revpimodio.exit().
HINWEIS: Die Aktualisierungszeit und die Laufzeit der Funktion duerfen
die eingestellte auto_refresh Zeit, bzw. uebergebene cycletime nicht
ueberschreiten!
Ueber den Parameter cycletime kann die Aktualisierungsrate fuer das
Prozessabbild gesetzt werden (selbe Funktion wie
set_refreshtime(milliseconds)).
@param func Funktion, die ausgefuehrt werden soll
@param cycletime auto_refresh Wert in Millisekunden
@return None
"""
# Prüfen ob ein Loop bereits läuft
if self._looprunning:
raise RuntimeError(
"can not start multiple loops mainloop/cycleloop"
)
# Prüfen ob Devices in auto_refresh sind
if len(self._lst_refresh) == 0:
raise RuntimeError("no device with auto_refresh activated")
# Prüfen ob Funktion callable ist
if not callable(func):
raise RuntimeError(
"registered function '{}' ist not callable".format(func)
)
# Zykluszeit übernehmen
if cycletime != self._imgwriter.refresh:
self._imgwriter.refresh = cycletime
# Cycleloop starten
self._looprunning = True
cycleinfo = helpermodule.Cycletools()
ec = None
while ec is None and not self._exit.is_set():
# Auf neue Daten warten und nur ausführen wenn set()
if not self._imgwriter.newdata.wait(2.5):
if not self._exit.is_set() and not self._imgwriter.is_alive():
raise RuntimeError("auto_refresh thread not running")
continue
self._imgwriter.newdata.clear()
# Vor Aufruf der Funktion auto_refresh sperren
self._imgwriter.lck_refresh.acquire()
# Funktion aufrufen und auswerten
ec = func(cycleinfo)
cycleinfo._docycle()
# auto_refresh freigeben
self._imgwriter.lck_refresh.release()
# Cycleloop beenden
self._looprunning = False
return ec
def exit(self, full=True):
"""Beendet mainloop() und optional auto_refresh.
Wenn sich das Programm im mainloop() befindet, wird durch Aufruf
von exit() die Kontrolle wieder an das Hauptprogramm zurueckgegeben.
Der Parameter full ist mit True vorbelegt und entfernt alle Devices aus
dem auto_refresh. Der Thread fuer die Prozessabbildsynchronisierung
wird dann gestoppt und das Programm kann sauber beendet werden.
@param full Entfernt auch alle Devices aus auto_refresh"""
self._exit.set()
self._waitexit.set()
if full:
if self._imgwriter.is_alive():
self._imgwriter.stop()
self._imgwriter.join(self._imgwriter._refresh)
while len(self._lst_refresh) > 0:
dev = self._lst_refresh.pop()
dev._selfupdate = False
if not self._monitoring:
self.writeprocimg(True, dev)
def get_jconfigrsc(self):
"""Laed die piCotry Konfiguration und erstellt ein dict().
@return dict() der piCtory Konfiguration"""
# piCtory Konfiguration prüfen
if self._configrsc is not None:
if not access(self._configrsc, F_OK | R_OK):
raise RuntimeError(
"can not access pictory configuration at {}".format(
self._configrsc))
else:
# piCtory Konfiguration an bekannten Stellen prüfen
lst_rsc = ["/etc/revpi/config.rsc", "/opt/KUNBUS/config.rsc"]
for rscfile in lst_rsc:
if access(rscfile, F_OK | R_OK):
self._configrsc = rscfile
break
if self._configrsc is None:
raise RuntimeError(
"can not access known pictory configurations at {} - "
"use 'configrsc' parameter so specify location"
"".format(", ".join(lst_rsc))
)
with open(self._configrsc, "r") as fhconfigrsc:
return jload(fhconfigrsc)
def handlesignalend(self, cleanupfunc=None):
"""Signalhandler fuer Programmende verwalten.
Wird diese Funktion aufgerufen, uebernimmt RevPiModIO die SignalHandler
fuer SIGINT und SIGTERM. Diese werden Empfangen, wenn das
Betriebssystem oder der Benutzer das Steuerungsprogramm sauber beenden
will.
Die optionale Funktion "cleanupfunc" wird als letztes nach dem letzten
Einlesen der Inputs ausgefuehrt. Dort gesetzte Outputs werden nach
Ablauf der Funktion ein letztes Mal geschrieben.
Gedacht ist dies fuer Aufraeumarbeiten, wie z.B. das abschalten der
LEDs am RevPi-Core.
Nach einmaligem Empfangen eines der Signale und dem Beenden der
RevPiModIO Thrads / Funktionen werden die SignalHandler wieder
freigegeben.
@param cleanupfunc Funktion wird nach dem letzten Lesen der Inputs
ausgefuehrt, gefolgt vom letzten Schreiben der Outputs
"""
# Prüfen ob Funktion callable ist
if not callable(cleanupfunc):
raise RuntimeError(
"registered function '{}' ist not callable".format(cleanupfunc)
)
self.__cleanupfunc = cleanupfunc
signal(SIGINT, self.__evt_exit)
signal(SIGTERM, self.__evt_exit)
def mainloop(self, freeze=False, blocking=True):
"""Startet den Mainloop mit Eventueberwachung.
Der aktuelle Programmthread wird hier bis Aufruf von
RevPiDevicelist.exit() "gefangen" (es sei denn blocking=False). Er
durchlaeuft die Eventueberwachung und prueft Aenderungen der, mit
einem Event registrierten, IOs. Wird eine Veraenderung erkannt,
fuert das Programm die dazugehoerigen Funktionen der Reihe nach aus.
Wenn der Parameter "freeze" mit True angegeben ist, wird die
Prozessabbildsynchronisierung angehalten bis alle Eventfunktionen
ausgefuehrt wurden. Inputs behalten fuer die gesamte Dauer ihren
aktuellen Wert und Outputs werden erst nach Durchlauf aller Funktionen
in das Prozessabbild geschrieben.
Wenn der Parameter "blocking" mit False angegeben wird, aktiviert
dies die Eventueberwachung und blockiert das Programm NICHT an der
Stelle des Aufrufs. Eignet sich gut fuer die GUI Programmierung, wenn
Events vom RevPi benoetigt werden, aber das Programm weiter ausgefuehrt
werden soll.
@param freeze Wenn True, Prozessabbildsynchronisierung anhalten
@param blocking Wenn False, blockiert das Programm NICHT
@return None
"""
# Prüfen ob ein Loop bereits läuft
if self._looprunning:
raise RuntimeError(
"can not start multiple loops mainloop/cycleloop"
)
# Prüfen ob Devices in auto_refresh sind
if len(self._lst_refresh) == 0:
raise RuntimeError("no device with auto_refresh activated")
# Thread erstellen, wenn nicht blockieren soll
if not blocking:
self._th_mainloop = Thread(
target=self.mainloop,
kwargs={"freeze": freeze, "blocking": True}
)
self._th_mainloop.start()
return
# Event säubern vor Eintritt in Mainloop
self._exit.clear()
self._looprunning = True
# Beim Eintritt in mainloop Bytecopy erstellen
for dev in self._lst_refresh:
dev._filelock.acquire()
dev._ba_datacp = dev._ba_devdata[:]
dev._filelock.release()
lst_fire = []
while not self._exit.is_set():
# Auf neue Daten warten und nur ausführen wenn set()
if not self._imgwriter.newdata.wait(2.5):
if not self._exit.is_set() and not self._imgwriter.is_alive():
raise RuntimeError("auto_refresh thread not running")
continue
self._imgwriter.newdata.clear()
# Während Auswertung refresh sperren
self._imgwriter.lck_refresh.acquire()
for dev in self._lst_refresh:
if len(dev._dict_events) == 0 \
or dev._ba_datacp == dev._ba_devdata:
continue
for io_event in dev._dict_events:
if dev._ba_datacp[io_event.slc_address] == \
dev._ba_devdata[io_event.slc_address]:
continue
if io_event._bitaddress >= 0:
boolcp = bool(int.from_bytes(
dev._ba_datacp[io_event.slc_address],
byteorder=io_event._byteorder
) & 1 << io_event._bitaddress)
boolor = bool(int.from_bytes(
dev._ba_devdata[io_event.slc_address],
byteorder=io_event._byteorder
) & 1 << io_event._bitaddress)
if boolor == boolcp:
continue
for regfunc in dev._dict_events[io_event]:
if regfunc[1] == BOTH \
or regfunc[1] == RISING and boolor \
or regfunc[1] == FALLING and not boolor:
lst_fire.append(
(regfunc, io_event.name, io_event.value)
)
else:
for regfunc in dev._dict_events[io_event]:
lst_fire.append(
(regfunc, io_event.name, io_event.value)
)
# Nach Verarbeitung aller IOs die Bytes kopieren
dev._filelock.acquire()
dev._ba_datacp = dev._ba_devdata[:]
dev._filelock.release()
# Refreshsperre aufheben wenn nicht freeze
if not freeze:
self._imgwriter.lck_refresh.release()
# Erst nach Datenübernahme alle Events feuern
while len(lst_fire) > 0:
tup_fire = lst_fire.pop()
event_func = tup_fire[0][0]
passname = tup_fire[1]
passvalue = tup_fire[2]
if tup_fire[0][2]:
th = helpermodule.EventCallback(
event_func, passname, passvalue
)
th.start()
else:
# Direct callen da Prüfung in RevPiDevice.reg_event ist
event_func(passname, passvalue)
# Refreshsperre aufheben wenn freeze
if freeze:
self._imgwriter.lck_refresh.release()
# Mainloop verlassen
self._looprunning = False
def readprocimg(self, force=False, device=None):
"""Einlesen aller Inputs aller/eines Devices vom Prozessabbild.
@param force auch Devices mit autoupdate=False
@param device nur auf einzelnes Device anwenden
@return True, wenn Arbeiten an allen Devices erfolgreich waren
"""
if device is None:
mylist = self._device
else:
# TODO: Devicesuchen ändern
dev = device if issubclass(type(device), devicemodule.Device) \
else self.device.__getitem__(device)
if dev._selfupdate:
raise RuntimeError(
"can not read process image, while device '{}|{}'"
"is in auto_refresh mode".format(dev.position, dev.name)
)
mylist = [dev]
# Daten komplett einlesen
try:
self._myfh.seek(0)
bytesbuff = self._myfh.read(self._length)
except IOError:
warnings.warn(
"read error on process image '{}'".format(self.myfh.name),
RuntimeWarning
)
return False
for dev in mylist:
if (force or dev.autoupdate) and not dev._selfupdate:
# FileHandler sperren
dev._filelock.acquire()
if self._monitoring:
# Alles vom Bus einlesen
dev._ba_devdata[:] = bytesbuff[dev.slc_devoff]
else:
# Inputs vom Bus einlesen
dev._ba_devdata[dev.slc_inp] = bytesbuff[dev.slc_inpoff]
# Mems vom Bus lesen
dev._ba_devdata[dev.slc_mem] = bytesbuff[dev.slc_memoff]
dev._filelock.release()
return True
def setdefaultvalues(self, force=False, device=None):
"""Alle Outputbuffer werden auf die piCtory default Werte gesetzt.
@param force auch Devices mit autoupdate=False
@param device nur auf einzelnes Device anwenden"""
if self._monitoring:
raise RuntimeError(
"can not set default values, while system is in monitoring "
"mode"
)
if device is None:
mylist = self._device
else:
dev = device if issubclass(type(device), devicemodule.Device) \
else self.__getitem__(device)
mylist = [dev]
for dev in mylist:
if (force or dev.autoupdate):
for io in dev.get_outs():
io.set_value(io.defaultvalue)
def syncoutputs(self, force=False, device=None):
"""Lesen aller aktuell gesetzten Outputs im Prozessabbild.
@param force auch Devices mit autoupdate=False
@param device nur auf einzelnes Device anwenden
@return True, wenn Arbeiten an allen Devices erfolgreich waren
"""
if device is None:
mylist = self._device
else:
dev = device if issubclass(type(device), devicemodule.Device) \
else self.__getitem__(device)
if dev._selfupdate:
raise RuntimeError(
"can not sync process image, while device '{}|{}'"
"is in auto_refresh mode".format(dev.position, dev.name)
)
mylist = [dev]
try:
self._myfh.seek(0)
bytesbuff = self._myfh.read(self._length)
except IOError:
warnings.warn(
"read error on process image '{}'".format(self._myfh.name),
RuntimeWarning
)
return False
for dev in mylist:
if (force or dev.autoupdate) and not dev._selfupdate:
dev._filelock.acquire()
# Outputs vom Bus einlesen
dev._ba_devdata[dev.slc_out] = bytesbuff[dev.slc_outoff]
dev._filelock.release()
return True
def writedefaultinputs(self, virtual_device):
"""Schreibt fuer ein virtuelles Device piCtory Defaultinputwerte.
Sollten in piCtory Defaultwerte fuer Inputs eines virtuellen Devices
angegeben sein, werden diese nur beim Systemstart oder einem piControl
Reset gesetzt. Sollte danach das Prozessabbild mit NULL ueberschrieben,
gehen diese Werte verloren.
Diese Funktion kann nur auf virtuelle Devices angewendet werden!
@param virtual_device Virtuelles Device fuer Wiederherstellung
@return True, wenn Arbeiten am virtuellen Device erfolgreich waren
"""
if self._monitoring:
raise RuntimeError(
"can not write process image, while system is in monitoring "
"mode"
)
# Device suchen
dev = virtual_device if issubclass(type(virtual_device), devicemodule.Device) \
else self.__getitem__(virtual_device)
# Prüfen ob es ein virtuelles Device ist
if not issubclass(type(dev), devicemodule.Virtual):
raise RuntimeError(
"this function can be used for virtual devices only"
)
workokay = True
dev._filelock.acquire()
for io in dev.get_inps():
dev._ba_devdata[io.slc_address] = io.defaultvalue
# Outpus auf Bus schreiben
try:
self._myfh.seek(dev.slc_inpoff.start)
self._myfh.write(dev._ba_devdata[dev.slc_inp])
if self._buffedwrite:
self._myfh.flush()
except IOError:
warnings.warn(
"write error on process image '{}'"
"".format(self._myfh.name),
RuntimeWarning
)
workokay = False
dev._filelock.release()
return workokay
def writeprocimg(self, force=False, device=None):
"""Schreiben aller Outputs aller Devices ins Prozessabbild.
@param force auch Devices mit autoupdate=False
@param device nur auf einzelnes Device anwenden
@return True, wenn Arbeiten an allen Devices erfolgreich waren
"""
if self._monitoring:
raise RuntimeError(
"can not write process image, while system is in monitoring "
"mode"
)
if device is None:
mylist = self._device
else:
dev = device if issubclass(type(device), devicemodule.Device) \
else self.__getitem__(device)
if dev._selfupdate:
raise RuntimeError(
"can not write process image, while device '{}|{}'"
"is in auto_refresh mode".format(dev.position, dev.name)
)
mylist = [dev]
workokay = True
for dev in mylist:
if (force or dev.autoupdate) and not dev._selfupdate:
dev._filelock.acquire()
# Outpus auf Bus schreiben
try:
self._myfh.seek(dev.slc_outoff.start)
self._myfh.write(dev._ba_devdata[dev.slc_out])
except IOError:
workokay = False
dev._filelock.release()
if self._buffedwrite:
try:
self._myfh.flush()
except IOError:
workokay = False
if not workokay:
warnings.warn(
"write error on process image '{}'"
"".format(self._myfh.name),
RuntimeWarning
)
return workokay
configrsc = property(_get_configrsc)
cycletime = property(_get_cycletime, _set_cycletime)
length = property(_get_length)
monitoring = property(_get_monitoring)
procimg = property(_get_procimg)
simulator = property(_get_simulator)
class RevPiModIOSelected(RevPiModIO):
"""Klasse fuer die Verwaltung einzelner Devices aus piCtory.
Diese Klasse uebernimmt nur angegebene Devices der piCtory Konfiguration
und bilded sie inkl. IOs ab. Sie uebernimmt die exklusive Verwaltung des
Adressbereichs im Prozessabbild an dem sich die angegebenen Devices
befinden und stellt sicher, dass die Daten synchron sind.
"""
def __init__(self, deviceselection, **kwargs):
"""Instantiiert nur fuer angegebene Devices die Grundfunktionen.
Der Parameter deviceselection kann eine einzelne
Device Position / einzelner Device Name sein oder eine Liste mit
mehreren Positionen / Namen
@param deviceselection Positionsnummer oder Devicename
@param kwargs Weitere Parameter
@see #RevPiModIO.__init__ RevPiModIO.__init__(...)
"""
super().__init__(**kwargs)
# Device liste erstellen
if type(deviceselection) == list:
for dev in deviceselection:
self._lst_devselect.append(dev)
else:
self._lst_devselect.append(deviceselection)
for vdev in self._lst_devselect:
if type(vdev) != int and type(vdev) != str:
raise ValueError(
"need device position as int() or device name as str()"
)
self._configure()
if len(self._device) == 0:
if type(self) == RevPiModIODriver:
raise RuntimeError(
"could not find any given VIRTUAL devices in config"
)
else:
raise RuntimeError(
"could not find any given devices in config"
)
elif len(self._device) != len(self._lst_devselect):
if type(self) == RevPiModIODriver:
raise RuntimeError(
"could not find all given VIRTUAL devices in config"
)
else:
raise RuntimeError(
"could not find all given devices in config"
)
class RevPiModIODriver(RevPiModIOSelected):
"""Klasse um eigene Treiber fuer die virtuellen Devices zu erstellen.
Mit dieser Klasse werden nur angegebene Virtuelle Devices mit RevPiModIO
verwaltet. Bei Instantiierung werden automatisch die Inputs und Outputs
verdreht, um das Schreiben der Inputs zu ermoeglichen. Die Daten koennen
dann ueber logiCAD an den Devices abgerufen werden.
"""
def __init__(self, vdev, **kwargs):
"""Instantiiert die Grundfunktionen.
@param vdev Virtuelles Device fuer die Verwendung / oder list()
@param kwargs Weitere Parameter (nicht monitoring und simulator)
@see #RevPiModIO.__init__ RevPiModIO.__init__(...)
"""
kwargs["monitoring"] = False
kwargs["simulator"] = True
super().__init__(vdev, **kwargs)

19
revpimodio2/summary.py Normal file
View File

@@ -0,0 +1,19 @@
#
# python3-RevPiModIO
#
# Webpage: https://revpimodio.org/
# (c) Sven Sager, License: LGPLv3
#
# -*- coding: utf-8 -*-
"""Bildet die Summary-Sektion von piCtory ab."""
class Summary(object):
"""Bildet die Summary-Sektion der config.rsc ab."""
def __init__(self, summary):
"""Instantiiert die RevPiSummary-Klasse.
@param summary piCtory Summaryinformationen"""
self.inptotal = summary["inpTotal"]
self.outtotal = summary["outTotal"]

41
setup.py Normal file
View File

@@ -0,0 +1,41 @@
#! /usr/bin/env python3
#
# (c) Sven Sager, License: LGPLv3
#
# -*- coding: utf-8 -*-
"""Setupscript fuer python3-revpimodio."""
from distutils.core import setup
setup(
author="Sven Sager",
author_email="akira@narux.de",
url="https://revpimodio.org",
maintainer="Sven Sager",
maintainer_email="akira@revpimodio.org",
license="LGPLv3",
name="revpimodio2",
version="2.0.0",
py_modules=["revpimodio2"],
description="Python3 Programmierung für Kunbus RevolutionPi",
long_description=""
"Das Modul stellt alle Devices und IOs aus der piCtory Konfiguration \n"
"in Python3 zur Verfügung. Es ermöglicht den direkten Zugriff auf die \n"
"Werte über deren vergebenen Namen. Lese- und Schreibaktionen mit dem \n"
"Prozessabbild werden von dem Modul selbst verwaltet, ohne dass sich \n"
"der Programmierer um Offsets und Adressen kümmern muss. Für die \n"
"Gatewaymodule wie ModbusTCP oder Profinet sind eigene 'Inputs' und \n"
"'Outputs' über einen bestimmten Adressbereich definierbar. Auf \n"
"diese IOs kann mit Python3 über den Namen direkt auf die Werte \n"
"zugegriffen werden.",
classifiers=[
"License :: OSI Approved :: "
"GNU Lesser General Public License v3 (LGPLv3)",
"Operating System :: POSIX :: Linux",
"Programming Language :: Python :: 3",
"Topic :: Software Development :: Libraries :: Python Modules"
],
)