# -*- coding: utf-8 -*- """ Библиотека поддержки Фронт-енд (общие методы не привязанные прямо к view) """ # pylint: disable=E0401,R6301 from time import time from browser import console, window, websocket, document, ajax from browser.timer import request_animation_frame, cancel_animation_frame, set_timeout, clear_timeout import cfg class GMeta(type): def __getattr__(cls, name): return None; # Не существующий атрибут в G возвращает None вместо исключения class G(metaclass=GMeta): # Глобальное Хранилище переменных в памяти pass G.language = window.navigator.language console.debug("navigator.language:", G.language) class MicReader: VOICE_MIME_TYPE = cfg.VOICE_MIME_TYPE VOICE_DURATION_MS = cfg.VOICE_DURATION_MS VOICE_SAMPLE_RATE = cfg.VOICE_SAMPLE_RATE VOICE_BIT_RATE = cfg.VOICE_BIT_RATE __slots__ = ['recorder', 'stream'] def denied(self, info=None): # overloadable """ Вызывается каждый раз если есть повод уведомить юзера обратить внимание на иконку микрофона или восклицательного знака (перед адресной строкой), где ему нужно вручную разрешить доступ и перезагрузить страницу, если он ранее (или случайно) запретил доступ, а теперь, нажимая старт стриминга с микрофона на сервер, терпит облом. """ console.debug("MicReader: mic denied", info) def ondataavailable(self, ev): # overloadable console.debug(f"MicReader: chunk {ev.data.size} bytes") def __init__(self): self.recorder = self.stream = None def start(self): if not window.MediaRecorder.isTypeSupported(self.VOICE_MIME_TYPE): console.error(f"MicReader: '{self.VOICE_MIME_TYPE}' not supported") return window.navigator.permissions.query({'name': 'microphone'}).then( self.permissions_query_cb ).catch(lambda e: [ console.error("MicReader: permissions query failed", e), ]) def permissions_query_cb(self, status): permission = status.state if permission not in ('prompt', 'granted'): # prompt, granted, denied console.error("MicReader: permission failed", permission) self.denied() return window.navigator.mediaDevices.getUserMedia({'audio': {'channelCount': 1, 'sampleRate': self.VOICE_SAMPLE_RATE, # 'sampleRate': 48000, # Родная для opus 'sampleSize': 16, # bits # 'latency': 60 / 1000; # sec 'echoCancellation': True, 'noiseSuppression': True, 'autoGainControl': True, }, 'video': False }).then( self.get_device_cb ).catch(lambda e: [ console.error("MicReader: stream request failed", e), self.denied() if getattr(e, 'name', None) == 'NotAllowedError' else None ]) def get_device_cb(self, stream): self.stream = stream try: self.recorder = window.MediaRecorder.new(self.stream, {'mimeType': self.VOICE_MIME_TYPE, 'audioBitsPerSecond': self.VOICE_BIT_RATE, }) self.recorder.ondataavailable = self.ondataavailable self.recorder.start(self.VOICE_DURATION_MS) except Exception as e: console.error("MicReader: Unexpected", e) for track in self.stream.getTracks(): track.stop() self.stream = None def stop(self): if self.recorder: if self.recorder.state != 'inactive': self.recorder.stop() if self.recorder.stream: for track in self.recorder.stream.getTracks(): track.stop() self.recorder = None self.stream = None class MicStreamer(MicReader): WebSocket_OPEN = window.WebSocket.OPEN __slots__ = ['wsroute', 'ws'] def __init__(self, wsroute: "socket route"): super().__init__() self.wsroute = wsroute; self.ws = None def start(self): super().start(); # MicReader уже может захватывать данные но пока нет сокета ondataavailable - молотит в холостую if not websocket.supported: console.error("Web Sockets are not supported") return self.ws = websocket.WebSocket(self.wsroute) self.ws.bind('open', self.ws_open_cb) self.ws.bind('close', self.ws_close_cb) self.ws.bind('message', self.ws_message_cb) self.ws.bind('error', self.ws_error_cb) def ondataavailable(self, ev): if self.ws and self.ws.readyState == self.WebSocket_OPEN and ev.data.size > 0: self.ws.send(ev.data) else: # super().ondataavailable(ev) pass def ws_error_cb(self, e): console.error("MicStreamer: Unexpected", e) def ws_open_cb(self, ev): console.debug("MicStreamer open:", ev) def ws_close_cb(self, ev): console.debug("MicStreamer close:", ev) def ws_message_cb(self, ev): console.debug("MicStreamer message:", ev) def stop(self): super().stop() if self.ws: if self.ws.readyState == window.WebSocket.OPEN: self.ws.close() self.ws = None class MicVisualiserMixin: FFT_SIZE = 32; # степень двойки (мин 32) ANIMATION_FPS = 10 __slots__ = ['actx', 'analyser', 'fft', 'frameid', 'frametime'] def updated(self, info: "volume" = None): # overloadable """ Вызывается для анимации активности микрофона в UI (fps ~ 10) """ console.debug("MicVisualiser: mic updated", info) def __init__(self, *args): super().__init__(*args) self.actx: "audioCtx" = None self.analyser: "audioContextAnalyser" = None self.fft: "Uint8Array" = None self.frameid = None; self.frametime = None def start(self): AudioContext = getattr(window, 'AudioContext', None) or getattr(window, 'webkitAudioContext', None) if not AudioContext: console.error("AudioContexts are not supported") return self.actx = AudioContext.new() self.analyser = self.actx.createAnalyser(); self.analyser.fftSize = self.FFT_SIZE self.analyser.smoothingTimeConstant = 0.0; # смещение окна в сторону прошлых данных (инерционность бинов fft) self.fft = window.Uint8Array.new(self.analyser.frequencyBinCount) super().start(); # подключится к self.stream сможем только как он появится def get_device_cb(self, stream): # overlaoded super().get_device_cb(stream); # Там self.stream = stream if self.actx: if self.actx.state == 'suspended': self.actx.resume(); # Resume context for iOS/mobile support self.actx.createMediaStreamSource(stream).connect(self.analyser) self.analyse() def analyse(self, t: "s" = None): """ Вызывается 60 раз в сек. Искусственно снижаем до ANIMATION_FPS """ if self.actx: if t is None or t > self.frametime: if self.actx.state == 'running' and self.analyser: self.analyser.getByteFrequencyData(fft := self.fft) # console.time("fftvolume") # Средняя громкость # volume = sum(fft) / fft.length # volume = (sum(f * f for f in fft) / fft.length)**0.5 volume = max(fft) # volume = (fft[0] + fft[fft.length // 2] + fft[fft.length-1]) // 3 # console.timeEnd("fftvolume") self.updated(volume * 100 / 256); # % self.frametime = (t or 0) + 1 / self.ANIMATION_FPS self.frameid = request_animation_frame(lambda _: self.analyse(time())) def stop(self): super().stop() if self.frameid is not None: cancel_animation_frame(self.frameid) self.frameid = None; self.frametime = None if self.actx: if self.actx.state != 'closed': self.actx.close() self.actx = None self.analyser = None self.fft = None def copy_to_clipboard(el): """ FIXME старый метод (для совместимости с устаревающими мобильными браузерами) """ # временно разблокируем для выделения, если нужно prevReadOnly = el.readOnly prevUserSelect = el.style.userSelect el.readOnly = False; # чтобы можно было выделить на некоторых мобильных el.style.userSelect = 'text' # запоминаем старую позицию выделения и выделяем весь текст start = el.selectionStart end = el.selectionEnd el.select() succeeded = False try: succeeded = document.execCommand('copy') except: console.error(f"Copy to clipboard failed: {el}") # восстановим состояние и позицию el.readOnly = prevReadOnly el.style.userSelect = prevUserSelect try: el.setSelectionRange(start, end) except: pass return succeeded float_epsilon = 2.220446049250313e-16; # sys.float_info.epsilon def debounce(ms=200): """ Декоратор для debounce обработчиков событий. 200ms - это примерная скорость печатания на клавиатуре (задерживает вызов обработчика до окончания серии событий; полезно для input/resize/search) """ def decor(func): timer = None def wrap(*args): nonlocal timer clear_timeout(timer) timer = set_timeout(func, ms, *args) wrap.__name__ = func.__name__ return wrap return decor _post_complete_handlers = []; _ajax_post = ajax.post; # Оригинальный ajax.post def ajax_post(*args, **kwargs): """ Враппер для перехвата oncomplete в eneter-ах """ oncomplete = kwargs.pop('oncomplete'); timeout = kwargs.pop('timeout') def _oncomplete(req, timeout=timeout): # Реализация oncomplete JustBry if req.status: # Вызов зарегистрированных oncomplete for handler in _post_complete_handlers: try: handler(req) except Exception as e: console.error(f"Oncomplete handler {handler} error: {e}") if oncomplete: oncomplete(req, timeout); # Вызов оригинальной реализации _ajax_post(*args, **kwargs, timeout=timeout, oncomplete = _oncomplete); # oncomplete Подменен ajax.post = ajax_post; # ajax.post Подменен (в цепочке) def register_post_complete(handler: "handler(req)"): _post_complete_handlers.append(handler) def unregister_post_complete(handler): try: _post_complete_handlers.remove(handler) except: pass