KittyPatch/kitty
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Finish implementation of clipboard writing

Browse Source
This commit is contained in:
Kovid Goyal 2022-12-03 08:02:27 +05:30
parent 71e09ba1fb
commit bde737fa38
No known key found for this signature in database
GPG Key ID: 06BC317B515ACE7C
4 changed files with 220 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
kitty/clipboard.py
View File

@ -320,8 +320,8 @@ class ClipboardRequestManager:
return
m[k] = v
typ = m.get('type', '')
payload = base64.standard_b64decode(epayload)
if typ == 'read':
payload = base64.standard_b64decode(epayload)
rr = ReadRequest(
is_primary_selection=m.get('loc', '') == 'primary',
mime_types=tuple(payload.decode('utf-8').split()),
@ -329,18 +329,11 @@ class ClipboardRequestManager:
)
self.handle_read_request(rr)
elif typ == 'write':
wr = self.in_flight_write_request
if wr is None:
wr = self.in_flight_write_request = WriteRequest(
is_primary_selection=m.get('loc', '') == 'primary',
protocol_type=ProtocolType.osc_5522, id=sanitize_id(m.get('id', ''))
)
self.handle_write_request(wr)
else:
w = get_boss().window_id_map.get(self.window_id)
if w is not None:
w.screen.send_escape_code_to_child(OSC, wr.encode_response(status='EBUSY'))
self.in_flight_write_request = WriteRequest(
is_primary_selection=m.get('loc', '') == 'primary',
protocol_type=ProtocolType.osc_5522, id=sanitize_id(m.get('id', ''))
)
self.handle_write_request(self.in_flight_write_request)
elif typ == 'wdata':
wr = self.in_flight_write_request
w = get_boss().window_id_map.get(self.window_id)
@ -349,7 +342,7 @@ class ClipboardRequestManager:
mime = m.get('mime', '')
if mime:
try:
wr.add_base64_data(payload, mime)
wr.add_base64_data(epayload, mime)
except OSError:
if w is not None:
w.screen.send_escape_code_to_child(OSC, wr.encode_response(status='EIO'))
@ -364,6 +357,8 @@ class ClipboardRequestManager:
wr.flush_base64_data()
wr.commit()
self.in_flight_write_request = None
if w is not None:
w.screen.send_escape_code_to_child(OSC, wr.encode_response(status='DONE'))
def parse_osc_52(self, data: str, is_partial: bool = False) -> None:
where, text = data.partition(';')[::2]
@ -395,7 +390,7 @@ class ClipboardRequestManager:
if not allowed or not cp.enabled:
self.in_flight_write_request = None
if w is not None:
w.screen.send_escape_code_to_child(OSC, wr.encode_response(status='EPERM' if not allowed else 'ENOCLIPBOARD'))
w.screen.send_escape_code_to_child(OSC, wr.encode_response(status='EPERM' if not allowed else 'ENOSYS'))
def fulfill_legacy_write_request(self, wr: WriteRequest, allowed: bool = True) -> None:
cp = get_boss().primary_selection if wr.is_primary_selection else get_boss().clipboard
@ -424,7 +419,7 @@ class ClipboardRequestManager:
return
cp = get_boss().primary_selection if rr.is_primary_selection else get_boss().clipboard
if not cp.enabled:
w.screen.send_escape_code_to_child(OSC, rr.encode_response(status='ENOCLIPBOARD'))
w.screen.send_escape_code_to_child(OSC, rr.encode_response(status='ENOSYS'))
return
if not allowed:
w.screen.send_escape_code_to_child(OSC, rr.encode_response(status='EPERM'))

68
tools/cmd/clipboard/read.go
View File

@ -7,7 +7,6 @@ import (
"encoding/base64"
"fmt"
"image"
"mime"
"os"
"path/filepath"
"strings"
@ -161,7 +160,7 @@ func (self *Output) assign_mime_type(available_mimes []string) (err error) {
return fmt.Errorf("The MIME type %s for %s not available on the clipboard", self.mime_type, self.arg)
}
func encode(metadata map[string]string, payload string) string {
func encode_bytes(metadata map[string]string, payload []byte) string {
ans := strings.Builder{}
ans.Grow(2048)
ans.WriteString("\x1b]")
@ -177,15 +176,19 @@ func encode(metadata map[string]string, payload string) string {
}
if len(payload) > 0 {
ans.WriteString(";")
ans.WriteString(base64.StdEncoding.EncodeToString(utils.UnsafeStringToBytes(payload)))
ans.WriteString(base64.StdEncoding.EncodeToString(payload))
}
ans.WriteString("\x1b\\")
return ans.String()
}
func encode(metadata map[string]string, payload string) string {
return encode_bytes(metadata, utils.UnsafeStringToBytes(payload))
}
func error_from_status(status string) error {
switch status {
case "ENOCLIPBOARD":
case "ENOSYS":
return fmt.Errorf("no primary selection available on this system")
case "EPERM":
return fmt.Errorf("permission denied")
@ -194,10 +197,37 @@ func error_from_status(status string) error {
}
}
func parse_escape_code(etype loop.EscapeCodeType, data []byte) (metadata map[string]string, payload []byte, err error) {
if etype != loop.OSC || !bytes.HasPrefix(data, utils.UnsafeStringToBytes(OSC_NUMBER+";")) {
return
}
parts := bytes.SplitN(data, utils.UnsafeStringToBytes(";"), 3)
metadata = make(map[string]string)
if len(parts) > 2 && len(parts[2]) > 0 {
payload, err = base64.StdEncoding.DecodeString(utils.UnsafeBytesToString(parts[2]))
if err != nil {
err = fmt.Errorf("Received OSC %s packet from terminal with invalid base64 encoded payload", OSC_NUMBER)
return
}
}
if len(parts) > 1 {
for _, record := range bytes.Split(parts[1], utils.UnsafeStringToBytes(":")) {
rp := bytes.SplitN(record, utils.UnsafeStringToBytes("="), 2)
v := ""
if len(rp) == 2 {
v = string(rp[1])
}
metadata[string(rp[0])] = v
}
}
return
}
func run_get_loop(opts *Options, args []string) (err error) {
lp, err := loop.New(loop.NoAlternateScreen, loop.NoRestoreColors, loop.NoMouseTracking)
if err != nil {
return
return err
}
var available_mimes []string
var wg sync.WaitGroup
@ -214,7 +244,7 @@ func run_get_loop(opts *Options, args []string) (err error) {
if outputs[i].arg_is_stream {
outputs[i].mime_type = "text/plain"
} else {
outputs[i].mime_type = mime.TypeByExtension(outputs[i].ext)
outputs[i].mime_type = utils.GuessMimeType(outputs[i].arg)
}
}
if outputs[i].mime_type == "" {
@ -235,28 +265,12 @@ func run_get_loop(opts *Options, args []string) (err error) {
}
lp.OnEscapeCode = func(etype loop.EscapeCodeType, data []byte) (err error) {
if etype != loop.OSC || !bytes.HasPrefix(data, utils.UnsafeStringToBytes(OSC_NUMBER+";")) {
return
metadata, payload, err := parse_escape_code(etype, data)
if err != nil {
return err
}
parts := bytes.SplitN(data, utils.UnsafeStringToBytes(";"), 3)
metadata := make(map[string]string)
var payload []byte
if len(parts) > 2 && len(parts[2]) > 0 {
payload, err = base64.StdEncoding.DecodeString(utils.UnsafeBytesToString(parts[2]))
if err != nil {
err = fmt.Errorf("Received OSC %s packet from terminal with invalid base64 encoded payload", OSC_NUMBER)
return
}
}
if len(parts) > 1 {
for _, record := range bytes.Split(parts[1], utils.UnsafeStringToBytes(":")) {
rp := bytes.SplitN(record, utils.UnsafeStringToBytes("="), 2)
v := ""
if len(rp) == 2 {
v = string(rp[1])
}
metadata[string(rp[0])] = v
}
if metadata == nil {
return nil
}
if reading_available_mimes {
switch metadata["status"] {

146
tools/cmd/clipboard/write.go
View File

@ -3,11 +3,153 @@
package clipboard
import (
"errors"
"fmt"
"io"
"os"
"path/filepath"
"kitty/tools/tui/loop"
"kitty/tools/utils"
)
var _ = fmt.Print
func run_set_loop(opts *Options, args []string) (err error) {
return fmt.Errorf("TODO: Implement me")
type Input struct {
src *os.File
arg string
ext string
is_stream bool
mime_type string
}
func write_loop(inputs []*Input) (err error) {
lp, err := loop.New(loop.NoAlternateScreen, loop.NoRestoreColors, loop.NoMouseTracking)
if err != nil {
return err
}
var waiting_for_write loop.IdType
var buf [4096]byte
lp.OnInitialize = func() (string, error) {
waiting_for_write = lp.QueueWriteString(encode(map[string]string{"type": "write"}, ""))
return "", nil
}
write_chunk := func() error {
if len(inputs) == 0 {
return nil
}
i := inputs[0]
n, err := i.src.Read(buf[:])
if n > 0 {
waiting_for_write = lp.QueueWriteString(encode_bytes(map[string]string{"type": "wdata", "mime": i.mime_type}, buf[:n]))
}
if err != nil {
if errors.Is(err, io.EOF) {
inputs = inputs[1:]
if len(inputs) == 0 {
lp.QueueWriteString(encode(map[string]string{"type": "wdata"}, ""))
waiting_for_write = 0
}
return lp.OnWriteComplete(waiting_for_write)
}
return fmt.Errorf("Failed to read from %s with error: %w", i.arg, err)
}
return nil
}
lp.OnWriteComplete = func(msg_id loop.IdType) error {
if waiting_for_write == msg_id {
return write_chunk()
}
return nil
}
lp.OnEscapeCode = func(etype loop.EscapeCodeType, data []byte) (err error) {
metadata, _, err := parse_escape_code(etype, data)
if err != nil {
return err
}
if metadata != nil && metadata["type"] == "write" {
switch metadata["status"] {
case "DONE":
lp.Quit(0)
case "EIO":
return fmt.Errorf("Could not write to clipboard an I/O error occurred while the terminal was processing the data")
case "EINVAL":
return fmt.Errorf("Could not write to clipboard base64 encoding invalid")
case "ENOSYS":
return fmt.Errorf("Could not write to primary selection as the system does not support it")
case "EPERM":
return fmt.Errorf("Could not write to clipboard as permission was denied")
default:
return fmt.Errorf("Could not write to clipboard unknowns status returned from terminal: %#v", metadata["status"])
}
}
return
}
esc_count := 0
lp.OnKeyEvent = func(event *loop.KeyEvent) error {
if event.MatchesPressOrRepeat("ctrl+c") || event.MatchesPressOrRepeat("esc") {
event.Handled = true
esc_count++
if esc_count < 2 {
key := "Esc"
if event.MatchesPressOrRepeat("ctrl+c") {
key = "Ctrl+C"
}
lp.QueueWriteString(fmt.Sprintf("Waiting for response from terminal, press %s again to abort. This could cause garbage to be spewed to the screen.\r\n", key))
} else {
return fmt.Errorf("Aborted by user!")
}
}
return nil
}
err = lp.Run()
if err != nil {
return
}
ds := lp.DeathSignalName()
if ds != "" {
fmt.Println("Killed by signal: ", ds)
lp.KillIfSignalled()
return
}
return
}
func run_set_loop(opts *Options, args []string) (err error) {
inputs := make([]*Input, len(args))
to_process := make([]*Input, len(args))
defer func() {
for _, i := range inputs {
if i.src != nil {
i.src.Close()
}
}
}()
for i, arg := range args {
f, err := os.Open(arg)
if err != nil {
return fmt.Errorf("Failed to open %s with error: %w", arg, err)
}
inputs[i] = &Input{arg: arg, src: f, ext: filepath.Ext(arg), is_stream: arg == "/dev/stdin"}
if i < len(opts.Mime) {
inputs[i].mime_type = opts.Mime[i]
} else if inputs[i].is_stream {
inputs[i].mime_type = "text/plain"
} else if inputs[i].ext != "" {
inputs[i].mime_type = utils.GuessMimeType(inputs[i].arg)
}
if inputs[i].mime_type == "" {
return fmt.Errorf("Could not guess MIME type for %s use the --mime option to specify a MIME type", arg)
}
to_process[i] = inputs[i]
}
return write_loop(to_process)
}

24
tools/utils/mimetypes.go Normal file
View File

@ -0,0 +1,24 @@
// License: GPLv3 Copyright: 2022, Kovid Goyal, <kovid at kovidgoyal.net>
package utils
import (
"fmt"
"mime"
"path/filepath"
)
var _ = fmt.Print
func GuessMimeType(filename string) string {
ext := filepath.Ext(filename)
mime_with_parameters := mime.TypeByExtension(ext)
if mime_with_parameters == "" {
return mime_with_parameters
}
ans, _, err := mime.ParseMediaType(mime_with_parameters)
if err != nil {
return ""
}
return ans
}