hat.gateway.devices.modbus.master.connection

  1import asyncio
  2import logging
  3import time
  4
  5from hat import json
  6from hat import aio
  7from hat.drivers import modbus
  8from hat.drivers import serial
  9from hat.drivers import tcp
 10
 11from hat.gateway.devices.modbus.master import common
 12
 13
 14mlog = logging.getLogger(__name__)
 15
 16
 17async def connect(conf: json.Data,
 18                  name: str,
 19                  request_queue_size: int = 1024
 20                  ) -> 'Connection':
 21    transport_conf = conf['transport']
 22    modbus_type = modbus.ModbusType[conf['modbus_type']]
 23
 24    if transport_conf['type'] == 'TCP':
 25        addr = tcp.Address(transport_conf['host'], transport_conf['port'])
 26        master = await modbus.create_tcp_master(
 27            modbus_type=modbus_type,
 28            addr=addr,
 29            response_timeout=conf['request_timeout'],
 30            name=name)
 31
 32    elif transport_conf['type'] == 'SERIAL':
 33        port = transport_conf['port']
 34        baudrate = transport_conf['baudrate']
 35        bytesize = serial.ByteSize[transport_conf['bytesize']]
 36        parity = serial.Parity[transport_conf['parity']]
 37        stopbits = serial.StopBits[transport_conf['stopbits']]
 38        xonxoff = transport_conf['flow_control']['xonxoff']
 39        rtscts = transport_conf['flow_control']['rtscts']
 40        dsrdtr = transport_conf['flow_control']['dsrdtr']
 41        silent_interval = transport_conf['silent_interval']
 42        master = await modbus.create_serial_master(
 43            modbus_type=modbus_type,
 44            port=port,
 45            baudrate=baudrate,
 46            bytesize=bytesize,
 47            parity=parity,
 48            stopbits=stopbits,
 49            xonxoff=xonxoff,
 50            rtscts=rtscts,
 51            dsrdtr=dsrdtr,
 52            silent_interval=silent_interval,
 53            response_timeout=conf['request_timeout'],
 54            name=name)
 55
 56    else:
 57        raise ValueError('unsupported link type')
 58
 59    return Connection(conf=conf,
 60                      master=master,
 61                      name=name,
 62                      request_queue_size=request_queue_size)
 63
 64
 65class Connection(aio.Resource):
 66
 67    def __init__(self,
 68                 conf: json.Data,
 69                 master: modbus.Master,
 70                 name: str,
 71                 request_queue_size: int):
 72        self._conf = conf
 73        self._master = master
 74        self._name = name
 75        self._request_queue = aio.Queue(request_queue_size)
 76        self._loop = asyncio.get_running_loop()
 77        self._log = common.create_device_logger_adapter(mlog, name)
 78
 79        self.async_group.spawn(self._request_loop)
 80
 81    @property
 82    def async_group(self) -> aio.Group:
 83        return self._master.async_group
 84
 85    async def send(self,
 86                   req: modbus.Request
 87                   ) -> modbus.Response | common.Timeout:
 88        self._log.debug('enqueuing request')
 89
 90        try:
 91            future = self._loop.create_future()
 92            delayed_count = self._conf['request_retry_delayed_count'] + 1
 93            await self._request_queue.put((req, delayed_count, future))
 94
 95            return await future
 96
 97        except aio.QueueClosedError:
 98            raise ConnectionError()
 99
100    async def _request_loop(self):
101        future = None
102        res_t = None
103
104        try:
105            self._log.debug('starting request loop')
106            while True:
107                req, delayed_count, future = await self._request_queue.get()  # NOQA
108                self._log.debug('dequed request')
109
110                if res_t is not None and self._conf['request_delay'] > 0:
111                    dt = time.monotonic() - res_t
112                    if dt < self._conf['request_delay']:
113                        await asyncio.sleep(self._conf['request_delay'] - dt)
114
115                if future.done():
116                    continue
117
118                delayed_count -= 1
119                immediate_count = (
120                    self._conf['request_retry_immediate_count'] + 1)
121
122                while True:
123                    try:
124                        immediate_count -= 1
125                        res = await self._master.send(req)
126                        res_t = time.monotonic()
127
128                        self._log.debug('received response %s', res)
129                        if not future.done():
130                            future.set_result(res)
131
132                        break
133
134                    except TimeoutError:
135                        self._log.debug('single request timeout')
136
137                        if immediate_count > 0:
138                            self._log.debug('immediate request retry')
139                            continue
140
141                        if delayed_count > 0:
142                            self._log.debug('delayed request retry')
143                            self.async_group.spawn(self._delay_request, req,
144                                                   delayed_count, future)
145                            future = None
146
147                        elif not future.done():
148                            self._log.debug('request resulting in timeout')
149                            future.set_result(common.Timeout())
150
151                        break
152
153                    except Exception as e:
154                        self._log.debug('setting request exception')
155                        if not future.done():
156                            future.set_exception(e)
157                        raise
158
159        except ConnectionError:
160            self._log.debug('connection closed')
161
162        except Exception as e:
163            self._log.error('request loop error: %s', e, exc_info=e)
164
165        finally:
166            self._log.debug('closing request loop')
167            self.close()
168            self._request_queue.close()
169
170            while True:
171                if future and not future.done():
172                    future.set_exception(ConnectionError())
173                if self._request_queue.empty():
174                    break
175                _, __, future = self._request_queue.get_nowait()
176
177    async def _delay_request(self, req, delayed_count, future):
178        try:
179            await asyncio.sleep(self._conf['request_retry_delay'])
180            await self._request_queue.put((req, delayed_count, future))
181            future = None
182
183        except aio.QueueClosedError:
184            pass
185
186        finally:
187            if future and not future.done():
188                future.set_exception(ConnectionError())