python 从HP 3050A打印机/扫描仪扫描图像

Posted 2021-05-09

"""
A Tkinter interface to the HP 3050A scanner.

Copyright (c) 2014, Adam Marchetti
All rights reserved.
 
Redistribution and use in source and binary forms, with or without modification, 
are permitted provided that the following conditions are met:
 
1. Redistributions of source code must retain the above copyright notice, this 
list of conditions and the following disclaimer.
 
2. Redistributions in binary form must reproduce the above copyright notice, 
this list of conditions and the following disclaimer in the documentation and/or
 other materials provided with the distribution.
 
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" 
AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 
IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE 
DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE 
FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL 
DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR 
SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR 
TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE 
OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
"""
from collections import namedtuple
import functools
import traceback
import urllib.request as urlrequest
try:
    import xml.etree.cElementTree as ElementTree
except ImportError:
    import xml.etree.ElementTree as ElementTree

import tkinter as tk
import tkinter.ttk as ttk
import tkinter.filedialog as tk_file

# This is used by the generator to tell the UI how things are going. The `time`
# attribute describes how soon the job should be run again (via `next()`), and
# the `message` attribute is what should be shown to the user. Note that, if
# this message is an error, than `time` should be None.
JobStatus = namedtuple('JobStatus', ['time', 'message'])

SCANNER_XML_NS = '{http://www.hp.com/schemas/imaging/con/cnx/scan/2008/08/19}'
SCAN_JOB_XML_NS = '{http://www.hp.com/schemas/imaging/con/ledm/jobs/2009/04/30}'

REQUEST_PAYLOAD = b"""
<scan:ScanJob 
    xmlns:scan="http://www.hp.com/schemas/imaging/con/cnx/scan/2008/08/19" 
    xmlns:dd="http://www.hp.com/schemas/imaging/con/dictionaries/1.0/">

    <scan:XResolution>300</scan:XResolution>
    <scan:YResolution>300</scan:YResolution>
    <scan:XStart>0</scan:XStart>
    <scan:YStart>0</scan:YStart>
    <scan:Width>2550</scan:Width>
    <scan:Height>3300</scan:Height>
    <scan:Format>Jpeg</scan:Format>
    <scan:CompressionQFactor>25</scan:CompressionQFactor>
    <scan:ColorSpace>Color</scan:ColorSpace>
    <scan:BitDepth>8</scan:BitDepth>
    <scan:InputSource>Platen</scan:InputSource>
    <scan:GrayRendering>NTSC</scan:GrayRendering>
    <scan:ToneMap>
        <scan:Gamma>1000</scan:Gamma>
        <scan:Brightness>1000</scan:Brightness>
        <scan:Contrast>1000</scan:Contrast>
        <scan:Highlite>179</scan:Highlite>
        <scan:Shadow>25</scan:Shadow>
    </scan:ToneMap>
    <scan:ContentType>Photo</scan:ContentType>
</scan:ScanJob>
"""

def make_url(*parts):
    "Constructs a http:// URL from a list of path elements."
    return 'http://' + '/'.join(parts)

def get_scan_job(ip_addr, output_file):
    """
    Runs a scan job on the scanner, generating a stream of JobStatus outputs to
    let the GUI know how we're doing.
    """
    SCAN_URL = ip_addr

    # First, ensure that the scanner is available
    with urlrequest.urlopen(make_url(SCAN_URL, 'Scan', 'Status')) as status_request:
        xml_result = ElementTree.parse(status_request)
        root_result = xml_result.getroot()

    # This is the first yield point, depending upon whether or not the scanner
    # is currently ready.
    scan_status = root_result.find(SCANNER_XML_NS + 'ScannerState').text
    if scan_status != 'Idle':
        yield JobStatus(None, 'Scanner is currently {}'.format(scan_status))
    else:
        yield JobStatus(0, 'Preparing to submit scan job...')

    # Secondly, issue the request via POST...
    content_handler = {'Content-Type': 'text/xml'}
    requester = urlrequest.Request(make_url(SCAN_URL, 'Scan', 'Jobs'),
        data=REQUEST_PAYLOAD, headers=content_handler)
    urlrequest.urlopen(requester)

    # ... and find out the information of the job we just submitted
    with urlrequest.urlopen(make_url(SCAN_URL, 'Jobs', 'JobList')) as job_location:
        _ = ElementTree.parse(job_location)
        root_result = _.getroot()

    job_url = ''
    image_url = ''
    for job in root_result:
        job_state = job.find(SCAN_JOB_XML_NS + 'JobState')
        if job_state.text == 'Processing':
            # It's ours! Pull out the status URL and the URL of our finished
            # image
            job_url = job.find(SCAN_JOB_XML_NS + 'JobUrl').text.lstrip('/')
            image_url = (job
                .find(SCANNER_XML_NS + 'ScanJob')
                .find(SCANNER_XML_NS + 'PreScanPage')
                .find(SCANNER_XML_NS + 'BinaryURL')).text.strip('/')
            break
    
    # The second yield point, when we either have the information we need to
    # poll the scanner, or we don't
    if not job_url or not image_url:
        yield JobStatus(None, 'Unable to retrieve information for our job')
    else:
        yield JobStatus(0, 'Waiting for scanner to finish our job...')

    # Poll the scanner until it says that it has finished scanning, and the
    # image is ready
    while True:
        with urlrequest.urlopen(make_url(SCAN_URL, job_url)) as job_status:
            _ = ElementTree.parse(job_status)
            root_result = _.getroot()

        scan_status = (root_result
            .find(SCANNER_XML_NS + 'ScanJob')
            .find(SCANNER_XML_NS + 'PreScanPage')
            .find(SCANNER_XML_NS + 'PageState'))

        if scan_status.text == 'ReadyToUpload':
            yield JobStatus(0, 'Downloading image...')
            break
        else:
            yield JobStatus(1000, 'Waiting for image...')
    
    # Finally, issue the request for the image, and save it on the filesystem
    urlrequest.urlretrieve(make_url(SCAN_URL, image_url), filename=output_file)

# Only one scan may go on at once. This is taken when `do_scan()` begins,
# and is only released when the scan fails or succeeds
SCAN_LOCK = False

def do_scan(status, ip_ctl, filename_ctl):
    """
    Launches the scan - note that this is not synchronous, so the call will
    return to the Tk event loop immediatey.
    
    :param status: The `Label` which shows the current status.
    :param ip_ctl: The `Entry` containing the scanner's IP address
    :param filename_ctl: The `Entry` containing the output filename.
    """
    global SCAN_LOCK
    if SCAN_LOCK:
        return
    else:
        SCAN_LOCK = True

    scanner_ip = ip_ctl.get()
    filename = filename_ctl.get()

    # Ensure that we can access the filename that we'll be outputting too,
    # so that we fail as quickly as possible.
    try:
        with open(filename, 'w'):
            pass
    except OSError:
        status.config(text='Cannot open the given filename for scanning')
        SCAN_LOCK = False
        return

    scan_gen = get_scan_job(scanner_ip, filename)
    
    def run_scan_generator(*_):
        "Runs the scanning generator, until it either succeeds or fails."
        global SCAN_LOCK
        try:
            result = next(scan_gen)
            if result.time is None:
                SCAN_LOCK = False
                status.config(text=result.message)
            else:
                status.config(text=result.message)
                status.after(result.time, run_scan_generator)
        except StopIteration:
            SCAN_LOCK = False
            status.config(text='Successfully completed scan')
        except Exception as exn:
            SCAN_LOCK = False
            status.config(text='Failure: ' + str(exn))
            traceback.print_exc()

    status.after(0, run_scan_generator)

def save_as(filename_entry, *_):
    "Opens up the SaveAs dialog and stores the filename."
    dialog = tk_file.SaveAs(filetypes=[('JPEG Image', '*.jpeg'), ('All Files', '*.*')])
    filename = dialog.show()
    
    # Correct the extension if the user doesn't put it in
    _ = filename.lower()
    if not _.endswith('.jpeg') or not _.endswith('.jpg'):
        filename += '.jpeg'

    filename_entry.delete('end')
    filename_entry.insert(0, filename)

def main():
    scan_win = tk.Tk()
    status_label = ttk.Label(scan_win, text='Not yet scanning')

    scanner_ip = ttk.Entry(scan_win)
    filename = ttk.Entry(scan_win)
    filename_picker = ttk.Button(scan_win, text='Save As...', 
        command=functools.partial(save_as, filename))

    run_scan = ttk.Button(scan_win, text='Scan...',
        command=functools.partial(do_scan, status_label, scanner_ip, filename))

    status_label.pack(expand=True, fill='x')
    scanner_ip.pack(expand=True, fill='x')
    filename.pack(expand=True, fill='x')
    filename_picker.pack(expand=True, fill='x')
    run_scan.pack(expand=True, fill='both')

    scanner_ip.insert(0, '192.168.1.90')

    scan_win.mainloop()

if __name__ == '__main__':
    main()

"""
Interfacing with the HP 3050A to automate scanning.

Copyright (c) 2014, Adam Marchetti
All rights reserved.
 
Redistribution and use in source and binary forms, with or without modification, 
are permitted provided that the following conditions are met:
 
1. Redistributions of source code must retain the above copyright notice, this 
list of conditions and the following disclaimer.
 
2. Redistributions in binary form must reproduce the above copyright notice, 
this list of conditions and the following disclaimer in the documentation and/or
 other materials provided with the distribution.
 
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" 
AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 
IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE 
DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE 
FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL 
DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR 
SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR 
TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE 
OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
"""

import sys
import time
import urllib.request as urlrequest
try:
    import xml.etree.cElementTree as ElementTree
except ImportError:
    import xml.etree.ElementTree as ElementTree

# HP uses these namespaces to describe the XML which this program uses - the braces
# are meant to conincide with how ElementTree represents namespaces
SCANNER_XML_NS = '{http://www.hp.com/schemas/imaging/con/cnx/scan/2008/08/19}'
SCAN_JOB_XML_NS = '{http://www.hp.com/schemas/imaging/con/ledm/jobs/2009/04/30}'

try:
    hostname = sys.argv[1]
    output = sys.argv[2]
except IndexError:
    print(sys.argv[0], '<printer>', '<output>')
    sys.exit(1)

# The printer's network name
base_url = 'http://' + hostname

# First, make sure that the scanner isn't busy
with urlrequest.urlopen('/'.join([base_url, 'Scan', 'Status'])) as status_request:
    root_result = ElementTree.parse(status_request).getroot()

# To give you an idea of what to expect, here's some sample output:

## <?xml version="1.0" encoding="UTF-8"?>
## <!-- THIS DATA SUBJECT TO DISCLAIMER(S) INCLUDED WITH THE PRODUCT OF ORIGIN. -->
## <ScanStatus xmlns="http://www.hp.com/schemas/imaging/con/cnx/scan/2008/08/19">
##
##  <!-- Important, since this tells us whether we can or cannot scan right now -->
## 	<ScannerState>Idle</ScannerState>
##
## </ScanStatus>


print(':: Getting scanner status')
status_parent = root_result.find(SCANNER_XML_NS + 'ScannerState')
scan_status = status_parent.text
if scan_status != 'Idle':
    print('Printer', hostname, 'is not ready!')
    sys.exit(2)

# Next, submit the scanning request
print(':: Sending scan request')
payload = b"""
<scan:ScanJob 
    xmlns:scan="http://www.hp.com/schemas/imaging/con/cnx/scan/2008/08/19" 
    xmlns:dd="http://www.hp.com/schemas/imaging/con/dictionaries/1.0/">

    <scan:XResolution>300</scan:XResolution>
    <scan:YResolution>300</scan:YResolution>
    <scan:XStart>0</scan:XStart>
    <scan:YStart>0</scan:YStart>
    <scan:Width>2550</scan:Width>
    <scan:Height>3300</scan:Height>
    <scan:Format>Jpeg</scan:Format>
    <scan:CompressionQFactor>25</scan:CompressionQFactor>
    <scan:ColorSpace>Color</scan:ColorSpace>
    <scan:BitDepth>8</scan:BitDepth>
    <scan:InputSource>Platen</scan:InputSource>
    <scan:GrayRendering>NTSC</scan:GrayRendering>
    <scan:ToneMap>
        <scan:Gamma>1000</scan:Gamma>
        <scan:Brightness>1000</scan:Brightness>
        <scan:Contrast>1000</scan:Contrast>
        <scan:Highlite>179</scan:Highlite>
        <scan:Shadow>25</scan:Shadow>
    </scan:ToneMap>
    <scan:ContentType>Photo</scan:ContentType>
</scan:ScanJob>
"""

# Usually, Python thinks we're sending some multipart form data. Tell it
# that we're interested in sending XML instead
content_header = {'Content-Type': 'text/xml'}
requester = urlrequest.Request('/'.join([base_url, 'Scan', 'Jobs']),
                               data=payload, headers=content_header)
urlrequest.urlopen(requester)

# Now, find out where the heck the printer put our job so we can query it

## <?xml version="1.0" encoding="UTF-8"?>
## <!-- THIS DATA SUBJECT TO DISCLAIMER(S) INCLUDED WITH THE PRODUCT OF ORIGIN. -->
## <j:JobList xmlns:j="http://www.hp.com/schemas/imaging/con/ledm/jobs/2009/04/30" 
##          xmlns:dd="http://www.hp.com/schemas/imaging/con/dictionaries/1.0/" 
##          xmlns:fax="http://www.hp.com/schemas/imaging/con/fax/2008/06/13" 
##          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 
##          xsi:schemaLocation="http://www.hp.com/schemas/imaging/con/ledm/jobs/2009/04/30 ../schemas/Jobs.xsd">
##   <j:Job>
##     <j:JobUrl>/Jobs/JobList/1</j:JobUrl>
##     <j:JobCategory>Scan</j:JobCategory>
##     <j:JobState>Completed</j:JobState>
##     <j:JobStateUpdate>56-2</j:JobStateUpdate>
##   </j:Job>
##   <j:Job>
##
##      <!-- Important, since this is where we can get additional updates on this job -->
##     <j:JobUrl>/Jobs/JobList/2</j:JobUrl>
##
##     <j:JobCategory>Scan</j:JobCategory>
##     <j:JobState>Processing</j:JobState>
##     <j:JobStateUpdate>56-3</j:JobStateUpdate>
##     <ScanJob xmlns="http://www.hp.com/schemas/imaging/con/cnx/scan/2008/08/19">
##       <PreScanPage>
##         <PageNumber>1</PageNumber>
##         <PageState>PreparingScan</PageState>
##         <BufferInfo>
##           <ScanSettings>
##             <XResolution>300</XResolution>
##             <YResolution>300</YResolution>
##             <XStart>0</XStart>
##             <YStart>0</YStart>
##             <Width>2550</Width>
##             <Height>3300</Height>
##             <Format>Jpeg</Format>
##             <CompressionQFactor>25</CompressionQFactor>
##             <ColorSpace>Color</ColorSpace>
##             <BitDepth>8</BitDepth>
##             <InputSource>Platen</InputSource>
##             <ContentType>Photo</ContentType>
##           </ScanSettings>
##           <ImageWidth>2550</ImageWidth>
##           <ImageHeight>3300</ImageHeight>
##           <BytesPerLine>7650</BytesPerLine>
##           <Cooked>enabled</Cooked>
##         </BufferInfo>
##
##        <!-- Important, since this the URL of the scanned image -->
##         <BinaryURL>/Scan/Jobs/2/Pages/1</BinaryURL>
##
##         <ImageOrientation>Normal</ImageOrientation>
##       </PreScanPage>
##     </ScanJob>
##   </j:Job>
## </j:JobList>

print(':: Finding our job...')
with urlrequest.urlopen('/'.join([base_url, 'Jobs', 'JobList'])) as job_location:
    root_location = ElementTree.parse(job_location).getroot()

# Search for a job that is currently processing to get the job status URL. We can use
# this to poll the printer for status updates regarding our scanning job.

# Also, get the URL for the page that is currently being scanned so we can download it
# later.
job_url = ''
image_url = ''

for job in root_location:
    job_state = job.find(SCAN_JOB_XML_NS + 'JobState')
    
    # The currently active job is the one we're interested in
    if job_state.text == 'Processing':
        job_url = job.find(SCAN_JOB_XML_NS + 'JobUrl').text.lstrip('/')
        print(':: Our job is', job_url)
        
        scan_status = job.find(SCANNER_XML_NS + 'ScanJob')
        print_status = scan_status.find(SCANNER_XML_NS + 'PreScanPage')
        image_url = print_status.find(SCANNER_XML_NS + 'BinaryURL').text.lstrip('/')
        print(':: Our image is', image_url)
        break

if not job_url:
    print('Unable to get the URL for the scanner job')
    sys.exit(3)

if not image_url:
    print('Unable to get the URL for the scanned image')
    sys.exit(4)

# Poll, waiting until the image is ready

##<?xml version="1.0" encoding="UTF-8"?>
##<!-- THIS DATA SUBJECT TO DISCLAIMER(S) INCLUDED WITH THE PRODUCT OF ORIGIN. -->
##<j:Job xmlns:j="http://www.hp.com/schemas/imaging/con/ledm/jobs/2009/04/30" xmlns:dd="http://www.hp.com/schemas/imaging/con/dictionaries/1.0/" xmlns:fax="http://www.hp.com/schemas/imaging/con/fax/2008/06/13" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.hp.com/schemas/imaging/con/ledm/jobs/2009/04/30 ../schemas/Jobs.xsd">
##	<j:JobUrl>/Jobs/JobList/1</j:JobUrl>
##	<j:JobCategory>Scan</j:JobCategory>
##	<j:JobState>Processing</j:JobState>
##	<j:JobStateUpdate>65-1</j:JobStateUpdate>
##	<ScanJob xmlns="http://www.hp.com/schemas/imaging/con/cnx/scan/2008/08/19">
##   <PreScanPage>
##     <PageNumber>1</PageNumber>
##
##      <!-- Important, since this indicates when we can get the image -->
##     <PageState>ReadyToUpload</PageState>
##
##     <BufferInfo>
##       <ScanSettings>
##         <XResolution>300</XResolution>
##         <YResolution>300</YResolution>
##         <XStart>0</XStart>
##         <YStart>0</YStart>
##         <Width>2550</Width>
##         <Height>3300</Height>
##         <Format>Jpeg</Format>
##         <CompressionQFactor>25</CompressionQFactor>
##         <ColorSpace>Color</ColorSpace>
##         <BitDepth>8</BitDepth>
##         <InputSource>Platen</InputSource>
##         <ContentType>Photo</ContentType>
##       </ScanSettings>
##       <ImageWidth>2550</ImageWidth>
##       <ImageHeight>3300</ImageHeight>
##       <BytesPerLine>7650</BytesPerLine>
##       <Cooked>enabled</Cooked>
##     </BufferInfo>
##     <BinaryURL>/Scan/Jobs/1/Pages/1</BinaryURL>
##     <ImageOrientation>Normal</ImageOrientation>
##   </PreScanPage>
##</ScanJob>
##</j:Job>

while True:
    with urlrequest.urlopen('/'.join([base_url, job_url])) as job_status:
        root_status = ElementTree.parse(job_status).getroot()

    scan_job = root_status.find(SCANNER_XML_NS + 'ScanJob')
    prescan_page = scan_job.find(SCANNER_XML_NS + 'PreScanPage')
    scan_status = prescan_page.find(SCANNER_XML_NS + 'PageState')

    if scan_status.text == 'ReadyToUpload':
        break

    print(":: Scanner is still processing")
    time.sleep(2)

print(':: Getting the scanned image')
(filename, headers) = urlrequest.urlretrieve('/'.join([base_url, image_url]), filename=output)
print(':: Scanned image now at', output)