python3-webengine-gtk3/usr/lib/python3/dist-packages/webengine/gtk3/_utils.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# @author yutent<yutent.io@gmail.com>
# @date 2023/07/28 14:39:33

import gi, threading
gi.require_version('Gtk', '3.0')
from gi.repository import GdkPixbuf, GObject

def noop():
  pass

def get_monitor_info(monitor):
  return {
    "model": monitor.props.model, 
    "scale_factor": monitor.props.scale_factor, 
    "manufacturer": monitor.props.manufacturer, 
    "refresh_rate": monitor.props.refresh_rate,
    "is_primary": monitor.is_primary(),
    "geometry": {
      "width": monitor.props.geometry.width, 
      "height": monitor.props.geometry.height, 
      "x": monitor.props.geometry.x, 
      "y": monitor.props.geometry.y, 
    }
  }


def pixbuf_to_dict(pixbuf, filename = ''):

  if pixbuf:
    has_alpha = pixbuf.get_has_alpha()

    # 没有apha通道时, get_pixels() 得到像素矩阵会出现极个别像素出现alpha通道
    # 所以, 这里强制设置alpha通道为false, 补全像素矩阵的数据
    if not has_alpha:
      pixbuf = pixbuf.add_alpha(False, 0, 0, 0)
      
    data = {
      "width": pixbuf.get_width(),
      "height": pixbuf.get_height(),
      "colorspace": pixbuf.get_colorspace(),
      "has_alpha": has_alpha,
      "bits_per_sample": pixbuf.get_bits_per_sample(),
      "rowstride": pixbuf.get_rowstride(),
      "filepath": filename,
      "bytes": list(pixbuf.get_pixels())
    }
  else:
    data = None

  return data


def dict_to_pixbuf(data):
  if data:
    image = GdkPixbuf.Pixbuf.new_from_data(
      data = bytes(data['bytes']),
      colorspace = data['colorspace'],
      has_alpha = data['has_alpha'],
      bits_per_sample = data['bits_per_sample'],
      width = data['width'],
      height = data['height'],
      rowstride = data['rowstride']
    )
  else:
    image = None

  return image


# 定义一个异步修饰器, 用于在子线程中运行一些会阻塞主线程的任务
def run_async(func):
  def wrapper(*args, **kwargs):
    thread = threading.Thread(target=func, args=args, kwargs=kwargs)
    thread.daemon = True
    thread.start()
    return thread
  return wrapper

# 类型js的settimeout的修饰器
def set_timeout(timeout = 0.5):
  def decorator(callback):
    def wrapper(*args):
      t = threading.Timer(timeout, callback, args=args)
      t.start()
      return t
    return wrapper
  
  return decorator


# 定义一个修饰器, 用于将当前方法转到主线程中运行 (子线程中调用方法时)
def idle(func):
  def wrapper(*args):
    GObject.idle_add(func, *args)
  return wrapper