Coverage for gws-app/gws/base/printer/worker.py: 19%
147 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-10-16 23:09 +0200
1from typing import Optional, cast
3import gws
4import gws.base.model
5import gws.base.job
6import gws.lib.crs
7import gws.gis.render
8import gws.lib.image
9import gws.lib.intl
10import gws.lib.mime
11import gws.lib.osx
12import gws.lib.style
13import gws.lib.uom
15_PAPER_COLOR = 'white'
18class Object(gws.base.job.worker.Object):
19 project: gws.Project
20 tri: gws.TemplateRenderInput
21 printer: gws.Printer
22 template: gws.Template
23 tmpDir: str
24 contentPath: str
26 @classmethod
27 def run(cls, root: gws.Root, job: gws.Job):
28 request = gws.u.unserialize_from_path(job.payload.get('requestPath'))
29 w = cls(root, job.user, job, request)
30 w.work()
32 def __init__(self, root: gws.Root, user: gws.User, job: Optional[gws.Job], request: gws.PrintRequest):
33 super().__init__(root, user, job)
34 self.request = request
35 self.page_number = 0
36 self.contentPath = ''
38 def work(self):
39 self.prepare()
40 res = self.template.render(self.tri)
41 self.contentPath = res.contentPath
42 self.update_job(state=gws.JobState.complete, payload=dict(outputPath=self.contentPath))
44 def prepare(self):
45 self.project = cast(gws.Project, self.user.require(self.request.projectUid, gws.ext.object.project))
47 self.tri = gws.TemplateRenderInput(
48 project=self.project,
49 user=self.user,
50 notify=self.notify,
51 )
53 fmt = self.request.outputFormat or 'pdf'
54 if fmt.lower() == 'pdf' or fmt == gws.lib.mime.PDF:
55 self.tri.mimeOut = gws.lib.mime.PDF
56 elif fmt.lower() == 'png' or fmt == gws.lib.mime.PNG:
57 self.tri.mimeOut = gws.lib.mime.PNG
58 else:
59 raise gws.Error(f'invalid outputFormat {fmt!r}')
61 self.tri.locale = gws.lib.intl.locale(self.request.localeUid, self.tri.project.localeUids)
62 self.tri.crs = gws.lib.crs.get(self.request.crs) or self.project.map.bounds.crs
63 self.tri.maps = [self.prepare_map(self.tri, m) for m in (self.request.maps or [])]
64 self.tri.dpi = int(min(gws.gis.render.MAX_DPI, max(self.request.dpi, gws.lib.uom.OGC_SCREEN_PPI)))
66 if self.request.type == 'template':
67 # @TODO check dpi against configured qualityLevels
68 self.printer = cast(gws.Printer, self.user.require(self.request.printerUid, gws.ext.object.printer))
69 self.template = self.printer.template
70 else:
71 mm = gws.lib.uom.size_px_to_mm(self.request.outputSize, gws.lib.uom.OGC_SCREEN_PPI)
72 px = gws.lib.uom.size_mm_to_px(mm, self.tri.dpi)
73 self.template = self.root.create_temporary(
74 gws.ext.object.template,
75 type='map',
76 pageSize=(px[0], px[1], gws.Uom.px))
78 extra = dict(
79 project=self.project,
80 user=self.user,
81 scale=self.tri.maps[0].scale if self.tri.maps else 0,
82 rotation=self.tri.maps[0].rotation if self.tri.maps else 0,
83 dpi=self.tri.dpi,
84 crs=self.tri.crs.srid,
85 )
87 # @TODO read the args feature from the request
88 self.tri.args = gws.u.merge(self.request.args, extra)
90 n = sum(len(mp.planes) for mp in self.tri.maps) + 1
91 self.update_job(numSteps=n)
93 def notify(self, event, details=None):
94 job = self.get_job()
95 if not job:
96 return
98 if event == 'begin_plane':
99 name = gws.u.get(details, 'layer.title')
100 self.update_job(step=job.step + 1, stepName=name or '')
101 return
103 if event == 'finalize_print':
104 self.update_job(step=job.step + 1)
105 return
107 if event == 'page_break':
108 self.page_number += 1
109 return
111 def prepare_map(self, tri: gws.TemplateRenderInput, mp: gws.PrintMap) -> gws.MapRenderInput:
112 planes = []
114 style_opts = gws.lib.style.parser.Options(
115 trusted=False,
116 strict=False,
117 imageDirs=[self.root.app.webMgr.sites[0].staticRoot.dir],
118 )
119 style_dct = {}
121 for p in (mp.styles or []):
122 style = gws.lib.style.from_props(p, style_opts)
123 if style:
124 style_dct[style.cssSelector] = style
126 if mp.planes:
127 for n, p in enumerate(mp.planes):
128 pp = self.prepare_map_plane(n, p, style_dct)
129 if pp:
130 planes.append(pp)
132 layers = gws.u.compact(
133 self.user.acquire(uid, gws.ext.object.layer)
134 for uid in (mp.visibleLayers or []))
136 return gws.MapRenderInput(
137 backgroundColor=_PAPER_COLOR,
138 bbox=mp.bbox,
139 center=mp.center,
140 crs=tri.crs,
141 dpi=tri.dpi,
142 notify=tri.notify,
143 planes=planes,
144 project=tri.project,
145 rotation=mp.rotation or 0,
146 scale=mp.scale,
147 user=tri.user,
148 visibleLayers=layers,
149 )
151 def prepare_map_plane(self, n, plane: gws.PrintPlane, style_dct) -> Optional[gws.MapRenderInputPlane]:
152 opacity = 1
153 s = plane.get('opacity')
154 if s is not None:
155 opacity = float(s)
156 if opacity == 0:
157 return
159 if plane.type == gws.PrintPlaneType.raster:
160 layer = cast(gws.Layer, self.user.acquire(plane.layerUid, gws.ext.object.layer))
161 if not layer:
162 gws.log.warning(f'PREPARE_FAILED: plane {n}: {plane.layerUid=} not found')
163 return
164 if not layer.canRenderBox:
165 gws.log.warning(f'PREPARE_FAILED: plane {n}: {plane.layerUid=} canRenderBox false')
166 return
167 return gws.MapRenderInputPlane(
168 type=gws.MapRenderInputPlaneType.imageLayer,
169 layer=layer,
170 opacity=opacity,
171 compositeLayerUids=plane.get('compositeLayerUids'),
172 )
174 if plane.type == gws.PrintPlaneType.vector:
175 layer = cast(gws.Layer, self.user.acquire(plane.layerUid, gws.ext.object.layer))
176 if not layer:
177 gws.log.warning(f'PREPARE_FAILED: plane {n}: {plane.layerUid=} not found')
178 return
179 if not layer.canRenderSvg:
180 gws.log.warning(f'PREPARE_FAILED: plane {n}: {plane.layerUid=} canRenderSvg false')
181 return
182 style = style_dct.get(plane.cssSelector)
183 return gws.MapRenderInputPlane(
184 type=gws.MapRenderInputPlaneType.svgLayer,
185 layer=layer,
186 opacity=opacity,
187 styles=[style] if style else [],
188 )
190 if plane.type == gws.PrintPlaneType.bitmap:
191 img = None
192 if plane.bitmapMode in ('RGBA', 'RGB'):
193 img = gws.lib.image.from_raw_data(
194 plane.bitmapData,
195 plane.bitmapMode,
196 (plane.bitmapWidth, plane.bitmapHeight))
197 if not img:
198 gws.log.warning(f'PREPARE_FAILED: plane {n}: bitmap error')
199 return
200 return gws.MapRenderInputPlane(
201 type=gws.MapRenderInputPlaneType.image,
202 image=img,
203 opacity=opacity,
204 )
206 if plane.type == gws.PrintPlaneType.url:
207 img = gws.lib.image.from_data_url(plane.url)
208 if not img:
209 gws.log.warning(f'PREPARE_FAILED: plane {n}: url error')
210 return
211 return gws.MapRenderInputPlane(
212 type=gws.MapRenderInputPlaneType.image,
213 image=img,
214 opacity=opacity,
215 )
217 if plane.type == gws.PrintPlaneType.features:
218 model = self.root.app.modelMgr.default_model()
219 used_styles = {}
221 mc = gws.ModelContext(op=gws.ModelOperation.read, target=gws.ModelReadTarget.map, user=self.user)
222 features = []
224 for props in plane.features:
225 feature = model.feature_from_props(props, mc)
226 if not feature or not feature.shape():
227 continue
228 feature.cssSelector = feature.cssSelector or plane.cssSelector
229 if feature.cssSelector in style_dct:
230 used_styles[feature.cssSelector] = style_dct[feature.cssSelector]
231 features.append(feature)
233 if not features:
234 gws.log.warning(f'PREPARE_FAILED: plane {n}: no features')
235 return
237 return gws.MapRenderInputPlane(
238 type=gws.MapRenderInputPlaneType.features,
239 features=features,
240 opacity=opacity,
241 styles=list(used_styles.values()),
242 )
244 if plane.type == gws.PrintPlaneType.soup:
245 return gws.MapRenderInputPlane(
246 type=gws.MapRenderInputPlaneType.svgSoup,
247 soupPoints=plane.soupPoints,
248 soupTags=plane.soupTags,
249 opacity=opacity,
250 )
252 raise gws.Error(f'invalid plane type {plane.type!r}')