Coverage for gws-app/gws/base/printer/worker.py: 19%
150 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-24 12:46 +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(
43 state=gws.JobState.complete,
44 result=gws.PrintResult(
45 path=self.contentPath,
46 mime=gws.lib.mime.for_path(self.contentPath),
47 ),
48 )
50 def prepare(self):
51 self.project = cast(gws.Project, self.user.require(self.request.projectUid, gws.ext.object.project))
53 self.tri = gws.TemplateRenderInput(
54 project=self.project,
55 user=self.user,
56 notify=self.notify,
57 )
59 fmt = self.request.outputFormat or 'pdf'
60 if fmt.lower() == 'pdf' or fmt == gws.lib.mime.PDF:
61 self.tri.mimeOut = gws.lib.mime.PDF
62 elif fmt.lower() == 'png' or fmt == gws.lib.mime.PNG:
63 self.tri.mimeOut = gws.lib.mime.PNG
64 else:
65 raise gws.Error(f'invalid outputFormat {fmt!r}')
67 self.tri.locale = gws.lib.intl.locale(self.request.localeUid, self.tri.project.localeUids)
68 self.tri.crs = gws.lib.crs.get(self.request.crs) or self.project.map.bounds.crs
69 self.tri.maps = [self.prepare_map(self.tri, m) for m in (self.request.maps or [])]
70 self.tri.dpi = int(min(gws.gis.render.MAX_DPI, max(self.request.dpi, gws.lib.uom.OGC_SCREEN_PPI)))
72 if self.request.type == 'template':
73 self.printer = cast(gws.Printer, self.user.require(self.request.printerUid, gws.ext.object.printer))
74 self.template = self.printer.template
75 dpis = [ql.dpi for ql in self.printer.qualityLevels if ql.dpi]
76 max_dpi = max(dpis) if dpis else gws.lib.uom.OGC_SCREEN_PPI
77 self.tri.dpi = int(max(gws.lib.uom.OGC_SCREEN_PPI, min(self.tri.dpi, max_dpi)))
78 else:
79 mm = gws.lib.uom.size_px_to_mm(self.request.outputSize, gws.lib.uom.OGC_SCREEN_PPI)
80 px = gws.lib.uom.size_mm_to_px(mm, self.tri.dpi)
81 self.template = self.root.create_temporary(
82 gws.ext.object.template,
83 type='map',
84 pageSize=(px[0], px[1], gws.Uom.px))
86 extra = dict(
87 project=self.project,
88 user=self.user,
89 scale=self.tri.maps[0].scale if self.tri.maps else 0,
90 rotation=self.tri.maps[0].rotation if self.tri.maps else 0,
91 dpi=self.tri.dpi,
92 crs=self.tri.crs.srid,
93 )
95 # @TODO read the args feature from the request
96 self.tri.args = gws.u.merge(self.request.args, extra)
98 n = sum(len(mp.planes) for mp in self.tri.maps) + 1
99 self.update_job(numSteps=n)
101 def notify(self, event, details=None):
102 job = self.get_job()
103 if not job:
104 return
106 if event == 'begin_plane':
107 name = gws.u.get(details, 'layer.title')
108 self.update_job(step=job.step + 1, stepName=name or '')
109 return
111 if event == 'finalize_print':
112 self.update_job(step=job.step + 1)
113 return
115 if event == 'page_break':
116 self.page_number += 1
117 return
119 def prepare_map(self, tri: gws.TemplateRenderInput, mp: gws.PrintMap) -> gws.MapRenderInput:
120 planes = []
122 style_opts = gws.lib.style.parser.Options(
123 trusted=False,
124 strict=False,
125 imageDirs=[self.root.app.webMgr.site.staticRoot.dir],
126 )
127 style_dct = {}
129 for p in (mp.styles or []):
130 style = gws.lib.style.from_props(p, style_opts)
131 if style:
132 style_dct[style.cssSelector] = style
134 if mp.planes:
135 for n, p in enumerate(mp.planes):
136 pp = self.prepare_map_plane(n, p, style_dct)
137 if pp:
138 planes.append(pp)
140 layers = gws.u.compact(
141 self.user.acquire(uid, gws.ext.object.layer)
142 for uid in (mp.visibleLayers or []))
144 return gws.MapRenderInput(
145 backgroundColor=_PAPER_COLOR,
146 bbox=mp.bbox,
147 center=mp.center,
148 crs=tri.crs,
149 dpi=tri.dpi,
150 notify=tri.notify,
151 planes=planes,
152 project=tri.project,
153 rotation=mp.rotation or 0,
154 scale=mp.scale,
155 user=tri.user,
156 visibleLayers=layers,
157 )
159 def prepare_map_plane(self, n, plane: gws.PrintPlane, style_dct) -> Optional[gws.MapRenderInputPlane]:
160 opacity = 1
161 s = plane.get('opacity')
162 if s is not None:
163 opacity = float(s)
164 if opacity == 0:
165 return
167 if plane.type == gws.PrintPlaneType.raster:
168 layer = cast(gws.Layer, self.user.acquire(plane.layerUid, gws.ext.object.layer))
169 if not layer:
170 gws.log.warning(f'PREPARE_FAILED: plane {n}: {plane.layerUid=} not found')
171 return
172 if not layer.canRenderBox:
173 gws.log.warning(f'PREPARE_FAILED: plane {n}: {plane.layerUid=} canRenderBox false')
174 return
175 return gws.MapRenderInputPlane(
176 type=gws.MapRenderInputPlaneType.imageLayer,
177 layer=layer,
178 opacity=opacity,
179 compositeLayerUids=plane.get('compositeLayerUids'),
180 )
182 if plane.type == gws.PrintPlaneType.vector:
183 layer = cast(gws.Layer, self.user.acquire(plane.layerUid, gws.ext.object.layer))
184 if not layer:
185 gws.log.warning(f'PREPARE_FAILED: plane {n}: {plane.layerUid=} not found')
186 return
187 if not layer.canRenderSvg:
188 gws.log.warning(f'PREPARE_FAILED: plane {n}: {plane.layerUid=} canRenderSvg false')
189 return
190 style = style_dct.get(plane.cssSelector)
191 return gws.MapRenderInputPlane(
192 type=gws.MapRenderInputPlaneType.svgLayer,
193 layer=layer,
194 opacity=opacity,
195 styles=[style] if style else [],
196 )
198 if plane.type == gws.PrintPlaneType.bitmap:
199 img = None
200 if plane.bitmapMode in ('RGBA', 'RGB'):
201 img = gws.lib.image.from_raw_data(
202 plane.bitmapData,
203 plane.bitmapMode,
204 (plane.bitmapWidth, plane.bitmapHeight))
205 if not img:
206 gws.log.warning(f'PREPARE_FAILED: plane {n}: bitmap error')
207 return
208 return gws.MapRenderInputPlane(
209 type=gws.MapRenderInputPlaneType.image,
210 image=img,
211 opacity=opacity,
212 )
214 if plane.type == gws.PrintPlaneType.url:
215 img = gws.lib.image.from_data_url(plane.url)
216 if not img:
217 gws.log.warning(f'PREPARE_FAILED: plane {n}: url error')
218 return
219 return gws.MapRenderInputPlane(
220 type=gws.MapRenderInputPlaneType.image,
221 image=img,
222 opacity=opacity,
223 )
225 if plane.type == gws.PrintPlaneType.features:
226 model = self.root.app.modelMgr.default_model()
227 used_styles = {}
229 mc = gws.ModelContext(op=gws.ModelOperation.read, target=gws.ModelReadTarget.map, user=self.user)
230 features = []
232 for props in plane.features:
233 feature = model.feature_from_props(props, mc)
234 if not feature or not feature.shape():
235 continue
236 feature.cssSelector = feature.cssSelector or plane.cssSelector
237 if feature.cssSelector in style_dct:
238 used_styles[feature.cssSelector] = style_dct[feature.cssSelector]
239 features.append(feature)
241 if not features:
242 gws.log.warning(f'PREPARE_FAILED: plane {n}: no features')
243 return
245 return gws.MapRenderInputPlane(
246 type=gws.MapRenderInputPlaneType.features,
247 features=features,
248 opacity=opacity,
249 styles=list(used_styles.values()),
250 )
252 if plane.type == gws.PrintPlaneType.soup:
253 return gws.MapRenderInputPlane(
254 type=gws.MapRenderInputPlaneType.svgSoup,
255 soupPoints=plane.soupPoints,
256 soupTags=plane.soupTags,
257 opacity=opacity,
258 )
260 raise gws.Error(f'invalid plane type {plane.type!r}')