To receive notifications about scheduled maintenance, please subscribe to the mailing-list gitlab-operations@sympa.ethz.ch. You can subscribe to the mailing-list at https://sympa.ethz.ch

Commit d88d28c6 authored by mikolajr's avatar mikolajr
Browse files

Merge branch 'release-0.1.0'

parents 5c850122 5d50c8c2
Pipeline #26510 passed with stages
in 3 minutes and 16 seconds
[flake8]
max-line-length = 88
......@@ -4,3 +4,7 @@
__pycache__
*.egg-info/
docs/_build/
.eggs
.tox
.coverage
htmlcov/
before_script:
- apk add --no-cache python3
- python3 -m ensurepip
- pip3 install -U pip setuptools tox
- if [[ ! -e /usr/bin/pip ]]; then ln -sf /usr/bin/pip3 /usr/bin/pip ; fi
- if [[ ! -e /usr/bin/python ]]; then ln -sf /usr/bin/python3 /usr/bin/python; fi
stages:
- style
- tests
tests:
script:
- tox -r -vv -e py36
stage: tests
style:
script:
- tox -r -vv -e flake8
allow_failure: true
stage: style
......@@ -5,7 +5,7 @@ Credits
Development Lead
----------------
* Mikolaj Rybiński <mikolaj.rybinski@id.ethz.ch>
* Mikołaj Rybiński <mikolaj.rybinski@id.ethz.ch>
* David Graber <graber@eeh.ee.ethz.ch>
Contributors
......
......@@ -102,7 +102,7 @@ Before you submit a merge request, check that it meets these guidelines:
2. If the merge request adds functionality, the docs should be updated. Put
your new functionality into a function with a docstring, and add the
feature to the list in README.rst.
3. The merge request should work for Python 3.7. Check
3. The merge request should work for Python 3.6 and 3.7. Check
https://gitlab.ethz.ch/hvl_priv/hvl_ccb/merge_requests
and make sure that the tests pass for all supported Python versions.
......@@ -122,7 +122,11 @@ Tips
$ python setup.py develop
* To generate a PDF version of the Sphinx documentation instead of HTML use::
$ rm -rf docs/hvl_ccb.rst docs/modules.rst docs/_build && sphinx-apidoc -o docs/ hvl_ccb && python -msphinx -M latexpdf docs/ docs/_build
This requires a local installation of a LaTeX distribution, e.g. MikTeX.
Deploying
---------
......
......@@ -2,7 +2,8 @@
History
=======
0.1.0 (2019-MM-DD)
0.1.0 (2019-02-06)
------------------
* TODO: first release on ETH GitLab.
* Communication protocol base and serial communication implementation.
* Device base and MBW973 implementation.
......@@ -23,7 +23,7 @@ Voltage Lab (HVL), D-ITET, ETH
Features
--------
* ...
* Control a MBW973 SF6 Analyzer / dew point mirror over a serial connection (COM-ports)
Credits
-------
......
......@@ -84,7 +84,7 @@ todo_include_todos = False
# The theme to use for HTML and HTML Help pages. See the documentation for
# a list of builtin themes.
#
html_theme = 'alabaster'
html_theme = 'sphinx_rtd_theme'
# Theme options are theme-specific and customize the look and feel of a
# theme further. For a list of options available for each theme, see the
......
hvl\_ccb.comm package
=====================
Submodules
----------
hvl\_ccb.comm.base module
-------------------------
.. automodule:: hvl_ccb.comm.base
:members:
:undoc-members:
:show-inheritance:
hvl\_ccb.comm.serial module
---------------------------
.. automodule:: hvl_ccb.comm.serial
:members:
:undoc-members:
:show-inheritance:
Module contents
---------------
.. automodule:: hvl_ccb.comm
:members:
:undoc-members:
:show-inheritance:
hvl\_ccb.dev package
====================
Submodules
----------
hvl\_ccb.dev.base module
------------------------
.. automodule:: hvl_ccb.dev.base
:members:
:undoc-members:
:show-inheritance:
hvl\_ccb.dev.mbw973 module
--------------------------
.. automodule:: hvl_ccb.dev.mbw973
:members:
:undoc-members:
:show-inheritance:
Module contents
---------------
.. automodule:: hvl_ccb.dev
:members:
:undoc-members:
:show-inheritance:
hvl\_ccb package
================
Subpackages
-----------
.. toctree::
hvl_ccb.comm
hvl_ccb.dev
Submodules
----------
......@@ -12,14 +20,6 @@ hvl\_ccb.cli module
:undoc-members:
:show-inheritance:
hvl\_ccb.hvl\_ccb module
------------------------
.. automodule:: hvl_ccb.hvl_ccb
:members:
:undoc-members:
:show-inheritance:
Module contents
---------------
......
"""
example script for the device MBW973
"""
import logging
from hvl_ccb.comm import SerialCommunication
from hvl_ccb.dev import MBW973
import serial
logging.basicConfig(level=logging.INFO)
# configuration dict with appropriate settings
comconfig = {"port": "COM3",
"baudrate": 9600,
"parity": serial.PARITY_NONE,
"stopbits": serial.STOPBITS_ONE,
"bytesize": serial.EIGHTBITS,
"terminator": b'\r',
"timeout": 3,
}
# create communication protocol object
comdevice = SerialCommunication(comconfig)
# create device object
mbw = MBW973(comdevice)
# start the device
mbw.start()
......@@ -2,6 +2,6 @@
"""Top-level package for HVL Common Code Base."""
__author__ = """Mikolaj Rybiński"""
__email__ = 'mikolaj.rybinski@id.ethz.ch'
__author__ = """Mikołaj Rybiński, David Graber"""
__email__ = 'mikolaj.rybinski@id.ethz.ch, graber@eeh.ee.ethz.ch'
__version__ = '0.1.0'
"""Communication protocols subpackage."""
from .base import CommunicationProtocol # noqa: F401
from .serial import SerialCommunication # noqa: F401
"""
Module with base classes for communication protocols.
"""
from abc import ABC, abstractmethod
from typing import Dict, Sequence
class CommunicationProtocol(ABC):
"""
Communication protocol abstract base class.
Specifies the methods to implement for communication protocol, as well as
implements some default settings and checks.
"""
def __init__(self, configuration: Dict[str, object]):
"""
Constructor for concrete CommunicationProtocol. Validates configuration.
:param configuration: Dictionary w/ string keys, containing comm
protocol's configuration.
"""
configuration = self._clean_required_configuration_keys(configuration)
configuration = self._clean_optional_configuration_keys(configuration)
self._configuration = self.clean_configuration_keys(configuration)
# Keep the configuration dictionary immutable
# by allowing for key access only
def configuration_value(self, key: str) -> object:
"""
Get configuration value.
:param key: Configuration key (str).
:return: Configuration value.
"""
return self._configuration[key]
def _clean_required_configuration_keys(self, configuration: Dict[str, object])\
-> Dict[str, object]:
"""
Validate if all required configuration keys are present.
:param configuration:
:return: Cleaned configuration dictionary.
"""
required_keys = set(self.required_configuration_keys())
missing_required_keys = required_keys.difference(
set(configuration.keys()))
if missing_required_keys:
raise ValueError(
"Missing required configuration keys: {}".format(list(
missing_required_keys)))
return configuration
def _clean_optional_configuration_keys(self, configuration: Dict[str, object])\
-> Dict[str, object]:
"""
Validate if there are no configuration keys other than the required
and the optional ones. Set default values for the optional keys.
Override, to add validation checks for specific configuration entries.
:param configuration:
:return: Cleaned configuration dictionary.
"""
config_keys = set(configuration.keys())
required_keys = set(self.required_configuration_keys())
optional_defaults = self.optional_configuration_defaults()
superfluous_keys = config_keys.difference(required_keys.union(
optional_defaults.keys()))
if superfluous_keys:
raise ValueError("Unrecognized configuration keys: {}".format(list(
superfluous_keys)))
for (opt_key, default_value) in optional_defaults.items():
if opt_key not in configuration:
configuration[opt_key] = default_value
return configuration
@classmethod
def clean_configuration_keys(cls, configuration: Dict[str, object])\
-> Dict[str, object]:
"""
Override to add validation checks for specific configuration entries.
Input `configuration` contains at this point all required and optional
keys. By default there are not checks done.
:param configuration: Configuration dictionary containing all required
and optional keys.
:return: Cleaned configuration dictionary.
"""
return configuration # pragma: no cover
@classmethod
@abstractmethod
def required_configuration_keys(cls) -> Sequence[str]:
"""
Required configuration keys checked on initalization of the object.
:return: Set of required configuration keys (strings).
"""
pass # pragma: no cover
@classmethod
@abstractmethod
def optional_configuration_defaults(cls) -> Dict[str, object]:
"""
Optional configuration keys w/ default values, set and checked on
initalization of the object.
:return: Dictionary of optional configuration entries w/ defualt values.
"""
pass # pragma: no cover
@abstractmethod
def open(self):
"""
Open communication protocol
"""
pass # pragma: no cover
@abstractmethod
def close(self):
"""
Close the communication protocol
"""
pass # pragma: no cover
def __enter__(self):
self.open()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.close()
"""
Communication protocol for serial ports. Makes use of the pySerial library.
"""
from typing import Dict
import serial
from .base import CommunicationProtocol
class SerialCommunication(CommunicationProtocol):
"""
Implements the Communication Protocol for serial ports.
"""
ENCODING = 'utf-8'
UNICODE_HANDLING = 'replace'
def __init__(self, configuration):
"""
Constructor for Serial
"""
super().__init__(configuration)
# create the serial port specified in the configuration
with serial.serial_for_url(self.configuration_value("port"),
do_not_open=True) as self.ser:
self.ser.baudrate = self.configuration_value("baudrate")
self.ser.parity = self.configuration_value("parity")
self.ser.stopbits = self.configuration_value("stopbits")
self.ser.bytesize = self.configuration_value("bytesize")
self.ser.timeout = self.configuration_value("timeout")
@classmethod
def required_configuration_keys(cls):
return (
'port',
'baudrate',
'parity',
'stopbits',
'bytesize',
)
@classmethod
def optional_configuration_defaults(cls):
return {
# default terminator is CRLF
'terminator': b'\r\n',
# default timeout is 2 seconds
'timeout': 2,
}
@classmethod
def clean_configuration_keys(cls, configuration: Dict[str, object])\
-> Dict[str, object]:
"""
Validation checks for specific configuration entries.
Input `configuration` contains at this point all required and optional
keys.
:param configuration: Configuration dictionary containing all required
and optional keys.
:return: Cleaned configuration dictionary.
"""
# port
if not isinstance(configuration['port'], str):
raise ValueError("Configuration value 'port' has to be of type "
"'string': {}"
.format(configuration['port']))
# baudrate
if not isinstance(configuration['baudrate'], int):
raise ValueError("Configuration value 'baudrate' has to be of type "
"'integer': {}"
.format(configuration['baudrate']))
# parity
available_parities = (serial.PARITY_EVEN,
serial.PARITY_MARK,
serial.PARITY_NAMES,
serial.PARITY_NONE,
serial.PARITY_ODD,
serial.PARITY_SPACE)
if configuration['parity'] not in available_parities:
raise ValueError("Configuration value 'parity' has to be out "
"of {} but is {}"
.format(available_parities,
configuration['parity']))
# stopbits
available_stopbits = (serial.STOPBITS_ONE,
serial.STOPBITS_ONE_POINT_FIVE,
serial.STOPBITS_TWO)
if configuration['stopbits'] not in available_stopbits:
raise ValueError("Configuration value 'stopbits' has to be "
"out of {} but is {}"
.format(available_stopbits,
configuration['stopbits']))
# bytesize
available_bytesizes = (serial.FIVEBITS,
serial.SIXBITS,
serial.SEVENBITS,
serial.EIGHTBITS)
if configuration['bytesize'] not in available_bytesizes:
raise ValueError("Configuration value 'bytesize' has to be "
"out of {} but is {}"
.format(available_bytesizes,
configuration['bytesize']))
# terminator
if not isinstance(configuration['terminator'], bytes):
raise ValueError("Configuration value 'terminator' has to be "
"of type 'bytes': {}"
.format(configuration['terminator']))
# timeout
if not isinstance(configuration['timeout'], (int, float)):
raise ValueError("Configuration value 'timeout' has to be "
"numeric: {}"
.format(configuration['timeout']))
return configuration
def open(self):
"""
Open the serial connection.
"""
# open the port
self.ser.open()
def close(self):
"""
Close the serial connection.
"""
# close the port
self.ser.close()
def write_text(self, text: str):
"""
Write text to the serial port. The text is encoded and terminated by
the configured terminator.
:param text: Text to send to the port.
"""
terminator = self.configuration_value('terminator')
self.ser.write(text.encode(self.ENCODING, self.UNICODE_HANDLING)
+ terminator)
def read_text(self) -> str:
"""
Read one line of text from the serial port. The input buffer may
hold additional data afterwards, since only one line is read.
:return: String read from the serial port.
"""
return self.ser.readline().decode(self.ENCODING)
"""Devices subpackage."""
from .base import Device # noqa: F401
from .mbw973 import ( # noqa: F401
MBW973,
MBW973ControlRunningException,
MBW973PumpRunningException,
)
"""
Module with base classes for devices.
"""
from abc import ABC, abstractmethod
from ..comm import CommunicationProtocol
class Device(ABC):
"""
Base class for devices. Implement this class for a concrete device,
such as measurement equipment or voltage sources.
Specifies the methods to implement for a device.
"""
def __init__(self, communication_protocol: CommunicationProtocol) -> None:
"""
Constructor for Device. Links the communication protocol.
:param communication_protocol: Communication protocol to be used with
this device.
"""
self.communication_protocol = communication_protocol
@abstractmethod
def start(self) -> None:
"""
Start or restart this Device. To be implemented in the subclass.
"""
pass # pragma: no cover
@abstractmethod
def stop(self) -> None:
"""
Stop this Device. To be implemented in the subclass.
"""
pass # pragma: no cover
def __enter__(self):
self.start()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.stop()
"""
Device class for controlling a MBW 973 SF6 Analyzer over a serial connection.
"""
import logging
from threading import Timer
from typing import Dict, Callable, Union
from serial import SerialException
from .base import Device
from ..comm import SerialCommunication
class MBW973ControlRunningException(Exception):
pass
class MBW973PumpRunningException(Exception):
pass
class MBW973(Device):
"""
MBW 973 dew point mirror device class.
"""
def __init__(self, communication_protocol: SerialCommunication,