hat.gateway.devices.modbus.master.device

  1from collections.abc import Collection
  2import asyncio
  3import contextlib
  4import logging
  5
  6from hat import aio
  7import hat.event.common
  8import hat.event.eventer
  9
 10from hat.gateway import common
 11from hat.gateway.devices.modbus.master.connection import connect
 12from hat.gateway.devices.modbus.master.eventer_client import (RemoteDeviceEnableReq,  # NOQA
 13                                                              RemoteDeviceWriteReq,  # NOQA
 14                                                              StatusRes,
 15                                                              RemoteDeviceStatusRes,  # NOQA
 16                                                              RemoteDeviceWriteRes,  # NOQA
 17                                                              EventerClientProxy)  # NOQA
 18from hat.gateway.devices.modbus.master.remote_device import RemoteDevice
 19
 20
 21mlog = logging.getLogger(__name__)
 22
 23
 24class ModbusMasterDevice(aio.Resource):
 25
 26    def __init__(self,
 27                 conf: common.DeviceConf,
 28                 eventer_client: hat.event.eventer.Client,
 29                 event_type_prefix: common.EventTypePrefix):
 30        self._conf = conf
 31        self._log_prefix = f"gateway device {conf['name']}"
 32        self._eventer_client = EventerClientProxy(eventer_client,
 33                                                  event_type_prefix,
 34                                                  self._log_prefix)
 35        self._enabled_devices = set()
 36        self._status = None
 37        self._conn = None
 38        self._devices = {}
 39        self._readers = {}
 40        self._async_group = aio.Group()
 41
 42        self.async_group.spawn(self._connection_loop)
 43
 44    @property
 45    def async_group(self) -> aio.Group:
 46        return self._async_group
 47
 48    async def process_events(self, events: Collection[hat.event.common.Event]):
 49        try:
 50            for request in self._eventer_client.process_events(events):
 51                if isinstance(request, RemoteDeviceEnableReq):
 52                    self._log(logging.DEBUG,
 53                              'received remote device enable request')
 54                    if request.enable:
 55                        self._enable_remote_device(request.device_id)
 56                    else:
 57                        await self._disable_remote_device(request.device_id)
 58
 59                elif isinstance(request, RemoteDeviceWriteReq):
 60                    self._log(logging.DEBUG,
 61                              'received remote device write request')
 62                    if self._conn and self._conn.is_open:
 63                        self._conn.async_group.spawn(
 64                            self._write, request.device_id, request.data_name,
 65                            request.request_id, request.value)
 66
 67                else:
 68                    raise ValueError('invalid request')
 69
 70        except Exception as e:
 71            self._log(logging.ERROR, 'process events error: %s', e, exc_info=e)
 72            self.close()
 73
 74    async def _connection_loop(self):
 75
 76        async def cleanup():
 77            if self._conn:
 78                await self._conn.async_close()
 79
 80            with contextlib.suppress(Exception):
 81                await self._set_status('DISCONNECTED')
 82
 83        try:
 84            self._log(logging.DEBUG, 'starting connection loop')
 85
 86            enabled_devices = await self._eventer_client.query_enabled_devices()  # NOQA
 87            self._enabled_devices.update(enabled_devices)
 88
 89            while True:
 90                await self._set_status('CONNECTING')
 91
 92                try:
 93                    self._conn = await aio.wait_for(
 94                        connect(self._conf['connection'],
 95                                self._log_prefix),
 96                        self._conf['connection']['connect_timeout'])
 97
 98                except aio.CancelledWithResultError as e:
 99                    self._conn = e.result
100                    raise
101
102                except Exception as e:
103                    self._log(logging.INFO, 'connecting error: %s', e,
104                              exc_info=e)
105                    await self._set_status('DISCONNECTED')
106                    await asyncio.sleep(
107                        self._conf['connection']['connect_delay'])
108                    continue
109
110                await self._set_status('CONNECTED')
111                self._devices = {}
112                self._readers = {}
113
114                self._log(logging.DEBUG, 'creating remote devices')
115                for device_conf in self._conf['remote_devices']:
116                    device = RemoteDevice(device_conf, self._conn,
117                                          self._log_prefix)
118                    self._devices[device.device_id] = device
119
120                    if device.device_id in self._enabled_devices:
121                        self._enable_remote_device(device.device_id)
122
123                    else:
124                        await self._notify_response(RemoteDeviceStatusRes(
125                            device_id=device.device_id,
126                            status='DISABLED'))
127
128                await self._conn.wait_closing()
129                await self._conn.async_close()
130
131                await self._set_status('DISCONNECTED')
132
133        except Exception as e:
134            self._log(logging.ERROR, 'connection loop error: %s', e,
135                      exc_info=e)
136
137        finally:
138            self._log(logging.DEBUG, 'closing connection loop')
139            self.close()
140            await aio.uncancellable(cleanup())
141
142    async def _notify_response(self, response):
143        await self._eventer_client.write([response])
144
145    async def _set_status(self, status):
146        if self._status == status:
147            return
148
149        self._log(logging.DEBUG, 'changing status: %s -> %s',
150                  self._status, status)
151        self._status = status
152        await self._notify_response(StatusRes(status))
153
154    def _enable_remote_device(self, device_id):
155        self._log(logging.DEBUG, 'enabling device %s', device_id)
156        self._enabled_devices.add(device_id)
157
158        device = self._devices.get(device_id)
159        if not device:
160            self._log(logging.DEBUG, 'device %s is not available', device_id)
161            return
162        if not device.conn.is_open:
163            self._log(logging.DEBUG, 'connection is not available')
164            return
165
166        reader = self._readers.get(device_id)
167        if reader and reader.is_open:
168            self._log(logging.DEBUG, 'reader %s is already running', device_id)
169            return
170
171        self._log(logging.DEBUG, 'creating reader for device %s', device_id)
172        reader = device.create_reader(self._notify_response)
173        self._readers[device.device_id] = reader
174
175    async def _disable_remote_device(self, device_id):
176        self._log(logging.DEBUG, 'disabling device %s', device_id)
177        self._enabled_devices.discard(device_id)
178
179        reader = self._readers.pop(device_id, None)
180        if not reader:
181            self._log(logging.DEBUG, 'device reader %s is not available',
182                      device_id)
183            return
184
185        await reader.async_close()
186
187    async def _write(self, device_id, data_name, request_id, value):
188        self._log(logging.DEBUG,
189                  'writing (device_id: %s; data_name: %s; value: %s)',
190                  device_id, data_name, value)
191
192        device = self._devices.get(device_id)
193        if not device:
194            self._log(logging.DEBUG, 'device %s is not available', device_id)
195            return
196
197        response = await device.write(data_name, request_id, value)
198        if response:
199            self._log(logging.DEBUG, 'writing result: %s', response.result)
200            await self._notify_response(response)
201
202    def _log(self, level, msg, *args, **kwargs):
203        if not mlog.isEnabledFor(level):
204            return
205
206        mlog.log(level, f"{self._log_prefix}: {msg}", *args, **kwargs)