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

Docstrings optimieren 1

Browse Source
This commit is contained in:
Sven Sager
2019-10-20 22:36:09 +02:00
parent 56da8e4cf6
commit 10af9a01d5
16 changed files with 828 additions and 587 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
.idea/dictionaries/akira.xml generated Normal file
View File

@@ -0,0 +1,3 @@
<component name="ProjectDictionaryState">
<dictionary name="akira" />
</component>

1
.idea/inspectionProfiles/profiles_settings.xml generated
View File

@@ -1,5 +1,6 @@
<component name="InspectionProjectProfileManager">
<settings>
<option name="PROJECT_PROFILE" value="Default" />
<option name="USE_PROJECT_PROFILE" value="false" />
<version value="1.0" />
</settings>

4
.idea/revpimodio2.iml generated
View File

@@ -1,7 +1,9 @@
<?xml version="1.0" encoding="UTF-8"?>
<module type="PYTHON_MODULE" version="4">
<component name="NewModuleRootManager">
<content url="file://$MODULE_DIR$" />
<content url="file://$MODULE_DIR$">
<sourceFolder url="file://$MODULE_DIR$/revpimodio2" isTestSource="false" />
</content>
<orderEntry type="jdk" jdkName="Python 3.6" jdkType="Python SDK" />
<orderEntry type="sourceFolder" forTests="false" />
</component>

7
docs/revpimodio2.app.rst Normal file
View File

@@ -0,0 +1,7 @@
revpimodio2\.app module
=======================
.. automodule:: revpimodio2.app
:members:
:undoc-members:
:show-inheritance:

7
docs/revpimodio2.device.rst Normal file
View File

@@ -0,0 +1,7 @@
revpimodio2\.device module
==========================
.. automodule:: revpimodio2.device
:members:
:undoc-members:
:show-inheritance:

7
docs/revpimodio2.helper.rst Normal file
View File

@@ -0,0 +1,7 @@
revpimodio2\.helper module
==========================
.. automodule:: revpimodio2.helper
:members:
:undoc-members:
:show-inheritance:

7
docs/revpimodio2.io.rst Normal file
View File

@@ -0,0 +1,7 @@
revpimodio2\.io module
======================
.. automodule:: revpimodio2.io
:members:
:undoc-members:
:show-inheritance:

7
docs/revpimodio2.modio.rst Normal file
View File

@@ -0,0 +1,7 @@
revpimodio2\.modio module
=========================
.. automodule:: revpimodio2.modio
:members:
:undoc-members:
:show-inheritance:

7
docs/revpimodio2.netio.rst Normal file
View File

@@ -0,0 +1,7 @@
revpimodio2\.netio module
=========================
.. automodule:: revpimodio2.netio
:members:
:undoc-members:
:show-inheritance:

63
docs/revpimodio2.rst
View File

@@ -4,62 +4,15 @@ revpimodio2 package
Submodules
----------
revpimodio2\.app module
-----------------------
.. automodule:: revpimodio2.app
:members:
:undoc-members:
:show-inheritance:
revpimodio2\.device module
--------------------------
.. automodule:: revpimodio2.device
:members:
:undoc-members:
:show-inheritance:
revpimodio2\.helper module
--------------------------
.. automodule:: revpimodio2.helper
:members:
:undoc-members:
:show-inheritance:
revpimodio2\.io module
----------------------
.. automodule:: revpimodio2.io
:members:
:undoc-members:
:show-inheritance:
revpimodio2\.modio module
-------------------------
.. automodule:: revpimodio2.modio
:members:
:undoc-members:
:show-inheritance:
revpimodio2\.netio module
-------------------------
.. automodule:: revpimodio2.netio
:members:
:undoc-members:
:show-inheritance:
revpimodio2\.summary module
---------------------------
.. automodule:: revpimodio2.summary
:members:
:undoc-members:
:show-inheritance:
.. toctree::
revpimodio2.app
revpimodio2.device
revpimodio2.helper
revpimodio2.io
revpimodio2.modio
revpimodio2.netio
revpimodio2.summary
Module contents
---------------

7
docs/revpimodio2.summary.rst Normal file
View File

@@ -0,0 +1,7 @@
revpimodio2\.summary module
===========================
.. automodule:: revpimodio2.summary
:members:
:undoc-members:
:show-inheritance:

34
revpimodio2/__init__.py
View File

@@ -1,5 +1,6 @@
# -*- coding: utf-8 -*-
"""Stellt alle Klassen fuer den RevolutionPi zur Verfuegung.
"""
Stellt alle Klassen fuer den RevolutionPi zur Verfuegung.
Webpage: https://revpimodio.org/
@@ -10,11 +11,10 @@ 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.
"""
__all__ = [
"RevPiModIO", "RevPiModIOSelected", "RevPiModIODriver",
"RevPiNetIO", "RevPiNetIOSelected", "RevPiNetIODriver",
"RevPiModIO", "RevPiModIODriver", "RevPiModIOSelected",
"RevPiNetIO", "RevPiNetIODriver", "RevPiNetIOSelected",
"Cycletools",
]
__author__ = "Sven Sager <akira@revpimodio.org>"
@@ -36,27 +36,25 @@ MEM = 302
class DeviceNotFoundError(Exception):
"""Fehler wenn ein Device nicht gefunden wird."""
pass
def acheck(check_type, **kwargs):
"""Check type of given arguments.
def acheck(check_type, **kwargs) -> None:
"""
Check type of given arguments.
Use the argument name as keyword and the argument itself as value.
@param check_type Type to check
@param kwargs Arguments to check
:param check_type: Type to check
:param kwargs: Arguments to check
"""
for var_name in kwargs:
none_okay = var_name.endswith("_noneok")
if not (isinstance(kwargs[var_name], check_type) or
none_okay and kwargs[var_name] is None):
msg = "Argument '{0}' must be {1}{2}".format(
var_name.rstrip("_noneok"), str(check_type),
" or <class 'NoneType'>" if none_okay else ""
@@ -64,14 +62,14 @@ def acheck(check_type, **kwargs):
raise TypeError(msg)
def consttostr(value):
"""Gibt <class 'str'> fuer Konstanten zurueck.
def consttostr(value) -> str:
"""
Gibt <class 'str'> fuer Konstanten zurueck.
Diese Funktion ist erforderlich, da enum in Python 3.2 nicht existiert.
@param value Konstantenwert
@return <class 'str'> Name der Konstanten
:param value: Konstantenwert
:return: <class 'str'> Name der Konstanten
"""
if value == 0:
return "OFF"
@@ -97,5 +95,5 @@ def consttostr(value):
# Benötigte Klassen importieren
from .helper import Cycletools
from .modio import RevPiModIO, RevPiModIOSelected, RevPiModIODriver
from .netio import RevPiNetIO, RevPiNetIOSelected, RevPiNetIODriver
from .modio import RevPiModIO, RevPiModIODriver, RevPiModIOSelected
from .netio import RevPiNetIO, RevPiNetIODriver, RevPiNetIOSelected

19
revpimodio2/app.py
View File

@@ -1,27 +1,36 @@
# -*- coding: utf-8 -*-
"""Bildet die App Sektion von piCtory ab."""
from time import strptime
__author__ = "Sven Sager"
__copyright__ = "Copyright (C) 2018 Sven Sager"
__license__ = "LGPLv3"
from time import strptime
class App(object):
"""Bildet die App Sektion der config.rsc ab."""
__slots__ = "name", "version", "language", "layout", "savets"
def __init__(self, app):
"""Instantiiert die App-Klasse.
@param app piCtory Appinformationen"""
"""
Instantiiert die App-Klasse.
:param app: piCtory Appinformationen
"""
self.name = app["name"]
"""Name of creating app"""
self.version = app["version"]
"""Version of creating app"""
self.language = app["language"]
"""Language of creating app"""
# Speicherungszeitpunkt laden, wenn vorhanden
self.savets = app.get("saveTS", None)
"""Timestamp of configuraiton"""
if self.savets is not None:
try:
self.savets = strptime(self.savets, "%Y%m%d%H%M%S")

571
revpimodio2/device.py
View File

@@ -1,15 +1,15 @@
# -*- coding: utf-8 -*-
"""Modul fuer die Verwaltung der Devices."""
from threading import Event, Lock, Thread
from .helper import ProcimgWriter
__author__ = "Sven Sager"
__copyright__ = "Copyright (C) 2018 Sven Sager"
__license__ = "LGPLv3"
from threading import Thread, Event, Lock
from .helper import ProcimgWriter
class DeviceList(object):
"""Basisklasse fuer direkten Zugriff auf Device Objekte."""
def __init__(self):
@@ -17,9 +17,12 @@ class DeviceList(object):
self.__dict_position = {}
def __contains__(self, key):
"""Prueft ob Device existiert.
@param key DeviceName <class 'str'> / Positionsnummer <class 'int'>
@return True, wenn Device vorhanden"""
"""
Prueft ob Device existiert.
:param key: DeviceName <class 'str'> / Positionsnummer <class 'int'>
:return: True, wenn Device vorhanden
"""
if type(key) == int:
return key in self.__dict_position
elif type(key) == str:
@@ -28,9 +31,12 @@ class DeviceList(object):
return key in self.__dict_position.values()
def __delattr__(self, key, delcomplete=True):
"""Entfernt angegebenes Device.
@param key Device zum entfernen
@param delcomplete Wenn True wird Device komplett entfernt"""
"""
Entfernt angegebenes Device.
:param key: Device zum entfernen
:param delcomplete: Wenn True wird Device komplett entfernt
"""
if delcomplete:
# Device finden
if type(key) == int:
@@ -51,16 +57,22 @@ class DeviceList(object):
object.__delattr__(self, key)
def __delitem__(self, key):
"""Entfernt Device an angegebener Position.
@param key Deviceposition zum entfernen"""
"""
Entfernt Device an angegebener Position.
:param key: Deviceposition zum entfernen
"""
if isinstance(key, Device):
key = key._position
self.__delattr__(key)
def __getitem__(self, key):
"""Gibt angegebenes Device zurueck.
@param key DeviceName <class 'str'> / Positionsnummer <class 'int'>
@return Gefundenes <class 'Device'>-Objekt"""
"""
Gibt angegebenes Device zurueck.
:param key: DeviceName <class 'str'> / Positionsnummer <class 'int'>
:return: Gefundenes <class 'Device'>-Objekt
"""
if type(key) == int:
if key not in self.__dict_position:
raise IndexError("no device on position {0}".format(key))
@@ -69,26 +81,33 @@ class DeviceList(object):
return getattr(self, key)
def __iter__(self):
"""Gibt Iterator aller Devices zurueck.
"""
Gibt Iterator aller Devices zurueck.
Die Reihenfolge ist nach Position im Prozessabbild sortiert und nicht
nach Positionsnummer (Dies entspricht der Positionierung aus piCtory)!
@return <class 'iter'> aller Devices"""
:return: <class 'iter'> aller Devices
"""
for dev in sorted(
self.__dict_position,
key=lambda key: self.__dict_position[key]._offset):
yield self.__dict_position[dev]
def __len__(self):
"""Gibt Anzahl der Devices zurueck.
return Anzahl der Devices"""
"""
Gibt Anzahl der Devices zurueck.
:return: Anzahl der Devices"""
return len(self.__dict_position)
def __setattr__(self, key, value):
"""Setzt Attribute nur wenn Device.
@param key Attributname
@param value Attributobjekt"""
"""
Setzt Attribute nur wenn Device.
:param key: Attributname
:param value: Attributobjekt
"""
if isinstance(value, Device):
object.__setattr__(self, key, value)
self.__dict_position[value._position] = value
@@ -97,13 +116,12 @@ class DeviceList(object):
class Device(object):
"""Basisklasse fuer alle Device-Objekte.
"""
Basisklasse fuer alle Device-Objekte.
Die Basisfunktionalitaet generiert bei Instantiierung alle IOs und
erweitert den Prozessabbildpuffer um die benoetigten Bytes. Sie verwaltet
ihren Prozessabbildpuffer und sorgt fuer die Aktualisierung der IO-Werte.
"""
__slots__ = "__my_io_list", "_ba_devdata", "_ba_datacp", \
@@ -114,12 +132,12 @@ class Device(object):
"guid", "id", "inpvariant", "outvariant", "type"
def __init__(self, parentmodio, dict_device, simulator=False):
"""Instantiierung der Device-Klasse.
@param parent RevpiModIO parent object
@param dict_device <class 'dict'> fuer dieses Device aus piCotry
@param simulator: Laedt das Modul als Simulator und vertauscht IOs
"""
Instantiierung der Device-Klasse.
:param parentmodio: RevpiModIO parent object
:param dict_device: <class 'dict'> fuer dieses Device aus piCotry
:param simulator: Laedt das Modul als Simulator und vertauscht IOs
"""
self._modio = parentmodio
@@ -187,14 +205,20 @@ class Device(object):
self._update_my_io_list()
def __bytes__(self):
"""Gibt alle Daten des Devices als <class 'bytes'> zurueck.
@return Devicedaten als <class 'bytes'>"""
"""
Gibt alle Daten des Devices als <class 'bytes'> zurueck.
:return: Devicedaten als <class 'bytes'>
"""
return bytes(self._ba_devdata)
def __contains__(self, key):
"""Prueft ob IO auf diesem Device liegt.
@param key IO-Name <class 'str'> / IO-Bytenummer <class 'int'>
@return True, wenn IO auf Device vorhanden"""
"""
Prueft ob IO auf diesem Device liegt.
:param key: IO-Name <class 'str'> / IO-Bytenummer <class 'int'>
:return: True, wenn IO auf Device vorhanden
"""
if isinstance(key, IOBase):
# Umwandlung für key
key = key._name
@@ -210,51 +234,66 @@ class Device(object):
and getattr(self._modio.io, key)._parentdevice == self
def __getitem__(self, key):
"""Gibt IO an angegebener Stelle zurueck.
@param key Index des IOs auf dem device als <class 'int'>
@return Gefundenes IO-Objekt"""
"""
Gibt IO an angegebener Stelle zurueck.
:param key: Index des IOs auf dem device als <class 'int'>
:return: Gefundenes IO-Objekt
"""
return self.__my_io_list[key]
def __int__(self):
"""Gibt die Positon im RevPi Bus zurueck.
@return Positionsnummer"""
"""
Gibt die Positon im RevPi Bus zurueck.
:return: Positionsnummer
"""
return self._position
def __iter__(self):
"""Gibt Iterator aller IOs zurueck.
@return <class 'iter'> aller IOs"""
"""
Gibt Iterator aller IOs zurueck.
:return: <class 'iter'> aller IOs
"""
return self.__getioiter(self._slc_devoff, None)
def __len__(self):
"""Gibt Anzahl der Bytes zurueck, die dieses Device belegt.
@return <class 'int'>"""
"""
Gibt Anzahl der Bytes zurueck, die dieses Device belegt.
:return: <class 'int'>
"""
return self._length
def __str__(self):
"""Gibt den Namen des Devices zurueck.
@return Devicename"""
"""
Gibt den Namen des Devices zurueck.
:return: Devicename
"""
return self._name
def __getioiter(self, ioslc, export):
"""Gibt <class 'iter'> mit allen IOs zurueck.
@param ioslc IO Abschnitt <class 'slice'>
@param export Filter fuer 'Export' Flag in piCtory
@return IOs als Iterator
def __getioiter(self, ioslc: slice, export):
"""
Gibt <class 'iter'> mit allen IOs zurueck.
:param ioslc: IO Abschnitt <class 'slice'>
:param export: Filter fuer 'Export' Flag in piCtory
:return: IOs als Iterator
"""
for lst_io in self._modio.io[ioslc]:
for io in lst_io:
if io is not None and (export is None or io.export == export):
yield io
def _buildio(self, dict_io, iotype):
"""Erstellt aus der piCtory-Liste die IOs fuer dieses Device.
@param dict_io <class 'dict'>-Objekt aus piCtory Konfiguration
@param iotype <class 'int'> Wert
@return <class 'slice'> mit Start und Stop Position dieser IOs
def _buildio(self, dict_io: dict, iotype: int) -> slice:
"""
Erstellt aus der piCtory-Liste die IOs fuer dieses Device.
:param dict_io: <class 'dict'>-Objekt aus piCtory Konfiguration
:param iotype: <class 'int'> Wert
:return: <class 'slice'> mit Start und Stop Position dieser IOs
"""
if len(dict_io) <= 0:
return slice(0, 0)
@@ -312,23 +351,32 @@ class Device(object):
"""Funktion zum ueberschreiben von abgeleiteten Klassen."""
pass
def _get_offset(self):
"""Gibt den Deviceoffset im Prozessabbild zurueck.
@return Deviceoffset"""
def _get_offset(self) -> int:
"""
Gibt den Deviceoffset im Prozessabbild zurueck.
:return: Deviceoffset
"""
return self._offset
def _get_producttype(self):
"""Gibt den Produkttypen des device zurueck.
@return Deviceprodukttyp"""
def _get_producttype(self) -> int:
"""
Gibt den Produkttypen des device zurueck.
:return: Deviceprodukttyp
"""
return self._producttype
def _update_my_io_list(self):
def _update_my_io_list(self) -> None:
"""Erzeugt eine neue IO Liste fuer schnellen Zugriff."""
self.__my_io_list = list(self.__iter__())
def autorefresh(self, activate=True):
"""Registriert dieses Device fuer die automatische Synchronisierung.
@param activate Default True fuegt Device zur Synchronisierung hinzu"""
def autorefresh(self, activate=True) -> None:
"""
Registriert dieses Device fuer die automatische Synchronisierung.
:param activate: Default True fuegt Device zur Synchronisierung hinzu
"""
if activate and self not in self._modio._lst_refresh:
# Daten bei Aufnahme direkt einlesen!
@@ -346,7 +394,6 @@ class Device(object):
# Thread starten, wenn er noch nicht läuft
if not self._modio._imgwriter.is_alive():
# Alte Einstellungen speichern
imgrefresh = self._modio._imgwriter.refresh
@@ -369,98 +416,97 @@ class Device(object):
if not self._modio._monitoring:
self._modio.writeprocimg(self)
def get_allios(self, export=None):
"""Gibt eine Liste aller Inputs und Outputs zurueck, keine MEMs.
def get_allios(self, export=None) -> list:
"""
Gibt eine Liste aller Inputs und Outputs zurueck, keine MEMs.
Bleibt Parameter 'export' auf None werden alle Inputs und Outputs
zurueckgegeben. Wird 'export' auf True/False gesetzt, werden nur Inputs
und Outputs zurueckgegeben, bei denen der Wert 'Export' in piCtory
uebereinstimmt.
@param export Nur In-/Outputs mit angegebenen 'Export' Wert in piCtory
@return <class 'list'> Input und Output, keine MEMs
:param export: Nur In-/Outputs mit angegebenen 'Export' Wert in piCtory
:return: <class 'list'> Input und Output, keine MEMs
"""
return list(self.__getioiter(
slice(self._slc_inpoff.start, self._slc_outoff.stop), export
))
def get_inputs(self, export=None):
"""Gibt eine Liste aller Inputs zurueck.
def get_inputs(self, export=None) -> list:
"""
Gibt eine Liste aller Inputs zurueck.
Bleibt Parameter 'export' auf None werden alle Inputs zurueckgegeben.
Wird 'export' auf True/False gesetzt, werden nur Inputs zurueckgegeben,
bei denen der Wert 'Export' in piCtory uebereinstimmt.
@param export Nur Inputs mit angegebenen 'Export' Wert in piCtory
@return <class 'list'> Inputs
:param export: Nur Inputs mit angegebenen 'Export' Wert in piCtory
:return: <class 'list'> Inputs
"""
return list(self.__getioiter(self._slc_inpoff, export))
def get_outputs(self, export=None):
"""Gibt eine Liste aller Outputs zurueck.
def get_outputs(self, export=None) -> list:
"""
Gibt eine Liste aller Outputs zurueck.
Bleibt Parameter 'export' auf None werden alle Outputs zurueckgegeben.
Wird 'export' auf True/False gesetzt, werden nur Outputs
zurueckgegeben, bei denen der Wert 'Export' in piCtory uebereinstimmt.
@param export Nur Outputs mit angegebenen 'Export' Wert in piCtory
@return <class 'list'> Outputs
:param export: Nur Outputs mit angegebenen 'Export' Wert in piCtory
:return: <class 'list'> Outputs
"""
return list(self.__getioiter(self._slc_outoff, export))
def get_memories(self, export=None):
"""Gibt eine Liste aller Memoryobjekte zurueck.
def get_memories(self, export=None) -> list:
"""
Gibt eine Liste aller Memoryobjekte zurueck.
Bleibt Parameter 'export' auf None werden alle Mems zurueckgegeben.
Wird 'export' auf True/False gesetzt, werden nur Mems zurueckgegeben,
bei denen der Wert 'Export' in piCtory uebereinstimmt.
@param export Nur Mems mit angegebenen 'Export' Wert in piCtory
@return <class 'list'> Mems
:param export: Nur Mems mit angegebenen 'Export' Wert in piCtory
:return: <class 'list'> Mems
"""
return list(self.__getioiter(self._slc_memoff, export))
def readprocimg(self):
"""Alle Inputs fuer dieses Device vom Prozessabbild einlesen.
def readprocimg(self) -> bool:
"""
Alle Inputs fuer dieses Device vom Prozessabbild einlesen.
@return True, wenn erfolgreich ausgefuehrt
@see revpimodio2.modio#RevPiModIO.readprocimg
RevPiModIO.readprocimg()
Same see
:return: True, wenn erfolgreich ausgefuehrt
:ref: :func:`revpimodio2.modio.RevPiModIO.readprocimg()`
"""
return self._modio.readprocimg(self)
def setdefaultvalues(self):
"""Alle Outputbuffer fuer dieses Device auf default Werte setzen.
@return True, wenn erfolgreich ausgefuehrt
@see revpimodio2.modio#RevPiModIO.setdefaultvalues
RevPiModIO.setdefaultvalues()
def setdefaultvalues(self) -> None:
"""
Alle Outputbuffer fuer dieses Device auf default Werte setzen.
:return: True, wenn erfolgreich ausgefuehrt
:ref: :func:`revpimodio2.modio.RevPiModIO.setdefaultvalues()`
"""
self._modio.setdefaultvalues(self)
def syncoutputs(self):
"""Lesen aller Outputs im Prozessabbild fuer dieses Device.
@return True, wenn erfolgreich ausgefuehrt
@see revpimodio2.modio#RevPiModIO.syncoutputs
RevPiModIO.syncoutputs()
def syncoutputs(self) -> bool:
"""
Lesen aller Outputs im Prozessabbild fuer dieses Device.
:return: True, wenn erfolgreich ausgefuehrt
:ref: :func:`revpimodio2.modio.RevPiModIO.syncoutputs()`
"""
return self._modio.syncoutputs(self)
def writeprocimg(self):
"""Schreiben aller Outputs dieses Devices ins Prozessabbild.
@return True, wenn erfolgreich ausgefuehrt
@see revpimodio2.modio#RevPiModIO.writeprocimg
RevPiModIO.writeprocimg()
def writeprocimg(self) -> bool:
"""
Schreiben aller Outputs dieses Devices ins Prozessabbild.
:return: True, wenn erfolgreich ausgefuehrt
:ref: :func:`revpimodio2.modio.RevPiModIO.writeprocimg()`
"""
return self._modio.writeprocimg(self)
@@ -472,7 +518,6 @@ class Device(object):
class Base(Device):
"""Klasse fuer alle Base-Devices wie Core / Connect usw."""
__slots__ = ()
@@ -481,11 +526,10 @@ class Base(Device):
class Core(Base):
"""Klasse fuer den RevPi Core.
"""
Klasse fuer den RevPi Core.
Stellt Funktionen fuer die LEDs und den Status zur Verfuegung.
"""
__slots__ = "_slc_cycle", "_slc_errorcnt", "_slc_statusbyte", \
@@ -502,7 +546,7 @@ class Core(Base):
else:
object.__setattr__(self, key, value)
def _devconfigure(self):
def _devconfigure(self) -> None:
"""Core-Klasse vorbereiten."""
# Statische IO Verknüpfungen je nach Core-Variante
@@ -564,10 +608,13 @@ class Core(Base):
exp_a2red, None, "LED_A2_RED", "3"
], OUT, "little", False)
def __errorlimit(self, slc_io, errorlimit):
"""Verwaltet das Schreiben der ErrorLimits.
@param slc_io Byte Slice vom ErrorLimit
@return Aktuellen ErrorLimit oder None wenn nicht verfuegbar"""
def __errorlimit(self, slc_io: slice, errorlimit: int) -> None:
"""
Verwaltet das Schreiben der ErrorLimits.
:param slc_io: Byte Slice vom ErrorLimit
:return: Aktuellen ErrorLimit oder None wenn nicht verfuegbar
"""
if 0 <= errorlimit <= 65535:
self._ba_devdata[slc_io] = \
errorlimit.to_bytes(2, byteorder="little")
@@ -576,16 +623,22 @@ class Core(Base):
"errorlimit value must be between 0 and 65535"
)
def _get_status(self):
"""Gibt den RevPi Core Status zurueck.
@return Status als <class 'int'>"""
def _get_status(self) -> int:
"""
Gibt den RevPi Core Status zurueck.
:return: Status als <class 'int'>
"""
return int.from_bytes(
self._ba_devdata[self._slc_statusbyte], byteorder="little"
)
def _get_leda1(self):
"""Gibt den Zustand der LED A1 vom Core zurueck.
@return 0=aus, 1=gruen, 2=rot"""
def _get_leda1(self) -> int:
"""
Gibt den Zustand der LED A1 vom Core zurueck.
:return: 0=aus, 1=gruen, 2=rot
"""
int_led = int.from_bytes(
self._ba_devdata[self._slc_led], byteorder="little"
)
@@ -593,9 +646,12 @@ class Core(Base):
led += int_led & 2
return led
def _get_leda2(self):
"""Gibt den Zustand der LED A2 vom Core zurueck.
@return 0=aus, 1=gruen, 2=rot"""
def _get_leda2(self) -> int:
"""
Gibt den Zustand der LED A2 vom Core zurueck.
:return: 0=aus, 1=gruen, 2=rot
"""
int_led = int.from_bytes(
self._ba_devdata[self._slc_led], byteorder="little"
) >> 2
@@ -603,10 +659,14 @@ class Core(Base):
led += int_led & 2
return led
def _set_calculatedled(self, addresslist, shifted_value):
"""Berechnet und setzt neuen Bytewert fuer LED byte.
@param addresslist Liste der Vergleicher
@param shifed_value Bits vergleichen"""
def _set_calculatedled(self, addresslist, shifted_value) -> None:
"""
Berechnet und setzt neuen Bytewert fuer LED byte.
:param addresslist: Liste der Vergleicher
:param shifted_value: Bits vergleichen
"""
# TODO: Docstring
# Byte als int holen
int_led = int.from_bytes(
self._ba_devdata[self._slc_led], byteorder="little"
@@ -625,17 +685,23 @@ class Core(Base):
self._ba_devdata[self._slc_led] = \
int_led.to_bytes(length=1, byteorder="little")
def _set_leda1(self, value):
"""Setzt den Zustand der LED A1 vom Core.
@param value 0=aus, 1=gruen, 2=rot"""
def _set_leda1(self, value: int) -> None:
"""
Setzt den Zustand der LED A1 vom Core.
:param value: 0=aus, 1=gruen, 2=rot
"""
if 0 <= value <= 3:
self._set_calculatedled([1, 2], value)
else:
raise ValueError("led status must be between 0 and 3")
def _set_leda2(self, value):
"""Setzt den Zustand der LED A2 vom Core.
@param value 0=aus, 1=gruen, 2=rot"""
def _set_leda2(self, value: int) -> None:
"""
Setzt den Zustand der LED A2 vom Core.
:param value: 0=aus, 1=gruen, 2=rot
"""
if 0 <= value <= 3:
self._set_calculatedled([4, 8], value << 2)
else:
@@ -646,97 +712,133 @@ class Core(Base):
status = property(_get_status)
@property
def picontrolrunning(self):
"""Statusbit fuer piControl-Treiber laeuft.
@return True, wenn Treiber laeuft"""
def picontrolrunning(self) -> bool:
"""
Statusbit fuer piControl-Treiber laeuft.
:return: True, wenn Treiber laeuft
"""
return bool(int.from_bytes(
self._ba_devdata[self._slc_statusbyte], byteorder="little"
) & 1)
@property
def unconfdevice(self):
"""Statusbit fuer ein IO-Modul nicht mit PiCtory konfiguriert.
@return True, wenn IO Modul nicht konfiguriert"""
def unconfdevice(self) -> bool:
"""
Statusbit fuer ein IO-Modul nicht mit PiCtory konfiguriert.
:return: True, wenn IO Modul nicht konfiguriert
"""
return bool(int.from_bytes(
self._ba_devdata[self._slc_statusbyte], byteorder="little"
) & 2)
@property
def missingdeviceorgate(self):
"""Statusbit fuer ein IO-Modul fehlt oder piGate konfiguriert.
@return True, wenn IO-Modul fehlt oder piGate konfiguriert"""
def missingdeviceorgate(self) -> bool:
"""
Statusbit fuer ein IO-Modul fehlt oder piGate konfiguriert.
:return: True, wenn IO-Modul fehlt oder piGate konfiguriert
"""
return bool(int.from_bytes(
self._ba_devdata[self._slc_statusbyte], byteorder="little"
) & 4)
@property
def overunderflow(self):
"""Statusbit Modul belegt mehr oder weniger Speicher als konfiguriert.
@return True, wenn falscher Speicher belegt ist"""
def overunderflow(self) -> bool:
"""
Statusbit Modul belegt mehr oder weniger Speicher als konfiguriert.
:return: True, wenn falscher Speicher belegt ist
"""
return bool(int.from_bytes(
self._ba_devdata[self._slc_statusbyte], byteorder="little"
) & 8)
@property
def leftgate(self):
"""Statusbit links vom RevPi ist ein piGate Modul angeschlossen.
@return True, wenn piGate links existiert"""
def leftgate(self) -> bool:
"""
Statusbit links vom RevPi ist ein piGate Modul angeschlossen.
:return: True, wenn piGate links existiert
"""
return bool(int.from_bytes(
self._ba_devdata[self._slc_statusbyte], byteorder="little"
) & 16)
@property
def rightgate(self):
"""Statusbit rechts vom RevPi ist ein piGate Modul angeschlossen.
@return True, wenn piGate rechts existiert"""
def rightgate(self) -> bool:
"""
Statusbit rechts vom RevPi ist ein piGate Modul angeschlossen.
:return: True, wenn piGate rechts existiert
"""
return bool(int.from_bytes(
self._ba_devdata[self._slc_statusbyte], byteorder="little"
) & 32)
@property
def iocycle(self):
"""Gibt Zykluszeit der Prozessabbildsynchronisierung zurueck.
@return Zykluszeit in ms"""
return None if self._slc_cycle is None else int.from_bytes(
def iocycle(self) -> int:
"""
Gibt Zykluszeit der Prozessabbildsynchronisierung zurueck.
:return: Zykluszeit in ms ( -1 wenn nicht verfuegbar)
"""
return -1 if self._slc_cycle is None else int.from_bytes(
self._ba_devdata[self._slc_cycle], byteorder="little"
)
@property
def temperature(self):
"""Gibt CPU-Temperatur zurueck.
@return CPU-Temperatur in Celsius"""
return None if self._slc_temperature is None else int.from_bytes(
def temperature(self) -> int:
"""
Gibt CPU-Temperatur zurueck.
:return: CPU-Temperatur in Celsius (-273 wenn nich verfuegbar)
"""
return -273 if self._slc_temperature is None else int.from_bytes(
self._ba_devdata[self._slc_temperature], byteorder="little"
)
@property
def frequency(self):
"""Gibt CPU Taktfrequenz zurueck.
@return CPU Taktfrequenz in MHz"""
return None if self._slc_frequency is None else int.from_bytes(
def frequency(self) -> int:
"""
Gibt CPU Taktfrequenz zurueck.
:return: CPU Taktfrequenz in MHz (-1 wenn nicht verfuegbar)
"""
return -1 if self._slc_frequency is None else int.from_bytes(
self._ba_devdata[self._slc_frequency], byteorder="little"
) * 10
@property
def ioerrorcount(self):
"""Gibt Fehleranzahl auf RS485 piBridge Bus zurueck.
@return Fehleranzahl der piBridge"""
return None if self._slc_errorcnt is None else int.from_bytes(
def ioerrorcount(self) -> int:
"""
Gibt Fehleranzahl auf RS485 piBridge Bus zurueck.
:return: Fehleranzahl der piBridge (-1 wenn nicht verfuegbar)
"""
return -1 if self._slc_errorcnt is None else int.from_bytes(
self._ba_devdata[self._slc_errorcnt], byteorder="little"
)
@property
def errorlimit1(self):
"""Gibt RS485 ErrorLimit1 Wert zurueck.
@return Aktueller Wert fuer ErrorLimit1"""
return None if self._slc_errorlimit1 is None else int.from_bytes(
def errorlimit1(self) -> int:
"""
Gibt RS485 ErrorLimit1 Wert zurueck.
:return: Aktueller Wert fuer ErrorLimit1 (-1 wenn nicht verfuegbar)
"""
return -1 if self._slc_errorlimit1 is None else int.from_bytes(
self._ba_devdata[self._slc_errorlimit1], byteorder="little"
)
@errorlimit1.setter
def errorlimit1(self, value):
"""Setzt RS485 ErrorLimit1 auf neuen Wert.
@param value Neuer ErrorLimit1 Wert"""
def errorlimit1(self, value: int) -> None:
"""
Setzt RS485 ErrorLimit1 auf neuen Wert.
:param value: Neuer ErrorLimit1 Wert
"""
if self._slc_errorlimit1 is None:
raise RuntimeError(
"selected core item in piCtory does not support errorlimit1"
@@ -745,17 +847,23 @@ class Core(Base):
self.__errorlimit(self._slc_errorlimit1, value)
@property
def errorlimit2(self):
"""Gibt RS485 ErrorLimit2 Wert zurueck.
@return Aktueller Wert fuer ErrorLimit2"""
return None if self._slc_errorlimit2 is None else int.from_bytes(
def errorlimit2(self) -> int:
"""
Gibt RS485 ErrorLimit2 Wert zurueck.
:return: Aktueller Wert fuer ErrorLimit2 (-1 wenn nicht verfuegbar)
"""
return -1 if self._slc_errorlimit2 is None else int.from_bytes(
self._ba_devdata[self._slc_errorlimit2], byteorder="little"
)
@errorlimit2.setter
def errorlimit2(self, value):
"""Setzt RS485 ErrorLimit2 auf neuen Wert.
@param value Neuer ErrorLimit2 Wert"""
def errorlimit2(self, value: int) -> None:
"""
Setzt RS485 ErrorLimit2 auf neuen Wert.
:param value: Neuer ErrorLimit2 Wert
"""
if self._slc_errorlimit2 is None:
raise RuntimeError(
"selected core item in piCtory does not support errorlimit2"
@@ -765,11 +873,9 @@ class Core(Base):
class Connect(Core):
"""Klasse fuer den RevPi Connect.
Stellt Funktionen fuer die LEDs, Watchdog und den Status zur Verfuegung.
"""
__slots__ = "__evt_wdtoggle", "__th_wdtoggle", "a3green", "a3red", "wd", \
@@ -786,12 +892,12 @@ class Connect(Core):
else:
object.__setattr__(self, key, value)
def __wdtoggle(self):
def __wdtoggle(self) -> None:
"""WD Ausgang alle 10 Sekunden automatisch toggeln."""
while not self.__evt_wdtoggle.wait(10):
self.wd.value = not self.wd.value
def _devconfigure(self):
def _devconfigure(self) -> None:
"""Connect-Klasse vorbereiten."""
super()._devconfigure()
@@ -841,9 +947,12 @@ class Connect(Core):
exp_x2out, None, "Connect_X2_OUT", "6"
], OUT, "little", False)
def _get_leda3(self):
"""Gibt den Zustand der LED A3 vom Connect zurueck.
@return 0=aus, 1=gruen, 2=rot"""
def _get_leda3(self) -> int:
"""
Gibt den Zustand der LED A3 vom Connect zurueck.
:return: 0=aus, 1=gruen, 2=rot
"""
int_led = int.from_bytes(
self._ba_devdata[self._slc_led], byteorder="little"
) >> 4
@@ -851,22 +960,28 @@ class Connect(Core):
led += int_led & 2
return led
def _get_wdtoggle(self):
"""Ruft den Wert fuer Autowatchdog ab.
@return True, wenn Autowatchdog aktiv ist"""
return self.__th_wdtoggle is not None \
and self.__th_wdtoggle.is_alive()
def _get_wdtoggle(self) -> bool:
"""
Ruft den Wert fuer Autowatchdog ab.
def _set_leda3(self, value):
"""Setzt den Zustand der LED A3 vom Connect.
@param value 0=aus, 1=gruen, 2=rot"""
:return: True, wenn Autowatchdog aktiv ist
"""
return self.__th_wdtoggle is not None and self.__th_wdtoggle.is_alive()
def _set_leda3(self, value: int) -> None:
"""
Setzt den Zustand der LED A3 vom Connect.
:param: value 0=aus, 1=gruen, 2=rot
"""
if 0 <= value <= 3:
self._set_calculatedled([16, 32], value << 4)
else:
raise ValueError("led status must be between 0 and 3")
def _set_wdtoggle(self, value):
"""Setzt den Wert fuer Autowatchdog.
def _set_wdtoggle(self, value: bool) -> None:
"""
Setzt den Wert fuer Autowatchdog.
Wird dieser Wert auf True gesetzt, wechselt im Hintergrund das noetige
Bit zum toggeln des Watchdogs alle 10 Sekunden zwichen True und False.
@@ -877,7 +992,8 @@ class Connect(Core):
.writeprocimg() aufgerufen werden, um den Wert in das
Prozessabbild zu schreiben!!!
@param value True zum aktivieren, Fals zum beenden"""
:param value: True zum aktivieren, Fals zum beenden
"""
if self._modio._monitoring:
raise RuntimeError(
"can not toggle watchdog, while system is in monitoring mode"
@@ -901,15 +1017,16 @@ class Connect(Core):
class DioModule(Device):
"""Stellt ein DIO / DI / DO Modul dar."""
__slots__ = ("_lst_counter")
__slots__ = "_lst_counter"
def __init__(self, parentmodio, dict_device, simulator=False):
"""Erweitert Device-Klasse zum Erkennen von IntIOCounter.
@see #Device.__init__ Device.__init__(...)"""
"""
Erweitert Device-Klasse zum Erkennen von IntIOCounter.
:rev: :func:`Device.__init__()`
"""
# Stringliste der Byteadressen (alle Module sind gleich)
self._lst_counter = list(map(str, range(6, 70, 4)))
@@ -918,23 +1035,26 @@ class DioModule(Device):
class Gateway(Device):
"""Klasse fuer die RevPi Gateway-Devices.
"""
Klasse fuer die RevPi Gateway-Devices.
Stellt neben den Funktionen von RevPiDevice weitere Funktionen fuer die
Gateways bereit. IOs auf diesem Device stellen die replace_io Funktion
zur verfuegung, ueber die eigene IOs definiert werden, die ein
RevPiStructIO-Objekt abbilden.
Dieser IO-Typ kann Werte ueber mehrere Bytes verarbeiten und zurueckgeben.
@see revpimodio2.io#IntIOReplaceable.replace_io replace_io(...)
:ref: :func:`revpimodio2.io.IntIOReplaceable.replace_io()`
"""
__slots__ = "_dict_slc"
def __init__(self, parent, dict_device, simulator=False):
"""Erweitert Device-Klasse um get_rawbytes-Funktionen.
@see #Device.__init__ Device.__init__(...)"""
"""
Erweitert Device-Klasse um get_rawbytes-Funktionen.
:ref: :func:`Device.__init__()`
"""
super().__init__(parent, dict_device, simulator)
self._dict_slc = {
@@ -943,28 +1063,32 @@ class Gateway(Device):
MEM: self._slc_mem
}
def get_rawbytes(self):
"""Gibt die Bytes aus, die dieses Device verwendet.
@return <class 'bytes'> des Devices"""
def get_rawbytes(self) -> bytes:
"""
Gibt die Bytes aus, die dieses Device verwendet.
:return: <class 'bytes'> des Devices
"""
return bytes(self._ba_devdata)
class Virtual(Gateway):
"""Klasse fuer die RevPi Virtual-Devices.
"""
Klasse fuer die RevPi Virtual-Devices.
Stellt die selben Funktionen wie Gateway zur Verfuegung. Es koennen
ueber die reg_*-Funktionen eigene IOs definiert werden, die ein
RevPiStructIO-Objekt abbilden.
Dieser IO-Typ kann Werte ueber mehrere Bytes verarbeiten und zurueckgeben.
@see #Gateway Gateway
:ref: :func:`Gateway`
"""
__slots__ = ()
def writeinputdefaults(self):
"""Schreibt fuer ein virtuelles Device piCtory Defaultinputwerte.
"""
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
@@ -972,8 +1096,7 @@ class Virtual(Gateway):
gehen diese Werte verloren.
Diese Funktion kann nur auf virtuelle Devices angewendet werden!
@return True, wenn Arbeiten am virtuellen Device erfolgreich waren
:return: True, wenn Arbeiten am virtuellen Device erfolgreich waren
"""
if self._modio._monitoring:
raise RuntimeError(

195
revpimodio2/helper.py
View File

@@ -1,19 +1,19 @@
# -*- coding: utf-8 -*-
"""RevPiModIO Helperklassen und Tools."""
__author__ = "Sven Sager"
__copyright__ = "Copyright (C) 2018 Sven Sager"
__license__ = "LGPLv3"
import queue
import warnings
from math import ceil
from threading import Event, Lock, Thread
from timeit import default_timer
from revpimodio2 import RISING, FALLING, BOTH
from revpimodio2 import BOTH, FALLING, RISING
__author__ = "Sven Sager"
__copyright__ = "Copyright (C) 2018 Sven Sager"
__license__ = "LGPLv3"
class EventCallback(Thread):
"""Thread fuer das interne Aufrufen von Event-Funktionen.
Der Eventfunktion, welche dieser Thread aufruft, wird der Thread selber
@@ -34,18 +34,17 @@ class EventCallback(Thread):
while not th.exit.is_set():
# IO-Arbeiten
th.exit.wait(0.5)
"""
__slots__ = "daemon", "exit", "func", "ioname", "iovalue"
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
def __init__(self, func, name: str, 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
@@ -63,9 +62,9 @@ class EventCallback(Thread):
self.exit.set()
class Cycletools():
"""Werkzeugkasten fuer Cycleloop-Funktion.
class Cycletools:
"""
Werkzeugkasten fuer Cycleloop-Funktion.
Diese Klasse enthaelt Werkzeuge fuer Zyklusfunktionen, wie Taktmerker
und Flankenmerker.
@@ -84,7 +83,6 @@ class Cycletools():
Diese Merker koennen z.B. verwendet werden um, an Outputs angeschlossene,
Lampen synchron blinken zu lassen.
"""
__slots__ = "__cycle", "__cycletime", "__ucycle", \
@@ -117,10 +115,13 @@ class Cycletools():
# Benutzerdaten
class Var:
"""Hier remanente Variablen anfuegen."""
pass
self.var = Var()
def _docycle(self):
def _docycle(self) -> None:
"""Zyklusarbeiten."""
# Einschaltverzoegerung
for tof in self.__dict_tof:
@@ -174,54 +175,66 @@ class Cycletools():
self.flag5c = not self.flag5c
self.__cycle = 0
def get_tof(self, name):
"""Wert der Ausschaltverzoegerung.
@param name Eindeutiger Name des Timers
@return Wert <class 'bool'> der Ausschaltverzoegerung"""
def get_tof(self, name: str) -> bool:
"""
Wert der Ausschaltverzoegerung.
:param name: Eindeutiger Name des Timers
:return: Wert <class 'bool'> der Ausschaltverzoegerung
"""
return self.__dict_tof.get(name, 0) > 0
def get_tofc(self, name):
"""Wert der Ausschaltverzoegerung.
@param name Eindeutiger Name des Timers
@return Wert <class 'bool'> der Ausschaltverzoegerung"""
def get_tofc(self, name: str) -> bool:
"""
Wert der Ausschaltverzoegerung.
:param name: Eindeutiger Name des Timers
:return: Wert <class 'bool'> der Ausschaltverzoegerung
"""
return self.__dict_tof.get(name, 0) > 0
def set_tof(self, name, milliseconds):
"""Startet bei Aufruf einen ausschaltverzoegerten Timer.
@param name Eindeutiger Name fuer Zugriff auf Timer
@param milliseconds Verzoegerung in Millisekunden
def set_tof(self, name: str, milliseconds: int) -> None:
"""
Startet bei Aufruf einen ausschaltverzoegerten Timer.
:param name: Eindeutiger Name fuer Zugriff auf Timer
:param milliseconds: Verzoegerung in Millisekunden
"""
self.__dict_tof[name] = ceil(milliseconds / self.__cycletime)
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
def set_tofc(self, name: str, cycles: int) -> None:
"""
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_ton(self, name):
"""Einschaltverzoegerung.
@param name Eindeutiger Name des Timers
@return Wert <class 'bool'> der Einschaltverzoegerung"""
def get_ton(self, name: str) -> bool:
"""
Einschaltverzoegerung.
:param name: Eindeutiger Name des Timers
:return: Wert <class 'bool'> der Einschaltverzoegerung
"""
return self.__dict_ton.get(name, [-1])[0] == 0
def get_tonc(self, name):
"""Einschaltverzoegerung.
@param name Eindeutiger Name des Timers
@return Wert <class 'bool'> der Einschaltverzoegerung"""
def get_tonc(self, name: str) -> bool:
"""
Einschaltverzoegerung.
:param name: Eindeutiger Name des Timers
:return: Wert <class 'bool'> der Einschaltverzoegerung
"""
return self.__dict_ton.get(name, [-1])[0] == 0
def set_ton(self, name, milliseconds):
"""Startet einen einschaltverzoegerten Timer.
@param name Eindeutiger Name fuer Zugriff auf Timer
@param milliseconds Millisekunden, der Verzoegerung wenn neu gestartet
def set_ton(self, name: str, milliseconds: int) -> None:
"""
Startet einen einschaltverzoegerten Timer.
:param name: Eindeutiger Name fuer Zugriff auf Timer
:param milliseconds: Millisekunden, der Verzoegerung wenn neu gestartet
"""
if self.__dict_ton.get(name, [-1])[0] == -1:
self.__dict_ton[name] = \
@@ -229,36 +242,42 @@ class Cycletools():
else:
self.__dict_ton[name][1] = True
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
def set_tonc(self, name: str, cycles: int) -> None:
"""
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_tp(self, name):
"""Impulstimer.
@param name Eindeutiger Name des Timers
@return Wert <class 'bool'> des Impulses"""
def get_tp(self, name: str) -> bool:
"""
Impulstimer.
:param name: Eindeutiger Name des Timers
:return: Wert <class 'bool'> des Impulses
"""
return self.__dict_tp.get(name, [-1])[0] > 0
def get_tpc(self, name):
"""Impulstimer.
@param name Eindeutiger Name des Timers
@return Wert <class 'bool'> des Impulses"""
def get_tpc(self, name: str) -> bool:
"""
Impulstimer.
:param name: Eindeutiger Name des Timers
:return: Wert <class 'bool'> des Impulses
"""
return self.__dict_tp.get(name, [-1])[0] > 0
def set_tp(self, name, milliseconds):
"""Startet einen Impuls Timer.
@param name Eindeutiger Name fuer Zugriff auf Timer
@param milliseconds Millisekunden, die der Impuls anstehen soll
def set_tp(self, name: str, milliseconds: int) -> None:
"""
Startet einen Impuls Timer.
:param name: Eindeutiger Name fuer Zugriff auf Timer
:param milliseconds: Millisekunden, die der Impuls anstehen soll
"""
if self.__dict_tp.get(name, [-1])[0] == -1:
self.__dict_tp[name] = \
@@ -266,12 +285,12 @@ class Cycletools():
else:
self.__dict_tp[name][1] = True
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
def set_tpc(self, name: str, cycles: int) -> None:
"""
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]
@@ -280,13 +299,12 @@ class Cycletools():
class ProcimgWriter(Thread):
"""Klasse fuer Synchroniseriungs-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.
"""
__slots__ = "__dict_delay", "__eventth", "_eventqth", "__eventwork", \
@@ -294,8 +312,7 @@ class ProcimgWriter(Thread):
"_refresh", "_work", "daemon", "lck_refresh", "newdata"
def __init__(self, parentmodio):
"""Init ProcimgWriter class.
@param parentmodio Parent Object"""
"""Init ProcimgWriter class."""
super().__init__()
self.__dict_delay = {}
self.__eventth = Thread(target=self.__exec_th)
@@ -311,7 +328,7 @@ class ProcimgWriter(Thread):
self.lck_refresh = Lock()
self.newdata = Event()
def __check_change(self, dev):
def __check_change(self, dev) -> None:
"""Findet Aenderungen fuer die Eventueberwachung."""
for io_event in dev._dict_events:
@@ -386,7 +403,7 @@ class ProcimgWriter(Thread):
# Nach Verarbeitung aller IOs die Bytes kopieren (Lock ist noch drauf)
dev._ba_datacp = dev._ba_devdata[:]
def __exec_th(self):
def __exec_th(self) -> None:
"""Laeuft als Thread, der Events als Thread startet."""
while self.__eventwork:
try:
@@ -398,10 +415,13 @@ class ProcimgWriter(Thread):
except queue.Empty:
pass
def _collect_events(self, value):
"""Aktiviert oder Deaktiviert die Eventueberwachung.
@param value True aktiviert / False deaktiviert
@return True, wenn Anforderung erfolgreich war"""
def _collect_events(self, value: bool) -> bool:
"""
Aktiviert oder Deaktiviert die Eventueberwachung.
:param value: True aktiviert / False deaktiviert
:return: True, wenn Anforderung erfolgreich war
"""
if type(value) != bool:
raise TypeError("value must be <class 'bool'>")
@@ -427,9 +447,12 @@ class ProcimgWriter(Thread):
return True
def get_refresh(self):
"""Gibt Zykluszeit zurueck.
@return <class 'int'> Zykluszeit in Millisekunden"""
def get_refresh(self) -> int:
"""
Gibt Zykluszeit zurueck.
:return: <class 'int'> Zykluszeit in Millisekunden
"""
return int(self._refresh * 1000)
def run(self):

428
revpimodio2/io.py
View File

@@ -1,13 +1,15 @@
# -*- coding: utf-8 -*-
"""RevPiModIO Modul fuer die Verwaltung der IOs."""
import struct
from re import match as rematch
from threading import Event
from revpimodio2 import BOTH, FALLING, INP, MEM, RISING, consttostr
__author__ = "Sven Sager"
__copyright__ = "Copyright (C) 2018 Sven Sager"
__license__ = "LGPLv3"
import struct
from re import match as rematch
from threading import Event
from revpimodio2 import RISING, FALLING, BOTH, INP, MEM, consttostr
try:
# Funktioniert nur auf Unix
from fcntl import ioctl
@@ -16,7 +18,6 @@ except Exception:
class IOEvent(object):
"""Basisklasse fuer IO-Events."""
__slots__ = "as_thread", "delay", "edge", "func", "overwrite", "prefire"
@@ -32,7 +33,6 @@ class IOEvent(object):
class IOList(object):
"""Basisklasse fuer direkten Zugriff auf IO Objekte."""
def __init__(self):
@@ -41,17 +41,23 @@ class IOList(object):
self.__dict_iorefname = {}
def __contains__(self, key):
"""Prueft ob IO existiert.
@param key IO-Name <class 'str'> oder Bytenummer <class 'int'>
@return True, wenn IO vorhanden / Byte belegt"""
"""
Prueft ob IO existiert.
:param key: IO-Name <class 'str'> oder Bytenummer <class 'int'>
:return: True, wenn IO vorhanden / Byte belegt
"""
if type(key) == int:
return len(self.__dict_iobyte.get(key, [])) > 0
else:
return hasattr(self, key) and type(getattr(self, key)) != DeadIO
def __delattr__(self, key):
"""Entfernt angegebenen IO.
@param key IO zum entfernen"""
"""
Entfernt angegebenen IO.
:param key: IO zum entfernen
"""
io_del = object.__getattribute__(self, key)
# Alte Events vom Device löschen
@@ -70,16 +76,20 @@ class IOList(object):
io_del._parentdevice._update_my_io_list()
def __getattr__(self, key):
"""Verwaltet geloeschte IOs (Attribute, die nicht existieren).
@param key Name oder Byte eines alten IOs
@return Alten IO, wenn in Ref-Listen"""
"""
Verwaltet geloeschte IOs (Attribute, die nicht existieren).
:param key: Name oder Byte eines alten IOs
:return: Alten IO, wenn in Ref-Listen
"""
if key in self.__dict_iorefname:
return self.__dict_iorefname[key]
else:
raise AttributeError("can not find io '{0}'".format(key))
def __getitem__(self, key):
"""Ruft angegebenen IO ab.
"""
Ruft angegebenen IO ab.
Wenn der Key <class 'str'> ist, wird ein einzelner IO geliefert. Wird
der Key als <class 'int'> uebergeben, wird eine <class 'list'>
@@ -87,9 +97,8 @@ class IOList(object):
Wird als Key <class 'slice'> gegeben, werden die Listen in einer Liste
zurueckgegeben.
@param key IO Name als <class 'str> oder Byte als <class 'int'>.
@return IO Objekt oder Liste der IOs
:param key: IO Name als <class 'str> oder Byte als <class 'int'>.
:return: IO Objekt oder Liste der IOs
"""
if type(key) == int:
if key not in self.__dict_iobyte:
@@ -106,16 +115,22 @@ class IOList(object):
return getattr(self, key)
def __iter__(self):
"""Gibt Iterator aller IOs zurueck.
@return Iterator aller IOs"""
"""
Gibt Iterator aller IOs zurueck.
:return: Iterator aller IOs
"""
for int_io in sorted(self.__dict_iobyte):
for io in self.__dict_iobyte[int_io]:
if io is not None:
yield io
def __len__(self):
"""Gibt die Anzahl aller IOs zurueck.
@return Anzahl aller IOs"""
"""
Gibt die Anzahl aller IOs zurueck.
:return: Anzahl aller IOs
"""
int_ios = 0
for int_io in self.__dict_iobyte:
for io in self.__dict_iobyte[int_io]:
@@ -135,10 +150,12 @@ class IOList(object):
"direct assignment is not supported - use .value Attribute"
)
def __private_replace_oldio_with_newio(self, io):
"""Ersetzt bestehende IOs durch den neu Registrierten.
@param io Neuer IO der eingefuegt werden soll"""
def __private_replace_oldio_with_newio(self, io) -> None:
"""
Ersetzt bestehende IOs durch den neu Registrierten.
:param io: Neuer IO der eingefuegt werden soll
"""
# Scanbereich festlegen
if io._bitaddress < 0:
scan_start = io.address
@@ -195,9 +212,12 @@ class IOList(object):
io._parentio_address - io.address
] & (1 << io._bitaddress))
def _private_register_new_io_object(self, new_io):
"""Registriert neues IO Objekt unabhaenging von __setattr__.
@param new_io Neues IO Objekt"""
def _private_register_new_io_object(self, new_io) -> None:
"""
Registriert neues IO Objekt unabhaenging von __setattr__.
:param new_io: Neues IO Objekt
"""
if isinstance(new_io, IOBase):
if hasattr(self, new_io._name):
raise AttributeError(
@@ -229,27 +249,32 @@ class IOList(object):
class DeadIO(object):
"""Klasse, mit der ersetzte IOs verwaltet werden."""
__slots__ = "__deadio"
def __init__(self, deadio):
"""Instantiierung der DeadIO-Klasse.
@param deadio IO, der ersetzt wurde"""
"""
Instantiierung der DeadIO-Klasse.
:param deadio: IO, der ersetzt wurde
"""
self.__deadio = deadio
def replace_io(self, name, frm, **kwargs):
"""Stellt Funktion fuer weiter Bit-Ersetzungen bereit.
@see #IntIOReplaceable.replace_io replace_io(...)"""
def replace_io(self, name: str, frm: str, **kwargs) -> None:
"""
Stellt Funktion fuer weiter Bit-Ersetzungen bereit.
:ref: :func:IntIOReplaceable.replace_io()
"""
self.__deadio.replace_io(name, frm, **kwargs)
_parentdevice = property(lambda self: None)
class IOBase(object):
"""Basisklasse fuer alle IO-Objekte.
"""
Basisklasse fuer alle IO-Objekte.
Die Basisfunktionalitaet ermoeglicht das Lesen und Schreiben der Werte
als <class bytes'> oder <class 'bool'>. Dies entscheidet sich bei der
@@ -259,7 +284,6 @@ class IOBase(object):
Diese Klasse dient als Basis fuer andere IO-Klassen mit denen die Werte
auch als <class 'int'> verwendet werden koennen.
"""
__slots__ = "__bit_ioctl_off", "__bit_ioctl_on", \
@@ -267,16 +291,16 @@ class IOBase(object):
"_iotype", "_length", "_name", "_parentdevice", \
"_signed", "_slc_address", "bmk", "export"
def __init__(self, parentdevice, valuelist, iotype, byteorder, signed):
"""Instantiierung der IOBase-Klasse.
def __init__(self, parentdevice, valuelist: list, iotype: int, byteorder: str, signed: bool):
"""
Instantiierung der IOBase-Klasse.
@param parentdevice Parentdevice auf dem der IO liegt
@param valuelist Datenliste fuer Instantiierung
:param parentdevice: Parentdevice auf dem der IO liegt
:param valuelist: Datenliste fuer Instantiierung
["name","defval","bitlen","startaddrdev",exp,"idx","bmk","bitaddr"]
@param iotype <class 'int'> Wert
@param byteorder Byteorder 'little'/'big' fuer <class 'int'> Berechnung
@param sigend Intberechnung mit Vorzeichen durchfuehren
:param iotype: <class 'int'> Wert
:param byteorder: Byteorder 'little'/'big' fuer <class 'int'> Berechnung
:param signed: Intberechnung mit Vorzeichen durchfuehren
"""
# ["name","defval","bitlen","startaddrdev",exp,"idx","bmk","bitaddr"]
# [ 0 , 1 , 2 , 3 , 4 , 5 , 6 , 7 ]
@@ -358,8 +382,11 @@ class IOBase(object):
self.__bit_ioctl_on = self.__bit_ioctl_off + b'\x01'
def __bool__(self):
"""<class 'bool'>-Wert der Klasse.
@return <class 'bool'> Nur False wenn False oder 0 sonst True"""
"""
<class 'bool'>-Wert der Klasse.
:return: <class 'bool'> Nur False wenn False oder 0 sonst True
"""
if self._bitaddress >= 0:
int_byte = int.from_bytes(
self._parentdevice._ba_devdata[self._slc_address],
@@ -371,25 +398,31 @@ class IOBase(object):
bytearray(self._length)
def __len__(self):
"""Gibt die Bytelaenge des IO zurueck.
@return Bytelaenge des IO - 0 bei BITs"""
"""
Gibt die Bytelaenge des IO zurueck.
:return: Bytelaenge des IO - 0 bei BITs
"""
return 0 if self._bitaddress > 0 else self._length
def __str__(self):
"""<class 'str'>-Wert der Klasse.
@return Namen des IOs"""
"""
<class 'str'>-Wert der Klasse.
:return: Namen des IOs
"""
return self._name
def __reg_xevent(self, func, delay, edge, as_thread, overwrite, prefire):
"""Verwaltet reg_event und reg_timerevent.
@param func Funktion die bei Aenderung aufgerufen werden soll
@param delay Verzoegerung in ms zum Ausloesen - auch bei Wertaenderung
@param edge Ausfuehren bei RISING, FALLING or BOTH Wertaenderung
@param as_thread Bei True, Funktion als EventCallback-Thread ausfuehren
@param overwrite Wenn True, wird Event bei ueberschrieben
@param prefire Ausloesen mit aktuellem Wert, wenn mainloop startet
def __reg_xevent(self, func, delay: int, edge: int, as_thread: bool, overwrite: bool, prefire: bool) -> None:
"""
Verwaltet reg_event und reg_timerevent.
:param func: Funktion die bei Aenderung aufgerufen werden soll
:param delay: Verzoegerung in ms zum Ausloesen - auch bei Wertaenderung
:param edge: Ausfuehren bei RISING, FALLING or BOTH Wertaenderung
:param as_thread: Bei True, Funktion als EventCallback-Thread ausfuehren
:param overwrite: Wenn True, wird Event bei ueberschrieben
:param prefire: Ausloesen mit aktuellem Wert, wenn mainloop startet
"""
# Prüfen ob Funktion callable ist
if not callable(func):
@@ -449,29 +482,44 @@ class IOBase(object):
IOEvent(func, edge, as_thread, delay, overwrite, prefire)
)
def _get_address(self):
"""Gibt die absolute Byteadresse im Prozessabbild zurueck.
@return Absolute Byteadresse"""
def _get_address(self) -> int:
"""
Gibt die absolute Byteadresse im Prozessabbild zurueck.
:return: Absolute Byteadresse
"""
return self._parentdevice._offset + self._slc_address.start
def _get_byteorder(self):
"""Gibt konfigurierte Byteorder zurueck.
@return <class 'str'> Byteorder"""
def _get_byteorder(self) -> str:
"""
Gibt konfigurierte Byteorder zurueck.
:return: <class 'str'> Byteorder
"""
return self._byteorder
def _get_iotype(self):
"""Gibt io type zurueck.
@return <class 'int'> io type"""
def _get_iotype(self) -> int:
"""
Gibt io type zurueck.
:return: <class 'int'> io type
"""
return self._iotype
def get_defaultvalue(self):
"""Gibt die Defaultvalue von piCtory zurueck.
@return Defaultvalue als <class 'byte'> oder <class 'bool'>"""
"""
Gibt die Defaultvalue von piCtory zurueck.
:return: Defaultvalue als <class 'byte'> oder <class 'bool'>
"""
return self._defaultvalue
def get_value(self):
"""Gibt den Wert des IOs zurueck.
@return IO-Wert als <class 'bytes'> oder <class 'bool'>"""
"""
Gibt den Wert des IOs zurueck.
:return: IO-Wert als <class 'bytes'> oder <class 'bool'>
"""
if self._bitaddress >= 0:
int_byte = int.from_bytes(
self._parentdevice._ba_devdata[self._slc_address],
@@ -483,7 +531,8 @@ class IOBase(object):
def reg_event(
self, func, delay=0, edge=BOTH, as_thread=False, prefire=False):
"""Registriert fuer IO ein Event bei der Eventueberwachung.
"""
Registriert fuer IO ein Event bei der Eventueberwachung.
Die uebergebene Funktion wird ausgefuehrt, wenn sich der IO Wert
aendert. Mit Angabe von optionalen Parametern kann das
@@ -492,17 +541,17 @@ class IOBase(object):
HINWEIS: Die delay-Zeit muss in die .cycletime passen, ist dies nicht
der Fall, wird IMMER aufgerundet!
@param func Funktion die bei Aenderung aufgerufen werden soll
@param delay Verzoegerung in ms zum Ausloesen wenn Wert gleich bleibt
@param edge Ausfuehren bei RISING, FALLING or BOTH Wertaenderung
@param as_thread Bei True, Funktion als EventCallback-Thread ausfuehren
@param prefire Ausloesen mit aktuellem Wert, wenn mainloop startet
:param func: Funktion die bei Aenderung aufgerufen werden soll
:param delay; Verzoegerung in ms zum Ausloesen wenn Wert gleich bleibt
:param edge: Ausfuehren bei RISING, FALLING or BOTH Wertaenderung
:param as_thread: Bei True, Funktion als EventCallback-Thread ausfuehren
:param prefire: Ausloesen mit aktuellem Wert, wenn mainloop startet
"""
self.__reg_xevent(func, delay, edge, as_thread, True, prefire)
def reg_timerevent(self, func, delay, edge=BOTH, as_thread=False):
"""Registriert fuer IO einen Timer, welcher nach delay func ausfuehrt.
"""
Registriert fuer IO einen Timer, welcher nach delay func ausfuehrt.
Der Timer wird gestartet, wenn sich der IO Wert aendert und fuehrt die
uebergebene Funktion aus - auch wenn sich der IO Wert in der
@@ -514,17 +563,19 @@ class IOBase(object):
HINWEIS: Die delay-Zeit muss in die .cycletime passen, ist dies nicht
der Fall, wird IMMER aufgerundet!
@param func Funktion die bei Aenderung aufgerufen werden soll
@param delay Verzoegerung in ms zum Ausloesen - auch bei Wertaenderung
@param edge Ausfuehren bei RISING, FALLING or BOTH Wertaenderung
@param as_thread Bei True, Funktion als EventCallback-Thread ausfuehren
:param func: Funktion die bei Aenderung aufgerufen werden soll
:param delay: Verzoegerung in ms zum Ausloesen - auch bei Wertaenderung
:param edge: Ausfuehren bei RISING, FALLING or BOTH Wertaenderung
:param as_thread: Bei True, Funktion als EventCallback-Thread ausfuehren
"""
self.__reg_xevent(func, delay, edge, as_thread, False, False)
def set_value(self, value):
"""Setzt den Wert des IOs.
@param value IO-Wert als <class bytes'> oder <class 'bool'>"""
def set_value(self, value) -> None:
"""
Setzt den Wert des IOs.
:param value: IO-Wert als <class bytes'> oder <class 'bool'>
"""
if self._iotype == INP:
if self._parentdevice._modio._simulator:
raise RuntimeError(
@@ -643,12 +694,12 @@ class IOBase(object):
else:
self._parentdevice._ba_devdata[self._slc_address] = value
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
def unreg_event(self, func=None, edge=None) -> 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:
@@ -659,7 +710,6 @@ class IOBase(object):
for regfunc in self._parentdevice._dict_events[self]:
if regfunc.func != func or edge is not None \
and regfunc.edge != edge:
newlist.append(regfunc)
# Wenn Funktionen übrig bleiben, diese übernehmen
@@ -669,8 +719,9 @@ class IOBase(object):
else:
del self._parentdevice._dict_events[self]
def wait(self, edge=BOTH, exitevent=None, okvalue=None, timeout=0):
"""Wartet auf Wertaenderung eines IOs.
def wait(self, edge=BOTH, exitevent=None, okvalue=None, timeout=0) -> int:
"""
Wartet auf Wertaenderung eines IOs.
Die Wertaenderung wird immer uerberprueft, wenn fuer Devices
mit aktiviertem autorefresh neue Daten gelesen wurden.
@@ -696,19 +747,18 @@ class IOBase(object):
der autorefresh Funktion berechnet, entspricht also nicht exakt den
angegeben Millisekunden! Es wird immer nach oben gerundet!)
@param edge Flanke RISING, FALLING, BOTH die eintreten muss
@param exitevent <class 'thrading.Event'> fuer vorzeitiges Beenden
@param okvalue IO-Wert, bei dem das Warten sofort beendet wird
@param timeout Zeit in ms nach der abgebrochen wird
@return <class '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
:param edge: Flanke RISING, FALLING, BOTH die eintreten muss
:param exitevent: <class 'thrading.Event'> fuer vorzeitiges Beenden
:param okvalue: IO-Wert, bei dem das Warten sofort beendet wird
:param timeout: Zeit in ms nach der abgebrochen wird
:return: <class '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 autorefresh ist
if not self._parentdevice._selfupdate:
@@ -791,68 +841,89 @@ class IOBase(object):
class IntIO(IOBase):
"""Klasse fuer den Zugriff auf die Daten mit Konvertierung in int.
"""
Klasse fuer den Zugriff auf die Daten mit Konvertierung in int.
Diese Klasse erweitert die Funktion von <class 'IOBase'> um Funktionen,
ueber die mit <class 'int'> Werten gearbeitet werden kann. Fuer die
Umwandlung koennen 'Byteorder' (Default 'little') und 'signed' (Default
False) als Parameter gesetzt werden.
@see #IOBase IOBase
:ref:`IOBase`
"""
__slots__ = ()
def __int__(self):
"""Gibt IO-Wert zurueck mit Beachtung byteorder/signed.
@return IO-Wert als <class 'int'>"""
"""
Gibt IO-Wert zurueck mit Beachtung byteorder/signed.
:return: IO-Wert als <class 'int'>
"""
return int.from_bytes(
self._parentdevice._ba_devdata[self._slc_address],
byteorder=self._byteorder,
signed=self._signed
)
def _get_signed(self):
"""Ruft ab, ob der Wert Vorzeichenbehaftet behandelt werden soll.
@return True, wenn Vorzeichenbehaftet"""
def _get_signed(self) -> bool:
"""
Ruft ab, ob der Wert Vorzeichenbehaftet behandelt werden soll.
:return: True, wenn Vorzeichenbehaftet
"""
return self._signed
def _set_byteorder(self, value):
"""Setzt Byteorder fuer <class 'int'> Umwandlung.
@param value <class 'str'> 'little' or 'big'"""
def _set_byteorder(self, value: str) -> None:
"""
Setzt Byteorder fuer <class 'int'> Umwandlung.
:param value: <class 'str'> 'little' or 'big'
"""
if not (value == "little" or value == "big"):
raise ValueError("byteorder must be 'little' or 'big'")
if self._byteorder != value:
self._byteorder = value
self._defaultvalue = self._defaultvalue[::-1]
def _set_signed(self, value):
"""Left fest, ob der Wert Vorzeichenbehaftet behandelt werden soll.
@param value True, wenn mit Vorzeichen behandel"""
def _set_signed(self, value: bool) -> None:
"""
Left fest, ob der Wert Vorzeichenbehaftet behandelt werden soll.
:param value: True, wenn mit Vorzeichen behandel
"""
if type(value) != bool:
raise TypeError("signed must be <class 'bool'> True or False")
self._signed = value
def get_intdefaultvalue(self):
"""Gibt die Defaultvalue als <class 'int'> zurueck.
@return <class 'int'> Defaultvalue"""
def get_intdefaultvalue(self) -> int:
"""
Gibt die Defaultvalue als <class 'int'> zurueck.
:return: <class 'int'> Defaultvalue
"""
return int.from_bytes(
self._defaultvalue, byteorder=self._byteorder, signed=self._signed
)
def get_intvalue(self):
"""Gibt IO-Wert zurueck mit Beachtung byteorder/signed.
@return IO-Wert als <class 'int'>"""
def get_intvalue(self) -> int:
"""
Gibt IO-Wert zurueck mit Beachtung byteorder/signed.
:return: IO-Wert als <class 'int'>
"""
return int.from_bytes(
self._parentdevice._ba_devdata[self._slc_address],
byteorder=self._byteorder,
signed=self._signed
)
def set_intvalue(self, value):
"""Setzt IO mit Beachtung byteorder/signed.
@param value <class 'int'> Wert"""
def set_intvalue(self, value: int) -> None:
"""
Setzt IO mit Beachtung byteorder/signed.
:param value: <class 'int'> Wert
"""
if type(value) == int:
self.set_value(value.to_bytes(
self._length,
@@ -872,19 +943,18 @@ class IntIO(IOBase):
class IntIOCounter(IntIO):
"""Erweitert die IntIO-Klasse um die .reset() Funktion fuer Counter."""
__slots__ = ("__ioctl_arg")
__slots__ = ("__ioctl_arg", )
def __init__(
self, counter_id,
parentdevice, valuelist, iotype, byteorder, signed):
"""Instantiierung der IntIOCounter-Klasse.
@param counter_id ID fuer den Counter, zu dem der IO gehoert (0-15)
@see #IOBase.__init__ IOBase.__init__(...)
"""
Instantiierung der IntIOCounter-Klasse.
:param counter_id: ID fuer den Counter, zu dem der IO gehoert (0-15)
:ref: :func:`IOBase.__init__(...)`
"""
if not isinstance(counter_id, int):
raise TypeError("counter_id must be <class 'int'>")
@@ -912,7 +982,7 @@ class IntIOCounter(IntIO):
# Basisklasse laden
super().__init__(parentdevice, valuelist, iotype, byteorder, signed)
def reset(self):
def reset(self) -> None:
"""Setzt den Counter des Inputs zurueck."""
if self._parentdevice._modio._monitoring:
raise RuntimeError(
@@ -957,13 +1027,13 @@ class IntIOCounter(IntIO):
class IntIOReplaceable(IntIO):
"""Erweitert die IntIO-Klasse um die .replace_io Funktion."""
__slots__ = ()
def replace_io(self, name, frm, **kwargs):
"""Ersetzt bestehenden IO mit Neuem.
def replace_io(self, name: str, frm: str, **kwargs) -> None:
"""
Ersetzt bestehenden IO mit Neuem.
Wenn die kwargs fuer byteorder und defaultvalue nicht angegeben werden,
uebernimmt das System die Daten aus dem ersetzten IO.
@@ -983,9 +1053,9 @@ class IntIOReplaceable(IntIO):
der urspruenglige IO hat, werden die nachfolgenden IOs ebenfalls
verwendet und entfernt.
@param name Name des neuen Inputs
@param frm struct formatierung (1 Zeichen) oder 'ANZAHLs' z.B. '8s'
@param kwargs Weitere Parameter:
:param name: Name des neuen Inputs
:param frm: struct formatierung (1 Zeichen) oder 'ANZAHLs' z.B. '8s'
:param kwargs: Weitere Parameter:
- bmk: interne Bezeichnung fuer IO
- bit: Registriert IO als <class 'bool'> am angegebenen Bit im Byte
- byteorder: Byteorder fuer den IO, Standardwert=little
@@ -995,10 +1065,7 @@ class IntIOReplaceable(IntIO):
- edge: Event ausfuehren bei RISING, FALLING or BOTH Wertaenderung
- as_thread: Fuehrt die event-Funktion als RevPiCallback-Thread aus
- prefire: Ausloesen mit aktuellem Wert, wenn mainloop startet
@see <a target="_blank"
href="https://docs.python.org/3/library/struct.html#format-characters"
>Python3 struct</a>
`https://docs.python.org/3/library/struct.html#format-characters`
"""
# StructIO erzeugen
io_new = StructIO(
@@ -1023,30 +1090,29 @@ class IntIOReplaceable(IntIO):
class StructIO(IOBase):
"""Klasse fuer den Zugriff auf Daten ueber ein definierten struct.
"""
Klasse fuer den Zugriff auf Daten ueber ein definierten struct.
Sie stellt ueber struct die Werte in der gewuenschten Formatierung
bereit. Der struct-Formatwert wird bei der Instantiierung festgelegt.
@see #IOBase IOBase
:ref:`IOBase`
"""
__slots__ = "__frm", "_parentio_address", "_parentio_defaultvalue", \
"_parentio_length", "_parentio_name"
def __init__(self, parentio, name, frm, **kwargs):
"""Erstellt einen IO mit struct-Formatierung.
def __init__(self, parentio, name: str, frm: str, **kwargs):
"""
Erstellt einen IO mit struct-Formatierung.
@param parentio ParentIO Objekt, welches ersetzt wird
@param name Name des neuen IO
@param frm struct formatierung (1 Zeichen) oder 'ANZAHLs' z.B. '8s'
@param kwargs Weitere Parameter:
:param parentio: ParentIO Objekt, welches ersetzt wird
:param name: Name des neuen IO
:param frm: struct formatierung (1 Zeichen) oder 'ANZAHLs' z.B. '8s'
:param kwargs: Weitere Parameter:
- bmk: Bezeichnung fuer IO
- bit: Registriert IO als <class 'bool'> am angegebenen Bit im Byte
- byteorder: Byteorder fuer IO, Standardwert vom ersetzten IO
- defaultvalue: Standardwert fuer IO, Standard vom ersetzten IO
"""
# Structformatierung prüfen
regex = rematch("^([0-9]*s|[cbB?hHiIlLqQefd])$", frm)
@@ -1118,40 +1184,54 @@ class StructIO(IOBase):
parentio._parentdevice._dict_slc[parentio._iotype].start and
self._slc_address.stop <=
parentio._parentdevice._dict_slc[parentio._iotype].stop):
raise BufferError(
"registered value does not fit process image scope"
)
def _get_frm(self):
"""Ruft die struct Formatierung ab.
@return struct Formatierung"""
def _get_frm(self) -> str:
"""
Ruft die struct Formatierung ab.
:return: struct Formatierung
"""
return self.__frm[1:]
def _get_signed(self):
"""Ruft ab, ob der Wert Vorzeichenbehaftet behandelt werden soll.
@return True, wenn Vorzeichenbehaftet"""
def _get_signed(self) -> bool:
"""
Ruft ab, ob der Wert Vorzeichenbehaftet behandelt werden soll.
:return: True, wenn Vorzeichenbehaftet
"""
return self._signed
def get_structdefaultvalue(self):
"""Gibt die Defaultvalue mit struct Formatierung zurueck.
@return Defaultvalue vom Typ der struct-Formatierung"""
"""
Gibt die Defaultvalue mit struct Formatierung zurueck.
:return: Defaultvalue vom Typ der struct-Formatierung
"""
if self._bitaddress >= 0:
return self._defaultvalue
else:
return struct.unpack(self.__frm, self._defaultvalue)[0]
def get_structvalue(self):
"""Gibt den Wert mit struct Formatierung zurueck.
@return Wert vom Typ der struct-Formatierung"""
"""
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"""
"""
Setzt den Wert mit struct Formatierung.
:param value: Wert vom Typ der struct-Formatierung
"""
if self._bitaddress >= 0:
self.set_value(value)
else: